From 3aebd90920f2a0a8b022b529044a942fd56ffcf9 Mon Sep 17 00:00:00 2001 From: onyx-and-iris <75868496+onyx-and-iris@users.noreply.github.com> Date: Fri, 29 Apr 2022 02:57:47 +0100 Subject: [PATCH] refactor meta functions 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 --- vbancmd/bus.py | 31 ++++--------- vbancmd/channel.py | 8 ++-- vbancmd/meta.py | 110 +++++++++++++++++++++------------------------ vbancmd/strip.py | 55 +++++------------------ vbancmd/util.py | 28 +++++++++--- vbancmd/vbancmd.py | 5 ++- 6 files changed, 99 insertions(+), 138 deletions(-) diff --git a/vbancmd/bus.py b/vbancmd/bus.py index f7e10d3..9ec69ee 100644 --- a/vbancmd/bus.py +++ b/vbancmd/bus.py @@ -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): diff --git a/vbancmd/channel.py b/vbancmd/channel.py index ac8fa21..7af1b4d 100644 --- a/vbancmd/channel.py +++ b/vbancmd/channel.py @@ -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): diff --git a/vbancmd/meta.py b/vbancmd/meta.py index fbde7c0..64f3ac2 100644 --- a/vbancmd/meta.py +++ b/vbancmd/meta.py @@ -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): diff --git a/vbancmd/strip.py b/vbancmd/strip.py index 4eeae67..ce78fc0 100644 --- a/vbancmd/strip.py +++ b/vbancmd/strip.py @@ -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): diff --git a/vbancmd/util.py b/vbancmd/util.py index 0b43f4c..7f55d97 100644 --- a/vbancmd/util.py +++ b/vbancmd/util.py @@ -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 diff --git a/vbancmd/vbancmd.py b/vbancmd/vbancmd.py index 11471d2..4a1bd54 100644 --- a/vbancmd/vbancmd.py +++ b/vbancmd/vbancmd.py @@ -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)