onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2025-01-18 18:40:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

clean up class names in packet module.

Browse Source 
add __init__ to vbanrtpacket class.

patch bump
This commit is contained in:
onyx-and-iris 2022-08-10 17:49:21 +01:00
parent 3e7bd14ba3
commit aea2be624e
4 changed files with 84 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "vban-cmd" name = "vban-cmd"
version = "1.3.2" version = "1.3.3"
description = "Python interface for the VBAN RT Packet Service (Sendtext)" description = "Python interface for the VBAN RT Packet Service (Sendtext)"
authors = ["onyx-and-iris <code@onyxandiris.online>"] authors = ["onyx-and-iris <code@onyxandiris.online>"]
license = "MIT" license = "MIT"

14
vban_cmd/base.py
View File

@ -1,17 +1,17 @@
import socket import socket
import time import time
from abc import ABCMeta, abstractmethod from abc import ABCMeta, abstractmethod
from typing import Iterable, NoReturn, Optional, Union from typing import Iterable, Optional, Union
from .misc import Event from .misc import Event
from .packet import TextRequestHeader from .packet import RequestHeader
from .subject import Subject from .subject import Subject
from .util import Socket, comp, script from .util import Socket, comp, script
from .worker import Subscriber, Updater from .worker import Subscriber, Updater
class VbanCmd(metaclass=ABCMeta): class VbanCmd(metaclass=ABCMeta):
"""Base class responsible for communicating over VBAN RT Service""" """Base class responsible for communicating with the VBAN RT Packet Service"""
DELAY = 0.001 DELAY = 0.001
# fmt: off # fmt: off
@ -26,7 +26,7 @@ class VbanCmd(metaclass=ABCMeta):
for attr, val in kwargs.items(): for attr, val in kwargs.items():
setattr(self, attr, val) setattr(self, attr, val)
self.text_header = TextRequestHeader( self.packet_request = RequestHeader(
name=self.streamname, name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps), bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel, channel=self.channel,
@ -66,11 +66,11 @@ class VbanCmd(metaclass=ABCMeta):
"""Sends a string request command over a network.""" """Sends a string request command over a network."""
cmd = id_ if not param else f"{id_}.{param}={val}" cmd = id_ if not param else f"{id_}.{param}={val}"
self.socks[Socket.request].sendto( self.socks[Socket.request].sendto(
self.text_header.header + cmd.encode(), self.packet_request.header + cmd.encode(),
(socket.gethostbyname(self.ip), self.port), (socket.gethostbyname(self.ip), self.port),
) )
count = int.from_bytes(self.text_header.framecounter, "little") + 1 count = int.from_bytes(self.packet_request.framecounter, "little") + 1
self.text_header.framecounter = count.to_bytes(4, "little") self.packet_request.framecounter = count.to_bytes(4, "little")
if param: if param:
self.cache[f"{id_}.{param}"] = val self.cache[f"{id_}.{param}"] = val
if self.sync: if self.sync:

59
vban_cmd/packet.py
View File

@ -8,31 +8,32 @@ HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16 + 4
@dataclass @dataclass
class VBAN_VMRT_Packet_Data: class VbanRtPacket:
"""Represents the structure of a VMRT data packet""" """Represents the body of a VBAN RT data packet"""
_voicemeeterType: bytes def __init__(self, data):
_reserved: bytes self._voicemeeterType: bytes = data[28:29]
_buffersize: bytes self._reserved: bytes = data[29:30]
_voicemeeterVersion: bytes self._buffersize: bytes = data[30:32]
_optionBits: bytes self._voicemeeterVersion: bytes = data[32:36]
_samplerate: bytes self._optionBits: bytes = data[36:40]
_inputLeveldB100: bytes self._samplerate: bytes = data[40:44]
_outputLeveldB100: bytes self._inputLeveldB100: bytes = data[44:112]
_TransportBit: bytes self._outputLeveldB100: bytes = data[112:240]
_stripState: bytes self._TransportBit: bytes = data[240:244]
_busState: bytes self._stripState: bytes = data[244:276]
_stripGaindB100Layer1: bytes self._busState: bytes = data[276:308]
_stripGaindB100Layer2: bytes self._stripGaindB100Layer1: bytes = data[308:324]
_stripGaindB100Layer3: bytes self._stripGaindB100Layer2: bytes = data[324:340]
_stripGaindB100Layer4: bytes self._stripGaindB100Layer3: bytes = data[340:356]
_stripGaindB100Layer5: bytes self._stripGaindB100Layer4: bytes = data[356:372]
_stripGaindB100Layer6: bytes self._stripGaindB100Layer5: bytes = data[372:388]
_stripGaindB100Layer7: bytes self._stripGaindB100Layer6: bytes = data[388:404]
_stripGaindB100Layer8: bytes self._stripGaindB100Layer7: bytes = data[404:420]
_busGaindB100: bytes self._stripGaindB100Layer8: bytes = data[420:436]
_stripLabelUTF8c60: bytes self._busGaindB100: bytes = data[436:452]
_busLabelUTF8c60: bytes self._stripLabelUTF8c60: bytes = data[452:932]
self._busLabelUTF8c60: bytes = data[932:1412]
def pdirty(self, other): def pdirty(self, other):
"""True iff any defined parameter has changed""" """True iff any defined parameter has changed"""
@ -201,8 +202,8 @@ class VBAN_VMRT_Packet_Data:
@dataclass @dataclass
class VBAN_VMRT_Packet_Header: class VbanRtPacketHeader:
"""Represents a RESPONSE RT PACKET header""" """Represents the header of VBAN RT data packet"""
name = "Voicemeeter-RTP" name = "Voicemeeter-RTP"
vban: bytes = "VBAN".encode() vban: bytes = "VBAN".encode()
@ -225,7 +226,7 @@ class VBAN_VMRT_Packet_Header:
@dataclass @dataclass
class TextRequestHeader: class RequestHeader:
"""Represents a REQUEST RT PACKET header""" """Represents a REQUEST RT PACKET header"""
name: str name: str
@ -262,8 +263,8 @@ class TextRequestHeader:
@dataclass @dataclass
class RegisterRTHeader: class SubscribeHeader:
"""Represents a REGISTER RT PACKET header""" """Represents a packet used to subscribe to the RT Packet Service"""
name = "Register RTP" name = "Register RTP"
timeout = 15 timeout = 15

129
vban_cmd/worker.py
View File

@ -1,15 +1,8 @@
import socket import socket
import threading import threading
import time import time
from enum import IntEnum
from typing import Optional
from .packet import ( from .packet import HEADER_SIZE, SubscribeHeader, VbanRtPacket, VbanRtPacketHeader
HEADER_SIZE,
RegisterRTHeader,
VBAN_VMRT_Packet_Data,
VBAN_VMRT_Packet_Header,
)
from .util import Socket from .util import Socket
@ -17,23 +10,23 @@ class Subscriber(threading.Thread):
"""fire a subscription packet every 10 seconds""" """fire a subscription packet every 10 seconds"""
def __init__(self, remote): def __init__(self, remote):
super().__init__(name="subscriber", target=self.register, daemon=True) super().__init__(name="subscriber", target=self.subscribe, daemon=True)
self._rem = remote self._remote = remote
self.register_header = RegisterRTHeader() self.packet = SubscribeHeader()
def register(self): def subscribe(self):
while self._rem.running: while self._remote.running:
try: try:
self._rem.socks[Socket.register].sendto( self._remote.socks[Socket.register].sendto(
self.register_header.header, self.packet.header,
(socket.gethostbyname(self._rem.ip), self._rem.port), (socket.gethostbyname(self._remote.ip), self._remote.port),
) )
count = int.from_bytes(self.register_header.framecounter, "little") + 1 count = int.from_bytes(self.packet.framecounter, "little") + 1
self.register_header.framecounter = count.to_bytes(4, "little") self.packet.framecounter = count.to_bytes(4, "little")
time.sleep(10) time.sleep(10)
except socket.gaierror as e: except socket.gaierror as e:
print(f"Unable to resolve hostname {self._rem.ip}") print(f"Unable to resolve hostname {self._remote.ip}")
self._rem.socks[Socket.register].close() self._remote.socks[Socket.register].close()
raise e raise e
@ -46,78 +39,48 @@ class Updater(threading.Thread):
def __init__(self, remote): def __init__(self, remote):
super().__init__(name="updater", target=self.update, daemon=True) super().__init__(name="updater", target=self.update, daemon=True)
self._rem = remote self._remote = remote
self._rem.socks[Socket.response].bind( self._remote.socks[Socket.response].bind(
(socket.gethostbyname(socket.gethostname()), self._rem.port) (socket.gethostbyname(socket.gethostname()), self._remote.port)
) )
self.expected_packet = VBAN_VMRT_Packet_Header() self.packet_expected = VbanRtPacketHeader()
self._rem._public_packet = self._get_rt() self._remote._public_packet = self._get_rt()
def _fetch_rt_packet(self) -> Optional[VBAN_VMRT_Packet_Data]: def _get_rt(self) -> VbanRtPacket:
"""Returns a valid RT Data Packet or None"""
data, _ = self._rem.socks[Socket.response].recvfrom(2048)
# check for packet data
if len(data) > HEADER_SIZE:
# check if packet is of type VBAN
if self.expected_packet.header == data[: HEADER_SIZE - 4]:
return VBAN_VMRT_Packet_Data(
_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 _get_rt(self) -> VBAN_VMRT_Packet_Data:
"""Attempt to fetch data packet until a valid one found""" """Attempt to fetch data packet until a valid one found"""
def fget(): while True:
data = False data, _ = self._remote.socks[Socket.response].recvfrom(2048)
while not data: # check for packet data
data = self._fetch_rt_packet() if len(data) > HEADER_SIZE:
time.sleep(self._rem.DELAY) # check if packet is of type rt packet response
return data if self.packet_expected.header == data[: HEADER_SIZE - 4]:
return VbanRtPacket(data)
return fget() time.sleep(self._remote.DELAY)
def update(self): def update(self):
print(f"Listening for {', '.join(self._rem.event.get())} events") print(f"Listening for {', '.join(self._remote.event.get())} events")
( (
self._rem.cache["strip_level"], self._remote.cache["strip_level"],
self._rem.cache["bus_level"], self._remote.cache["bus_level"],
) = self._rem._get_levels(self._rem.public_packet) ) = self._remote._get_levels(self._remote.public_packet)
while self._rem.running: while self._remote.running:
start = time.time() start = time.time()
_pp = self._get_rt() _pp = self._get_rt()
self._rem._strip_buf, self._rem._bus_buf = self._rem._get_levels(_pp) self._remote._strip_buf, self._remote._bus_buf = self._remote._get_levels(
self._rem._pdirty = _pp.pdirty(self._rem.public_packet) _pp
)
self._remote._pdirty = _pp.pdirty(self._remote.public_packet)
if self._rem.event.ldirty and self._rem.ldirty: if self._remote.event.ldirty and self._remote.ldirty:
self._rem.cache["strip_level"] = self._rem._strip_buf self._remote.cache["strip_level"] = self._remote._strip_buf
self._rem.cache["bus_level"] = self._rem._bus_buf self._remote.cache["bus_level"] = self._remote._bus_buf
self._rem.subject.notify("ldirty") self._remote.subject.notify("ldirty")
if self._rem.public_packet != _pp: if self._remote.public_packet != _pp:
self._rem._public_packet = _pp self._remote._public_packet = _pp
if self._rem.event.pdirty and self._rem.pdirty: if self._remote.event.pdirty and self._remote.pdirty:
self._rem.subject.notify("pdirty") self._remote.subject.notify("pdirty")
elapsed = time.time() - start elapsed = time.time() - start
if self._rem.ratelimit - elapsed > 0: if self._remote.ratelimit - elapsed > 0:
time.sleep(self._rem.ratelimit - elapsed) time.sleep(self._remote.ratelimit - elapsed)