From 312b5c5842c6d9f76db91bb9067bb5769974c5ef Mon Sep 17 00:00:00 2001 From: onyx-and-iris Date: Sun, 18 Jan 2026 11:43:43 +0000 Subject: [PATCH] refactor header dataclasses --- vban_cmd/packet.py | 96 ++++++++++++++++++++++++++++++---------------- 1 file changed, 63 insertions(+), 33 deletions(-) diff --git a/vban_cmd/packet.py b/vban_cmd/packet.py index afd29fe..ad78729 100644 --- a/vban_cmd/packet.py +++ b/vban_cmd/packet.py @@ -422,19 +422,38 @@ class VbanRtPacketNBS1(VbanRtPacket): class SubscribeHeader: """Represents the header of an RT subscription packet""" - ident: NBS = NBS.zero - name = 'Register-RTP' - timeout = 15 - vban: bytes = 'VBAN'.encode() - format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little') - format_nbs: bytes = (ident.value & 0xFF).to_bytes(1, 'little') - format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, 'little') - format_bit: bytes = (timeout & 0xFF).to_bytes(1, 'little') # timeout - streamname: bytes = name.encode('ascii') + bytes(16 - len(name)) + nbs: NBS = NBS.zero + name: str = 'Register-RTP' + timeout: int = 15 + + @property + def vban(self) -> bytes: + return b'VBAN' + + @property + def format_sr(self) -> bytes: + return VBAN_PROTOCOL_SERVICE.to_bytes(1, 'little') + + @property + def format_nbs(self) -> bytes: + return (self.nbs.value & 0xFF).to_bytes(1, 'little') + + @property + def format_nbc(self) -> bytes: + return VBAN_SERVICE_RTPACKETREGISTER.to_bytes(1, 'little') + + @property + def format_bit(self) -> bytes: + return (self.timeout & 0xFF).to_bytes(1, 'little') + + @property + def streamname(self) -> bytes: + return self.name.encode('ascii') + bytes(16 - len(self.name)) @classmethod def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes: - header = cls(ident=nbs) + header = cls(nbs=nbs) + data = bytearray() data.extend(header.vban) data.extend(header.format_sr) @@ -451,30 +470,31 @@ class VbanRtPacketHeader: """Represents the header of an RT response packet""" name: str = 'Voicemeeter-RTP' - vban: bytes = 'VBAN'.encode() - format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little') - format_nbs: bytes = (0).to_bytes(1, 'little') - format_nbc: bytes = (VBAN_SERVICE_RTPACKET).to_bytes(1, 'little') - format_bit: bytes = (0).to_bytes(1, 'little') - streamname: bytes = name.encode('ascii') + bytes(16 - len(name)) + format_sr: int = VBAN_PROTOCOL_SERVICE + format_nbs: int = 0 + format_nbc: int = VBAN_SERVICE_RTPACKET + format_bit: int = 0 + + @property + def vban(self) -> bytes: + return b'VBAN' + + @property + def streamname(self) -> bytes: + return self.name.encode('ascii') + bytes(16 - len(self.name)) @classmethod def from_bytes(cls, data: bytes): if len(data) < HEADER_SIZE: raise ValueError('Data is too short to be a valid VbanRTPPacketHeader') - vban = data[0:4] - format_sr = data[4] - format_nbs = data[5] - format_nbc = data[6] - format_bit = data[7] + name = data[8:24].rstrip(b'\x00').decode('utf-8') return cls( name=name, - vban=vban, - format_sr=format_sr & VBAN_SERVICE_MASK, - format_nbs=format_nbs, - format_nbc=format_nbc, - format_bit=format_bit, + format_sr=data[4] & VBAN_SERVICE_MASK, + format_nbs=data[5], + format_nbc=data[6], + format_bit=data[7], ) @@ -485,21 +505,30 @@ class RequestHeader: name: str bps_index: int channel: int - vban: bytes = 'VBAN'.encode() - nbs: bytes = (0).to_bytes(1, 'little') - bit: bytes = (0x10).to_bytes(1, 'little') - framecounter: bytes = (0).to_bytes(4, 'little') + framecounter: int = 0 @property - def sr(self): + def vban(self) -> bytes: + return b'VBAN' + + @property + def sr(self) -> bytes: return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little') @property - def nbc(self): + def nbs(self) -> bytes: + return (0).to_bytes(1, 'little') + + @property + def nbc(self) -> bytes: return (self.channel).to_bytes(1, 'little') @property - def streamname(self): + def bit(self) -> bytes: + return (0x10).to_bytes(1, 'little') + + @property + def streamname(self) -> bytes: return self.name.encode() + bytes(16 - len(self.name)) @classmethod @@ -509,6 +538,7 @@ class RequestHeader: header = cls( name=name, bps_index=bps_index, channel=channel, framecounter=framecounter ) + data = bytearray() data.extend(header.vban) data.extend(header.sr)