improve to_bytes efficiency with struct.pack

This commit is contained in:
onyx-and-iris 2026-03-07 14:20:31 +00:00
parent 9f43ee18d3
commit 5363584940
2 changed files with 57 additions and 51 deletions

View File

@ -1,3 +1,4 @@
import struct
from dataclasses import dataclass from dataclasses import dataclass
from vban_cmd.enums import NBS from vban_cmd.enums import NBS
@ -34,15 +35,16 @@ class VbanPingHeader:
"""Creates the PING header bytes only.""" """Creates the PING header bytes only."""
header = cls(framecounter=framecounter) header = cls(framecounter=framecounter)
data = bytearray() return struct.pack(
data.extend(header.vban) '<4s4B16sI',
data.extend(header.format_sr.to_bytes(1, 'little')) header.vban,
data.extend(header.format_nbs.to_bytes(1, 'little')) header.format_sr,
data.extend(header.format_nbc.to_bytes(1, 'little')) header.format_nbs,
data.extend(header.format_bit.to_bytes(1, 'little')) header.format_nbc,
data.extend(header.streamname) header.format_bit,
data.extend(header.framecounter.to_bytes(4, 'little')) header.streamname,
return bytes(data) header.framecounter,
)
@dataclass @dataclass
@ -163,15 +165,16 @@ class VbanRTSubscribeHeader:
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes: def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(nbs=nbs) header = cls(nbs=nbs)
data = bytearray() return struct.pack(
data.extend(header.vban) '<4s4B16sI',
data.extend(header.format_sr) header.vban,
data.extend(header.format_nbs) header.format_sr[0],
data.extend(header.format_nbc) header.format_nbs[0],
data.extend(header.format_bit) header.format_nbc[0],
data.extend(header.streamname) header.format_bit[0],
data.extend(framecounter.to_bytes(4, 'little')) header.streamname,
return bytes(data) framecounter,
)
@dataclass @dataclass
@ -215,15 +218,16 @@ class VbanRTRequestHeader:
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
) )
data = bytearray() return struct.pack(
data.extend(header.vban) '<4s4B16sI',
data.extend(header.sr) header.vban,
data.extend(header.nbs) header.sr[0],
data.extend(header.nbc) header.nbs[0],
data.extend(header.bit) header.nbc[0],
data.extend(header.streamname) header.bit[0],
data.extend(header.framecounter.to_bytes(4, 'little')) header.streamname,
return bytes(data) header.framecounter,
)
@classmethod @classmethod
def encode_with_payload( def encode_with_payload(

View File

@ -1,3 +1,4 @@
import struct
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum from enum import Enum
@ -65,30 +66,31 @@ class VbanPing0Payload:
"""Convert payload to bytes""" """Convert payload to bytes"""
payload = cls() payload = cls()
data = bytearray() return struct.pack(
data.extend(payload.bit_type.to_bytes(4, 'little')) '<7I4s8s8s8s8s64s32s2H64s64s64s64s128s128s',
data.extend(payload.bit_feature.to_bytes(4, 'little')) payload.bit_type,
data.extend(payload.bit_feature_ex.to_bytes(4, 'little')) payload.bit_feature,
data.extend(payload.preferred_rate.to_bytes(4, 'little')) payload.bit_feature_ex,
data.extend(payload.min_rate.to_bytes(4, 'little')) payload.preferred_rate,
data.extend(payload.max_rate.to_bytes(4, 'little')) payload.min_rate,
data.extend(payload.color_rgb.to_bytes(4, 'little')) payload.max_rate,
data.extend(payload.version) payload.color_rgb,
data.extend(payload.gps_position) payload.version,
data.extend(payload.user_position) payload.gps_position,
data.extend(payload.lang_code) payload.user_position,
data.extend(payload.reserved) payload.lang_code,
data.extend(payload.reserved_ex) payload.reserved,
data.extend(payload.distant_ip) payload.reserved_ex,
data.extend(payload.distant_port.to_bytes(2, 'little')) payload.distant_ip,
data.extend(payload.distant_reserved.to_bytes(2, 'little')) payload.distant_port,
data.extend(payload.device_name) payload.distant_reserved,
data.extend(payload.manufacturer_name) payload.device_name,
data.extend(payload.application_name) payload.manufacturer_name,
data.extend(payload.host_name) payload.application_name,
data.extend(payload.user_name) payload.host_name,
data.extend(payload.user_comment) payload.user_name,
return bytes(data) payload.user_comment,
)
@classmethod @classmethod
def create_packet(cls, framecounter: int) -> bytes: def create_packet(cls, framecounter: int) -> bytes: