diff --git a/vban_cmd/packet.py b/vban_cmd/packet.py index f6adf8a..62a8c4e 100644 --- a/vban_cmd/packet.py +++ b/vban_cmd/packet.py @@ -51,6 +51,35 @@ class VbanRtPacketNBS0(VbanRtPacket): _stripLabelUTF8c60: bytes # data[452:932] _busLabelUTF8c60: bytes # data[932:1412] + @classmethod + def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes): + return cls( + nbs=nbs, + _kind=kind, + _voicemeeterType=data[28:29], + _reserved=data[29:30], + _buffersize=data[30:32], + _voicemeeterVersion=data[32:36], + _optionBits=data[36:40], + _samplerate=data[40:44], + _inputLeveldB100=data[44:112], + _outputLeveldB100=data[112:240], + _TransportBit=data[240:244], + _stripState=data[244:276], + _busState=data[276:308], + _stripGaindB100Layer1=data[308:324], + _stripGaindB100Layer2=data[324:340], + _stripGaindB100Layer3=data[340:356], + _stripGaindB100Layer4=data[356:372], + _stripGaindB100Layer5=data[372:388], + _stripGaindB100Layer6=data[388:404], + _stripGaindB100Layer7=data[404:420], + _stripGaindB100Layer8=data[420:436], + _busGaindB100=data[436:452], + _stripLabelUTF8c60=data[452:932], + _busLabelUTF8c60=data[932:1412], + ) + def _generate_levels(self, levelarray) -> tuple: return tuple( int.from_bytes(levelarray[i : i + 2], 'little') @@ -358,6 +387,30 @@ class VbanRtPacketNBS1(VbanRtPacket): strips: tuple[VbanVMParamStrip, ...] + @classmethod + def from_bytes( + cls, + nbs: NBS, + kind: KindMapClass, + data: bytes, + ): + return cls( + nbs=nbs, + _kind=kind, + _voicemeeterType=data[28:29], + _reserved=data[29:30], + _buffersize=data[30:32], + _voicemeeterVersion=data[32:36], + _optionBits=data[36:40], + _samplerate=data[40:44], + strips=tuple( + VbanVMParamStrip.from_bytes( + data[44 + i * VMPARAMSTRIP_SIZE : 44 + (i + 1) * VMPARAMSTRIP_SIZE] + ) + for i in range(16) + ), + ) + @dataclass class SubscribeHeader: diff --git a/vban_cmd/worker.py b/vban_cmd/worker.py index 25848ee..26d382b 100644 --- a/vban_cmd/worker.py +++ b/vban_cmd/worker.py @@ -9,13 +9,11 @@ from .packet import ( HEADER_SIZE, VBAN_PROTOCOL_SERVICE, VBAN_SERVICE_RTPACKET, - VMPARAMSTRIP_SIZE, SubscribeHeader, VbanRtPacket, VbanRtPacketHeader, VbanRtPacketNBS0, VbanRtPacketNBS1, - VbanVMParamStrip, ) from .util import bump_framecounter @@ -112,31 +110,8 @@ class Producer(threading.Thread): ) """ - return VbanRtPacketNBS0( - nbs=NBS.zero, - _kind=self._remote.kind, - _voicemeeterType=data[28:29], - _reserved=data[29:30], - _buffersize=data[30:32], - _voicemeeterVersion=data[32:36], - _optionBits=data[36:40], - _samplerate=data[40:44], - _inputLeveldB100=data[44:112], - _outputLeveldB100=data[112:240], - _TransportBit=data[240:244], - _stripState=data[244:276], - _busState=data[276:308], - _stripGaindB100Layer1=data[308:324], - _stripGaindB100Layer2=data[324:340], - _stripGaindB100Layer3=data[340:356], - _stripGaindB100Layer4=data[356:372], - _stripGaindB100Layer5=data[372:388], - _stripGaindB100Layer6=data[388:404], - _stripGaindB100Layer7=data[404:420], - _stripGaindB100Layer8=data[420:436], - _busGaindB100=data[436:452], - _stripLabelUTF8c60=data[452:932], - _busLabelUTF8c60=data[932:1412], + return VbanRtPacketNBS0.from_bytes( + nbs=NBS.zero, kind=self._remote.kind, data=data ) case NBS.one: @@ -148,24 +123,8 @@ class Producer(threading.Thread): ) """ - return VbanRtPacketNBS1( - nbs=NBS.one, - _kind=self._remote.kind, - _voicemeeterType=data[28:29], - _reserved=data[29:30], - _buffersize=data[30:32], - _voicemeeterVersion=data[32:36], - _optionBits=data[36:40], - _samplerate=data[40:44], - strips=tuple( - VbanVMParamStrip.from_bytes( - data[ - 44 + i * VMPARAMSTRIP_SIZE : 44 - + (i + 1) * VMPARAMSTRIP_SIZE - ] - ) - for i in range(self._remote.kind.num_strip) - ), + return VbanRtPacketNBS1.from_bytes( + nbs=NBS.one, kind=self._remote.kind, data=data ) return None except TimeoutError as e: