mirror of
https://github.com/onyx-and-iris/voicemeeter-api-python.git
synced 2024-11-15 16:40:46 +00:00
285 lines
9.1 KiB
Python
285 lines
9.1 KiB
Python
import ctypes as ct
|
|
import time
|
|
from abc import abstractmethod
|
|
from functools import partial
|
|
from threading import Thread
|
|
from typing import Iterable, NoReturn, Optional, Self, Union
|
|
|
|
from .cbindings import CBindings
|
|
from .error import VMError
|
|
from .kinds import KindId
|
|
from .subject import Subject
|
|
from .util import polling, script
|
|
|
|
|
|
class Remote(CBindings):
|
|
"""Base class responsible for wrapping the C Remote API"""
|
|
|
|
DELAY = 0.001
|
|
|
|
def __init__(self, **kwargs):
|
|
self.cache = {}
|
|
self.subject = Subject()
|
|
self._strip_levels, self._bus_levels = self.all_levels
|
|
|
|
for attr, val in kwargs.items():
|
|
setattr(self, attr, val)
|
|
|
|
def __enter__(self) -> Self:
|
|
"""setup procedures"""
|
|
self.login()
|
|
self.init_thread()
|
|
return self
|
|
|
|
@abstractmethod
|
|
def __str__(self):
|
|
"""Ensure subclasses override str magic method"""
|
|
pass
|
|
|
|
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."""
|
|
while self.running:
|
|
if self.pdirty:
|
|
self.subject.notify("pdirty")
|
|
if self.ldirty:
|
|
self._strip_levels = self.strip_buf
|
|
self._bus_levels = self.bus_buf
|
|
self.subject.notify(
|
|
"ldirty",
|
|
(
|
|
self._strip_levels,
|
|
self._strip_comp,
|
|
self._bus_levels,
|
|
self._bus_comp,
|
|
),
|
|
)
|
|
time.sleep(self.ratelimit)
|
|
|
|
def login(self) -> NoReturn:
|
|
"""Login to the API, initialize dirty parameters"""
|
|
res = self.vm_login()
|
|
if res == 0:
|
|
print(f"Successfully logged into {self}")
|
|
elif res == 1:
|
|
self.run_voicemeeter(self.kind.name)
|
|
self.clear_dirty()
|
|
|
|
def run_voicemeeter(self, kind_id: str) -> NoReturn:
|
|
if kind_id not in (kind.name.lower() for kind in KindId):
|
|
raise VMError(f"Unexpected Voicemeeter type: '{kind_id}'")
|
|
if kind_id == "potato" and ct.sizeof(ct.c_voidp) == 8:
|
|
value = KindId[kind_id.upper()].value + 3
|
|
else:
|
|
value = KindId[kind_id.upper()].value
|
|
self.vm_runvm(value)
|
|
time.sleep(1)
|
|
|
|
@property
|
|
def type(self) -> str:
|
|
"""Returns the type of Voicemeeter installation (basic, banana, potato)."""
|
|
type_ = ct.c_long()
|
|
self.vm_get_type(ct.byref(type_))
|
|
return KindId(type_.value).name.lower()
|
|
|
|
@property
|
|
def version(self) -> str:
|
|
"""Returns Voicemeeter's version as a string"""
|
|
ver = ct.c_long()
|
|
self.vm_get_version(ct.byref(ver))
|
|
v1 = (ver.value & 0xFF000000) >> 24
|
|
v2 = (ver.value & 0x00FF0000) >> 16
|
|
v3 = (ver.value & 0x0000FF00) >> 8
|
|
v4 = ver.value & 0x000000FF
|
|
return f"{v1}.{v2}.{v3}.{v4}"
|
|
|
|
@property
|
|
def pdirty(self) -> bool:
|
|
"""True iff UI parameters have been updated."""
|
|
return self.vm_pdirty() == 1
|
|
|
|
@property
|
|
def mdirty(self) -> bool:
|
|
"""True iff MB parameters have been updated."""
|
|
return self.vm_mdirty() == 1
|
|
|
|
@property
|
|
def ldirty(self) -> bool:
|
|
"""True iff levels have been updated."""
|
|
self.strip_buf, self.bus_buf = self.all_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,
|
|
)
|
|
)
|
|
|
|
def clear_dirty(self):
|
|
while self.pdirty or self.mdirty:
|
|
pass
|
|
|
|
@polling
|
|
def get(self, param: str, is_string: Optional[bool] = False) -> Union[str, float]:
|
|
"""Gets a string or float parameter"""
|
|
if is_string:
|
|
buf = ct.create_unicode_buffer(512)
|
|
self.call(
|
|
partial(self.vm_get_parameter_string, param.encode(), ct.byref(buf))
|
|
)
|
|
else:
|
|
buf = ct.c_float()
|
|
self.call(
|
|
partial(self.vm_get_parameter_float, param.encode(), ct.byref(buf))
|
|
)
|
|
return buf.value
|
|
|
|
def set(self, param: str, val: Union[str, float]) -> NoReturn:
|
|
"""Sets a string or float parameter. Caches value"""
|
|
if isinstance(val, str):
|
|
if len(val) >= 512:
|
|
raise VMError("String is too long")
|
|
self.call(
|
|
partial(self.vm_set_parameter_string, param.encode(), ct.c_wchar_p(val))
|
|
)
|
|
else:
|
|
self.call(
|
|
partial(
|
|
self.vm_set_parameter_float, param.encode(), ct.c_float(float(val))
|
|
)
|
|
)
|
|
self.cache[param] = val
|
|
|
|
@polling
|
|
def get_buttonstatus(self, id: int, mode: int) -> int:
|
|
"""Gets a macrobutton parameter"""
|
|
state = ct.c_float()
|
|
self.call(
|
|
partial(
|
|
self.vm_get_buttonstatus,
|
|
ct.c_long(id),
|
|
ct.byref(state),
|
|
ct.c_long(mode),
|
|
)
|
|
)
|
|
return int(state.value)
|
|
|
|
def set_buttonstatus(self, id: int, state: int, mode: int) -> NoReturn:
|
|
"""Sets a macrobutton parameter. Caches value"""
|
|
c_state = ct.c_float(float(state))
|
|
self.call(
|
|
partial(self.vm_set_buttonstatus, ct.c_long(id), c_state, ct.c_long(mode))
|
|
)
|
|
self.cache[f"mb_{id}_{mode}"] = int(c_state.value)
|
|
|
|
def get_num_devices(self, direction: str = None) -> int:
|
|
"""Retrieves number of physical devices connected"""
|
|
if direction not in ("in", "out"):
|
|
raise VMError("Expected a direction: in or out")
|
|
func = getattr(self, f"vm_get_num_{direction}devices")
|
|
return func()
|
|
|
|
def get_device_description(self, index: int, direction: str = None) -> tuple:
|
|
"""Returns a tuple of device parameters"""
|
|
if direction not in ("in", "out"):
|
|
raise VMError("Expected a direction: in or out")
|
|
type_ = ct.c_long()
|
|
name = ct.create_unicode_buffer(256)
|
|
hwid = ct.create_unicode_buffer(256)
|
|
func = getattr(self, f"vm_get_desc_{direction}devices")
|
|
func(
|
|
ct.c_long(index),
|
|
ct.byref(type_),
|
|
ct.byref(name),
|
|
ct.byref(hwid),
|
|
)
|
|
return (name.value, type_.value, hwid.value)
|
|
|
|
@property
|
|
def all_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)
|
|
for i in range(2 * self.kind.phys_in + 8 * self.kind.virt_in)
|
|
),
|
|
tuple(
|
|
self.get_level(3, i)
|
|
for i in range(8 * (self.kind.phys_out + self.kind.virt_out))
|
|
),
|
|
)
|
|
|
|
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
|
|
|
|
@script
|
|
def sendtext(self, script: str):
|
|
"""Sets many parameters from a script"""
|
|
if len(script) > 48000:
|
|
raise ValueError("Script too large, max size 48kB")
|
|
self.call(partial(self.vm_set_parameter_multi, script))
|
|
time.sleep(self.DELAY)
|
|
|
|
def apply(self, data: dict):
|
|
"""
|
|
Sets all parameters of a dict
|
|
|
|
minor delay between each recursion
|
|
"""
|
|
|
|
def param(key):
|
|
obj, m2, *rem = key.split("-")
|
|
index = int(m2) if m2.isnumeric() else int(*rem)
|
|
if obj in ("strip", "bus", "button"):
|
|
return getattr(self, obj)[index]
|
|
elif obj == "vban":
|
|
return getattr(getattr(self, obj), f"{m2}stream")[index]
|
|
else:
|
|
raise ValueError(obj)
|
|
|
|
[param(key).apply(datum).then_wait() for key, datum in data.items()]
|
|
|
|
def apply_config(self, name):
|
|
"""applies a config from memory"""
|
|
error_msg = (
|
|
f"No config with name '{name}' is loaded into memory",
|
|
f"Known configs: {list(self.configs.keys())}",
|
|
)
|
|
try:
|
|
self.apply(self.configs[name])
|
|
except KeyError as e:
|
|
print(("\n").join(error_msg))
|
|
print(f"Profile '{name}' applied!")
|
|
|
|
def logout(self) -> NoReturn:
|
|
"""Wait for dirty parameters to clear, then logout of the API"""
|
|
self.clear_dirty()
|
|
time.sleep(0.1)
|
|
res = self.vm_logout()
|
|
if res == 0:
|
|
print(f"Successfully logged out of {self}")
|
|
|
|
def end_thread(self):
|
|
self.running = False
|
|
|
|
def __exit__(self, exc_type, exc_value, exc_traceback) -> NoReturn:
|
|
"""teardown procedures"""
|
|
self.end_thread()
|
|
self.logout()
|