onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2024-11-15 17:10:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

inputlevels, ouputlevels now return generators

Browse Source 
'strip_level', 'bus_level' added to cache.

defautl ratelimit set to 0.01
ratelimit adjusted to wait by ratelimit - elapsed time.

is_updated now returns a bool

minor ver bump
This commit is contained in:
onyx-and-iris 2022-07-06 13:40:46 +01:00
parent a47a29d9f7
commit 86b7fb5d32
6 changed files with 69 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[tool.poetry]
name = "vban-cmd"
version = "1.0.8"
version = "1.1.0"
description = "Python interface for the VBAN RT Packet Service (Sendtext)"
authors = ["onyx-and-iris <code@onyxandiris.online>"]
license = "MIT"

73
vban_cmd/base.py
View File

@ -1,9 +1,10 @@
import dataclasses
import socket
import time
from abc import ABCMeta, abstractmethod
from enum import IntEnum
from threading import Thread
from typing import NoReturn, Optional, Union
from typing import Iterable, NoReturn, Optional, Union
from .packet import (
HEADER_SIZE,
@ -165,12 +166,22 @@ class VbanCmd(metaclass=ABCMeta):
@property
def pdirty(self):
"""True iff a parameter has changed"""
return self._pdirty
return self._pp.pdirty(self.public_packet)
@property
def ldirty(self):
"""True iff a level value has changed."""
return self._ldirty
self._strip_comp, self._bus_comp = (
tuple(
not a == b
for a, b in zip(self._public_packet.inputlevels, self._strip_buf)
),
tuple(
not a == b
for a, b in zip(self._public_packet.outputlevels, self._bus_buf)
),
)
return any(any(l) for l in (self._strip_comp, self._bus_comp))
@property
def public_packet(self):
@ -181,49 +192,35 @@ class VbanCmd(metaclass=ABCMeta):
pass
def _updates(self) -> NoReturn:
while self.running:
private_packet = self._get_rt()
self._strip_comp, self._bus_comp = (
tuple(
not a == b
for a, b in zip(
private_packet.inputlevels, self.public_packet.inputlevels
)
),
tuple(
not a == b
for a, b in zip(
private_packet.outputlevels, self.public_packet.outputlevels
)
),
)
self._pdirty = private_packet.pdirty(self.public_packet)
self._ldirty = any(
any(list_) for list_ in (self._strip_comp, self._bus_comp)
self.cache["strip_level"], self.cache["bus_level"] = self._get_levels(
self.public_packet
)
if self._public_packet != private_packet:
self._public_packet = private_packet
while self.running:
start = time.time()
self._pp = self._get_rt()
self._strip_buf, self._bus_buf = self._get_levels(self._pp)
if self.pdirty:
self._public_packet = self._pp
self.subject.notify("pdirty")
if self.ldirty:
self.cache["strip_level"] = tuple(self._strip_buf)
self.cache["bus_level"] = tuple(self._bus_buf)
self.subject.notify("ldirty")
time.sleep(self.ratelimit)
elapsed = time.time() - start
if self.ratelimit - elapsed > 0:
time.sleep(self.ratelimit - elapsed)
@property
def strip_levels(self):
"""Returns the full strip level array for a kind, PREFADER mode, before math conversion"""
return tuple(
list(filter(lambda x: x != ((1 << 16) - 1), self.public_packet.inputlevels))
)
def _get_levels(self, packet) -> Iterable:
"""
returns both level arrays (strip_levels, bus_levels) BEFORE math conversion
@property
def bus_levels(self):
"""Returns the full bus level array for a kind, before math conversion"""
return tuple(
list(
filter(lambda x: x != ((1 << 16) - 1), self.public_packet.outputlevels)
)
strip levels in PREFADER mode.
"""
return (
[val for val in packet.inputlevels],
[val for val in packet.outputlevels],
)
def apply(self, data: dict):

8
vban_cmd/bus.py
View File

@ -66,13 +66,15 @@ class BusLevel(IRemote):
(i, i + 8)
for i in range(0, (remote.kind.phys_out + remote.kind.virt_out) * 8, 8)
)
self.range = self.level_map[self.index]
def getter(self):
"""Returns a tuple of level values for the channel."""
range_ = self.level_map[self.index]
return tuple(
round(-i * 0.01, 1) for i in self._remote.bus_levels[range_[0] : range_[-1]]
round(-i * 0.01, 1)
for i in self._remote.cache["bus_level"][self.range[0] : self.range[-1]]
)
@property
@ -84,8 +86,8 @@ class BusLevel(IRemote):
return self.getter()
@property
def updated(self) -> tuple:
return self._remote._bus_comp
def is_updated(self) -> bool:
return any(self._remote._bus_comp[self.range[0] : self.range[-1]])
def _make_bus_mode_mixin():

2
vban_cmd/factory.py
View File

@ -64,7 +64,7 @@ class FactoryBase(VbanCmd):
"streamname": "Command1",
"bps": 0,
"channel": 0,
"ratelimit": 0,
"ratelimit": 0.01,
"sync": False,
}
kwargs = defaultkwargs | kwargs

22
vban_cmd/packet.py
View File

@ -1,4 +1,5 @@
from dataclasses import dataclass
from typing import Generator
VBAN_SERVICE_RTPACKETREGISTER = 32
VBAN_SERVICE_RTPACKET = 33
@ -76,21 +77,24 @@ class VBAN_VMRT_Packet_Data:
return int.from_bytes(self._samplerate, "little")
@property
def inputlevels(self) -> tuple:
def inputlevels(self) -> Generator[float, None, None]:
"""returns the entire level array across all inputs"""
return tuple(
((1 << 16) - 1) - int.from_bytes(self._inputLeveldB100[i : i + 2], "little")
for i in range(0, 68, 2)
for i in range(0, 68, 2):
val = ((1 << 16) - 1) - int.from_bytes(
self._inputLeveldB100[i : i + 2], "little"
)
if val != ((1 << 16) - 1):
yield val
@property
def outputlevels(self) -> tuple:
def outputlevels(self) -> Generator[float, None, None]:
"""returns the entire level array across all outputs"""
return tuple(
((1 << 16) - 1)
- int.from_bytes(self._outputLeveldB100[i : i + 2], "little")
for i in range(0, 128, 2)
for i in range(0, 128, 2):
val = ((1 << 16) - 1) - int.from_bytes(
self._outputLeveldB100[i : i + 2], "little"
)
if val != ((1 << 16) - 1):
yield val
@property
def stripstate(self) -> tuple:

12
vban_cmd/strip.py
View File

@ -100,12 +100,12 @@ class StripLevel(IRemote):
)
)
self.level_map = phys_map + virt_map
self.range = self.level_map[self.index]
def getter_prefader(self):
range_ = self.level_map[self.index]
def getter(self):
return tuple(
round(-i * 0.01, 1)
for i in self._remote.strip_levels[range_[0] : range_[-1]]
for i in self._remote.cache["strip_level"][self.range[0] : self.range[-1]]
)
@property
@ -114,7 +114,7 @@ class StripLevel(IRemote):
@property
def prefader(self) -> tuple:
return self.getter_prefader()
return self.getter()
@property
def postfader(self) -> tuple:
@ -125,8 +125,8 @@ class StripLevel(IRemote):
return
@property
def updated(self) -> tuple:
return self._remote._strip_comp
def is_updated(self) -> bool:
return any(self._remote._strip_comp[self.range[0] : self.range[-1]])
class GainLayer(IRemote):