Compare commits

..

No commits in common. "7f3b0ac7c9cf915233389bc83d144a9c1c3ad14b" and "ed8e281f7ffc5899749af5cca40a0a84bd1d31af" have entirely different histories.

9 changed files with 134 additions and 514 deletions

View File

@ -1,5 +1,4 @@
import logging
import os
import tkinter as tk
from tkinter import ttk
@ -101,14 +100,7 @@ class App(tk.Tk):
def main():
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:
with vban_cmd.api('banana', ldirty=True) as vban:
app = App(vban)
app.mainloop()

View File

@ -1,4 +1,3 @@
import os
import threading
from logging import config
@ -93,13 +92,8 @@ class Observer:
def main():
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, **conn) as vban:
with vban_cmd.api(KIND_ID) as vban:
stop_event = threading.Event()
with Observer(vban, stop_event):

View File

@ -1,5 +1,4 @@
import logging
import os
import vban_cmd
@ -24,13 +23,8 @@ class App:
def main():
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, **conn) as vban:
with vban_cmd.api(KIND_ID, pdirty=True, ldirty=True) as vban:
App(vban)
while _ := input('Press <Enter> to exit\n'):

View File

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

View File

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

View File

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

View File

@ -1,6 +1,4 @@
import struct
from dataclasses import dataclass
from typing import NamedTuple
from .enums import NBS
from .kinds import KindMapClass
@ -218,77 +216,6 @@ 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
comprate: float
threshold: float
c_enabled: bool
makeup: bool
gain_out: float
class GateSettings(NamedTuple):
dBThreshold_in: float
dBDamping_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
class VbanVMParamStrip:
"""Represents the VBAN_VMPARAMSTRIP_PACKET structure"""
@ -406,133 +333,59 @@ class VbanVMParamStrip:
return int.from_bytes(self._mode, 'little')
@property
def audibility(self) -> Audibility:
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),
def position_pan(self) -> tuple[int, int]:
return (
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),
)
@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_y, 'little', signed=True) * 0.01, 2),
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_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_y, 'little', signed=True) * 0.01, 2),
)
@property
def eqgains(self) -> EqGains:
return EqGains(
*[
round(
int.from_bytes(getattr(self, f'_EQgain{i}'), 'little', signed=True)
* 0.01,
2,
)
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(
def send_reverb(self) -> tuple[float, float]:
return (
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(
int.from_bytes(getattr(self, f'_EQgain{i}'), 'little', signed=True)
* 0.01,
2,
)
for i in range(1, 4)
)
@property
def karaoke(self) -> int:
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),
comprate=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(
dBThreshold_in=round(
int.from_bytes(self._GATE_dBThreshold_in, 'little', signed=True) * 0.01,
2,
),
dBDamping_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
class VbanRtPacketNBS1(VbanRtPacket):
@ -569,38 +422,19 @@ class VbanRtPacketNBS1(VbanRtPacket):
class SubscribeHeader:
"""Represents the header of an RT subscription packet"""
nbs: NBS = NBS.zero
name: str = 'Register-RTP'
timeout: int = 15
@property
def vban(self) -> bytes:
return b'VBAN'
@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))
ident: NBS = NBS.zero
name = 'Register-RTP'
timeout = 15
vban: bytes = 'VBAN'.encode()
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little')
format_nbs: bytes = (ident.value & 0xFF).to_bytes(1, 'little')
format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, 'little')
format_bit: bytes = (timeout & 0xFF).to_bytes(1, 'little') # timeout
streamname: bytes = name.encode('ascii') + bytes(16 - len(name))
@classmethod
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(nbs=nbs)
header = cls(ident=nbs)
data = bytearray()
data.extend(header.vban)
data.extend(header.format_sr)
@ -617,31 +451,30 @@ class VbanRtPacketHeader:
"""Represents the header of an RT response packet"""
name: str = 'Voicemeeter-RTP'
format_sr: int = VBAN_PROTOCOL_SERVICE
format_nbs: int = 0
format_nbc: int = VBAN_SERVICE_RTPACKET
format_bit: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
vban: bytes = 'VBAN'.encode()
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little')
format_nbs: bytes = (0).to_bytes(1, 'little')
format_nbc: bytes = (VBAN_SERVICE_RTPACKET).to_bytes(1, 'little')
format_bit: bytes = (0).to_bytes(1, 'little')
streamname: bytes = name.encode('ascii') + bytes(16 - len(name))
@classmethod
def from_bytes(cls, data: bytes):
if len(data) < HEADER_SIZE:
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')
return cls(
name=name,
format_sr=data[4] & VBAN_SERVICE_MASK,
format_nbs=data[5],
format_nbc=data[6],
format_bit=data[7],
vban=vban,
format_sr=format_sr & VBAN_SERVICE_MASK,
format_nbs=format_nbs,
format_nbc=format_nbc,
format_bit=format_bit,
)
@ -652,30 +485,21 @@ class RequestHeader:
name: str
bps_index: int
channel: int
framecounter: int = 0
vban: bytes = 'VBAN'.encode()
nbs: bytes = (0).to_bytes(1, 'little')
bit: bytes = (0x10).to_bytes(1, 'little')
framecounter: bytes = (0).to_bytes(4, 'little')
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def sr(self) -> bytes:
def sr(self):
return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
@property
def nbs(self) -> bytes:
return (0).to_bytes(1, 'little')
@property
def nbc(self) -> bytes:
def nbc(self):
return (self.channel).to_bytes(1, 'little')
@property
def bit(self) -> bytes:
return (0x10).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
def streamname(self):
return self.name.encode() + bytes(16 - len(self.name))
@classmethod
@ -685,7 +509,6 @@ class RequestHeader:
header = cls(
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
)
data = bytearray()
data.extend(header.vban)
data.extend(header.sr)

