mirror of
https://github.com/onyx-and-iris/voicemeeter-api-python.git
synced 2024-11-15 16:40:46 +00:00
add support for all strip level modes to observers
is_updated now returns a bool (if that channel is updated (dirty)) observer example updated. minor ver bump
This commit is contained in:
parent
1b75d53cfb
commit
9af2fc98aa
@ -6,7 +6,20 @@ class Observer:
|
||||
self.vm = vm
|
||||
|
||||
def on_update(self, subject):
|
||||
print(subject)
|
||||
if subject == "pdirty":
|
||||
print("pdirty!")
|
||||
if subject == "mdirty":
|
||||
print("mdirty!")
|
||||
if subject == "ldirty":
|
||||
info = (
|
||||
"levels changed:",
|
||||
f"[strip 0 {self.vm.strip[0].levels.is_updated}]",
|
||||
f"[strip 1 {self.vm.strip[1].levels.is_updated}]",
|
||||
f"[strip 2 {self.vm.strip[2].levels.is_updated}]",
|
||||
f"[strip 3 {self.vm.strip[3].levels.is_updated}]",
|
||||
f"[strip 4 {self.vm.strip[4].levels.is_updated}]",
|
||||
)
|
||||
print(" ".join(info))
|
||||
|
||||
|
||||
def main():
|
||||
|
@ -1,6 +1,6 @@
|
||||
[tool.poetry]
|
||||
name = "voicemeeter-api"
|
||||
version = "0.1.10"
|
||||
version = "0.2.0"
|
||||
description = "A Python wrapper for the Voiceemeter API"
|
||||
authors = ["onyx-and-iris <code@onyxandiris.online>"]
|
||||
license = "MIT"
|
||||
|
@ -20,7 +20,8 @@ class Remote(CBindings):
|
||||
def __init__(self, **kwargs):
|
||||
self.cache = {}
|
||||
self.subject = Subject()
|
||||
self._strip_levels, self._bus_levels = self.all_levels
|
||||
self.strip_mode = 0
|
||||
self.running = True
|
||||
|
||||
for attr, val in kwargs.items():
|
||||
setattr(self, attr, val)
|
||||
@ -38,21 +39,23 @@ class Remote(CBindings):
|
||||
|
||||
def init_thread(self):
|
||||
"""Starts updates thread."""
|
||||
self.running = True
|
||||
t = Thread(target=self._updates, daemon=True)
|
||||
t.start()
|
||||
|
||||
def _updates(self):
|
||||
"""Continously update observers of dirty states."""
|
||||
self.cache["strip_level"], self.cache["bus_level"] = self._get_levels()
|
||||
|
||||
while self.running:
|
||||
if self.pdirty:
|
||||
self.subject.notify("pdirty")
|
||||
if self.mdirty:
|
||||
self.subject.notify("mdirty")
|
||||
if self.ldirty:
|
||||
self._strip_levels = self.strip_buf
|
||||
self._bus_levels = self.bus_buf
|
||||
self.cache["strip_level"] = self._strip_buf
|
||||
self.cache["bus_level"] = self._bus_buf
|
||||
self.subject.notify("ldirty")
|
||||
|
||||
time.sleep(self.ratelimit)
|
||||
|
||||
def login(self) -> NoReturn:
|
||||
@ -105,18 +108,17 @@ class Remote(CBindings):
|
||||
@property
|
||||
def ldirty(self) -> bool:
|
||||
"""True iff levels have been updated."""
|
||||
self.strip_buf, self.bus_buf = self.all_levels
|
||||
self._strip_buf, self._bus_buf = self._get_levels()
|
||||
self._strip_comp, self._bus_comp = (
|
||||
tuple(not a == b for a, b in zip(self.strip_buf, self._strip_levels)),
|
||||
tuple(not a == b for a, b in zip(self.bus_buf, self._bus_levels)),
|
||||
)
|
||||
return any(
|
||||
any(l)
|
||||
for l in (
|
||||
self._strip_comp,
|
||||
self._bus_comp,
|
||||
)
|
||||
tuple(
|
||||
not a == b
|
||||
for a, b in zip(self.cache.get("strip_level"), self._strip_buf)
|
||||
),
|
||||
tuple(
|
||||
not a == b for a, b in zip(self.cache.get("bus_level"), self._bus_buf)
|
||||
),
|
||||
)
|
||||
return any(any(l) for l in (self._strip_comp, self._bus_comp))
|
||||
|
||||
def clear_dirty(self):
|
||||
while self.pdirty or self.mdirty:
|
||||
@ -198,16 +200,19 @@ class Remote(CBindings):
|
||||
)
|
||||
return (name.value, type_.value, hwid.value)
|
||||
|
||||
@property
|
||||
def all_levels(self) -> Iterable:
|
||||
def get_level(self, type_: int, index: int) -> float:
|
||||
"""Retrieves a single level value"""
|
||||
val = ct.c_float()
|
||||
self.vm_get_level(ct.c_long(type_), ct.c_long(index), ct.byref(val))
|
||||
return val.value
|
||||
|
||||
def _get_levels(self) -> Iterable:
|
||||
"""
|
||||
returns both level arrays (strip_levels, bus_levels) BEFORE math conversion
|
||||
|
||||
strip levels in PREFADER mode.
|
||||
"""
|
||||
return (
|
||||
tuple(
|
||||
self.get_level(0, i)
|
||||
self.get_level(self.strip_mode, i)
|
||||
for i in range(2 * self.kind.phys_in + 8 * self.kind.virt_in)
|
||||
),
|
||||
tuple(
|
||||
@ -216,20 +221,6 @@ class Remote(CBindings):
|
||||
),
|
||||
)
|
||||
|
||||
def get_level(self, type_: int, index: int) -> float:
|
||||
"""Retrieves a single level value"""
|
||||
val = ct.c_float()
|
||||
self.vm_get_level(ct.c_long(type_), ct.c_long(index), ct.byref(val))
|
||||
return val.value
|
||||
|
||||
@property
|
||||
def strip_levels(self):
|
||||
return self._strip_levels
|
||||
|
||||
@property
|
||||
def bus_levels(self):
|
||||
return self._bus_levels
|
||||
|
||||
@script
|
||||
def sendtext(self, script: str):
|
||||
"""Sets many parameters from a script"""
|
||||
|
@ -113,15 +113,19 @@ 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):
|
||||
def getter(self, mode):
|
||||
"""Returns a tuple of level values for the channel."""
|
||||
|
||||
def fget(i):
|
||||
return round(20 * log(i, 10), 1) if i > 0 else -200.0
|
||||
if self._remote.running and "bus_level" in self._remote.cache:
|
||||
res = self._remote.cache["bus_level"][i]
|
||||
else:
|
||||
res = self._remote.get_level(mode, i)
|
||||
return round(20 * log(res, 10), 1) if res > 0 else -200.0
|
||||
|
||||
range_ = self.level_map[self.index]
|
||||
return tuple(fget(i) for i in self._remote._bus_levels[range_[0] : range_[-1]])
|
||||
return tuple(fget(i) for i in range(*self.range))
|
||||
|
||||
@property
|
||||
def identifier(self) -> str:
|
||||
@ -129,11 +133,11 @@ class BusLevel(IRemote):
|
||||
|
||||
@property
|
||||
def all(self) -> tuple:
|
||||
return self.getter()
|
||||
return self.getter(3)
|
||||
|
||||
@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():
|
||||
|
@ -186,25 +186,19 @@ class StripLevel(IRemote):
|
||||
)
|
||||
)
|
||||
self.level_map = phys_map + virt_map
|
||||
self.range = self.level_map[self.index]
|
||||
|
||||
def getter(self, mode):
|
||||
"""Returns a tuple of level values for the channel."""
|
||||
|
||||
def fget(i):
|
||||
res = self._remote.get_level(mode, i)
|
||||
if self._remote.running and "strip_level" in self._remote.cache:
|
||||
res = self._remote.cache["strip_level"][i]
|
||||
else:
|
||||
res = self._remote.get_level(mode, i)
|
||||
return round(20 * log(res, 10), 1) if res > 0 else -200.0
|
||||
|
||||
range_ = self.level_map[self.index]
|
||||
return tuple(fget(i) for i in range(*range_))
|
||||
|
||||
def getter_prefader(self):
|
||||
def fget(i):
|
||||
return round(20 * log(i, 10), 1) if i > 0 else -200.0
|
||||
|
||||
range_ = self.level_map[self.index]
|
||||
return tuple(
|
||||
fget(i) for i in self._remote._strip_levels[range_[0] : range_[-1]]
|
||||
)
|
||||
return tuple(fget(i) for i in range(*self.range))
|
||||
|
||||
@property
|
||||
def identifier(self) -> str:
|
||||
@ -212,19 +206,22 @@ class StripLevel(IRemote):
|
||||
|
||||
@property
|
||||
def prefader(self) -> tuple:
|
||||
return self.getter_prefader()
|
||||
self._remote.strip_mode = 0
|
||||
return self.getter(0)
|
||||
|
||||
@property
|
||||
def postfader(self) -> tuple:
|
||||
self._remote.strip_mode = 1
|
||||
return self.getter(1)
|
||||
|
||||
@property
|
||||
def postmute(self) -> tuple:
|
||||
self._remote.strip_mode = 2
|
||||
return self.getter(2)
|
||||
|
||||
@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):
|
||||
|
Loading…
Reference in New Issue
Block a user