change to bus_mode_prop meta function.

change to bus_mode_prop meta function.

save gain to cache as gain.

remove kwarg ratelimiter (used for test units)
This commit is contained in:
onyx-and-iris 2022-04-26 07:35:41 +01:00
parent d1555663b3
commit 2bc9fa9d57
5 changed files with 31 additions and 57 deletions

View File

@ -123,7 +123,7 @@ def _make_bus_mode_mixin(cls):
(cls,), (cls,),
{ {
**{ **{
f"{mode.lower()}": bus_mode_prop(mode) f"{mode.lower()}": bus_mode_prop(mode.lower())
for mode in [ for mode in [
"normal", "normal",
"Amix", "Amix",

View File

@ -13,7 +13,6 @@ class Modes:
_mono: hex = 0x00000004 _mono: hex = 0x00000004
_mutec: hex = 0x00000008 _mutec: hex = 0x00000008
_normal: hex = 0x00000000
_amix: hex = 0x00000010 _amix: hex = 0x00000010
_repeat: hex = 0x00000020 _repeat: hex = 0x00000020
_bmix: hex = 0x00000030 _bmix: hex = 0x00000030
@ -62,7 +61,6 @@ class Modes:
return ( return (
val val
for val in [ for val in [
self._normal,
self._amix, self._amix,
self._repeat, self._repeat,
self._bmix, self._bmix,

View File

@ -73,62 +73,35 @@ def strip_output_prop(param):
def bus_mode_prop(param): def bus_mode_prop(param):
"""A bus mode prop.""" """A bus mode prop."""
# fmt: off
def fget(self): def fget(self):
data = self.public_packet
modes = {
"normal": (
False, False, False, False, False, False, False, False, False, False, False, False,
),
"amix": (
False, True, False, True, False, True, False, True, False, True, False, True,
),
"repeat": (
False, False, True, True, False, False, True, True, False, False, True, True,
),
"bmix": (
False, True, True, True, False, True, True, True, False, True, True, True,
),
"composite": (
False, False, False, False, True, True, True, True, False, False, False, False,
),
"tvmix": (
False, True, False, True, True, True, True, True, False, True, False, True,
),
"upmix21": (
False, False, True, True, True, True, True, True, False, False, True, True,
),
"upmix41": (
False, True, True, True, True, True, True, True, False, True, True, True,
),
"upmix61": (
False, False, False, False, False, False, False, False, True, True, True, True,
),
"centeronly": (
False, True, False, True, False, True, False, True, True, True, True, True,
),
"lfeonly": (
False, False, True, True, False, False, True, True, True, True, True, True,
),
"rearonly": (
False, True, True, True, False, True, True, True, True, True, True, True,
),
}
vals = tuple(
not int.from_bytes(data.busstate[self.index], "little") & val == 0
for val in self._modes.modevals
)
val = self.getter(f"mode.{param}") val = self.getter(f"mode.{param}")
if val is None: if val is None:
val = vals == modes[param.lower()] if param == "normal":
return not any(
not int.from_bytes(
self.public_packet.busstate[self.index], "little"
)
& val
== 0
for val in self._modes.modevals
)
else:
val = (
not int.from_bytes(
self.public_packet.busstate[self.index], "little"
)
& getattr(self._modes, f"_{param}")
== 0
)
self._remote.cache[f"{self.identifier}.mode.{param}"] = [val, False] self._remote.cache[f"{self.identifier}.mode.{param}"] = [val, False]
return val return val
return val == 1 return val == 1
# fmt: on
def fset(self, val): def fset(self, val):
if not isinstance(val, bool) and val not in (0, 1): if not isinstance(val, bool) and val not in (0, 1):
raise VMCMDErrors(f"mode.{param} is a boolean parameter") raise VMCMDErrors(f"mode.{param} is a boolean parameter")
self.setter(f"mode.{param}", 1 if val else 0) self.setter(f"mode.{param}", 1)
return property(fget, fset) return property(fget, fset)

View File

@ -60,10 +60,10 @@ class InputStrip(Channel):
@property @property
def gain(self) -> float: def gain(self) -> float:
val = self.getter("GainLayer[0]") val = self.getter("gain")
if val is None: if val is None:
val = self.gainlayer[0].gain val = self.gainlayer[0].gain
self._remote.cache[f"{self.identifier}.GainLayer[0]"] = [val, False] self._remote.cache[f"{self.identifier}.gain"] = [val, False]
return round(val, 1) return round(val, 1)
@gain.setter @gain.setter

View File

@ -28,7 +28,6 @@ class VbanCmd(abc.ABC):
self._bps = kwargs["bps"] self._bps = kwargs["bps"]
self._channel = kwargs["channel"] self._channel = kwargs["channel"]
self._delay = kwargs["delay"] self._delay = kwargs["delay"]
self._ratelimiter = kwargs["ratelimiter"]
self._sync = kwargs["sync"] self._sync = kwargs["sync"]
# fmt: off # fmt: off
self._bps_opts = [ self._bps_opts = [
@ -89,6 +88,7 @@ class VbanCmd(abc.ABC):
self._public_packet = self._get_rt() self._public_packet = self._get_rt()
worker2 = Thread(target=self._keepupdated, daemon=True) worker2 = Thread(target=self._keepupdated, daemon=True)
worker2.start() worker2.start()
self._clear_dirty()
def _send_register_rt(self): def _send_register_rt(self):
""" """
@ -150,6 +150,10 @@ class VbanCmd(abc.ABC):
def public_packet(self): def public_packet(self):
return self._public_packet return self._public_packet
def _clear_dirty(self):
while self.pdirty:
pass
@public_packet.setter @public_packet.setter
def public_packet(self, val): def public_packet(self, val):
self._public_packet = val self._public_packet = val
@ -198,7 +202,7 @@ class VbanCmd(abc.ABC):
self._text_header.framecounter = count.to_bytes(4, "little") self._text_header.framecounter = count.to_bytes(4, "little")
self.cache[f"{id_}.{param}"] = [val, True] self.cache[f"{id_}.{param}"] = [val, True]
if self._sync: if self._sync:
sleep(self._ratelimiter) sleep(self._delay)
def sendtext(self, cmd): def sendtext(self, cmd):
"""Sends a multiple parameter string over a network.""" """Sends a multiple parameter string over a network."""
@ -243,9 +247,10 @@ class VbanCmd(abc.ABC):
else: else:
raise ValueError(obj) raise ValueError(obj)
target.apply(submapping) target.apply(submapping)
if not self._sync:
sleep(self._ratelimiter)
def apply_profile(self, name: str): def apply_profile(self, name: str):
self._sync = True
try: try:
profile = self.profiles[name] profile = self.profiles[name]
if "extends" in profile: if "extends" in profile:
@ -260,7 +265,6 @@ class VbanCmd(abc.ABC):
self.apply(profile) self.apply(profile)
except KeyError: except KeyError:
raise VMCMDErrors(f"Unknown profile: {self.kind.id}/{name}") raise VMCMDErrors(f"Unknown profile: {self.kind.id}/{name}")
self._sync = False
def reset(self) -> NoReturn: def reset(self) -> NoReturn:
self.apply_profile("base") self.apply_profile("base")
@ -309,7 +313,6 @@ def _make_remote(kind: NamedTuple) -> VbanCmd:
"bps": 0, "bps": 0,
"channel": 0, "channel": 0,
"delay": 0.001, "delay": 0.001,
"ratelimiter": 0.018,
"sync": False, "sync": False,
} }
kwargs = defaultkwargs | kwargs kwargs = defaultkwargs | kwargs