15 Commits

17 changed files with 664 additions and 242 deletions

View File

@@ -11,6 +11,13 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
- [x] - [x]
## [2.6.0] - 2026-02-26
### Added
- support for packet with [ident:1](https://github.com/onyx-and-iris/Voicemeeter-SDK/blob/3be2c1c36563afbd6df3da8436406c77d2cc1f10/VoicemeeterRemote.h#L982) in VBAN TEXT subprotocol.
- This includes Strip 3D, PEQ, comp, gate, denoiser and pitch parameters.
## [2.5.2] - 2025-01-25 ## [2.5.2] - 2025-01-25
### Changed ### Changed

View File

@@ -171,9 +171,7 @@ example:
print(vban.strip[4].comp.knob) print(vban.strip[4].comp.knob)
``` ```
Strip Comp properties are defined as write only. Strip Comp `knob` is defined for all versions, all other parameters potato only.
`knob` defined for all versions, all other parameters potato only.
##### Strip.Gate ##### Strip.Gate
@@ -193,9 +191,7 @@ example:
vban.strip[2].gate.attack = 300.8 vban.strip[2].gate.attack = 300.8
``` ```
Strip Gate properties are defined as write only, potato version only. Strip Gate `knob` is defined for all versions, all other parameters potato only.
`knob` defined for all versions, all other parameters potato only.
##### Strip.Denoiser ##### Strip.Denoiser
@@ -212,7 +208,32 @@ The following properties are available.
- `on`: boolean - `on`: boolean
- `ab`: boolean - `ab`: boolean
Strip EQ properties are defined as write only, potato version only. example:
```python
vban.strip[0].eq.ab = True
```
##### Strip.EQ.Channel.Cell
The following properties are available.
- `on`: boolean
- `type`: int, from 0 up to 6
- `f`: float, from 20.0 up to 20_000.0
- `gain`: float, from -36.0 up to 18.0
- `q`: float, from 0.3 up to 100
example:
```python
vban.strip[0].eq.channel[0].cell[2].on = True
vban.strip[1].eq.channel[0].cell[2].f = 5000
```
Strip EQ parameters are defined for PhysicalStrips, potato version only.
Only channel[0] properties are readable over VBAN.
##### Gainlayers ##### Gainlayers
@@ -400,8 +421,8 @@ You just need to define a key `extends` in the config TOML, that names the confi
Three example 'extender' configs are included with the repo. You may load them with: Three example 'extender' configs are included with the repo. You may load them with:
```python ```python
import voicemeeterlib import vban_cmd
with voicemeeterlib.api('banana') as vm: with vban_cmd.api('banana') as vm:
vm.apply_config('extender') vm.apply_config('extender')
``` ```
@@ -528,13 +549,15 @@ with vban_cmd.api('banana', **opts) as vban:
... ...
``` ```
## Tests ### Run tests
First make sure you installed the [development dependencies](https://github.com/onyx-and-iris/vban-cmd-python#installation) Install [poetry](https://python-poetry.org/docs/#installation) and then:
Then from tests directory: ```powershell
poetry poe test-basic
`pytest -v` poetry poe test-banana
poetry poe test-potato
```
## Resources ## Resources

View File

@@ -1,4 +1,5 @@
import logging import logging
import os
import tkinter as tk import tkinter as tk
from tkinter import ttk from tkinter import ttk
@@ -100,7 +101,14 @@ class App(tk.Tk):
def main(): def main():
with vban_cmd.api('banana', ldirty=True) as vban: KIND_ID = 'banana'
conn = {
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
}
with vban_cmd.api(KIND_ID, ldirty=True, **conn) as vban:
app = App(vban) app = App(vban)
app.mainloop() app.mainloop()

View File

@@ -1,3 +1,4 @@
import os
import threading import threading
from logging import config from logging import config
@@ -92,8 +93,13 @@ class Observer:
def main(): def main():
KIND_ID = 'potato' KIND_ID = 'potato'
conn = {
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
}
with vban_cmd.api(KIND_ID) as vban: with vban_cmd.api(KIND_ID, **conn) as vban:
stop_event = threading.Event() stop_event = threading.Event()
with Observer(vban, stop_event): with Observer(vban, stop_event):

View File

@@ -1,4 +1,5 @@
import logging import logging
import os
import vban_cmd import vban_cmd
@@ -23,8 +24,13 @@ class App:
def main(): def main():
KIND_ID = 'banana' KIND_ID = 'banana'
conn = {
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
}
with vban_cmd.api(KIND_ID, pdirty=True, ldirty=True) as vban: with vban_cmd.api(KIND_ID, pdirty=True, ldirty=True, **conn) as vban:
App(vban) App(vban)
while _ := input('Press <Enter> to exit\n'): while _ := input('Press <Enter> to exit\n'):

View File

@@ -1,6 +1,6 @@
[project] [project]
name = "vban-cmd" name = "vban-cmd"
version = "2.5.2" version = "2.6.0"
description = "Python interface for the VBAN RT Packet Service (Sendtext)" description = "Python interface for the VBAN RT Packet Service (Sendtext)"
authors = [{ name = "Onyx and Iris", email = "code@onyxandiris.online" }] authors = [{ name = "Onyx and Iris", email = "code@onyxandiris.online" }]
license = { text = "MIT" } license = { text = "MIT" }

View File

@@ -1,5 +1,5 @@
import abc
import time import time
from abc import abstractmethod
from typing import Union from typing import Union
from .enums import NBS, BusModes from .enums import NBS, BusModes
@@ -14,7 +14,7 @@ class Bus(IRemote):
Defines concrete implementation for bus Defines concrete implementation for bus
""" """
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
pass pass

View File

@@ -1,4 +1,11 @@
from enum import IntEnum from enum import Enum, IntEnum, unique
@unique
class KindId(Enum):
BASIC = 1
BANANA = 2
POTATO = 3
class NBS(IntEnum): class NBS(IntEnum):
@@ -11,5 +18,3 @@ BusModes = IntEnum(
'normal amix bmix repeat composite tvmix upmix21 upmix41 upmix61 centeronly lfeonly rearonly', 'normal amix bmix repeat composite tvmix upmix21 upmix41 upmix61 centeronly lfeonly rearonly',
start=0, start=0,
) )
EQGains = IntEnum('EQGains', 'bass mid treble', start=0)

View File

@@ -1,5 +1,5 @@
import abc
import logging import logging
from abc import abstractmethod
from enum import IntEnum from enum import IntEnum
from functools import cached_property from functools import cached_property
from typing import Iterable from typing import Iterable
@@ -115,7 +115,7 @@ class FactoryBase(VbanCmd):
) )
@property @property
@abstractmethod @abc.abstractmethod
def steps(self): def steps(self):
pass pass

View File

@@ -1,6 +1,6 @@
import abc
import logging import logging
import time import time
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -78,7 +78,7 @@ class Modes:
) )
class IRemote(metaclass=ABCMeta): class IRemote(abc.ABC):
""" """
Common interface between base class and extended (higher) classes Common interface between base class and extended (higher) classes
@@ -111,7 +111,7 @@ class IRemote(metaclass=ABCMeta):
return ''.join(cmd) return ''.join(cmd)
@property @property
@abstractmethod @abc.abstractmethod
def identifier(self): def identifier(self):
pass pass

View File

@@ -1,16 +1,9 @@
from dataclasses import dataclass from dataclasses import dataclass
from enum import Enum, unique
from .enums import KindId
from .error import VBANCMDError from .error import VBANCMDError
@unique
class KindId(Enum):
BASIC = 1
BANANA = 2
POTATO = 3
class SingletonType(type): class SingletonType(type):
"""ensure only a single instance of a kind map object""" """ensure only a single instance of a kind map object"""
@@ -22,12 +15,15 @@ class SingletonType(type):
return cls._instances[cls] return cls._instances[cls]
@dataclass @dataclass(frozen=True)
class KindMapClass(metaclass=SingletonType): class KindMapClass(metaclass=SingletonType):
name: str name: str
ins: tuple ins: tuple
outs: tuple outs: tuple
vban: tuple vban: tuple
strip_channels: int
bus_channels: int
cells: int
@property @property
def phys_in(self): def phys_in(self):
@@ -65,28 +61,37 @@ class KindMapClass(metaclass=SingletonType):
return self.name.capitalize() return self.name.capitalize()
@dataclass @dataclass(frozen=True)
class BasicMap(KindMapClass): class BasicMap(KindMapClass):
name: str name: str
ins: tuple = (2, 1) ins: tuple = (2, 1)
outs: tuple = (1, 1) outs: tuple = (1, 1)
vban: tuple = (4, 4, 1, 1) vban: tuple = (4, 4, 1, 1)
strip_channels: int = 0
bus_channels: int = 0
cells: int = 0
@dataclass @dataclass(frozen=True)
class BananaMap(KindMapClass): class BananaMap(KindMapClass):
name: str name: str
ins: tuple = (3, 2) ins: tuple = (3, 2)
outs: tuple = (3, 2) outs: tuple = (3, 2)
vban: tuple = (8, 8, 1, 1) vban: tuple = (8, 8, 1, 1)
strip_channels: int = 0
bus_channels: int = 8
cells: int = 6
@dataclass @dataclass(frozen=True)
class PotatoMap(KindMapClass): class PotatoMap(KindMapClass):
name: str name: str
ins: tuple = (5, 3) ins: tuple = (5, 3)
outs: tuple = (5, 3) outs: tuple = (5, 3)
vban: tuple = (8, 8, 1, 1) vban: tuple = (8, 8, 1, 1)
strip_channels: int = 2
bus_channels: int = 8
cells: int = 6
def kind_factory(kind_id): def kind_factory(kind_id):

View File

@@ -106,13 +106,23 @@ def xy_prop(param):
def fget(self): def fget(self):
cmd = self._cmd(param) cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}') self.logger.debug(f'getter: {cmd}')
_type, axis = param.split('_')
if self.public_packets[NBS.one] is None: if self.public_packets[NBS.one] is None:
return 0.0 return 0.0
x, y = getattr(
self.public_packets[NBS.one].strips[self.index], f'position_{_type.lower()}' positions = self.public_packets[NBS.one].strips[self.index].positions
) match param:
return x if axis == 'x' else y case 'pan_x':
return positions.pan_x
case 'pan_y':
return positions.pan_y
case 'color_x':
return positions.color_x
case 'color_y':
return positions.color_y
case 'fx1':
return positions.fx1
case 'fx2':
return positions.fx2
def fset(self, val): def fset(self, val):
self.setter(param, val) self.setter(param, val)
@@ -129,12 +139,17 @@ def send_prop(param):
self.logger.debug(f'getter: {cmd}') self.logger.debug(f'getter: {cmd}')
if self.public_packets[NBS.one] is None: if self.public_packets[NBS.one] is None:
return 0.0 return 0.0
val = getattr(self.public_packets[NBS.one].strips[self.index], f'send_{param}')
sends = self.public_packets[NBS.one].strips[self.index].sends
match param: match param:
case 'reverb' | 'fx1': case 'reverb':
return val[0] return sends.reverb
case 'delay' | 'fx2': case 'delay':
return val[1] return sends.delay
case 'fx1':
return sends.fx1
case 'fx2':
return sends.fx2
def fset(self, val): def fset(self, val):
self.setter(param, val) self.setter(param, val)

