mirror of
https://github.com/onyx-and-iris/vban-cmd-python.git
synced 2026-01-24 17:27:48 +00:00
246 lines
9.0 KiB
Python
246 lines
9.0 KiB
Python
import logging
|
|
import socket
|
|
import threading
|
|
import time
|
|
|
|
from .enums import NBS
|
|
from .error import VBANCMDConnectionError
|
|
from .packet import (
|
|
HEADER_SIZE,
|
|
VBAN_PROTOCOL_SERVICE,
|
|
VBAN_SERVICE_RTPACKET,
|
|
VMPARAMSTRIP_SIZE,
|
|
SubscribeHeader,
|
|
VbanRtPacket,
|
|
VbanRtPacketHeader,
|
|
VbanRtPacketNBS0,
|
|
VbanRtPacketNBS1,
|
|
VbanVMParamStrip,
|
|
)
|
|
from .util import bump_framecounter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Subscriber(threading.Thread):
|
|
"""fire a subscription packet every 10 seconds"""
|
|
|
|
def __init__(self, remote, stop_event):
|
|
super().__init__(name='subscriber', daemon=False)
|
|
self._remote = remote
|
|
self.stop_event = stop_event
|
|
self.logger = logger.getChild(self.__class__.__name__)
|
|
self._framecounter = 0
|
|
|
|
def run(self):
|
|
while not self.stopped():
|
|
try:
|
|
for nbs in NBS:
|
|
sub_packet = SubscribeHeader().to_bytes(nbs, self._framecounter)
|
|
self._remote.sock.sendto(
|
|
sub_packet, (self._remote.ip, self._remote.port)
|
|
)
|
|
self._framecounter = bump_framecounter(self._framecounter)
|
|
self.logger.debug(
|
|
f'sent subscription for NBS {nbs.name} to {self._remote.ip}:{self._remote.port}'
|
|
)
|
|
|
|
self.wait_until_stopped(10)
|
|
except socket.gaierror as e:
|
|
self.logger.exception(f'{type(e).__name__}: {e}')
|
|
raise VBANCMDConnectionError(
|
|
f'unable to resolve hostname {self._remote.ip}'
|
|
) from e
|
|
self.logger.debug(f'terminating {self.name} thread')
|
|
|
|
def stopped(self):
|
|
return self.stop_event.is_set()
|
|
|
|
def wait_until_stopped(self, timeout, period=0.2):
|
|
must_end = time.time() + timeout
|
|
while time.time() < must_end:
|
|
if self.stopped():
|
|
break
|
|
time.sleep(period)
|
|
|
|
|
|
class Producer(threading.Thread):
|
|
"""Continously send job queue to the Updater thread at a rate of self._remote.ratelimit."""
|
|
|
|
def __init__(self, remote, queue, stop_event):
|
|
super().__init__(name='producer', daemon=False)
|
|
self._remote = remote
|
|
self.queue = queue
|
|
self.stop_event = stop_event
|
|
self.logger = logger.getChild(self.__class__.__name__)
|
|
self._remote.sock.settimeout(self._remote.timeout)
|
|
self._remote._public_packets = [None] * (max(NBS) + 1)
|
|
_pp = self._get_rt()
|
|
self._remote._public_packets[_pp.nbs] = _pp
|
|
(
|
|
self._remote.cache['strip_level'],
|
|
self._remote.cache['bus_level'],
|
|
) = self._remote._get_levels(self._remote.public_packets[NBS.zero])
|
|
|
|
def _get_rt(self) -> VbanRtPacket:
|
|
"""Attempt to fetch data packet until a valid one found"""
|
|
|
|
while True:
|
|
if resp := self._fetch_rt_packet():
|
|
return resp
|
|
|
|
def _fetch_rt_packet(self) -> VbanRtPacket | None:
|
|
try:
|
|
data, _ = self._remote.sock.recvfrom(2048)
|
|
if len(data) < HEADER_SIZE:
|
|
return
|
|
|
|
response_header = VbanRtPacketHeader.from_bytes(data[:HEADER_SIZE])
|
|
if (
|
|
response_header.format_sr != VBAN_PROTOCOL_SERVICE
|
|
or response_header.format_nbc != VBAN_SERVICE_RTPACKET
|
|
):
|
|
return
|
|
|
|
match response_header.format_nbs:
|
|
case NBS.zero:
|
|
"""
|
|
self.logger.debug(
|
|
'Received NB0 RTP Packet from %s, Size: %d bytes',
|
|
addr,
|
|
len(data),
|
|
)
|
|
"""
|
|
|
|
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],
|
|
)
|
|
|
|
case NBS.one:
|
|
"""
|
|
self.logger.debug(
|
|
'Received NB1 RTP Packet from %s, Size: %d bytes',
|
|
addr,
|
|
len(data),
|
|
)
|
|
"""
|
|
|
|
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 None
|
|
except TimeoutError as e:
|
|
self.logger.exception(f'{type(e).__name__}: {e}')
|
|
raise VBANCMDConnectionError(
|
|
f'timeout waiting for RtPacket from {self._remote.ip}'
|
|
) from e
|
|
|
|
def stopped(self):
|
|
return self.stop_event.is_set()
|
|
|
|
def run(self):
|
|
while not self.stopped():
|
|
pdirty = ldirty = False
|
|
_pp = self._get_rt()
|
|
match _pp.nbs:
|
|
case NBS.zero:
|
|
ldirty = _pp.ldirty(
|
|
self._remote.cache['strip_level'],
|
|
self._remote.cache['bus_level'],
|
|
)
|
|
pdirty = _pp.pdirty(self._remote.public_packets[NBS.zero])
|
|
case NBS.one:
|
|
pdirty = True
|
|
|
|
if pdirty or ldirty:
|
|
self._remote._public_packets[_pp.nbs] = _pp
|
|
self._remote._pdirty = pdirty
|
|
self._remote._ldirty = ldirty
|
|
|
|
if self._remote.event.pdirty:
|
|
self.queue.put('pdirty')
|
|
if self._remote.event.ldirty:
|
|
self.queue.put('ldirty')
|
|
time.sleep(self._remote.ratelimit)
|
|
self.logger.debug(f'terminating {self.name} thread')
|
|
self.queue.put(None)
|
|
|
|
|
|
class Updater(threading.Thread):
|
|
"""
|
|
continously updates the public packet
|
|
|
|
notifies observers of event updates
|
|
"""
|
|
|
|
def __init__(self, remote, queue):
|
|
super().__init__(name='updater', daemon=True)
|
|
self._remote = remote
|
|
self.queue = queue
|
|
self.logger = logger.getChild(self.__class__.__name__)
|
|
self._remote._strip_comp = [False] * (self._remote.kind.num_strip_levels)
|
|
self._remote._bus_comp = [False] * (self._remote.kind.num_bus_levels)
|
|
|
|
def run(self):
|
|
"""
|
|
Continously update observers of dirty states.
|
|
|
|
Generate _strip_comp, _bus_comp and update level cache if ldirty.
|
|
"""
|
|
while event := self.queue.get():
|
|
if event == 'pdirty' and self._remote.pdirty:
|
|
self._remote.subject.notify(event)
|
|
elif event == 'ldirty' and self._remote.ldirty:
|
|
self._remote._strip_comp, self._remote._bus_comp = (
|
|
self._remote._public_packets[NBS.zero]._strip_comp,
|
|
self._remote._public_packets[NBS.zero]._bus_comp,
|
|
)
|
|
(
|
|
self._remote.cache['strip_level'],
|
|
self._remote.cache['bus_level'],
|
|
) = (
|
|
self._remote._public_packets[NBS.zero].inputlevels,
|
|
self._remote._public_packets[NBS.zero].outputlevels,
|
|
)
|
|
self._remote.subject.notify(event)
|
|
self.logger.debug(f'terminating {self.name} thread')
|