onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2025-10-31 05:31:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor meta functions

Browse Source 
add channel_bool_prop and channel_label_prop to strip/bus
self.identifier in strip bus now returning only class name

add psuedo decorators cache_bool and cache_string to util.

fix bug with cache
todo. add to cache on multi-set
This commit is contained in:
onyx-and-iris 2022-04-29 02:57:47 +01:00
parent 98c15967d1
commit 3aebd90920
6 changed files with 99 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
vbancmd/bus.py
View File

@ -1,8 +1,7 @@
from .errors import VMCMDErrors
from . import channel
from .channel import Channel
from . import kinds
from .meta import bus_mode_prop, bus_bool_prop
from .meta import bus_mode_prop, channel_bool_prop, channel_label_prop
class OutputBus(Channel):
@ -22,34 +21,20 @@ class OutputBus(Channel):
{
"levels": BusLevel(remote, index),
"mode": BusModeMixin(remote, index),
**{param: channel_bool_prop(param) for param in ["mute", "mono"]},
},
)
return OB_cls(remote, index, *args, **kwargs)
@property
def identifier(self):
return f"Bus[{self.index}]"
return "bus"
mute = bus_bool_prop("mute")
eq = channel_bool_prop("eq.On")
mono = bus_bool_prop("mono")
eq_ab = channel_bool_prop("eq.ab")
eq = bus_bool_prop("eq.On")
eq_ab = bus_bool_prop("eq.ab")
@property
def label(self) -> str:
val = self.getter("label")
if val is None:
val = self.public_packet.buslabels[self.index]
return val
@label.setter
def label(self, val: str):
if not isinstance(val, str):
raise VMCMDErrors("label is a string parameter")
self.setter("label", val)
label = channel_label_prop()
@property
def gain(self) -> float:
@ -62,10 +47,10 @@ class OutputBus(Channel):
else:
return ((1 << 16) - 1) - val
val = self.getter("gain")
val = round(self.getter("gain"), 1)
if val is None:
val = round((fget() * 0.01), 1)
return round(val, 1)
return val
@gain.setter
def gain(self, val: float):

8
vbancmd/channel.py
View File

@ -11,7 +11,7 @@ class Modes:
_mute: hex = 0x00000001
_solo: hex = 0x00000002
_mono: hex = 0x00000004
_mutec: hex = 0x00000008
_mc: hex = 0x00000008
_amix: hex = 0x00000010
_repeat: hex = 0x00000020
@ -85,13 +85,13 @@ class Channel(abc.ABC):
self._modes = Modes()
def getter(self, param):
cmd = f"{self.identifier}.{param}"
cmd = f"{self.identifier}[{self.index}].{param}"
if cmd in self._remote.cache:
return self._remote.cache.pop(f"{self.identifier}.{param}")
return self._remote.cache.pop(cmd)
def setter(self, param, val):
"""Sends a string request RT packet."""
self._remote.set_rt(f"{self.identifier}", param, val)
self._remote.set_rt(f"{self.identifier}[{self.index}]", param, val)
@abc.abstractmethod
def identifier(self):

110
vbancmd/meta.py
View File