View File

@@ -1,4 +1,6 @@
import struct
from dataclasses import dataclass from dataclasses import dataclass
from typing import NamedTuple
from .enums import NBS from .enums import NBS
from .kinds import KindMapClass from .kinds import KindMapClass
@@ -22,34 +24,34 @@ class VbanRtPacket:
nbs: NBS nbs: NBS
_kind: KindMapClass _kind: KindMapClass
_voicemeeterType: bytes # data[28:29] _voicemeeterType: bytes
_reserved: bytes # data[29:30] _reserved: bytes
_buffersize: bytes # data[30:32] _buffersize: bytes
_voicemeeterVersion: bytes # data[32:36] _voicemeeterVersion: bytes
_optionBits: bytes # data[36:40] _optionBits: bytes
_samplerate: bytes # data[40:44] _samplerate: bytes
@dataclass @dataclass
class VbanRtPacketNBS0(VbanRtPacket): class VbanRtPacketNBS0(VbanRtPacket):
"""Represents the body of a VBAN RT data packet with NBS 0""" """Represents the body of a VBAN RT data packet with NBS 0"""
_inputLeveldB100: bytes # data[44:112] _inputLeveldB100: bytes
_outputLeveldB100: bytes # data[112:240] _outputLeveldB100: bytes
_TransportBit: bytes # data[240:244] _TransportBit: bytes
_stripState: bytes # data[244:276] _stripState: bytes
_busState: bytes # data[276:308] _busState: bytes
_stripGaindB100Layer1: bytes # data[308:324] _stripGaindB100Layer1: bytes
_stripGaindB100Layer2: bytes # data[324:340] _stripGaindB100Layer2: bytes
_stripGaindB100Layer3: bytes # data[340:356] _stripGaindB100Layer3: bytes
_stripGaindB100Layer4: bytes # data[356:372] _stripGaindB100Layer4: bytes
_stripGaindB100Layer5: bytes # data[372:388] _stripGaindB100Layer5: bytes
_stripGaindB100Layer6: bytes # data[388:404] _stripGaindB100Layer6: bytes
_stripGaindB100Layer7: bytes # data[404:420] _stripGaindB100Layer7: bytes
_stripGaindB100Layer8: bytes # data[420:436] _stripGaindB100Layer8: bytes
_busGaindB100: bytes # data[436:452] _busGaindB100: bytes
_stripLabelUTF8c60: bytes # data[452:932] _stripLabelUTF8c60: bytes
_busLabelUTF8c60: bytes # data[932:1412] _busLabelUTF8c60: bytes
@classmethod @classmethod
def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes): def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes):
@@ -216,64 +218,135 @@ class VbanRtPacketNBS0(VbanRtPacket):
) )
class Audibility(NamedTuple):
knob: float
comp: float
gate: float
denoiser: float
class Positions(NamedTuple):
pan_x: float
pan_y: float
color_x: float
color_y: float
fx1: float
fx2: float
class EqGains(NamedTuple):
bass: float
mid: float
treble: float
class ParametricEQSettings(NamedTuple):
on: bool
type: int
gain: float
freq: float
q: float
class Sends(NamedTuple):
reverb: float
delay: float
fx1: float
fx2: float
class CompressorSettings(NamedTuple):
gain_in: float
attack_ms: float
release_ms: float
n_knee: float
ratio: float
threshold: float
c_enabled: bool
makeup: bool
gain_out: float
class GateSettings(NamedTuple):
threshold_in: float
damping_max: float
bp_sidechain: bool
attack_ms: float
hold_ms: float
release_ms: float
class DenoiserSettings(NamedTuple):
threshold: float
class PitchSettings(NamedTuple):
enabled: bool
dry_wet: float
value: float
formant_lo: float
formant_med: float
formant_high: float
@dataclass @dataclass
class VbanVMParamStrip: class VbanVMParamStrip:
"""Represents the VBAN_VMPARAMSTRIP_PACKET structure""" """Represents the VBAN_VMPARAMSTRIP_PACKET structure"""
_mode: bytes # long = 4 bytes data[0:4] _mode: bytes
_dblevel: bytes # float = 4 bytes data[4:8] _dblevel: bytes
_audibility: bytes # short = 2 bytes data[8:10] _audibility: bytes
_pos3D_x: bytes # short = 2 bytes data[10:12] _pos3D_x: bytes
_pos3D_y: bytes # short = 2 bytes data[12:14] _pos3D_y: bytes
_posColor_x: bytes # short = 2 bytes data[14:16] _posColor_x: bytes
_posColor_y: bytes # short = 2 bytes data[16:18] _posColor_y: bytes
_EQgain1: bytes # short = 2 bytes data[18:20] _EQgain1: bytes
_EQgain2: bytes # short = 2 bytes data[20:22] _EQgain2: bytes
_EQgain3: bytes # short = 2 bytes data[22:24] _EQgain3: bytes
# First channel parametric EQ # First channel parametric EQ
_PEQ_eqOn: bytes # 6 * char = 6 bytes data[24:30] _PEQ_eqOn: bytes
_PEQ_eqtype: bytes # 6 * char = 6 bytes data[30:36] _PEQ_eqtype: bytes
_PEQ_eqgain: bytes # 6 * float = 24 bytes data[36:60] _PEQ_eqgain: bytes
_PEQ_eqfreq: bytes # 6 * float = 24 bytes data[60:84] _PEQ_eqfreq: bytes
_PEQ_eqq: bytes # 6 * float = 24 bytes data[84:108] _PEQ_eqq: bytes
_audibility_c: bytes # short = 2 bytes data[108:110] _audibility_c: bytes
_audibility_g: bytes # short = 2 bytes data[110:112] _audibility_g: bytes
_audibility_d: bytes # short = 2 bytes data[112:114] _audibility_d: bytes
_posMod_x: bytes # short = 2 bytes data[114:116] _posMod_x: bytes
_posMod_y: bytes # short = 2 bytes data[116:118] _posMod_y: bytes
_send_reverb: bytes # short = 2 bytes data[118:120] _send_reverb: bytes
_send_delay: bytes # short = 2 bytes data[120:122] _send_delay: bytes
_send_fx1: bytes # short = 2 bytes data[122:124] _send_fx1: bytes
_send_fx2: bytes # short = 2 bytes data[124:126] _send_fx2: bytes
_dblimit: bytes # short = 2 bytes data[126:128] _dblimit: bytes
_nKaraoke: bytes # short = 2 bytes data[128:130] _nKaraoke: bytes
_COMP_gain_in: bytes # short = 2 bytes data[130:132] _COMP_gain_in: bytes
_COMP_attack_ms: bytes # short = 2 bytes data[132:134] _COMP_attack_ms: bytes
_COMP_release_ms: bytes # short = 2 bytes data[134:136] _COMP_release_ms: bytes
_COMP_n_knee: bytes # short = 2 bytes data[136:138] _COMP_n_knee: bytes
_COMP_comprate: bytes # short = 2 bytes data[138:140] _COMP_comprate: bytes
_COMP_threshold: bytes # short = 2 bytes data[140:142] _COMP_threshold: bytes
_COMP_c_enabled: bytes # short = 2 bytes data[142:144] _COMP_c_enabled: bytes
_COMP_c_auto: bytes # short = 2 bytes data[144:146] _COMP_c_auto: bytes
_COMP_gain_out: bytes # short = 2 bytes data[146:148] _COMP_gain_out: bytes
_GATE_dBThreshold_in: bytes # short = 2 bytes data[148:150] _GATE_dBThreshold_in: bytes
_GATE_dBDamping_max: bytes # short = 2 bytes data[150:152] _GATE_dBDamping_max: bytes
_GATE_BP_Sidechain: bytes # short = 2 bytes data[152:154] _GATE_BP_Sidechain: bytes
_GATE_attack_ms: bytes # short = 2 bytes data[154:156] _GATE_attack_ms: bytes
_GATE_hold_ms: bytes # short = 2 bytes data[156:158] _GATE_hold_ms: bytes
_GATE_release_ms: bytes # short = 2 bytes data[158:160] _GATE_release_ms: bytes
_DenoiserThreshold: bytes # short = 2 bytes data[160:162] _DenoiserThreshold: bytes
_PitchEnabled: bytes # short = 2 bytes data[162:164] _PitchEnabled: bytes
_Pitch_DryWet: bytes # short = 2 bytes data[164:166] _Pitch_DryWet: bytes
_Pitch_Value: bytes # short = 2 bytes data[166:168] _Pitch_Value: bytes
_Pitch_formant_lo: bytes # short = 2 bytes data[168:170] _Pitch_formant_lo: bytes
_Pitch_formant_med: bytes # short = 2 bytes data[170:172] _Pitch_formant_med: bytes
_Pitch_formant_high: bytes # short = 2 bytes data[172:174] _Pitch_formant_high: bytes
@classmethod @classmethod
def from_bytes(cls, data: bytes): def from_bytes(cls, data: bytes):
@@ -333,59 +406,133 @@ class VbanVMParamStrip:
return int.from_bytes(self._mode, 'little') return int.from_bytes(self._mode, 'little')
@property @property
def position_pan(self) -> tuple[int, int]: def audibility(self) -> Audibility:
return ( return Audibility(
round(int.from_bytes(self._audibility, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._audibility_c, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._audibility_g, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._audibility_d, 'little', signed=True) * 0.01, 2),
)
@property
def positions(self) -> Positions:
return Positions(
round(int.from_bytes(self._pos3D_x, 'little', signed=True) * 0.01, 2), round(int.from_bytes(self._pos3D_x, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._pos3D_y, 'little', signed=True) * 0.01, 2), round(int.from_bytes(self._pos3D_y, 'little', signed=True) * 0.01, 2),
)
@property
def position_color(self) -> tuple[int, int]:
return (
round(int.from_bytes(self._posColor_x, 'little', signed=True) * 0.01, 2), round(int.from_bytes(self._posColor_x, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._posColor_y, 'little', signed=True) * 0.01, 2), round(int.from_bytes(self._posColor_y, 'little', signed=True) * 0.01, 2),
)
@property
def position_fx(self) -> tuple[int, int]:
return (
round(int.from_bytes(self._posMod_x, 'little', signed=True) * 0.01, 2), round(int.from_bytes(self._posMod_x, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._posMod_y, 'little', signed=True) * 0.01, 2), round(int.from_bytes(self._posMod_y, 'little', signed=True) * 0.01, 2),
) )
@property @property
def send_reverb(self) -> tuple[float, float]: def eqgains(self) -> EqGains:
return ( return EqGains(
round(int.from_bytes(self._send_reverb, 'little', signed=True) * 0.01, 2), *[
round(int.from_bytes(self._send_delay, 'little', signed=True) * 0.01, 2),
)
send_delay = send_reverb
@property
def send_fx1(self) -> tuple[float, float]:
return (
round(int.from_bytes(self._send_fx1, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._send_fx2, 'little', signed=True) * 0.01, 2),
)
send_fx2 = send_fx1
@property
def eqgains(self) -> tuple[float, float, float]:
return tuple(
round( round(
int.from_bytes(getattr(self, f'_EQgain{i}'), 'little', signed=True) int.from_bytes(getattr(self, f'_EQgain{i}'), 'little', signed=True)
* 0.01, * 0.01,
2, 2,
) )
for i in range(1, 4) for i in range(1, 4)
]
)
@property
def parametric_eq(self) -> tuple[ParametricEQSettings, ...]:
return tuple(
ParametricEQSettings(
on=bool(int.from_bytes(self._PEQ_eqOn[i : i + 1], 'little')),
type=int.from_bytes(self._PEQ_eqtype[i : i + 1], 'little'),
freq=struct.unpack('<f', self._PEQ_eqfreq[i * 4 : (i + 1) * 4])[0],
gain=struct.unpack('<f', self._PEQ_eqgain[i * 4 : (i + 1) * 4])[0],
q=struct.unpack('<f', self._PEQ_eqq[i * 4 : (i + 1) * 4])[0],
)
for i in range(6)
)
@property
def sends(self) -> Sends:
return Sends(
round(int.from_bytes(self._send_reverb, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._send_delay, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._send_fx1, 'little', signed=True) * 0.01, 2),
round(int.from_bytes(self._send_fx2, 'little', signed=True) * 0.01, 2),
) )
@property @property
def karaoke(self) -> int: def karaoke(self) -> int:
return int.from_bytes(self._nKaraoke, 'little') return int.from_bytes(self._nKaraoke, 'little')
@property
def compressor(self) -> CompressorSettings:
return CompressorSettings(
gain_in=round(
int.from_bytes(self._COMP_gain_in, 'little', signed=True) * 0.01, 2
),
attack_ms=round(int.from_bytes(self._COMP_attack_ms, 'little') * 0.1, 2),
release_ms=round(int.from_bytes(self._COMP_release_ms, 'little') * 0.1, 2),
n_knee=round(int.from_bytes(self._COMP_n_knee, 'little') * 0.01, 2),
ratio=round(int.from_bytes(self._COMP_comprate, 'little') * 0.01, 2),
threshold=round(
int.from_bytes(self._COMP_threshold, 'little', signed=True) * 0.01, 2
),
c_enabled=bool(int.from_bytes(self._COMP_c_enabled, 'little')),
makeup=bool(int.from_bytes(self._COMP_c_auto, 'little')),
gain_out=round(
int.from_bytes(self._COMP_gain_out, 'little', signed=True) * 0.01, 2
),
)
@property
def gate(self) -> GateSettings:
return GateSettings(
threshold_in=round(
int.from_bytes(self._GATE_dBThreshold_in, 'little', signed=True) * 0.01,
2,
),
damping_max=round(
int.from_bytes(self._GATE_dBDamping_max, 'little', signed=True) * 0.01,
2,
),
bp_sidechain=round(
int.from_bytes(self._GATE_BP_Sidechain, 'little') * 0.1, 2
),
attack_ms=round(int.from_bytes(self._GATE_attack_ms, 'little') * 0.1, 2),
hold_ms=round(int.from_bytes(self._GATE_hold_ms, 'little') * 0.1, 2),
release_ms=round(int.from_bytes(self._GATE_release_ms, 'little') * 0.1, 2),
)
@property
def denoiser(self) -> DenoiserSettings:
return DenoiserSettings(
threshold=round(
int.from_bytes(self._DenoiserThreshold, 'little', signed=True) * 0.01, 2
)
)
@property
def pitch(self) -> PitchSettings:
return PitchSettings(
enabled=bool(int.from_bytes(self._PitchEnabled, 'little')),
dry_wet=round(
int.from_bytes(self._Pitch_DryWet, 'little', signed=True) * 0.01, 2
),
value=round(
int.from_bytes(self._Pitch_Value, 'little', signed=True) * 0.01, 2
),
formant_lo=round(
int.from_bytes(self._Pitch_formant_lo, 'little', signed=True) * 0.01, 2
),
formant_med=round(
int.from_bytes(self._Pitch_formant_med, 'little', signed=True) * 0.01, 2
),
formant_high=round(
int.from_bytes(self._Pitch_formant_high, 'little', signed=True) * 0.01,
2,
),
)
@dataclass @dataclass
class VbanRtPacketNBS1(VbanRtPacket): class VbanRtPacketNBS1(VbanRtPacket):
@@ -422,19 +569,38 @@ class VbanRtPacketNBS1(VbanRtPacket):
class SubscribeHeader: class SubscribeHeader:
"""Represents the header of an RT subscription packet""" """Represents the header of an RT subscription packet"""
ident: NBS = NBS.zero nbs: NBS = NBS.zero
name = 'Register-RTP' name: str = 'Register-RTP'
timeout = 15 timeout: int = 15
vban: bytes = 'VBAN'.encode()
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little') @property
format_nbs: bytes = (ident.value & 0xFF).to_bytes(1, 'little') def vban(self) -> bytes:
format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, 'little') return b'VBAN'
format_bit: bytes = (timeout & 0xFF).to_bytes(1, 'little') # timeout
streamname: bytes = name.encode('ascii') + bytes(16 - len(name)) @property
def format_sr(self) -> bytes:
return VBAN_PROTOCOL_SERVICE.to_bytes(1, 'little')
@property
def format_nbs(self) -> bytes:
return (self.nbs.value & 0xFF).to_bytes(1, 'little')
@property
def format_nbc(self) -> bytes:
return VBAN_SERVICE_RTPACKETREGISTER.to_bytes(1, 'little')
@property
def format_bit(self) -> bytes:
return (self.timeout & 0xFF).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod @classmethod
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes: def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(ident=nbs) header = cls(nbs=nbs)
data = bytearray() data = bytearray()
data.extend(header.vban) data.extend(header.vban)
data.extend(header.format_sr) data.extend(header.format_sr)
@@ -451,30 +617,31 @@ class VbanRtPacketHeader:
"""Represents the header of an RT response packet""" """Represents the header of an RT response packet"""
name: str = 'Voicemeeter-RTP' name: str = 'Voicemeeter-RTP'
vban: bytes = 'VBAN'.encode() format_sr: int = VBAN_PROTOCOL_SERVICE
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little') format_nbs: int = 0
format_nbs: bytes = (0).to_bytes(1, 'little') format_nbc: int = VBAN_SERVICE_RTPACKET
format_nbc: bytes = (VBAN_SERVICE_RTPACKET).to_bytes(1, 'little') format_bit: int = 0
format_bit: bytes = (0).to_bytes(1, 'little')
streamname: bytes = name.encode('ascii') + bytes(16 - len(name)) @property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod @classmethod
def from_bytes(cls, data: bytes): def from_bytes(cls, data: bytes):
if len(data) < HEADER_SIZE: if len(data) < HEADER_SIZE:
raise ValueError('Data is too short to be a valid VbanRTPPacketHeader') raise ValueError('Data is too short to be a valid VbanRTPPacketHeader')
vban = data[0:4]
format_sr = data[4]
format_nbs = data[5]
format_nbc = data[6]
format_bit = data[7]
name = data[8:24].rstrip(b'\x00').decode('utf-8') name = data[8:24].rstrip(b'\x00').decode('utf-8')
return cls( return cls(
name=name, name=name,
vban=vban, format_sr=data[4] & VBAN_SERVICE_MASK,
format_sr=format_sr & VBAN_SERVICE_MASK, format_nbs=data[5],
format_nbs=format_nbs, format_nbc=data[6],
format_nbc=format_nbc, format_bit=data[7],
format_bit=format_bit,
) )
@@ -485,21 +652,30 @@ class RequestHeader:
name: str name: str
bps_index: int bps_index: int
channel: int channel: int
vban: bytes = 'VBAN'.encode() framecounter: int = 0
nbs: bytes = (0).to_bytes(1, 'little')
bit: bytes = (0x10).to_bytes(1, 'little')
framecounter: bytes = (0).to_bytes(4, 'little')
@property @property
def sr(self): def vban(self) -> bytes:
return b'VBAN'
@property
def sr(self) -> bytes:
return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little') return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
@property @property
def nbc(self): def nbs(self) -> bytes:
return (0).to_bytes(1, 'little')
@property
def nbc(self) -> bytes:
return (self.channel).to_bytes(1, 'little') return (self.channel).to_bytes(1, 'little')
@property @property
def streamname(self): def bit(self) -> bytes:
return (0x10).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
return self.name.encode() + bytes(16 - len(self.name)) return self.name.encode() + bytes(16 - len(self.name))
@classmethod @classmethod
@@ -509,6 +685,7 @@ class RequestHeader:
header = cls( header = cls(
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
) )
data = bytearray() data = bytearray()
data.extend(header.vban) data.extend(header.vban)
data.extend(header.sr) data.extend(header.sr)

