onyx_online/voicemeeter-api-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/voicemeeter-api-python.git synced 2024-11-22 19:10:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

all CAPI calls wrapped by call().

Browse Source 
raise CAPIError if macrobutton fns are not bound

producer thread added to init_thread()
This commit is contained in:
onyx-and-iris 2023-06-23 01:36:02 +01:00
parent 6ddfe3044e
commit 7d4d09ff29
1 changed files with 69 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
voicemeeterlib/remote.py
View File

@ -2,7 +2,7 @@ import ctypes as ct
import logging import logging
import time import time
from abc import abstractmethod from abc import abstractmethod
from functools import partial from queue import Queue
from typing import Iterable, NoReturn, Optional, Union from typing import Iterable, NoReturn, Optional, Union
from .cbindings import CBindings from .cbindings import CBindings
@ -12,33 +12,36 @@ from .inst import bits
from .kinds import KindId from .kinds import KindId
from .misc import Midi from .misc import Midi
from .subject import Subject from .subject import Subject
from .updater import Updater from .updater import Producer, Updater
from .util import grouper, polling, script from .util import grouper, polling, script
logger = logging.getLogger(__name__)
class Remote(CBindings): class Remote(CBindings):
"""Base class responsible for wrapping the C Remote API""" """Base class responsible for wrapping the C Remote API"""
logger = logging.getLogger("remote.remote")
DELAY = 0.001 DELAY = 0.001
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.strip_mode = 0 self.strip_mode = 0
self.cache = {} self.cache = {}
self.cache["strip_level"], self.cache["bus_level"] = self._get_levels()
self.midi = Midi() self.midi = Midi()
self.subject = Subject() self.subject = self.observer = Subject()
self.running = None self.running = None
self.event = Event(
{k: kwargs.pop(k) for k in ("pdirty", "mdirty", "midi", "ldirty")}
)
self.logger = logger.getChild(self.__class__.__name__)
for attr, val in kwargs.items(): for attr, val in kwargs.items():
setattr(self, attr, val) setattr(self, attr, val)
self.event = Event(self.subs)
def __enter__(self): def __enter__(self):
"""setup procedures""" """setup procedures"""
self.login() self.login()
self.init_thread() if self.event.any():
self.init_thread()
return self return self
@abstractmethod @abstractmethod
@ -50,17 +53,24 @@ class Remote(CBindings):
"""Starts updates thread.""" """Starts updates thread."""
self.running = True self.running = True
self.event.info() self.event.info()
self.cache["strip_level"], self.cache["bus_level"] = self._get_levels()
self.updater = Updater(self) queue = Queue()
self.updater = Updater(self, queue)
self.updater.start() self.updater.start()
self.producer = Producer(self, queue)
self.producer.start()
self.logger.debug("events thread initiated!")
def login(self) -> NoReturn: def login(self) -> NoReturn:
"""Login to the API, initialize dirty parameters""" """Login to the API, initialize dirty parameters"""
res = self.vm_login() res = self.call(self.vm_login, ok=(0, 1))
if res == 1: if res == 1:
self.logger.info(
"Voicemeeter engine running but GUI not launched. Launching the GUI now."
)
self.run_voicemeeter(self.kind.name) self.run_voicemeeter(self.kind.name)
elif res != 0:
raise CAPIError(f"VBVMR_Login returned {res}")
self.logger.info(f"{type(self).__name__}: Successfully logged into {self}") self.logger.info(f"{type(self).__name__}: Successfully logged into {self}")
self.clear_dirty() self.clear_dirty()
@ -71,21 +81,21 @@ class Remote(CBindings):
value = KindId[kind_id.upper()].value + 3 value = KindId[kind_id.upper()].value + 3
else: else:
value = KindId[kind_id.upper()].value value = KindId[kind_id.upper()].value
self.vm_runvm(value) self.call(self.vm_runvm, value)
time.sleep(1) time.sleep(1)
@property @property
def type(self) -> str: def type(self) -> str:
"""Returns the type of Voicemeeter installation (basic, banana, potato).""" """Returns the type of Voicemeeter installation (basic, banana, potato)."""
type_ = ct.c_long() type_ = ct.c_long()
self.vm_get_type(ct.byref(type_)) self.call(self.vm_get_type, ct.byref(type_))
return KindId(type_.value).name.lower() return KindId(type_.value).name.lower()
@property @property
def version(self) -> str: def version(self) -> str:
"""Returns Voicemeeter's version as a string""" """Returns Voicemeeter's version as a string"""
ver = ct.c_long() ver = ct.c_long()
self.vm_get_version(ct.byref(ver)) self.call(self.vm_get_version, ct.byref(ver))
return "{}.{}.{}.{}".format( return "{}.{}.{}.{}".format(
(ver.value & 0xFF000000) >> 24, (ver.value & 0xFF000000) >> 24,
(ver.value & 0x00FF0000) >> 16, (ver.value & 0x00FF0000) >> 16,
@ -96,12 +106,18 @@ class Remote(CBindings):
@property @property
def pdirty(self) -> bool: def pdirty(self) -> bool:
"""True iff UI parameters have been updated.""" """True iff UI parameters have been updated."""
return self.vm_pdirty() == 1 return self.call(self.vm_pdirty, ok=(0, 1)) == 1
@property @property
def mdirty(self) -> bool: def mdirty(self) -> bool:
"""True iff MB parameters have been updated.""" """True iff MB parameters have been updated."""
return self.vm_mdirty() == 1 try:
return self.call(self.vm_mdirty, ok=(0, 1)) == 1
except AttributeError as e:
self.logger.exception(f"{type(e).__name__}: {e}")
raise CAPIError(
"no bind for VBVMR_MacroButton_IsDirty. are you using an old version of the API?"
) from e
@property @property
def ldirty(self) -> bool: def ldirty(self) -> bool:
@ -112,23 +128,24 @@ class Remote(CBindings):
and self.cache.get("bus_level") == self._bus_buf and self.cache.get("bus_level") == self._bus_buf
) )
def clear_dirty(self): def clear_dirty(self) -> NoReturn:
while self.pdirty or self.mdirty: try:
pass while self.pdirty or self.mdirty:
pass
except CAPIError:
self.logger.error("no bind for mdirty, clearing pdirty only")
while self.pdirty:
pass
@polling @polling
def get(self, param: str, is_string: Optional[bool] = False) -> Union[str, float]: def get(self, param: str, is_string: Optional[bool] = False) -> Union[str, float]:
"""Gets a string or float parameter""" """Gets a string or float parameter"""
if is_string: if is_string:
buf = ct.create_unicode_buffer(512) buf = ct.create_unicode_buffer(512)
self.call( self.call(self.vm_get_parameter_string, param.encode(), ct.byref(buf))
partial(self.vm_get_parameter_string, param.encode(), ct.byref(buf))
)
else: else:
buf = ct.c_float() buf = ct.c_float()
self.call( self.call(self.vm_get_parameter_float, param.encode(), ct.byref(buf))
partial(self.vm_get_parameter_float, param.encode(), ct.byref(buf))
)
return buf.value return buf.value
def set(self, param: str, val: Union[str, float]) -> NoReturn: def set(self, param: str, val: Union[str, float]) -> NoReturn:
@ -136,14 +153,10 @@ class Remote(CBindings):
if isinstance(val, str): if isinstance(val, str):
if len(val) >= 512: if len(val) >= 512:
raise VMError("String is too long") raise VMError("String is too long")
self.call( self.call(self.vm_set_parameter_string, param.encode(), ct.c_wchar_p(val))
partial(self.vm_set_parameter_string, param.encode(), ct.c_wchar_p(val))
)
else: else:
self.call( self.call(
partial( self.vm_set_parameter_float, param.encode(), ct.c_float(float(val))
self.vm_set_parameter_float, param.encode(), ct.c_float(float(val))
)
) )
self.cache[param] = val self.cache[param] = val
@ -151,22 +164,30 @@ class Remote(CBindings):
def get_buttonstatus(self, id: int, mode: int) -> int: def get_buttonstatus(self, id: int, mode: int) -> int:
"""Gets a macrobutton parameter""" """Gets a macrobutton parameter"""
state = ct.c_float() state = ct.c_float()
self.call( try:
partial( self.call(
self.vm_get_buttonstatus, self.vm_get_buttonstatus,
ct.c_long(id), ct.c_long(id),
ct.byref(state), ct.byref(state),
ct.c_long(mode), ct.c_long(mode),
) )
) except AttributeError as e:
self.logger.exception(f"{type(e).__name__}: {e}")
raise CAPIError(
"no bind for VBVMR_MacroButton_GetStatus. are you using an old version of the API?"
) from e
return int(state.value) return int(state.value)
def set_buttonstatus(self, id: int, state: int, mode: int) -> NoReturn: def set_buttonstatus(self, id: int, state: int, mode: int) -> NoReturn:
"""Sets a macrobutton parameter. Caches value""" """Sets a macrobutton parameter. Caches value"""
c_state = ct.c_float(float(state)) c_state = ct.c_float(float(state))
self.call( try:
partial(self.vm_set_buttonstatus, ct.c_long(id), c_state, ct.c_long(mode)) self.call(self.vm_set_buttonstatus, ct.c_long(id), c_state, ct.c_long(mode))
) except AttributeError as e:
self.logger.exception(f"{type(e).__name__}: {e}")
raise CAPIError(
"no bind for VBVMR_MacroButton_SetStatus. are you using an old version of the API?"
) from e
self.cache[f"mb_{id}_{mode}"] = int(c_state.value) self.cache[f"mb_{id}_{mode}"] = int(c_state.value)
def get_num_devices(self, direction: str = None) -> int: def get_num_devices(self, direction: str = None) -> int:
@ -174,7 +195,8 @@ class Remote(CBindings):
if direction not in ("in", "out"): if direction not in ("in", "out"):
raise VMError("Expected a direction: in or out") raise VMError("Expected a direction: in or out")
func = getattr(self, f"vm_get_num_{direction}devices") func = getattr(self, f"vm_get_num_{direction}devices")
return func() res = self.call(func, ok_exp=lambda r: r >= 0)
return res
def get_device_description(self, index: int, direction: str = None) -> tuple: def get_device_description(self, index: int, direction: str = None) -> tuple:
"""Returns a tuple of device parameters""" """Returns a tuple of device parameters"""
@ -184,7 +206,8 @@ class Remote(CBindings):
name = ct.create_unicode_buffer(256) name = ct.create_unicode_buffer(256)
hwid = ct.create_unicode_buffer(256) hwid = ct.create_unicode_buffer(256)
func = getattr(self, f"vm_get_desc_{direction}devices") func = getattr(self, f"vm_get_desc_{direction}devices")
func( self.call(
func,
ct.c_long(index), ct.c_long(index),
ct.byref(type_), ct.byref(type_),
ct.byref(name), ct.byref(name),
@ -195,7 +218,7 @@ class Remote(CBindings):
def get_level(self, type_: int, index: int) -> float: def get_level(self, type_: int, index: int) -> float:
"""Retrieves a single level value""" """Retrieves a single level value"""
val = ct.c_float() val = ct.c_float()
self.vm_get_level(ct.c_long(type_), ct.c_long(index), ct.byref(val)) self.call(self.vm_get_level, ct.c_long(type_), ct.c_long(index), ct.byref(val))
return val.value return val.value
def _get_levels(self) -> Iterable: def _get_levels(self) -> Iterable:
@ -216,7 +239,7 @@ class Remote(CBindings):
def get_midi_message(self): def get_midi_message(self):
n = ct.c_long(1024) n = ct.c_long(1024)
buf = ct.create_string_buffer(1024) buf = ct.create_string_buffer(1024)
res = self.vm_get_midi_message(ct.byref(buf), n) res = self.vm_get_midi_message(ct.byref(buf), n, ok_exp=lambda r: r >= 0)
if res > 0: if res > 0:
vals = tuple( vals = tuple(
grouper(3, (int.from_bytes(buf[i], "little") for i in range(res))) grouper(3, (int.from_bytes(buf[i], "little") for i in range(res)))
@ -228,15 +251,13 @@ class Remote(CBindings):
self.midi._most_recent = pitch self.midi._most_recent = pitch
self.midi._set(pitch, vel) self.midi._set(pitch, vel)
return True return True
elif res == -1 or res == -2:
raise CAPIError(f"VBVMR_GetMidiMessage returned {res}")
@script @script
def sendtext(self, script: str): def sendtext(self, script: str):
"""Sets many parameters from a script""" """Sets many parameters from a script"""
if len(script) > 48000: if len(script) > 48000:
raise ValueError("Script too large, max size 48kB") raise ValueError("Script too large, max size 48kB")
self.call(partial(self.vm_set_parameter_multi, script.encode())) self.call(self.vm_set_parameter_multi, script.encode())
time.sleep(self.DELAY * 5) time.sleep(self.DELAY * 5)
def apply(self, data: dict): def apply(self, data: dict):
@ -266,20 +287,19 @@ class Remote(CBindings):
try: try:
self.apply(self.configs[name]) self.apply(self.configs[name])
self.logger.info(f"Profile '{name}' applied!") self.logger.info(f"Profile '{name}' applied!")
except KeyError as e: except KeyError:
self.logger.error(("\n").join(error_msg)) self.logger.error(("\n").join(error_msg))
def logout(self) -> NoReturn: def logout(self) -> NoReturn:
"""Wait for dirty parameters to clear, then logout of the API""" """Wait for dirty parameters to clear, then logout of the API"""
self.clear_dirty() self.clear_dirty()
time.sleep(0.1) time.sleep(0.1)
res = self.vm_logout() self.call(self.vm_logout)
if res != 0:
raise CAPIError(f"VBVMR_Logout returned {res}")
self.logger.info(f"{type(self).__name__}: Successfully logged out of {self}") self.logger.info(f"{type(self).__name__}: Successfully logged out of {self}")
def end_thread(self): def end_thread(self):
self.running = False self.running = False
self.logger.debug("events thread stopped")
def __exit__(self, exc_type, exc_value, exc_traceback) -> NoReturn: def __exit__(self, exc_type, exc_value, exc_traceback) -> NoReturn:
"""teardown procedures""" """teardown procedures"""