@ -1,20 +1,22 @@
from .util import cache_bool, cache_string
from .errors import VMCMDErrors
from time import sleep
from functools import partial
def strip_bool_prop(param):
"""A strip bool prop."""
def channel_bool_prop(param):
"""A channel bool prop. (strip|bus)"""
@partial(cache_bool, param=param)
def fget(self):
val = self.getter(param)
if val is None:
val = (
not int.from_bytes(self.public_packet.stripstate[self.index], "little")
& getattr(self._modes, f"_{param}")
== 0
return (
not int.from_bytes(
getattr(self.public_packet, f"{self.identifier}state")[self.index],
"little",
)
return val
return val == 1
& getattr(self._modes, f"_{param}")
== 0
)
def fset(self, val):
if not isinstance(val, bool) and val not in (0, 1):
@ -24,41 +26,31 @@ def strip_bool_prop(param):
return property(fget, fset)
def bus_bool_prop(param):
"""A bus bool prop."""
def channel_label_prop():
"""A channel label prop. (strip|bus)"""
def fget(self):
val = self.getter(param)
if val is None:
val = (
not int.from_bytes(self.public_packet.busstate[self.index], "little")
& getattr(self._modes, f'_{param.replace(".", "_").lower()}')
== 0
)
return val
return val == 1
@partial(cache_string, param="label")
def fget(self) -> str:
return getattr(self.public_packet, f"{self.identifier}labels")[self.index]
def fset(self, val):
if not isinstance(val, bool) and val not in (0, 1):
raise VMCMDErrors(f"{param} is a boolean parameter")
self.setter(param, 1 if val else 0)
def fset(self, val: str):
if not isinstance(val, str):
raise VMCMDErrors("label is a string parameter")
self.setter("label", val)
return property(fget, fset)
def strip_output_prop(param):
"""A strip output prop."""
"""A strip output prop. (A1-A5, B1-B3)"""
@partial(cache_bool, param=param)
def fget(self):
val = self.getter(param)
if val is None:
val = (
not int.from_bytes(self.public_packet.stripstate[self.index], "little")
& getattr(self._modes, f"_bus{param.lower()}")
== 0
)
return val
return val == 1
return (
not int.from_bytes(self.public_packet.stripstate[self.index], "little")
& getattr(self._modes, f"_bus{param.lower()}")
== 0
)
def fset(self, val):
if not isinstance(val, bool) and val not in (0, 1):
@ -71,30 +63,28 @@ def strip_output_prop(param):
def bus_mode_prop(param):
"""A bus mode prop."""
@partial(cache_bool, param=f"mode.{param}")
def fget(self):
val = self.getter(f"mode.{param}")
if val is None:
modelist = {
"amix": (1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1),
"repeat": (0, 2, 2, 0, 0, 2, 2, 0, 0, 2, 2),
"bmix": (1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3),
"composite": (0, 0, 0, 4, 4, 4, 4, 0, 0, 0, 0),
"tvmix": (1, 0, 1, 4, 5, 4, 5, 0, 1, 0, 1),
"upmix21": (0, 2, 2, 4, 4, 6, 6, 0, 0, 2, 2),
"upmix41": (1, 2, 3, 4, 5, 6, 7, 0, 1, 2, 3),
"upmix61": (0, 0, 0, 0, 0, 0, 0, 8, 8, 8, 8),
"centeronly": (1, 0, 1, 0, 1, 0, 1, 8, 9, 8, 9),
"lfeonly": (0, 2, 2, 0, 0, 2, 2, 8, 8, 10, 10),
"rearonly": (1, 2, 3, 0, 1, 2, 3, 8, 9, 10, 11),
}
vals = (
int.from_bytes(self.public_packet.busstate[self.index], "little") & val
for val in self._modes.modevals
)
if param == "normal":
return not any(val for val in vals)
return tuple(round(val / 16) for val in vals) == modelist[param]
return val == 1
modelist = {
"amix": (1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1),
"repeat": (0, 2, 2, 0, 0, 2, 2, 0, 0, 2, 2),
"bmix": (1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3),
"composite": (0, 0, 0, 4, 4, 4, 4, 0, 0, 0, 0),
"tvmix": (1, 0, 1, 4, 5, 4, 5, 0, 1, 0, 1),
"upmix21": (0, 2, 2, 4, 4, 6, 6, 0, 0, 2, 2),
"upmix41": (1, 2, 3, 4, 5, 6, 7, 0, 1, 2, 3),
"upmix61": (0, 0, 0, 0, 0, 0, 0, 8, 8, 8, 8),
"centeronly": (1, 0, 1, 0, 1, 0, 1, 8, 9, 8, 9),
"lfeonly": (0, 2, 2, 0, 0, 2, 2, 8, 8, 10, 10),
"rearonly": (1, 2, 3, 0, 1, 2, 3, 8, 9, 10, 11),
}
vals = (
int.from_bytes(self.public_packet.busstate[self.index], "little") & val
for val in self._modes.modevals
)
if param == "normal":
return not any(vals)
return tuple(round(val / 16) for val in vals) == modelist[param]
def fset(self, val):
if not isinstance(val, bool) and val not in (0, 1):

55
vbancmd/strip.py
View File

@ -1,8 +1,7 @@
from .errors import VMCMDErrors
from . import channel
from .channel import Channel
from . import kinds
from .meta import strip_output_prop, strip_bool_prop
from .meta import strip_output_prop, channel_bool_prop, channel_label_prop
class InputStrip(Channel):
@ -22,19 +21,19 @@ class InputStrip(Channel):
(InputStrip, GainLayerMixin),
{
"levels": StripLevel(remote, index),
**{
param: channel_bool_prop(param)
for param in ["mono", "solo", "mute"]
},
},
)
return IS_cls(remote, index, **kwargs)
@property
def identifier(self):
return f"Strip[{self.index}]"
return "strip"
mono = strip_bool_prop("mono")
solo = strip_bool_prop("solo")
mute = strip_bool_prop("mute")
label = channel_label_prop()
@property
def limit(self) -> int:
@ -44,25 +43,12 @@ class InputStrip(Channel):
def limit(self, val: int):
self.setter("limit", val)
@property
def label(self) -> str:
val = self.getter("label")
if val is None:
val = self.public_packet.striplabels[self.index]
return val
@label.setter
def label(self, val: str):
if not isinstance(val, str):
raise VMCMDErrors("label is a string parameter")
self.setter("label", val)
@property
def gain(self) -> float:
val = self.getter("gain")
val = round(self.getter("gain"), 1)
if val is None:
val = self.gainlayer[0].gain
return round(val, 1)
return val
@gain.setter
def gain(self, val: float):
@ -96,23 +82,7 @@ class PhysicalInputStrip(InputStrip):
class VirtualInputStrip(InputStrip):
@property
def mc(self) -> bool:
val = self.getter("mc")
if val is None:
val = (
not int.from_bytes(self.public_packet.stripstate[self.index], "little")
& getattr(self._modes, f"_mutec")
== 0
)
return val
return val == 1
@mc.setter
def mc(self, val: bool):
if not isinstance(val, bool) and val not in (0, 1):
raise VMCMDErrors("mc is a boolean parameter")
self.setter("mc", 1 if val else 0)
mc = channel_bool_prop("mc")
mono = mc
@ -169,11 +139,10 @@ class GainLayer(InputStrip):
else:
return ((1 << 16) - 1) - val
val = self.getter(f"GainLayer[{self._i}]")
val = round(self.getter(f"GainLayer[{self._i}]"), 1)
if val is None:
val = round((fget() * 0.01), 1)
return val
return round(val, 1)
return val
@gain.setter
def gain(self, val: float):