View File

@@ -1,9 +1,9 @@
import abc
import time import time
from abc import abstractmethod
from typing import Union from typing import Union
from . import kinds from . import kinds
from .enums import NBS, EQGains from .enums import NBS
from .iremote import IRemote from .iremote import IRemote
from .meta import ( from .meta import (
channel_bool_prop, channel_bool_prop,
@@ -21,7 +21,7 @@ class Strip(IRemote):
Defines concrete implementation for strip Defines concrete implementation for strip
""" """
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
pass pass
@@ -68,7 +68,7 @@ class PhysicalStrip(Strip):
'comp': StripComp(remote, index), 'comp': StripComp(remote, index),
'gate': StripGate(remote, index), 'gate': StripGate(remote, index),
'denoiser': StripDenoiser(remote, index), 'denoiser': StripDenoiser(remote, index),
'eq': StripEQ(remote, index), 'eq': StripEQ.make(remote, index),
}, },
) )
@@ -76,12 +76,14 @@ class PhysicalStrip(Strip):
return f'{type(self).__name__}{self.index}' return f'{type(self).__name__}{self.index}'
@property @property
def device(self): def audibility(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.knob
@property @audibility.setter
def sr(self): def audibility(self, val: float):
return self.setter('audibility', val)
class StripComp(IRemote): class StripComp(IRemote):
@@ -91,7 +93,9 @@ class StripComp(IRemote):
@property @property
def knob(self) -> float: def knob(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.comp
@knob.setter @knob.setter
def knob(self, val: float): def knob(self, val: float):
@@ -99,7 +103,9 @@ class StripComp(IRemote):
@property @property
def gainin(self) -> float: def gainin(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.gain_in
@gainin.setter @gainin.setter
def gainin(self, val: float): def gainin(self, val: float):
@@ -107,7 +113,9 @@ class StripComp(IRemote):
@property @property
def ratio(self) -> float: def ratio(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.ratio
@ratio.setter @ratio.setter
def ratio(self, val: float): def ratio(self, val: float):
@@ -115,7 +123,9 @@ class StripComp(IRemote):
@property @property
def threshold(self) -> float: def threshold(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.threshold
@threshold.setter @threshold.setter
def threshold(self, val: float): def threshold(self, val: float):
@@ -123,7 +133,9 @@ class StripComp(IRemote):
@property @property
def attack(self) -> float: def attack(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.attack_ms
@attack.setter @attack.setter
def attack(self, val: float): def attack(self, val: float):
@@ -131,7 +143,9 @@ class StripComp(IRemote):
@property @property
def release(self) -> float: def release(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.release_ms
@release.setter @release.setter
def release(self, val: float): def release(self, val: float):
@@ -139,7 +153,9 @@ class StripComp(IRemote):
@property @property
def knee(self) -> float: def knee(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.n_knee
@knee.setter @knee.setter
def knee(self, val: float): def knee(self, val: float):
@@ -147,7 +163,9 @@ class StripComp(IRemote):
@property @property
def gainout(self) -> float: def gainout(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.gain_out
@gainout.setter @gainout.setter
def gainout(self, val: float): def gainout(self, val: float):
@@ -155,7 +173,9 @@ class StripComp(IRemote):
@property @property
def makeup(self) -> bool: def makeup(self) -> bool:
return if self.public_packets[NBS.one] is None:
return False
return bool(self.public_packets[NBS.one].strips[self.index].compressor.makeup)
@makeup.setter @makeup.setter
def makeup(self, val: bool): def makeup(self, val: bool):
@@ -169,7 +189,9 @@ class StripGate(IRemote):
@property @property
def knob(self) -> float: def knob(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.gate
@knob.setter @knob.setter
def knob(self, val: float): def knob(self, val: float):
@@ -177,7 +199,9 @@ class StripGate(IRemote):
@property @property
def threshold(self) -> float: def threshold(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].gate.threshold_in
@threshold.setter @threshold.setter
def threshold(self, val: float): def threshold(self, val: float):
@@ -185,7 +209,9 @@ class StripGate(IRemote):
@property @property
def damping(self) -> float: def damping(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].gate.damping_max
@damping.setter @damping.setter
def damping(self, val: float): def damping(self, val: float):
@@ -193,7 +219,9 @@ class StripGate(IRemote):
@property @property
def bpsidechain(self) -> int: def bpsidechain(self) -> int:
return if self.public_packets[NBS.one] is None:
return 0
return self.public_packets[NBS.one].strips[self.index].gate.bp_sidechain
@bpsidechain.setter @bpsidechain.setter
def bpsidechain(self, val: int): def bpsidechain(self, val: int):
@@ -201,7 +229,9 @@ class StripGate(IRemote):
@property @property
def attack(self) -> float: def attack(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].gate.attack_ms
@attack.setter @attack.setter
def attack(self, val: float): def attack(self, val: float):
@@ -209,7 +239,9 @@ class StripGate(IRemote):
@property @property
def hold(self) -> float: def hold(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].gate.hold_ms
@hold.setter @hold.setter
def hold(self, val: float): def hold(self, val: float):
@@ -217,7 +249,9 @@ class StripGate(IRemote):
@property @property
def release(self) -> float: def release(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].gate.release_ms
@release.setter @release.setter
def release(self, val: float): def release(self, val: float):
@@ -231,7 +265,9 @@ class StripDenoiser(IRemote):
@property @property
def knob(self) -> float: def knob(self) -> float:
return if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.denoiser
@knob.setter @knob.setter
def knob(self, val: float): def knob(self, val: float):
@@ -239,6 +275,25 @@ class StripDenoiser(IRemote):
class StripEQ(IRemote): class StripEQ(IRemote):
@classmethod
def make(cls, remote, i):
"""
Factory method for Strip EQ.
Returns a StripEQ class.
"""
STRIPEQ_cls = type(
'StripEQ',
(cls,),
{
'channel': tuple(
StripEQCh.make(remote, i, j)
for j in range(remote.kind.strip_channels)
)
},
)
return STRIPEQ_cls(remote, i)
@property @property
def identifier(self) -> str: def identifier(self) -> str:
return f'strip[{self.index}].eq' return f'strip[{self.index}].eq'
@@ -260,6 +315,140 @@ class StripEQ(IRemote):
self.setter('ab', 1 if val else 0) self.setter('ab', 1 if val else 0)
class StripEQCh(IRemote):
@classmethod
def make(cls, remote, i, j):
"""
Factory method for Strip EQ channel.
Returns a StripEQCh class.
"""
StripEQCh_cls = type(
'StripEQCh',
(cls,),
{
'cell': tuple(
StripEQChCell(remote, i, j, k) for k in range(remote.kind.cells)
)
},
)
return StripEQCh_cls(remote, i, j)
def __init__(self, remote, i, j):
super().__init__(remote, i)
self.channel_index = j
@property
def identifier(self) -> str:
return f'Strip[{self.index}].eq.channel[{self.channel_index}]'
class StripEQChCell(IRemote):
def __init__(self, remote, i, j, k):
super().__init__(remote, i)
self.channel_index = j
self.cell_index = k
@property
def identifier(self) -> str:
return f'Strip[{self.index}].eq.channel[{self.channel_index}].cell[{self.cell_index}]'
@property
def on(self) -> bool:
if self.channel_index > 0:
self.logger.warning(
'Only channel 0 is supported over VBAN for Strip EQ cells'
)
if self.public_packets[NBS.one] is None:
return False
return (
self.public_packets[NBS.one]
.strips[self.index]
.parametric_eq[self.cell_index]
.on
)
@on.setter
def on(self, val: bool):
self.setter('on', 1 if val else 0)
@property
def type(self) -> int:
if self.channel_index > 0:
self.logger.warning(
'Only channel 0 is supported over VBAN for Strip EQ cells'
)
if self.public_packets[NBS.one] is None:
return 0
return (
self.public_packets[NBS.one]
.strips[self.index]
.parametric_eq[self.cell_index]
.type
)
@type.setter
def type(self, val: int):
self.setter('type', val)
@property
def f(self) -> float:
if self.channel_index > 0:
self.logger.warning(
'Only channel 0 is supported over VBAN for Strip EQ cells'
)
if self.public_packets[NBS.one] is None:
return 0.0
return (
self.public_packets[NBS.one]
.strips[self.index]
.parametric_eq[self.cell_index]
.freq
)
@f.setter
def f(self, val: float):
self.setter('f', val)
@property
def gain(self) -> float:
if self.channel_index > 0:
self.logger.warning(
'Only channel 0 is supported over VBAN for Strip EQ cells'
)
if self.public_packets[NBS.one] is None:
return 0.0
return (
self.public_packets[NBS.one]
.strips[self.index]
.parametric_eq[self.cell_index]
.gain
)
@gain.setter
def gain(self, val: float):
self.setter('gain', val)
@property
def q(self) -> float:
if self.channel_index > 0:
self.logger.warning(
'Only channel 0 is supported over VBAN for Strip EQ cells'
)
if self.public_packets[NBS.one] is None:
return 0.0
return (
self.public_packets[NBS.one]
.strips[self.index]
.parametric_eq[self.cell_index]
.q
)
@q.setter
def q(self, val: float):
self.setter('q', val)
class VirtualStrip(Strip): class VirtualStrip(Strip):
@classmethod @classmethod
def make(cls, remote, i, is_phys): def make(cls, remote, i, is_phys):
@@ -296,7 +485,7 @@ class VirtualStrip(Strip):
def bass(self) -> float: def bass(self) -> float:
if self.public_packets[NBS.one] is None: if self.public_packets[NBS.one] is None:
return 0.0 return 0.0
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.bass] return self.public_packets[NBS.one].strips[self.index].eqgains.bass
@bass.setter @bass.setter
def bass(self, val: float): def bass(self, val: float):
@@ -306,7 +495,7 @@ class VirtualStrip(Strip):
def mid(self) -> float: def mid(self) -> float:
if self.public_packets[NBS.one] is None: if self.public_packets[NBS.one] is None:
return 0.0 return 0.0
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.mid] return self.public_packets[NBS.one].strips[self.index].eqgains.mid
@mid.setter @mid.setter
def mid(self, val: float): def mid(self, val: float):
@@ -318,7 +507,7 @@ class VirtualStrip(Strip):
def treble(self) -> float: def treble(self) -> float:
if self.public_packets[NBS.one] is None: if self.public_packets[NBS.one] is None:
return 0.0 return 0.0
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.treble] return self.public_packets[NBS.one].strips[self.index].eqgains.treble
@treble.setter @treble.setter
def treble(self, val: float): def treble(self, val: float):

View File

@@ -1,4 +1,4 @@
from abc import abstractmethod import abc
from . import kinds from . import kinds
from .iremote import IRemote from .iremote import IRemote
@@ -11,7 +11,7 @@ class VbanStream(IRemote):
Defines concrete implementation for vban stream Defines concrete implementation for vban stream
""" """
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
pass pass

View File

@@ -1,8 +1,8 @@
import abc
import logging import logging
import socket import socket
import threading import threading
import time import time
from abc import ABCMeta, abstractmethod
from pathlib import Path from pathlib import Path
from queue import Queue from queue import Queue
from typing import Iterable, Union from typing import Iterable, Union
@@ -18,8 +18,8 @@ from .worker import Producer, Subscriber, Updater
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class VbanCmd(metaclass=ABCMeta): class VbanCmd(abc.ABC):
"""Base class responsible for communicating with the VBAN RT Packet Service""" """Abstract Base Class for Voicemeeter VBAN Command Interfaces"""
DELAY = 0.001 DELAY = 0.001
# fmt: off # fmt: off
@@ -49,7 +49,7 @@ class VbanCmd(metaclass=ABCMeta):
self.stop_event = None self.stop_event = None
self.producer = None self.producer = None
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
"""Ensure subclasses override str magic method""" """Ensure subclasses override str magic method"""
pass pass
@@ -58,7 +58,7 @@ class VbanCmd(metaclass=ABCMeta):
try: try:
import tomllib import tomllib
except ModuleNotFoundError: except ModuleNotFoundError:
import tomli as tomllib import tomli as tomllib # type: ignore[import]
def get_filepath(): def get_filepath():
for pn in ( for pn in (

View File

@@ -39,9 +39,6 @@ class Subscriber(threading.Thread):
sub_packet, (self._remote.ip, self._remote.port) sub_packet, (self._remote.ip, self._remote.port)
) )
self._framecounter = bump_framecounter(self._framecounter) 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) self.wait_until_stopped(10)
except socket.gaierror as e: except socket.gaierror as e:
@@ -102,27 +99,11 @@ class Producer(threading.Thread):
match response_header.format_nbs: match response_header.format_nbs:
case NBS.zero: case NBS.zero:
"""
self.logger.debug(
'Received NB0 RTP Packet from %s, Size: %d bytes',
addr,
len(data),
)
"""
return VbanRtPacketNBS0.from_bytes( return VbanRtPacketNBS0.from_bytes(
nbs=NBS.zero, kind=self._remote.kind, data=data nbs=NBS.zero, kind=self._remote.kind, data=data
) )
case NBS.one: case NBS.one:
"""
self.logger.debug(
'Received NB1 RTP Packet from %s, Size: %d bytes',
addr,
len(data),
)
"""
return VbanRtPacketNBS1.from_bytes( return VbanRtPacketNBS1.from_bytes(
nbs=NBS.one, kind=self._remote.kind, data=data nbs=NBS.one, kind=self._remote.kind, data=data
) )