Compare commits

..

No commits in common. "98ec9b715f44655ead52c53d7f40db9ab8711049" and "9f43ee18d367b4b9797a07550a68d466dab3c7fe" have entirely different histories.

9 changed files with 148 additions and 121 deletions

View File

@ -103,7 +103,7 @@ class App(tk.Tk):
def main():
KIND_ID = 'banana'
conn = {
'host': os.environ.get('VBANCMD_HOST', 'localhost'),
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
}

View File

@ -94,7 +94,7 @@ class Observer:
def main():
KIND_ID = 'potato'
conn = {
'host': os.environ.get('VBANCMD_HOST', 'localhost'),
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
}

View File

@ -25,7 +25,7 @@ class App:
def main():
KIND_ID = 'banana'
conn = {
'host': os.environ.get('VBANCMD_HOST', 'localhost'),
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
}

View File

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

View File

@ -1,4 +1,3 @@
import struct
from dataclasses import dataclass
from vban_cmd.enums import NBS
@ -35,16 +34,15 @@ class VbanPingHeader:
"""Creates the PING header bytes only."""
header = cls(framecounter=framecounter)
return struct.pack(
'<4s4B16sI',
header.vban,
header.format_sr,
header.format_nbs,
header.format_nbc,
header.format_bit,
header.streamname,
header.framecounter,
)
data = bytearray()
data.extend(header.vban)
data.extend(header.format_sr.to_bytes(1, 'little'))
data.extend(header.format_nbs.to_bytes(1, 'little'))
data.extend(header.format_nbc.to_bytes(1, 'little'))
data.extend(header.format_bit.to_bytes(1, 'little'))
data.extend(header.streamname)
data.extend(header.framecounter.to_bytes(4, 'little'))
return bytes(data)
@dataclass
@ -165,16 +163,15 @@ class VbanRTSubscribeHeader:
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(nbs=nbs)
return struct.pack(
'<4s4B16sI',
header.vban,
header.format_sr[0],
header.format_nbs[0],
header.format_nbc[0],
header.format_bit[0],
header.streamname,
framecounter,
)
data = bytearray()
data.extend(header.vban)
data.extend(header.format_sr)
data.extend(header.format_nbs)
data.extend(header.format_nbc)
data.extend(header.format_bit)
data.extend(header.streamname)
data.extend(framecounter.to_bytes(4, 'little'))
return bytes(data)
@dataclass
@ -218,16 +215,15 @@ class VbanRTRequestHeader:
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
)
return struct.pack(
'<4s4B16sI',
header.vban,
header.sr[0],
header.nbs[0],
header.nbc[0],
header.bit[0],
header.streamname,
header.framecounter,
)
data = bytearray()
data.extend(header.vban)
data.extend(header.sr)
data.extend(header.nbs)
data.extend(header.nbc)
data.extend(header.bit)
data.extend(header.streamname)
data.extend(header.framecounter.to_bytes(4, 'little'))
return bytes(data)
@classmethod
def encode_with_payload(

View File

@ -1,6 +1,4 @@
import struct
from dataclasses import dataclass
from functools import cached_property
from typing import NamedTuple
from vban_cmd.enums import NBS
@ -23,13 +21,6 @@ class ChannelState:
# Convert 4-byte state to integer once for efficient lookups
self._state = int.from_bytes(state_bytes, 'little')
@classmethod
def from_int(cls, state_int: int):
"""Create ChannelState directly from integer for efficiency"""
instance = cls.__new__(cls)
instance._state = state_int
return instance
def get_mode(self, mode_value: int) -> bool:
"""Get boolean state for a specific mode"""
return (self._state & mode_value) != 0
@ -156,17 +147,32 @@ class VbanRTPacketNBS0(VbanRTPacket):
def pdirty(self, other) -> bool:
"""True iff any defined parameter has changed"""
self_gains = (
self._stripGaindB100Layer1
+ self._stripGaindB100Layer2
+ self._stripGaindB100Layer3
+ self._stripGaindB100Layer4
+ self._stripGaindB100Layer5
+ self._stripGaindB100Layer6
+ self._stripGaindB100Layer7
+ self._stripGaindB100Layer8
)
other_gains = (
other._stripGaindB100Layer1
+ other._stripGaindB100Layer2
+ other._stripGaindB100Layer3
+ other._stripGaindB100Layer4
+ other._stripGaindB100Layer5
+ other._stripGaindB100Layer6
+ other._stripGaindB100Layer7
+ other._stripGaindB100Layer8
)
return (
self._stripState != other._stripState
or self._busState != other._busState
or self._stripGaindB100Layer1 != other._stripGaindB100Layer1
or self._stripGaindB100Layer2 != other._stripGaindB100Layer2
or self._stripGaindB100Layer3 != other._stripGaindB100Layer3
or self._stripGaindB100Layer4 != other._stripGaindB100Layer4
or self._stripGaindB100Layer5 != other._stripGaindB100Layer5
or self._stripGaindB100Layer6 != other._stripGaindB100Layer6
or self._stripGaindB100Layer7 != other._stripGaindB100Layer7
or self._stripGaindB100Layer8 != other._stripGaindB100Layer8
or self_gains != other_gains
or self._busGaindB100 != other._busGaindB100
or self._stripLabelUTF8c60 != other._stripLabelUTF8c60
or self._busLabelUTF8c60 != other._busLabelUTF8c60
@ -180,54 +186,77 @@ class VbanRTPacketNBS0(VbanRTPacket):
)
return any(self._strip_comp) or any(self._bus_comp)
@cached_property
@property
def strip_levels(self) -> tuple[float, ...]:
"""Returns strip levels in dB"""
strip_raw = struct.unpack('<34h', self._inputLeveldB100)
return tuple(round(val * 0.01, 1) for val in strip_raw)[
: self._kind.num_strip_levels
]
return tuple(
round(
int.from_bytes(self._inputLeveldB100[i : i + 2], 'little', signed=True)
* 0.01,
1,
)
for i in range(0, len(self._inputLeveldB100), 2)
)[: self._kind.num_strip_levels]
@cached_property
@property
def bus_levels(self) -> tuple[float, ...]:
"""Returns bus levels in dB"""
bus_raw = struct.unpack('<64h', self._outputLeveldB100)
return tuple(round(val * 0.01, 1) for val in bus_raw)[
: self._kind.num_bus_levels
]
return tuple(
round(
int.from_bytes(self._outputLeveldB100[i : i + 2], 'little', signed=True)
* 0.01,
1,
)
for i in range(0, len(self._outputLeveldB100), 2)
)[: self._kind.num_bus_levels]
@property
def levels(self) -> Levels:
"""Returns strip and bus levels as a namedtuple"""
return Levels(strip=self.strip_levels, bus=self.bus_levels)
@cached_property
@property
def states(self) -> States:
"""returns States object with processed strip and bus channel states"""
strip_states = struct.unpack('<8I', self._stripState)
bus_states = struct.unpack('<8I', self._busState)
return States(
strip=tuple(ChannelState.from_int(state) for state in strip_states),
bus=tuple(ChannelState.from_int(state) for state in bus_states),
strip=tuple(
ChannelState(self._stripState[i : i + 4]) for i in range(0, 32, 4)
),
bus=tuple(ChannelState(self._busState[i : i + 4]) for i in range(0, 32, 4)),
)
@cached_property
@property
def gainlayers(self) -> tuple:
"""returns tuple of all strip gain layers as tuples"""
layer_data = []
for layer in range(1, 9):
layer_bytes = getattr(self, f'_stripGaindB100Layer{layer}')
layer_raw = struct.unpack('<8h', layer_bytes)
layer_data.append(tuple(round(val * 0.01, 2) for val in layer_raw))
return tuple(layer_data)
return tuple(
tuple(
round(
int.from_bytes(
getattr(self, f'_stripGaindB100Layer{layer}')[i : i + 2],
'little',
signed=True,
)
* 0.01,
2,
)
for i in range(0, 16, 2)
)
for layer in range(1, 9)
)
@cached_property
@property
def busgain(self) -> tuple:
"""returns tuple of bus gains"""
bus_gain_raw = struct.unpack('<8h', self._busGaindB100)
return tuple(round(val * 0.01, 2) for val in bus_gain_raw)
return tuple(
round(
int.from_bytes(self._busGaindB100[i : i + 2], 'little', signed=True)
* 0.01,
2,
)
for i in range(0, 16, 2)
)
@cached_property
@property
def labels(self) -> Labels:
"""returns Labels namedtuple of strip and bus labels"""

View File

@ -1,6 +1,5 @@
import struct
from dataclasses import dataclass
from functools import cached_property
from typing import NamedTuple
from vban_cmd.enums import NBS
@ -194,15 +193,11 @@ class VbanVMParamStrip:
_Pitch_formant_high=data[172:174],
)
@cached_property
@property
def mode(self) -> int:
return int.from_bytes(self._mode, 'little')
@cached_property
def karaoke(self) -> int:
return int.from_bytes(self._nKaraoke, 'little')
@cached_property
@property
def audibility(self) -> Audibility:
return Audibility(
round(int.from_bytes(self._audibility, 'little', signed=True) * 0.01, 2),
@ -211,7 +206,7 @@ class VbanVMParamStrip:
round(int.from_bytes(self._audibility_d, 'little', signed=True) * 0.01, 2),
)
@cached_property
@property
def positions(self) -> Positions:
return Positions(
round(int.from_bytes(self._pos3D_x, 'little', signed=True) * 0.01, 2),
@ -222,7 +217,7 @@ class VbanVMParamStrip:
round(int.from_bytes(self._posMod_y, 'little', signed=True) * 0.01, 2),
)
@cached_property
@property
def eqgains(self) -> EqGains:
return EqGains(
*[
@ -235,7 +230,7 @@ class VbanVMParamStrip:
]
)
@cached_property
@property
def parametric_eq(self) -> tuple[ParametricEQSettings, ...]:
return tuple(
ParametricEQSettings(
@ -248,7 +243,7 @@ class VbanVMParamStrip:
for i in range(6)
)
@cached_property
@property
def sends(self) -> Sends:
return Sends(
round(int.from_bytes(self._send_reverb, 'little', signed=True) * 0.01, 2),
@ -257,7 +252,11 @@ class VbanVMParamStrip:
round(int.from_bytes(self._send_fx2, 'little', signed=True) * 0.01, 2),
)
@cached_property
@property
def karaoke(self) -> int:
return int.from_bytes(self._nKaraoke, 'little')
@property
def compressor(self) -> CompressorSettings:
return CompressorSettings(
gain_in=round(
@ -277,7 +276,7 @@ class VbanVMParamStrip:
),
)
@cached_property
@property
def gate(self) -> GateSettings:
return GateSettings(
threshold_in=round(
@ -296,7 +295,7 @@ class VbanVMParamStrip:
release_ms=round(int.from_bytes(self._GATE_release_ms, 'little') * 0.1, 2),
)
@cached_property
@property
def denoiser(self) -> DenoiserSettings:
return DenoiserSettings(
threshold=round(
@ -304,7 +303,7 @@ class VbanVMParamStrip:
)
)
@cached_property
@property
def pitch(self) -> PitchSettings:
return PitchSettings(
enabled=bool(int.from_bytes(self._PitchEnabled, 'little')),

View File

@ -1,4 +1,3 @@
import struct
from dataclasses import dataclass
from enum import Enum
@ -66,31 +65,30 @@ class VbanPing0Payload:
"""Convert payload to bytes"""
payload = cls()
return struct.pack(
'<7I4s8s8s8s8s64s32s2H64s64s64s64s128s128s',
payload.bit_type,
payload.bit_feature,
payload.bit_feature_ex,
payload.preferred_rate,
payload.min_rate,
payload.max_rate,
payload.color_rgb,
payload.version,
payload.gps_position,
payload.user_position,
payload.lang_code,
payload.reserved,
payload.reserved_ex,
payload.distant_ip,
payload.distant_port,
payload.distant_reserved,
payload.device_name,
payload.manufacturer_name,
payload.application_name,
payload.host_name,
payload.user_name,
payload.user_comment,
)
data = bytearray()
data.extend(payload.bit_type.to_bytes(4, 'little'))
data.extend(payload.bit_feature.to_bytes(4, 'little'))
data.extend(payload.bit_feature_ex.to_bytes(4, 'little'))
data.extend(payload.preferred_rate.to_bytes(4, 'little'))
data.extend(payload.min_rate.to_bytes(4, 'little'))
data.extend(payload.max_rate.to_bytes(4, 'little'))
data.extend(payload.color_rgb.to_bytes(4, 'little'))
data.extend(payload.version)
data.extend(payload.gps_position)
data.extend(payload.user_position)
data.extend(payload.lang_code)
data.extend(payload.reserved)
data.extend(payload.reserved_ex)
data.extend(payload.distant_ip)
data.extend(payload.distant_port.to_bytes(2, 'little'))
data.extend(payload.distant_reserved.to_bytes(2, 'little'))
data.extend(payload.device_name)
data.extend(payload.manufacturer_name)
data.extend(payload.application_name)
data.extend(payload.host_name)
data.extend(payload.user_name)
data.extend(payload.user_comment)
return bytes(data)
@classmethod
def create_packet(cls, framecounter: int) -> bytes:

View File

@ -124,11 +124,16 @@ def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
"""
Generator function, accepts two tuples of dB values.
Returns True when levels are equal (no change), False when different.
Evaluates equality of each member in both tuples.
Only ignores changes when levels are very quiet (below -72 dB).
"""
for a, b in zip(t0, t1):
yield a == b
# If both values are very quiet (below -72dB), ignore small changes
if a <= -72.0 and b <= -72.0:
yield a == b # Both quiet, check if they're equal
else:
yield a != b # At least one has significant level, detect changes
def deep_merge(dict1, dict2):