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

upd the interface to read/write multiple private/public packets.

Browse Source 
{VirtualStrip}.bass/mid/treble implemented reading from public packet NBS=1
This commit is contained in:
onyx-and-iris 2026-01-17 09:37:31 +00:00
parent 51394c0076
commit 96e9d6f4fd
7 changed files with 117 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
vban_cmd/bus.py
View File

@ -1,17 +1,11 @@
import time
from abc import abstractmethod
from enum import IntEnum
from typing import Union
from .enums import NBS, BusModes
from .iremote import IRemote
from .meta import bus_mode_prop, channel_bool_prop, channel_label_prop
BusModes = IntEnum(
'BusModes',
'normal amix bmix repeat composite tvmix upmix21 upmix41 upmix61 centeronly lfeonly rearonly',
start=0,
)
class Bus(IRemote):
"""
@ -31,7 +25,7 @@ class Bus(IRemote):
@property
def gain(self) -> float:
def fget():
val = self.public_packet.busgain[self.index]
val = self.public_packets[NBS.zero].busgain[self.index]
if 0 <= val <= 1200:
return val * 0.01
return (((1 << 16) - 1) - val) * -0.01
@ -109,7 +103,7 @@ class BusLevel(IRemote):
)
return tuple(
fget(i)
for i in self._remote._get_levels(self.public_packet)[1][
for i in self._remote._get_levels(self.public_packets[NBS.zero])[1][
self.range[0] : self.range[-1]
]
)
@ -157,7 +151,12 @@ def _make_bus_mode_mixin():
def get(self):
states = [
(int.from_bytes(self.public_packet.busstate[self.index], 'little') & val)
(
int.from_bytes(
self.public_packets[NBS.zero].busstate[self.index], 'little'
)
& val
)
>> 4
for val in self._modes.modevals
]

15
vban_cmd/enums.py Normal file
View File

@ -0,0 +1,15 @@
from enum import IntEnum
class NBS(IntEnum):
zero = 0
one = 1
BusModes = IntEnum(
'BusModes',
'normal amix bmix repeat composite tvmix upmix21 upmix41 upmix61 centeronly lfeonly rearonly',
start=0,
)
EQGains = IntEnum('EQGains', 'bass mid treble', start=0)

4
vban_cmd/iremote.py
View File

@ -116,9 +116,9 @@ class IRemote(metaclass=ABCMeta):
pass
@property
def public_packet(self):
def public_packets(self):
"""Returns an RT data packet."""
return self._remote.public_packet
return self._remote.public_packets
def apply(self, data):
"""Sets all parameters of a dict for the channel."""

16
vban_cmd/meta.py
View File

@ -1,5 +1,6 @@
from functools import partial
from .enums import NBS
from .util import cache_bool, cache_string
@ -13,7 +14,7 @@ def channel_bool_prop(param):
return (
not int.from_bytes(
getattr(
self.public_packet,
self.public_packets[NBS.zero],
f'{"strip" if "strip" in type(self).__name__.lower() else "bus"}state',
)[self.index],
'little',
@ -34,7 +35,7 @@ def channel_label_prop():
@partial(cache_string, param='label')
def fget(self) -> str:
return getattr(
self.public_packet,
self.public_packets[NBS.zero],
f'{"strip" if "strip" in type(self).__name__.lower() else "bus"}labels',
)[self.index]
@ -52,7 +53,9 @@ def strip_output_prop(param):
cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}')
return (
not int.from_bytes(self.public_packet.stripstate[self.index], 'little')
not int.from_bytes(
self.public_packets[NBS.zero].stripstate[self.index], 'little'
)
& getattr(self._modes, f'_bus{param.lower()}')
== 0
)
@ -71,7 +74,12 @@ def bus_mode_prop(param):
cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}')
return [
(int.from_bytes(self.public_packet.busstate[self.index], 'little') & val)
(
int.from_bytes(
self.public_packets[NBS.zero].busstate[self.index], 'little'
)
& val
)
>> 4
for val in self._modes.modevals
] == self.modestates[param]

49
vban_cmd/strip.py
View File

@ -2,6 +2,7 @@ import time
from abc import abstractmethod
from typing import Union
from .enums import NBS, EQGains
from .iremote import IRemote
from .kinds import kinds_all
from .meta import channel_bool_prop, channel_label_prop, strip_output_prop
@ -34,7 +35,7 @@ class Strip(IRemote):
def gain(self) -> float:
val = self.getter('gain')
if val is None:
val = self.gainlayer[0].gain
val = max(layer.gain for layer in self.gainlayer)
return round(val, 1)
@gain.setter
@ -262,12 +263,48 @@ class VirtualStrip(Strip):
@property
def k(self) -> int:
return
if self.public_packets[NBS.one] is None:
return 0
return self.public_packets[NBS.one].strips[self.index].karaoke
@k.setter
def k(self, val: int):
self.setter('karaoke', val)
@property
def bass(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.bass]
@bass.setter
def bass(self, val: float):
self.setter('EQGain1', val)
@property
def mid(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.mid]
@mid.setter
def mid(self, val: float):
self.setter('EQGain2', val)
med = mid
@property
def treble(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.treble]
@treble.setter
def treble(self, val: float):
self.setter('EQGain3', val)
high = treble
def appgain(self, name: str, gain: float):
self.setter('AppGain', f'("{name}", {gain})')
@ -305,7 +342,7 @@ class StripLevel(IRemote):
)
return tuple(
fget(i)
for i in self._remote._get_levels(self.public_packet)[0][
for i in self._remote._get_levels(self.public_packets[NBS.zero])[0][
self.range[0] : self.range[-1]
]
)
@ -350,9 +387,9 @@ class GainLayer(IRemote):
@property
def gain(self) -> float:
def fget():
val = getattr(self.public_packet, f'stripgainlayer{self._i + 1}')[
self.index
]
val = getattr(
self.public_packets[NBS.zero], f'stripgainlayer{self._i + 1}'
)[self.index]
if 0 <= val <= 1200:
return val * 0.01
return (((1 << 16) - 1) - val) * -0.01

8
vban_cmd/util.py
View File

@ -82,3 +82,11 @@ def deep_merge(dict1, dict2):
yield k, dict1[k]
else:
yield k, dict2[k]
def bump_framecounter(framecounter: int) -> int:
"""Increment framecounter with rollover at 0xFFFFFFFF."""
if framecounter > 0xFFFFFFFF:
return 0
else:
return framecounter + 1

46
vban_cmd/vbancmd.py
View File

@ -7,11 +7,12 @@ from pathlib import Path
from queue import Queue
from typing import Iterable, Union
from .enums import NBS
from .error import VBANCMDError
from .event import Event
from .packet import RequestHeader
from .subject import Subject
from .util import deep_merge, script
from .util import bump_framecounter, deep_merge, script
from .worker import Producer, Subscriber, Updater
logger = logging.getLogger(__name__)
@ -37,12 +38,9 @@ class VbanCmd(metaclass=ABCMeta):
for attr, val in kwargs.items():
setattr(self, attr, val)
self.packet_request = RequestHeader(
name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel,
)
self._framecounter = 0
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.subject = self.observer = Subject()
self.cache = {}
self._pdirty = False
@ -124,37 +122,49 @@ class VbanCmd(metaclass=ABCMeta):
def _set_rt(self, cmd: str, val: Union[str, float]):
"""Sends a string request command over a network."""
req_packet = RequestHeader.to_bytes(
name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel,
framecounter=self._framecounter,
)
self.sock.sendto(
self.packet_request.header + f'{cmd}={val};'.encode(),
req_packet + f'{cmd}={val};'.encode(),
(socket.gethostbyname(self.ip), self.port),
)
self.packet_request.framecounter = (
int.from_bytes(self.packet_request.framecounter, 'little') + 1
).to_bytes(4, 'little')
self._framecounter = bump_framecounter(self._framecounter)
self.cache[cmd] = val
@script
def sendtext(self, script):
"""Sends a multiple parameter string over a network."""
req_packet = RequestHeader.to_bytes(
name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel,
framecounter=self._framecounter,
)
self.sock.sendto(
self.packet_request.header + script.encode(),
req_packet + script.encode(),
(socket.gethostbyname(self.ip), self.port),
)
self.packet_request.framecounter = (
int.from_bytes(self.packet_request.framecounter, 'little') + 1
).to_bytes(4, 'little')
self._framecounter = bump_framecounter(self._framecounter)
self.logger.debug(f'sendtext: {script}')
time.sleep(self.DELAY)
@property
def type(self) -> str:
"""Returns the type of Voicemeeter installation."""
return self.public_packet.voicemeetertype
return self.public_packets[NBS.zero].voicemeetertype
@property
def version(self) -> str:
"""Returns Voicemeeter's version as a string"""
return '{0}.{1}.{2}.{3}'.format(*self.public_packet.voicemeeterversion)
return '{0}.{1}.{2}.{3}'.format(
*self.public_packets[NBS.zero].voicemeeterversion
)
@property
def pdirty(self):
@ -167,8 +177,8 @@ class VbanCmd(metaclass=ABCMeta):
return self._ldirty
@property
def public_packet(self):
return self._public_packet
def public_packets(self):
return self._public_packets
def clear_dirty(self) -> None:
while self.pdirty: