onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2026-01-24 17:27:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

add classmethod from_bytes() to both RT packets NBS0/NBS1

Browse Source
This commit is contained in:
onyx-and-iris 2026-01-17 10:06:28 +00:00
parent 1babf85a89
commit ecbdd2778f
2 changed files with 57 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
vban_cmd/packet.py
View File

@ -51,6 +51,35 @@ class VbanRtPacketNBS0(VbanRtPacket):
_stripLabelUTF8c60: bytes # data[452:932] _stripLabelUTF8c60: bytes # data[452:932]
_busLabelUTF8c60: bytes # data[932:1412] _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: def _generate_levels(self, levelarray) -> tuple:
return tuple( return tuple(
int.from_bytes(levelarray[i : i + 2], 'little') int.from_bytes(levelarray[i : i + 2], 'little')
@ -358,6 +387,30 @@ class VbanRtPacketNBS1(VbanRtPacket):
strips: tuple[VbanVMParamStrip, ...] 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 @dataclass
class SubscribeHeader: class SubscribeHeader:

49
vban_cmd/worker.py
View File

@ -9,13 +9,11 @@ from .packet import (
HEADER_SIZE, HEADER_SIZE,
VBAN_PROTOCOL_SERVICE, VBAN_PROTOCOL_SERVICE,
VBAN_SERVICE_RTPACKET, VBAN_SERVICE_RTPACKET,
VMPARAMSTRIP_SIZE,
SubscribeHeader, SubscribeHeader,
VbanRtPacket, VbanRtPacket,
VbanRtPacketHeader, VbanRtPacketHeader,
VbanRtPacketNBS0, VbanRtPacketNBS0,
VbanRtPacketNBS1, VbanRtPacketNBS1,
VbanVMParamStrip,
) )
from .util import bump_framecounter from .util import bump_framecounter
@ -112,31 +110,8 @@ class Producer(threading.Thread):
) )
""" """
return VbanRtPacketNBS0( return VbanRtPacketNBS0.from_bytes(
nbs=NBS.zero, nbs=NBS.zero, kind=self._remote.kind, data=data
_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],
) )
case NBS.one: case NBS.one:
@ -148,24 +123,8 @@ class Producer(threading.Thread):
) )
""" """
return VbanRtPacketNBS1( return VbanRtPacketNBS1.from_bytes(
nbs=NBS.one, nbs=NBS.one, kind=self._remote.kind, data=data
_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 None return None
except TimeoutError as e: except TimeoutError as e: