remote is now our ABC - as it should be because it is the launching point of the interface.

it no longer inherits from CBindings.

move steps abstract method into Remote class. This is a much more meaningful abstraction - because it is the principle behaviour that distinguishes each kind of Remote.

add wrapper methods to CBindings. This provides a cleaner api for the Remote class.

import abc as namespace throughout the package.
This commit is contained in:
onyx-and-iris 2026-03-15 22:02:17 +00:00
parent 84b4426e44
commit 842feb2407
8 changed files with 139 additions and 62 deletions

View File

@ -1,5 +1,5 @@
import abc
import time import time
from abc import abstractmethod
from enum import IntEnum from enum import IntEnum
from math import log from math import log
from typing import Union from typing import Union
@ -22,7 +22,7 @@ class Bus(IRemote):
Defines concrete implementation for bus Defines concrete implementation for bus
""" """
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
pass pass

View File

@ -1,6 +1,5 @@
import ctypes as ct import ctypes as ct
import logging import logging
from abc import ABCMeta
from ctypes.wintypes import CHAR, FLOAT, LONG, WCHAR from ctypes.wintypes import CHAR, FLOAT, LONG, WCHAR
from .error import CAPIError from .error import CAPIError
@ -9,11 +8,10 @@ from .inst import libc
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class CBindings(metaclass=ABCMeta): class CBindings:
""" """Class responsible for defining C function bindings.
C bindings defined here.
Maps expected ctype argument and res types for each binding. Wrapper methods are provided for each C function to handle error checking and logging.
""" """
logger_cbindings = logger.getChild('CBindings') logger_cbindings = logger.getChild('CBindings')
@ -111,7 +109,8 @@ class CBindings(metaclass=ABCMeta):
bind_get_midi_message.restype = LONG bind_get_midi_message.restype = LONG
bind_get_midi_message.argtypes = [ct.POINTER(CHAR * 1024), LONG] bind_get_midi_message.argtypes = [ct.POINTER(CHAR * 1024), LONG]
def call(self, func, *args, ok=(0,), ok_exp=None): def _call(self, func, *args, ok=(0,), ok_exp=None):
"""Call a C function and handle errors."""
try: try:
res = func(*args) res = func(*args)
if ok_exp is None: if ok_exp is None:
@ -123,3 +122,93 @@ class CBindings(metaclass=ABCMeta):
except CAPIError as e: except CAPIError as e:
self.logger_cbindings.exception(f'{type(e).__name__}: {e}') self.logger_cbindings.exception(f'{type(e).__name__}: {e}')
raise raise
def login(self, **kwargs):
"""Login to Voicemeeter API"""
return self._call(self.bind_login, **kwargs)
def logout(self):
"""Logout from Voicemeeter API"""
return self._call(self.bind_logout)
def run_voicemeeter(self, value):
"""Run Voicemeeter with specified type"""
return self._call(self.bind_run_voicemeeter, value)
def get_voicemeeter_type(self, type_ref):
"""Get Voicemeeter type"""
return self._call(self.bind_get_voicemeeter_type, type_ref)
def get_voicemeeter_version(self, version_ref):
"""Get Voicemeeter version"""
return self._call(self.bind_get_voicemeeter_version, version_ref)
def is_parameters_dirty(self, **kwargs):
"""Check if parameters are dirty"""
return self._call(self.bind_is_parameters_dirty, **kwargs)
def macro_button_is_dirty(self, **kwargs):
"""Check if macro button parameters are dirty"""
if hasattr(self, 'bind_macro_button_is_dirty'):
return self._call(self.bind_macro_button_is_dirty, **kwargs)
raise AttributeError('macro_button_is_dirty not available')
def get_parameter_float(self, param_name, value_ref):
"""Get float parameter value"""
return self._call(self.bind_get_parameter_float, param_name, value_ref)
def set_parameter_float(self, param_name, value):
"""Set float parameter value"""
return self._call(self.bind_set_parameter_float, param_name, value)
def get_parameter_string_w(self, param_name, buffer_ref):
"""Get string parameter value (Unicode)"""
return self._call(self.bind_get_parameter_string_w, param_name, buffer_ref)
def set_parameter_string_w(self, param_name, value):
"""Set string parameter value (Unicode)"""
return self._call(self.bind_set_parameter_string_w, param_name, value)
def macro_button_get_status(self, id_, state_ref, mode):
"""Get macro button status"""
if hasattr(self, 'bind_macro_button_get_status'):
return self._call(self.bind_macro_button_get_status, id_, state_ref, mode)
raise AttributeError('macro_button_get_status not available')
def macro_button_set_status(self, id_, state, mode):
"""Set macro button status"""
if hasattr(self, 'bind_macro_button_set_status'):
return self._call(self.bind_macro_button_set_status, id_, state, mode)
raise AttributeError('macro_button_set_status not available')
def get_level(self, type_, index, value_ref):
"""Get audio level"""
return self._call(self.bind_get_level, type_, index, value_ref)
def input_get_device_number(self, **kwargs):
"""Get number of input devices"""
return self._call(self.bind_input_get_device_number, **kwargs)
def output_get_device_number(self, **kwargs):
"""Get number of output devices"""
return self._call(self.bind_output_get_device_number, **kwargs)
def input_get_device_desc_w(self, index, type_ref, name_ref, hwid_ref):
"""Get input device description"""
return self._call(
self.bind_input_get_device_desc_w, index, type_ref, name_ref, hwid_ref
)
def output_get_device_desc_w(self, index, type_ref, name_ref, hwid_ref):
"""Get output device description"""
return self._call(
self.bind_output_get_device_desc_w, index, type_ref, name_ref, hwid_ref
)
def get_midi_message(self, buffer_ref, length, **kwargs):
"""Get MIDI message"""
return self._call(self.bind_get_midi_message, buffer_ref, length, **kwargs)
def set_parameters(self, script):
"""Set multiple parameters via script"""
return self._call(self.bind_set_parameters, script)

View File

@ -1,4 +1,4 @@
from abc import abstractmethod import abc
from typing import Union from typing import Union
from .iremote import IRemote from .iremote import IRemote
@ -7,19 +7,19 @@ from .iremote import IRemote
class Adapter(IRemote): class Adapter(IRemote):
"""Adapter to the common interface.""" """Adapter to the common interface."""
@abstractmethod @abc.abstractmethod
def ins(self): def ins(self):
pass pass
@abstractmethod @abc.abstractmethod
def outs(self): def outs(self):
pass pass
@abstractmethod @abc.abstractmethod
def input(self): def input(self):
pass pass
@abstractmethod @abc.abstractmethod
def output(self): def output(self):
pass pass

View File

@ -1,5 +1,4 @@
import logging import logging
from abc import abstractmethod
from enum import IntEnum from enum import IntEnum
from functools import cached_property from functools import cached_property
from typing import Iterable from typing import Iterable
@ -137,11 +136,6 @@ class FactoryBase(Remote):
def __str__(self) -> str: def __str__(self) -> str:
return f'Voicemeeter {self.kind}' return f'Voicemeeter {self.kind}'
@property
@abstractmethod
def steps(self):
pass
@cached_property @cached_property
def configs(self): def configs(self):
self._configs = configs(self.kind.name) self._configs = configs(self.kind.name)

View File

@ -1,11 +1,11 @@
import abc
import logging import logging
import time import time
from abc import ABCMeta, abstractmethod
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class IRemote(metaclass=ABCMeta): class IRemote(abc.ABC):
""" """
Common interface between base class and extended (higher) classes Common interface between base class and extended (higher) classes
@ -33,7 +33,7 @@ class IRemote(metaclass=ABCMeta):
cmd += (f'.{param}',) cmd += (f'.{param}',)
return ''.join(cmd) return ''.join(cmd)
@abstractmethod @abc.abstractmethod
def identifier(self): def identifier(self):
pass pass

View File

@ -1,8 +1,8 @@
import abc
import ctypes as ct import ctypes as ct
import logging import logging
import threading import threading
import time import time
from abc import abstractmethod
from queue import Queue from queue import Queue
from typing import Iterable, Optional, Union from typing import Iterable, Optional, Union
@ -19,12 +19,13 @@ from .util import deep_merge, grouper, polling, script, timeout
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class Remote(CBindings): class Remote(abc.ABC):
"""Base class responsible for wrapping the C Remote API""" """An abstract base class for Voicemeeter Remote API wrappers. Defines common methods and properties."""
DELAY = 0.001 DELAY = 0.001
def __init__(self, **kwargs): def __init__(self, **kwargs):
self._bindings = CBindings()
self.strip_mode = 0 self.strip_mode = 0
self.cache = {} self.cache = {}
self.midi = Midi() self.midi = Midi()
@ -52,10 +53,10 @@ class Remote(CBindings):
self.init_thread() self.init_thread()
return self return self
@abstractmethod @property
def __str__(self): @abc.abstractmethod
"""Ensure subclasses override str magic method""" def steps(self):
pass """Steps required to build the interface for this Voicemeeter kind"""
def init_thread(self): def init_thread(self):
"""Starts updates thread.""" """Starts updates thread."""
@ -76,7 +77,7 @@ class Remote(CBindings):
@timeout @timeout
def login(self) -> None: def login(self) -> None:
"""Login to the API, initialize dirty parameters""" """Login to the API, initialize dirty parameters"""
self.gui.launched = self.call(self.bind_login, ok=(0, 1)) == 0 self.gui.launched = self._bindings.login(ok=(0, 1)) == 0
if not self.gui.launched: if not self.gui.launched:
self.logger.info( self.logger.info(
'Voicemeeter engine running but GUI not launched. Launching the GUI now.' 'Voicemeeter engine running but GUI not launched. Launching the GUI now.'
@ -89,20 +90,20 @@ class Remote(CBindings):
value = KindId[kind_id.upper()].value value = KindId[kind_id.upper()].value
if BITS == 64 and self.bits == 64: if BITS == 64 and self.bits == 64:
value += 3 value += 3
self.call(self.bind_run_voicemeeter, value) self._bindings.run_voicemeeter(value)
@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.call(self.bind_get_voicemeeter_type, ct.byref(type_)) self._bindings.get_voicemeeter_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.call(self.bind_get_voicemeeter_version, ct.byref(ver)) self._bindings.get_voicemeeter_version(ct.byref(ver))
return '{}.{}.{}.{}'.format( return '{}.{}.{}.{}'.format(
(ver.value & 0xFF000000) >> 24, (ver.value & 0xFF000000) >> 24,
(ver.value & 0x00FF0000) >> 16, (ver.value & 0x00FF0000) >> 16,
@ -113,13 +114,13 @@ 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.call(self.bind_is_parameters_dirty, ok=(0, 1)) == 1 return self._bindings.is_parameters_dirty(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."""
try: try:
return self.call(self.bind_macro_button_is_dirty, ok=(0, 1)) == 1 return self._bindings.macro_button_is_dirty(ok=(0, 1)) == 1
except AttributeError as e: except AttributeError as e:
self.logger.exception(f'{type(e).__name__}: {e}') self.logger.exception(f'{type(e).__name__}: {e}')
raise CAPIError('VBVMR_MacroButton_IsDirty', -9) from e raise CAPIError('VBVMR_MacroButton_IsDirty', -9) from e
@ -149,10 +150,10 @@ class Remote(CBindings):
"""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.bind_get_parameter_string_w, param.encode(), ct.byref(buf)) self._bindings.get_parameter_string_w(param.encode(), ct.byref(buf))
else: else:
buf = ct.c_float() buf = ct.c_float()
self.call(self.bind_get_parameter_float, param.encode(), ct.byref(buf)) self._bindings.get_parameter_float(param.encode(), ct.byref(buf))
return buf.value return buf.value
def set(self, param: str, val: Union[str, float]) -> None: def set(self, param: str, val: Union[str, float]) -> None:
@ -160,12 +161,11 @@ 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._bindings.set_parameter_string_w(param.encode(), ct.c_wchar_p(val))
self.bind_set_parameter_string_w, param.encode(), ct.c_wchar_p(val)
)
else: else:
self.call( self._bindings.set_parameter_float(
self.bind_set_parameter_float, param.encode(), ct.c_float(float(val)) param.encode(),
ct.c_float(float(val)),
) )
self.cache[param] = val self.cache[param] = val
@ -174,8 +174,7 @@ class Remote(CBindings):
"""Gets a macrobutton parameter""" """Gets a macrobutton parameter"""
c_state = ct.c_float() c_state = ct.c_float()
try: try:
self.call( self._bindings.macro_button_get_status(
self.bind_macro_button_get_status,
ct.c_long(id_), ct.c_long(id_),
ct.byref(c_state), ct.byref(c_state),
ct.c_long(mode), ct.c_long(mode),
@ -189,8 +188,7 @@ class Remote(CBindings):
"""Sets a macrobutton parameter. Caches value""" """Sets a macrobutton parameter. Caches value"""
c_state = ct.c_float(float(val)) c_state = ct.c_float(float(val))
try: try:
self.call( self._bindings.macro_button_set_status(
self.bind_macro_button_set_status,
ct.c_long(id_), ct.c_long(id_),
c_state, c_state,
ct.c_long(mode), ct.c_long(mode),
@ -204,8 +202,8 @@ class Remote(CBindings):
"""Retrieves number of physical devices connected""" """Retrieves number of physical devices connected"""
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'bind_{direction}put_get_device_number') func = getattr(self._bindings, f'{direction}put_get_device_number')
res = self.call(func, ok_exp=lambda r: r >= 0) res = func(ok_exp=lambda r: r >= 0)
return res return res
def get_device_description(self, index: int, direction: str = None) -> tuple: def get_device_description(self, index: int, direction: str = None) -> tuple:
@ -215,9 +213,8 @@ class Remote(CBindings):
type_ = ct.c_long() type_ = ct.c_long()
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'bind_{direction}put_get_device_desc_w') func = getattr(self._bindings, f'{direction}put_get_device_desc_w')
self.call( func(
func,
ct.c_long(index), ct.c_long(index),
ct.byref(type_), ct.byref(type_),
ct.byref(name), ct.byref(name),
@ -228,9 +225,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.call( self._bindings.get_level(ct.c_long(type_), ct.c_long(index), ct.byref(val))
self.bind_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:
@ -248,8 +243,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.call( res = self._bindings.get_midi_message(
self.bind_get_midi_message,
ct.byref(buf), ct.byref(buf),
n, n,
ok=(-5, -6), # no data received from midi device ok=(-5, -6), # no data received from midi device
@ -272,7 +266,7 @@ class Remote(CBindings):
"""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(self.bind_set_parameters, script.encode()) self._bindings.set_parameters(script.encode())
time.sleep(self.DELAY * 5) time.sleep(self.DELAY * 5)
def apply(self, data: dict): def apply(self, data: dict):
@ -339,7 +333,7 @@ class Remote(CBindings):
def logout(self) -> None: def logout(self) -> None:
"""Logout of the API""" """Logout of the API"""
time.sleep(0.1) time.sleep(0.1)
self.call(self.bind_logout) self._bindings.logout()
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 __exit__(self, exc_type, exc_value, exc_traceback) -> None: def __exit__(self, exc_type, exc_value, exc_traceback) -> None:

View File

@ -1,5 +1,5 @@
import abc
import time import time
from abc import abstractmethod
from math import log from math import log
from typing import Union from typing import Union
@ -15,7 +15,7 @@ class Strip(IRemote):
Defines concrete implementation for strip Defines concrete implementation for strip
""" """
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
pass pass

View File

@ -1,4 +1,4 @@
from abc import abstractmethod import abc
from . import kinds from . import kinds
from .iremote import IRemote from .iremote import IRemote
@ -11,7 +11,7 @@ class VbanStream(IRemote):
Defines concrete implementation for vban stream Defines concrete implementation for vban stream
""" """
@abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
pass pass