refactor header dataclasses

This commit is contained in:
onyx-and-iris 2026-01-18 11:43:43 +00:00
parent ed8e281f7f
commit 312b5c5842

View File

@ -422,19 +422,38 @@ class VbanRtPacketNBS1(VbanRtPacket):
class SubscribeHeader: class SubscribeHeader:
"""Represents the header of an RT subscription packet""" """Represents the header of an RT subscription packet"""
ident: NBS = NBS.zero nbs: NBS = NBS.zero
name = 'Register-RTP' name: str = 'Register-RTP'
timeout = 15 timeout: int = 15
vban: bytes = 'VBAN'.encode()
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little') @property
format_nbs: bytes = (ident.value & 0xFF).to_bytes(1, 'little') def vban(self) -> bytes:
format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, 'little') return b'VBAN'
format_bit: bytes = (timeout & 0xFF).to_bytes(1, 'little') # timeout
streamname: bytes = name.encode('ascii') + bytes(16 - len(name)) @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 @classmethod
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes: def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(ident=nbs) header = cls(nbs=nbs)
data = bytearray() data = bytearray()
data.extend(header.vban) data.extend(header.vban)
data.extend(header.format_sr) data.extend(header.format_sr)
@ -451,30 +470,31 @@ class VbanRtPacketHeader:
"""Represents the header of an RT response packet""" """Represents the header of an RT response packet"""
name: str = 'Voicemeeter-RTP' name: str = 'Voicemeeter-RTP'
vban: bytes = 'VBAN'.encode() format_sr: int = VBAN_PROTOCOL_SERVICE
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little') format_nbs: int = 0
format_nbs: bytes = (0).to_bytes(1, 'little') format_nbc: int = VBAN_SERVICE_RTPACKET
format_nbc: bytes = (VBAN_SERVICE_RTPACKET).to_bytes(1, 'little') format_bit: int = 0
format_bit: bytes = (0).to_bytes(1, 'little')
streamname: bytes = name.encode('ascii') + bytes(16 - len(name)) @property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod @classmethod
def from_bytes(cls, data: bytes): def from_bytes(cls, data: bytes):
if len(data) < HEADER_SIZE: if len(data) < HEADER_SIZE:
raise ValueError('Data is too short to be a valid VbanRTPPacketHeader') 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') name = data[8:24].rstrip(b'\x00').decode('utf-8')
return cls( return cls(
name=name, name=name,
vban=vban, format_sr=data[4] & VBAN_SERVICE_MASK,
format_sr=format_sr & VBAN_SERVICE_MASK, format_nbs=data[5],
format_nbs=format_nbs, format_nbc=data[6],
format_nbc=format_nbc, format_bit=data[7],
format_bit=format_bit,
) )
@ -485,21 +505,30 @@ class RequestHeader:
name: str name: str
bps_index: int bps_index: int
channel: int channel: int
vban: bytes = 'VBAN'.encode() framecounter: int = 0
nbs: bytes = (0).to_bytes(1, 'little')
bit: bytes = (0x10).to_bytes(1, 'little')
framecounter: bytes = (0).to_bytes(4, 'little')
@property @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') return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
@property @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') return (self.channel).to_bytes(1, 'little')
@property @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)) return self.name.encode() + bytes(16 - len(self.name))
@classmethod @classmethod
@ -509,6 +538,7 @@ class RequestHeader:
header = cls( header = cls(
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
) )
data = bytearray() data = bytearray()
data.extend(header.vban) data.extend(header.vban)
data.extend(header.sr) data.extend(header.sr)