mirror of
https://github.com/onyx-and-iris/vban-cmd-python.git
synced 2025-01-18 10:30:48 +00:00
vbanrtpacket refactored
_generate_levels method added ldirty method added. moved initialize strip_level, bus_level cache into updater init() initialize comps in updater init()
This commit is contained in:
onyx-and-iris
parent
bfa1a718f9
commit
a029011012
@ -1,6 +1,8 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Generator
|
||||
|
||||
from .util import comp
|
||||
|
||||
VBAN_SERVICE_RTPACKETREGISTER = 32
|
||||
VBAN_SERVICE_RTPACKET = 33
|
||||
MAX_PACKET_SIZE = 1436
|
||||
@ -11,37 +13,24 @@ HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16 + 4
|
||||
class VbanRtPacket:
|
||||
"""Represents the body of a VBAN RT data packet"""
|
||||
|
||||
_voicemeeterType: bytes
|
||||
_reserved: bytes
|
||||
_buffersize: bytes
|
||||
_voicemeeterVersion: bytes
|
||||
_optionBits: bytes
|
||||
_samplerate: bytes
|
||||
_inputLeveldB100: bytes
|
||||
_outputLeveldB100: bytes
|
||||
_TransportBit: bytes
|
||||
_stripState: bytes
|
||||
_busState: bytes
|
||||
_stripGaindB100Layer1: bytes
|
||||
_stripGaindB100Layer2: bytes
|
||||
_stripGaindB100Layer3: bytes
|
||||
_stripGaindB100Layer4: bytes
|
||||
_stripGaindB100Layer5: bytes
|
||||
_stripGaindB100Layer6: bytes
|
||||
_stripGaindB100Layer7: bytes
|
||||
_stripGaindB100Layer8: bytes
|
||||
_busGaindB100: bytes
|
||||
_stripLabelUTF8c60: bytes
|
||||
_busLabelUTF8c60: bytes
|
||||
def __init__(self, **kwargs):
|
||||
for k, v in kwargs.items():
|
||||
setattr(self, k, v)
|
||||
self._strip_level = self._generate_levels(self._inputLeveldB100)
|
||||
self._bus_level = self._generate_levels(self._outputLeveldB100)
|
||||
|
||||
def pdirty(self, other):
|
||||
def _generate_levels(self, levelarray) -> tuple:
|
||||
return tuple(
|
||||
int.from_bytes(levelarray[i : i + 2], "little")
|
||||
for i in range(0, len(levelarray), 2)
|
||||
)
|
||||
|
||||
def pdirty(self, other) -> bool:
|
||||
"""True iff any defined parameter has changed"""
|
||||
|
||||
return not (
|
||||
self._stripState == other._stripState
|
||||
and self._busState == other._busState
|
||||
and self._stripLabelUTF8c60 == other._stripLabelUTF8c60
|
||||
and self._busLabelUTF8c60 == other._busLabelUTF8c60
|
||||
and self._stripGaindB100Layer1 == other._stripGaindB100Layer1
|
||||
and self._stripGaindB100Layer2 == other._stripGaindB100Layer2
|
||||
and self._stripGaindB100Layer3 == other._stripGaindB100Layer3
|
||||
@ -51,8 +40,17 @@ class VbanRtPacket:
|
||||
and self._stripGaindB100Layer7 == other._stripGaindB100Layer7
|
||||
and self._stripGaindB100Layer8 == other._stripGaindB100Layer8
|
||||
and self._busGaindB100 == other._busGaindB100
|
||||
and self._stripLabelUTF8c60 == other._stripLabelUTF8c60
|
||||
and self._busLabelUTF8c60 == other._busLabelUTF8c60
|
||||
)
|
||||
|
||||
def ldirty(self, strip_cache, bus_cache) -> bool:
|
||||
self._strip_comp, self._bus_comp = (
|
||||
tuple(not val for val in comp(strip_cache, self._strip_level)),
|
||||
tuple(not val for val in comp(bus_cache, self._bus_level)),
|
||||
)
|
||||
return any(any(l) for l in (self._strip_comp, self._bus_comp))
|
||||
|
||||
@property
|
||||
def voicemeetertype(self) -> str:
|
||||
"""returns voicemeeter type as a string"""
|
||||
@ -77,20 +75,14 @@ class VbanRtPacket:
|
||||
return int.from_bytes(self._samplerate, "little")
|
||||
|
||||
@property
|
||||
def inputlevels(self) -> Generator[float, None, None]:
|
||||
"""returns the entire level array across all inputs"""
|
||||
for i in range(0, 68, 2):
|
||||
val = int.from_bytes(self._inputLeveldB100[i : i + 2], "little")
|
||||
if val != ((1 << 16) - 1):
|
||||
yield val
|
||||
def inputlevels(self) -> tuple:
|
||||
"""returns the entire level array across all inputs for a kind"""
|
||||
return self._strip_level[0 : (2 * self._kind.phys_in + 8 * self._kind.virt_in)]
|
||||
|
||||
@property
|
||||
def outputlevels(self) -> Generator[float, None, None]:
|
||||
"""returns the entire level array across all outputs"""
|
||||
for i in range(0, 128, 2):
|
||||
val = int.from_bytes(self._outputLeveldB100[i : i + 2], "little")
|
||||
if val != ((1 << 16) - 1):
|
||||
yield val
|
||||
def outputlevels(self) -> tuple:
|
||||
"""returns the entire level array across all outputs for a kind"""
|
||||
return self._bus_level[0 : 8 * self._kind.num_bus]
|
||||
|
||||
@property
|
||||
def stripstate(self) -> tuple:
|
||||
|
||||
@ -6,7 +6,7 @@ from typing import Optional
|
||||
|
||||
from .error import VBANCMDError
|
||||
from .packet import HEADER_SIZE, SubscribeHeader, VbanRtPacket, VbanRtPacketHeader
|
||||
from .util import Socket
|
||||
from .util import Socket, comp
|
||||
|
||||
|
||||
class Subscriber(threading.Thread):
|
||||
@ -51,6 +51,14 @@ class Updater(threading.Thread):
|
||||
)
|
||||
self.packet_expected = VbanRtPacketHeader()
|
||||
self._remote._public_packet = self._get_rt()
|
||||
(
|
||||
self._remote.cache["strip_level"],
|
||||
self._remote.cache["bus_level"],
|
||||
) = self._remote._get_levels(self._remote.public_packet)
|
||||
self._remote._strip_comp = [False] * (
|
||||
2 * self._remote.kind.phys_in + 8 * self._remote.kind.virt_in
|
||||
)
|
||||
self._remote._bus_comp = [False] * (self._remote.kind.num_bus * 8)
|
||||
|
||||
def _fetch_rt_packet(self) -> Optional[VbanRtPacket]:
|
||||
try:
|
||||
@ -59,7 +67,9 @@ class Updater(threading.Thread):
|
||||
if len(data) > HEADER_SIZE:
|
||||
# check if packet is of type rt packet response
|
||||
if self.packet_expected.header == data[: HEADER_SIZE - 4]:
|
||||
self.logger.debug("valid packet received")
|
||||
return VbanRtPacket(
|
||||
_kind=self._remote.kind,
|
||||
_voicemeeterType=data[28:29],
|
||||
_reserved=data[29:30],
|
||||
_buffersize=data[30:32],
|
||||
@ -92,7 +102,7 @@ class Updater(threading.Thread):
|
||||
"""Attempt to fetch data packet until a valid one found"""
|
||||
|
||||
def fget():
|
||||
data = False
|
||||
data = None
|
||||
while not data:
|
||||
data = self._fetch_rt_packet()
|
||||
time.sleep(self._remote.DELAY)
|
||||
@ -100,28 +110,33 @@ class Updater(threading.Thread):
|
||||
|
||||
return fget()
|
||||
|
||||
def update(self):
|
||||
(
|
||||
self._remote.cache["strip_level"],
|
||||
self._remote.cache["bus_level"],
|
||||
) = self._remote._get_levels(self._remote.public_packet)
|
||||
def _update_comps(self, _pp):
|
||||
self._remote._strip_comp = _pp._strip_comp
|
||||
self._remote._bus_comp = _pp._bus_comp
|
||||
|
||||
def update(self):
|
||||
while self._remote.running:
|
||||
start = time.time()
|
||||
_pp = self._get_rt()
|
||||
self._remote._strip_buf, self._remote._bus_buf = self._remote._get_levels(
|
||||
_pp
|
||||
)
|
||||
self._remote._pdirty = _pp.pdirty(self._remote.public_packet)
|
||||
self._remote._ldirty = _pp.ldirty(
|
||||
self._remote.cache["strip_level"], self._remote.cache["bus_level"]
|
||||
)
|
||||
|
||||
if self._remote.event.ldirty and self._remote.ldirty:
|
||||
self._remote.cache["strip_level"] = self._remote._strip_buf
|
||||
self._remote.cache["bus_level"] = self._remote._bus_buf
|
||||
self._remote.subject.notify("ldirty")
|
||||
if self._remote.public_packet != _pp:
|
||||
if self._remote.pdirty or self._remote.ldirty:
|
||||
self.logger.debug("dirty state, updating public packet")
|
||||
self._remote._public_packet = _pp
|
||||
|
||||
if self._remote.event.pdirty and self._remote.pdirty:
|
||||
self._remote.subject.notify("pdirty")
|
||||
if self._remote.event.ldirty and self._remote.ldirty:
|
||||
self._update_comps(_pp)
|
||||
self._remote.cache["strip_level"], self._remote.cache["bus_level"] = (
|
||||
_pp.inputlevels,
|
||||
_pp.outputlevels,
|
||||
)
|
||||
self._remote.subject.notify("ldirty")
|
||||
|
||||
elapsed = time.time() - start
|
||||
if self._remote.ratelimit - elapsed > 0:
|
||||
time.sleep(self._remote.ratelimit - elapsed)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user