28
vbancmd/util.py
View File

@ -1,5 +1,6 @@
from pathlib import Path
PROJECT_DIR = str(Path(__file__).parents[1])
@ -7,14 +8,29 @@ def project_path():
return PROJECT_DIR
def cache(func):
"""check if recently cached was an updated value"""
def cache_bool(func, param):
"""Check cache for a bool prop"""
def wrapper(*args, **kwargs):
# setup cache check
res = func(*args, **kwargs)
# update cache
return res
self, *rem = args
cmd = f"{self.identifier}[{self.index}].{param}"
if cmd in self._remote.cache:
print(self._remote.cache[cmd] == 1)
return self._remote.cache.pop(cmd) == 1
return func(*args, **kwargs)
return wrapper
def cache_string(func, param):
"""Check cache for a string prop"""
def wrapper(*args, **kwargs):
self, *rem = args
cmd = f"{self.identifier}[{self.index}].{param}"
if cmd in self._remote.cache:
return self._remote.cache.pop(cmd)
return func(*args, **kwargs)
return wrapper

5
vbancmd/vbancmd.py
View File

@ -194,7 +194,7 @@ class VbanCmd(abc.ABC):
val: Optional[Union[int, float]] = None,
):
"""Sends a string request command over a network."""
cmd = id_ if not param and val else f"{id_}.{param}={val}"
cmd = id_ if not param else f"{id_}.{param}={val}"
if self._sendrequest_string_socket in self.ready_to_write:
self._sendrequest_string_socket.sendto(
self._text_header.header + cmd.encode(),
@ -202,7 +202,8 @@ class VbanCmd(abc.ABC):
)
count = int.from_bytes(self._text_header.framecounter, "little") + 1
self._text_header.framecounter = count.to_bytes(4, "little")
self.cache[f"{id_}.{param}"] = val
if param:
self.cache[f"{id_}.{param}"] = val
if self._sync or self.in_apply:
sleep(self._delay)