5 Commits

8 changed files with 119 additions and 141 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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