View File

@ -3,7 +3,7 @@ from abc import abstractmethod
from typing import Union
from . import kinds
from .enums import NBS
from .enums import NBS, EQGains
from .iremote import IRemote
from .meta import (
channel_bool_prop,
@ -68,7 +68,7 @@ class PhysicalStrip(Strip):
'comp': StripComp(remote, index),
'gate': StripGate(remote, index),
'denoiser': StripDenoiser(remote, index),
'eq': StripEQ.make(remote, index),
'eq': StripEQ(remote, index),
},
)
@ -76,14 +76,12 @@ class PhysicalStrip(Strip):
return f'{type(self).__name__}{self.index}'
@property
def audibility(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.knob
def device(self):
return
@audibility.setter
def audibility(self, val: float):
self.setter('audibility', val)
@property
def sr(self):
return
class StripComp(IRemote):
@ -93,9 +91,7 @@ class StripComp(IRemote):
@property
def knob(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.comp
return
@knob.setter
def knob(self, val: float):
@ -103,9 +99,7 @@ class StripComp(IRemote):
@property
def gainin(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.gain_in
return
@gainin.setter
def gainin(self, val: float):
@ -113,9 +107,7 @@ class StripComp(IRemote):
@property
def ratio(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.comprate
return
@ratio.setter
def ratio(self, val: float):
@ -123,9 +115,7 @@ class StripComp(IRemote):
@property
def threshold(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.threshold
return
@threshold.setter
def threshold(self, val: float):
@ -133,9 +123,7 @@ class StripComp(IRemote):
@property
def attack(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.attack_ms
return
@attack.setter
def attack(self, val: float):
@ -143,9 +131,7 @@ class StripComp(IRemote):
@property
def release(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.release_ms
return
@release.setter
def release(self, val: float):
@ -153,9 +139,7 @@ class StripComp(IRemote):
@property
def knee(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.n_knee
return
@knee.setter
def knee(self, val: float):
@ -163,9 +147,7 @@ class StripComp(IRemote):
@property
def gainout(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].compressor.gain_out
return
@gainout.setter
def gainout(self, val: float):
@ -173,9 +155,7 @@ class StripComp(IRemote):
@property
def makeup(self) -> bool:
if self.public_packets[NBS.one] is None:
return False
return bool(self.public_packets[NBS.one].strips[self.index].compressor.makeup)
return
@makeup.setter
def makeup(self, val: bool):
@ -189,9 +169,7 @@ class StripGate(IRemote):
@property
def knob(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.gate
return
@knob.setter
def knob(self, val: float):
@ -253,9 +231,7 @@ class StripDenoiser(IRemote):
@property
def knob(self) -> float:
if self.public_packets[NBS.one] is None:
return 0.0
return self.public_packets[NBS.one].strips[self.index].audibility.denoiser
return
@knob.setter
def knob(self, val: float):
@ -263,25 +239,6 @@ class StripDenoiser(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
def identifier(self) -> str:
return f'strip[{self.index}].eq'
@ -303,140 +260,6 @@ class StripEQ(IRemote):
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):
@classmethod
def make(cls, remote, i, is_phys):
@ -473,7 +296,7 @@ class VirtualStrip(Strip):
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.bass
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.bass]
@bass.setter
def bass(self, val: float):
@ -483,7 +306,7 @@ class VirtualStrip(Strip):
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.mid
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.mid]
@mid.setter
def mid(self, val: float):
@ -495,7 +318,7 @@ class VirtualStrip(Strip):
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.treble
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.treble]
@treble.setter
def treble(self, val: float):

View File

@ -39,6 +39,9 @@ class Subscriber(threading.Thread):
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:
@ -99,11 +102,27 @@ class Producer(threading.Thread):
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.from_bytes(
nbs=NBS.zero, kind=self._remote.kind, data=data
)
case NBS.one:
"""
self.logger.debug(
'Received NB1 RTP Packet from %s, Size: %d bytes',
addr,
len(data),
)
"""
return VbanRtPacketNBS1.from_bytes(
nbs=NBS.one, kind=self._remote.kind, data=data
)