mirror of
https://github.com/onyx-and-iris/vban-cmd-python.git
synced 2026-03-12 12:59:13 +00:00
Compare commits
No commits in common. "98ec9b715f44655ead52c53d7f40db9ab8711049" and "9f43ee18d367b4b9797a07550a68d466dab3c7fe" have entirely different histories.
98ec9b715f
...
9f43ee18d3
@ -103,7 +103,7 @@ class App(tk.Tk):
|
|||||||
def main():
|
def main():
|
||||||
KIND_ID = 'banana'
|
KIND_ID = 'banana'
|
||||||
conn = {
|
conn = {
|
||||||
'host': os.environ.get('VBANCMD_HOST', 'localhost'),
|
'ip': os.environ.get('VBANCMD_IP', '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'),
|
||||||
}
|
}
|
||||||
|
|||||||
@ -94,7 +94,7 @@ class Observer:
|
|||||||
def main():
|
def main():
|
||||||
KIND_ID = 'potato'
|
KIND_ID = 'potato'
|
||||||
conn = {
|
conn = {
|
||||||
'host': os.environ.get('VBANCMD_HOST', 'localhost'),
|
'ip': os.environ.get('VBANCMD_IP', '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'),
|
||||||
}
|
}
|
||||||
|
|||||||
@ -25,7 +25,7 @@ class App:
|
|||||||
def main():
|
def main():
|
||||||
KIND_ID = 'banana'
|
KIND_ID = 'banana'
|
||||||
conn = {
|
conn = {
|
||||||
'host': os.environ.get('VBANCMD_HOST', 'localhost'),
|
'ip': os.environ.get('VBANCMD_IP', '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'),
|
||||||
}
|
}
|
||||||
|
|||||||
@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "vban-cmd"
|
name = "vban-cmd"
|
||||||
version = "2.10.1"
|
version = "2.9.7"
|
||||||
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" }
|
||||||
|
|||||||
@ -1,4 +1,3 @@
|
|||||||
import struct
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
from vban_cmd.enums import NBS
|
from vban_cmd.enums import NBS
|
||||||
@ -35,16 +34,15 @@ class VbanPingHeader:
|
|||||||
"""Creates the PING header bytes only."""
|
"""Creates the PING header bytes only."""
|
||||||
header = cls(framecounter=framecounter)
|
header = cls(framecounter=framecounter)
|
||||||
|
|
||||||
return struct.pack(
|
data = bytearray()
|
||||||
'<4s4B16sI',
|
data.extend(header.vban)
|
||||||
header.vban,
|
data.extend(header.format_sr.to_bytes(1, 'little'))
|
||||||
header.format_sr,
|
data.extend(header.format_nbs.to_bytes(1, 'little'))
|
||||||
header.format_nbs,
|
data.extend(header.format_nbc.to_bytes(1, 'little'))
|
||||||
header.format_nbc,
|
data.extend(header.format_bit.to_bytes(1, 'little'))
|
||||||
header.format_bit,
|
data.extend(header.streamname)
|
||||||
header.streamname,
|
data.extend(header.framecounter.to_bytes(4, 'little'))
|
||||||
header.framecounter,
|
return bytes(data)
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@ -165,16 +163,15 @@ 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)
|
||||||
|
|
||||||
return struct.pack(
|
data = bytearray()
|
||||||
'<4s4B16sI',
|
data.extend(header.vban)
|
||||||
header.vban,
|
data.extend(header.format_sr)
|
||||||
header.format_sr[0],
|
data.extend(header.format_nbs)
|
||||||
header.format_nbs[0],
|
data.extend(header.format_nbc)
|
||||||
header.format_nbc[0],
|
data.extend(header.format_bit)
|
||||||
header.format_bit[0],
|
data.extend(header.streamname)
|
||||||
header.streamname,
|
data.extend(framecounter.to_bytes(4, 'little'))
|
||||||
framecounter,
|
return bytes(data)
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@ -218,16 +215,15 @@ class VbanRTRequestHeader:
|
|||||||
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
|
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
|
||||||
)
|
)
|
||||||
|
|
||||||
return struct.pack(
|
data = bytearray()
|
||||||
'<4s4B16sI',
|
data.extend(header.vban)
|
||||||
header.vban,
|
data.extend(header.sr)
|
||||||
header.sr[0],
|
data.extend(header.nbs)
|
||||||
header.nbs[0],
|
data.extend(header.nbc)
|
||||||
header.nbc[0],
|
data.extend(header.bit)
|
||||||
header.bit[0],
|
data.extend(header.streamname)
|
||||||
header.streamname,
|
data.extend(header.framecounter.to_bytes(4, 'little'))
|
||||||
header.framecounter,
|
return bytes(data)
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def encode_with_payload(
|
def encode_with_payload(
|
||||||
|
|||||||
@ -1,6 +1,4 @@
|
|||||||
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
|
||||||
@ -23,13 +21,6 @@ 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
|
||||||
@ -156,17 +147,32 @@ 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._stripGaindB100Layer1 != other._stripGaindB100Layer1
|
or self_gains != other_gains
|
||||||
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
|
||||||
@ -180,54 +186,77 @@ class VbanRTPacketNBS0(VbanRTPacket):
|
|||||||
)
|
)
|
||||||
return any(self._strip_comp) or any(self._bus_comp)
|
return any(self._strip_comp) or any(self._bus_comp)
|
||||||
|
|
||||||
@cached_property
|
@property
|
||||||
def strip_levels(self) -> tuple[float, ...]:
|
def strip_levels(self) -> tuple[float, ...]:
|
||||||
"""Returns strip levels in dB"""
|
"""Returns strip levels in dB"""
|
||||||
strip_raw = struct.unpack('<34h', self._inputLeveldB100)
|
return tuple(
|
||||||
return tuple(round(val * 0.01, 1) for val in strip_raw)[
|
round(
|
||||||
: self._kind.num_strip_levels
|
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, ...]:
|
def bus_levels(self) -> tuple[float, ...]:
|
||||||
"""Returns bus levels in dB"""
|
"""Returns bus levels in dB"""
|
||||||
bus_raw = struct.unpack('<64h', self._outputLeveldB100)
|
return tuple(
|
||||||
return tuple(round(val * 0.01, 1) for val in bus_raw)[
|
round(
|
||||||
: self._kind.num_bus_levels
|
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
|
@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)
|
||||||
|
|
||||||
@cached_property
|
@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(ChannelState.from_int(state) for state in strip_states),
|
strip=tuple(
|
||||||
bus=tuple(ChannelState.from_int(state) for state in bus_states),
|
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:
|
def gainlayers(self) -> tuple:
|
||||||
"""returns tuple of all strip gain layers as tuples"""
|
"""returns tuple of all strip gain layers as tuples"""
|
||||||
layer_data = []
|
return tuple(
|
||||||
for layer in range(1, 9):
|
tuple(
|
||||||
layer_bytes = getattr(self, f'_stripGaindB100Layer{layer}')
|
round(
|
||||||
layer_raw = struct.unpack('<8h', layer_bytes)
|
int.from_bytes(
|
||||||
layer_data.append(tuple(round(val * 0.01, 2) for val in layer_raw))
|
getattr(self, f'_stripGaindB100Layer{layer}')[i : i + 2],
|
||||||
return tuple(layer_data)
|
'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:
|
def busgain(self) -> tuple:
|
||||||
"""returns tuple of bus gains"""
|
"""returns tuple of bus gains"""
|
||||||
bus_gain_raw = struct.unpack('<8h', self._busGaindB100)
|
return tuple(
|
||||||
return tuple(round(val * 0.01, 2) for val in bus_gain_raw)
|
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:
|
def labels(self) -> Labels:
|
||||||
"""returns Labels namedtuple of strip and bus labels"""
|
"""returns Labels namedtuple of strip and bus labels"""
|
||||||
|
|
||||||
|
|||||||
@ -1,6 +1,5 @@
|
|||||||
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
|
||||||
@ -194,15 +193,11 @@ class VbanVMParamStrip:
|
|||||||
_Pitch_formant_high=data[172:174],
|
_Pitch_formant_high=data[172:174],
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@property
|
||||||
def mode(self) -> int:
|
def mode(self) -> int:
|
||||||
return int.from_bytes(self._mode, 'little')
|
return int.from_bytes(self._mode, 'little')
|
||||||
|
|
||||||
@cached_property
|
@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),
|
||||||
@ -211,7 +206,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),
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@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),
|
||||||
@ -222,7 +217,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),
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@property
|
||||||
def eqgains(self) -> EqGains:
|
def eqgains(self) -> EqGains:
|
||||||
return EqGains(
|
return EqGains(
|
||||||
*[
|
*[
|
||||||
@ -235,7 +230,7 @@ class VbanVMParamStrip:
|
|||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@property
|
||||||
def parametric_eq(self) -> tuple[ParametricEQSettings, ...]:
|
def parametric_eq(self) -> tuple[ParametricEQSettings, ...]:
|
||||||
return tuple(
|
return tuple(
|
||||||
ParametricEQSettings(
|
ParametricEQSettings(
|
||||||
@ -248,7 +243,7 @@ class VbanVMParamStrip:
|
|||||||
for i in range(6)
|
for i in range(6)
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@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),
|
||||||
@ -257,7 +252,11 @@ 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),
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@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(
|
||||||
@ -277,7 +276,7 @@ class VbanVMParamStrip:
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@property
|
||||||
def gate(self) -> GateSettings:
|
def gate(self) -> GateSettings:
|
||||||
return GateSettings(
|
return GateSettings(
|
||||||
threshold_in=round(
|
threshold_in=round(
|
||||||
@ -296,7 +295,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),
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@property
|
||||||
def denoiser(self) -> DenoiserSettings:
|
def denoiser(self) -> DenoiserSettings:
|
||||||
return DenoiserSettings(
|
return DenoiserSettings(
|
||||||
threshold=round(
|
threshold=round(
|
||||||
@ -304,7 +303,7 @@ class VbanVMParamStrip:
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
@cached_property
|
@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')),
|
||||||
|
|||||||
@ -1,4 +1,3 @@
|
|||||||
import struct
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
|
|
||||||
@ -66,31 +65,30 @@ class VbanPing0Payload:
|
|||||||
"""Convert payload to bytes"""
|
"""Convert payload to bytes"""
|
||||||
payload = cls()
|
payload = cls()
|
||||||
|
|
||||||
return struct.pack(
|
data = bytearray()
|
||||||
'<7I4s8s8s8s8s64s32s2H64s64s64s64s128s128s',
|
data.extend(payload.bit_type.to_bytes(4, 'little'))
|
||||||
payload.bit_type,
|
data.extend(payload.bit_feature.to_bytes(4, 'little'))
|
||||||
payload.bit_feature,
|
data.extend(payload.bit_feature_ex.to_bytes(4, 'little'))
|
||||||
payload.bit_feature_ex,
|
data.extend(payload.preferred_rate.to_bytes(4, 'little'))
|
||||||
payload.preferred_rate,
|
data.extend(payload.min_rate.to_bytes(4, 'little'))
|
||||||
payload.min_rate,
|
data.extend(payload.max_rate.to_bytes(4, 'little'))
|
||||||
payload.max_rate,
|
data.extend(payload.color_rgb.to_bytes(4, 'little'))
|
||||||
payload.color_rgb,
|
data.extend(payload.version)
|
||||||
payload.version,
|
data.extend(payload.gps_position)
|
||||||
payload.gps_position,
|
data.extend(payload.user_position)
|
||||||
payload.user_position,
|
data.extend(payload.lang_code)
|
||||||
payload.lang_code,
|
data.extend(payload.reserved)
|
||||||
payload.reserved,
|
data.extend(payload.reserved_ex)
|
||||||
payload.reserved_ex,
|
data.extend(payload.distant_ip)
|
||||||
payload.distant_ip,
|
data.extend(payload.distant_port.to_bytes(2, 'little'))
|
||||||
payload.distant_port,
|
data.extend(payload.distant_reserved.to_bytes(2, 'little'))
|
||||||
payload.distant_reserved,
|
data.extend(payload.device_name)
|
||||||
payload.device_name,
|
data.extend(payload.manufacturer_name)
|
||||||
payload.manufacturer_name,
|
data.extend(payload.application_name)
|
||||||
payload.application_name,
|
data.extend(payload.host_name)
|
||||||
payload.host_name,
|
data.extend(payload.user_name)
|
||||||
payload.user_name,
|
data.extend(payload.user_comment)
|
||||||
payload.user_comment,
|
return bytes(data)
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def create_packet(cls, framecounter: int) -> bytes:
|
def create_packet(cls, framecounter: int) -> bytes:
|
||||||
|
|||||||
@ -124,11 +124,16 @@ def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
|
|||||||
"""
|
"""
|
||||||
Generator function, accepts two tuples of dB values.
|
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):
|
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):
|
def deep_merge(dict1, dict2):
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user