42 Commits

Author SHA1 Message Date
708a7e6d8e reword 2023-08-05 13:05:29 +01:00
409d2deea9 patch bump 2023-08-05 13:02:56 +01:00
0ee3a223ec stopped() checks if stop_event object is None.
In case the events thread was not initiated.
2023-08-05 13:02:04 +01:00
6bfd18c1ac call on_midi_press()
only if midi.current == MIDI_BUTTON
2023-08-05 13:00:45 +01:00
103355d265 use Threading.Event object to terminate producer 2023-08-04 23:13:54 +01:00
09cb62ecfa patch bump 2023-08-04 16:21:21 +01:00
cddd04974b use walrus 2023-08-04 16:21:07 +01:00
50e95d6b8d remove unused imports 2023-08-04 15:19:49 +01:00
b33926f304 replace generator function with factory function
patch bump
2023-08-03 12:09:34 +01:00
58a26e89a8 Correct type annotations None type.
Fixes 'code unreachable'
2023-08-02 17:17:59 +01:00
e96151cd5a InstallError and CAPIError classes
now subclass VMError

minor version bump
2023-08-02 15:42:45 +01:00
6b79c091e8 should the loader attempt to load an invalid toml config
log as error but allow the loader to continue

patch bump
2023-08-01 18:18:02 +01:00
bf77ded007 fix bind name for get_num_devices.
patch bump
2023-07-21 12:56:12 +01:00
236125d095 patch bump 2023-07-21 12:51:01 +01:00
7841dfe10f add ButtonModes IntEnum to macrobutton
debug log getter,setter for macrobutton
2023-07-21 12:49:01 +01:00
bdf8dc489a rename bindings to match CAPI functions
use prefix bind_
2023-07-21 12:47:53 +01:00
160a6f89f9 patch bump 2023-07-20 11:12:34 +01:00
4fcb2f93ca remove unused import 2023-07-20 11:11:44 +01:00
8acd0b1385 add missing type annotations 2023-07-20 11:11:34 +01:00
89866bb87b remove redundant __str__ overrides 2023-07-20 11:10:37 +01:00
f996fc0d9c num_strip_levels, num_bus_levesl added to KindMaps 2023-07-20 11:10:05 +01:00
68177c3c6c md fix 2023-07-13 08:54:21 +01:00
54a1938694 Added Errors and Logging sections to README. 2023-07-13 08:50:41 +01:00
9a4205ce64 I don't think this is necessary on logout.
patch bump
2023-07-13 01:09:01 +01:00
9b2e38aab3 implement midi, text vban streams
kindmaps updated

factory tests updated.

closes #7
2023-07-12 09:45:33 +01:00
278566c2e0 deep_merge implemented
recursively merges dicts in profiles

patch bump
2023-07-12 04:52:42 +01:00
b0acde6a52 fix weird code? 2023-07-11 19:45:43 +01:00
07b04d16d8 add vban-in-3 example to extender configs 2023-07-11 19:39:49 +01:00
f854ec7875 Adds ability to extend one config with another
apply_config() checks for 'extends' in TOML config

2.3.0 section added to README

three example extender.toml configs added
2023-07-11 19:34:43 +01:00
5640f54e65 rethrow if not mdirty error code -9, immediately.
patch bump
2023-07-10 20:17:06 +01:00
4569e8c760 accept incoming change 2023-07-10 17:45:38 +01:00
5e39461966 2.2.0 section added to changelog
mino version bump
2023-07-10 16:20:59 +01:00
6de78a4037 check for error code -9 in clear_dirty()
re-raise error if not AttributeError
otherwise clear pdirty only

add -5,-6 response to ok in get_midi_message().
2023-07-10 16:20:13 +01:00
bafaa58507 extends error class
now accepts a custom message

fn_name and error code stored as class attributes
2023-07-10 15:36:38 +01:00
af368b4b0a patch bump 2023-07-10 15:18:11 +01:00
32527e37bd patch bump 2023-07-09 01:45:27 +01:00
c21b04e1a8 add version number to login logger.info string 2023-07-09 01:44:44 +01:00
76960f36d0 if a wrong user config is requested,
this error should be exposed to the consumer.

patch bump.
2023-07-08 07:57:39 +01:00
2849b37670 remove redundant if test 2023-07-04 19:52:55 +01:00
7732a26c40 issue where subprocess not inheriting virtual env
see SO python-subprocess-doesnt-inherit-virtual-environment
2023-07-04 19:52:24 +01:00
c1e23ab250 typo 2023-07-01 20:26:44 +01:00
c2daba1a62 when out of bounds values are passed, log warnings
bump to version 2.1.1

closes #6
2023-07-01 19:50:54 +01:00
24 changed files with 484 additions and 230 deletions

View File

@@ -11,7 +11,37 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
- [x] - [x]
## [2.1.0] - 2023-07-01 ## [2.3.2] - 2023-07-12
### Added
- vban.{instream,outstream} tuples now contain classes that represent MIDI and TEXT streams.
### Fixed
- apply_config() now performs a deep merge when extending a config with another.
## [2.3.0] - 2023-07-11
### Added
- user configs may now extend other user configs. check `config extends` section in README.
## [2.2.0] - 2023-07-10
### Added
- CAPIError class now stores fn_name, error code and message as class attributes.
### Changed
- macrobutton capi calls now use error code -9 on AttributeError (using an old version of the API).
### Fixed
- call to `self.vm_get_midi_message` now wrapped by {CBindings}.call.
## [2.1.1] - 2023-07-01
### Added ### Added
@@ -24,6 +54,10 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
- Recorder.loop removed from documentation - Recorder.loop removed from documentation
### Changed
- When out of bounds values are passed, log warnings instead of raising Errors. See [Issue #6][Issue 6].
## [2.0.0] - 2023-06-25 ## [2.0.0] - 2023-06-25
Where possible I've attempted to make the changes backwards compatible. The breaking changes affect two higher classes, Strip and Bus, as well as the behaviour of events. All other changes are additive or QOL aimed at giving more options to the developer. For example, every low-level CAPI call is now logged and error raised on Exception, you can now register callback functions as well as observer classes, extra examples to demonstrate different use cases etc. Where possible I've attempted to make the changes backwards compatible. The breaking changes affect two higher classes, Strip and Bus, as well as the behaviour of events. All other changes are additive or QOL aimed at giving more options to the developer. For example, every low-level CAPI call is now logged and error raised on Exception, you can now register callback functions as well as observer classes, extra examples to demonstrate different use cases etc.
@@ -376,3 +410,4 @@ I will move this commit to a separate branch in preparation for version 2.0.
- project packaged with poetry and added to pypi. - project packaged with poetry and added to pypi.
[issue 4]: https://github.com/onyx-and-iris/voicemeeter-api-python/issues/4 [issue 4]: https://github.com/onyx-and-iris/voicemeeter-api-python/issues/4
[Issue 6]: https://github.com/onyx-and-iris/voicemeeter-api-python/issues/6

View File

@@ -405,7 +405,7 @@ The following methods are available
The following properties are available The following properties are available
- `A1 - A5`: boolean - `A1 - A5`: boolean
- `B1 - A3`: boolean - `B1 - B3`: boolean
- `samplerate`: int, (22050, 24000, 32000, 44100, 48000, 88200, 96000, 176400, 192000) - `samplerate`: int, (22050, 24000, 32000, 44100, 48000, 88200, 96000, 176400, 192000)
- `bitresolution`: int, (8, 16, 24, 32) - `bitresolution`: int, (8, 16, 24, 32)
- `channel`: int, from 1 to 8 - `channel`: int, from 1 to 8
@@ -629,7 +629,9 @@ vm.option.sr = 48000
The following methods are available: The following methods are available:
- `buffer(driver, buffer)` : Set buffer size for particular audio driver. - `buffer(driver, buf)` : Set buffer size for particular audio driver.
- buf: int, from 128 to 2048
- driver:str, ("mme", "wdm", "ks", "asio")
example: example:
@@ -637,10 +639,6 @@ example:
vm.option.buffer("wdm", 512) vm.option.buffer("wdm", 512)
``` ```
driver defined as one of ("mme", "wdm", "ks", "asio")
buffer, from 128 to 2048
##### delay[i] ##### delay[i]
- `get()`: int - `get()`: int
@@ -693,8 +691,8 @@ vm.apply(
Or for each class you may do: Or for each class you may do:
```python ```python
vm.strip[0].apply(mute: True, gain: 3.2, A1: True) vm.strip[0].apply({"mute": True, "gain": 3.2, "A1": True})
vm.vban.outstream[0].apply(on: True, name: 'streamname', bit: 24) vm.vban.outstream[0].apply({"on": True, "name": "streamname", "bit": 24})
``` ```
## Config Files ## Config Files
@@ -703,7 +701,7 @@ vm.vban.outstream[0].apply(on: True, name: 'streamname', bit: 24)
You may load config files in TOML format. You may load config files in TOML format.
Three example configs have been included with the package. Remember to save Three example configs have been included with the package. Remember to save
current settings before loading a user config. To set one you may do: current settings before loading a user config. To load one you may do:
```python ```python
import voicemeeterlib import voicemeeterlib
@@ -711,7 +709,26 @@ with voicemeeterlib.api('banana') as vm:
vm.apply_config('example') vm.apply_config('example')
``` ```
will load a user config file at configs/banana/example.toml for Voicemeeter Banana. Your configs may be located in one of the following paths:
- \<current working directory\> / "configs" / kind_id
- \<user home directory\> / ".config" / "voicemeeter" / kind_id
- \<user home directory\> / "Documents" / "Voicemeeter" / "configs" / kind_id
If a config with the same name is located in multiple locations, only the first one found is loaded into memory, in the above order.
#### `config extends`
You may also load a config that extends another config with overrides or additional parameters.
You just need to define a key `extends` in the config TOML, that names the config to be extended.
Three example 'extender' configs are included with the repo. You may load them with:
```python
import voicemeeterlib
with voicemeeterlib.api('banana') as vm:
vm.apply_config('extender')
```
## Events ## Events
@@ -723,7 +740,7 @@ example:
import voicemeeterlib import voicemeeterlib
# Set event updates to occur every 50ms # Set event updates to occur every 50ms
# Listen for level updates only # Listen for level updates only
with voicemeeterlib.api('banana', ratelimit=0.05, ldirty=True}) as vm: with voicemeeterlib.api('banana', ratelimit=0.05, ldirty=True) as vm:
... ...
``` ```
@@ -810,19 +827,42 @@ vm.set('Strip[0].Gain', -3.6)
Access to lower level polling functions are provided with the following property objects: Access to lower level polling functions are provided with the following property objects:
#### `vm.pdirty` ##### `vm.pdirty`
True iff a parameter has been updated. True iff a parameter has been updated.
#### `vm.mdirty` ##### `vm.mdirty`
True iff a macrobutton has been updated. True iff a macrobutton has been updated.
#### `vm.ldirty` ##### `vm.ldirty`
True iff a level has been updated. True iff a level has been updated.
### Errors
- `errors.VMError`: Exception raised when general errors occur.
- `errors.InstallError`: Exception raised when installation errors occur.
- `errors.CAPIError`: Exception raised when the C-API returns error values.
- Error codes are stored in {Exception Class}.code. For a full list of error codes [check the VoicemeeterRemote header file][Voicemeeter Remote Header].
### Logging
It's possible to see the messages sent by the interface's setters and getters, may be useful for debugging.
example:
```python
import voicemeeterlib
logging.basicConfig(level=logging.DEBUG)
with voicemeeterlib.api("banana") as vm:
...
```
### Run tests ### Run tests
To run all tests: To run all tests:
@@ -834,3 +874,6 @@ pytest -v
### Official Documentation ### Official Documentation
- [Voicemeeter Remote C API](https://github.com/onyx-and-iris/Voicemeeter-SDK/blob/update-docs/VoicemeeterRemoteAPI.pdf) - [Voicemeeter Remote C API](https://github.com/onyx-and-iris/Voicemeeter-SDK/blob/update-docs/VoicemeeterRemoteAPI.pdf)
[Voicemeeter Remote Header]: https://github.com/onyx-and-iris/Voicemeeter-SDK/blob/update-docs/VoicemeeterRemote.h

View File

@@ -0,0 +1,12 @@
extends = "example"
[strip-0]
label = "strip0_extended"
A1 = false
gain = 0.0
[bus-0]
label = "bus0_extended"
mute = false
[vban-in-3]
name = "vban_extended"

View File

@@ -0,0 +1,12 @@
extends = "example"
[strip-0]
label = "strip0_extended"
A1 = false
gain = 0.0
[bus-0]
label = "bus0_extended"
mute = false
[vban-in-3]
name = "vban_extended"

View File

@@ -0,0 +1,12 @@
extends = "example"
[strip-0]
label = "strip0_extended"
A1 = false
gain = 0.0
[bus-0]
label = "bus0_extended"
mute = false
[vban-in-3]
name = "vban_extended"

View File

@@ -63,8 +63,6 @@ class Parser:
def interactive_mode(parser): def interactive_mode(parser):
while cmd := input("Please enter command (Press <Enter> to exit)\n"): while cmd := input("Please enter command (Press <Enter> to exit)\n"):
if not cmd:
break
if res := parser.parse((cmd,)): if res := parser.parse((cmd,)):
print(res) print(res)

View File

@@ -1,7 +1,5 @@
import json
import logging import logging
import time import time
from logging import config
import voicemeeterlib import voicemeeterlib
@@ -18,6 +16,7 @@ class App:
def __enter__(self): def __enter__(self):
self.vm.init_thread() self.vm.init_thread()
return self
def __exit__(self, exc_type, exc_value, traceback): def __exit__(self, exc_type, exc_value, traceback):
self.vm.end_thread() self.vm.end_thread()

View File

@@ -14,19 +14,20 @@ class App:
self.vm.observer.add(self.on_midi) self.vm.observer.add(self.on_midi)
def on_midi(self): def on_midi(self):
self.get_info() if self.get_info() == self.MIDI_BUTTON:
self.on_midi_press() self.on_midi_press()
def get_info(self): def get_info(self):
current = self.vm.midi.current current = self.vm.midi.current
print(f"Value of midi button {current} is {self.vm.midi.get(current)}") print(f"Value of midi button {current} is {self.vm.midi.get(current)}")
return current
def on_midi_press(self): def on_midi_press(self):
"""if strip 3 level max > -40 and midi button 48 is pressed, then set trigger for macrobutton 0""" """if midi button 48 is pressed and strip 3 level max > -40, then set trigger for macrobutton 0"""
if ( if (
max(self.vm.strip[3].levels.postfader) > -40 self.vm.midi.get(self.MIDI_BUTTON) == 127
and self.vm.midi.get(self.MIDI_BUTTON) == 127 and max(self.vm.strip[3].levels.postfader) > -40
): ):
print( print(
f"Strip 3 level max is greater than -40 and midi button {self.MIDI_BUTTON} is pressed" f"Strip 3 level max is greater than -40 and midi button {self.MIDI_BUTTON} is pressed"

View File

@@ -1,15 +1,13 @@
[tool.poetry] [tool.poetry]
name = "voicemeeter-api" name = "voicemeeter-api"
version = "2.0.2" version = "2.4.4"
description = "A Python wrapper for the Voiceemeter API" description = "A Python wrapper for the Voiceemeter API"
authors = ["onyx-and-iris <code@onyxandiris.online>"] authors = ["onyx-and-iris <code@onyxandiris.online>"]
license = "MIT" license = "MIT"
readme = "README.md" readme = "README.md"
repository = "https://github.com/onyx-and-iris/voicemeeter-api-python" repository = "https://github.com/onyx-and-iris/voicemeeter-api-python"
packages = [ packages = [{ include = "voicemeeterlib" }]
{ include = "voicemeeterlib" },
]
[tool.poetry.dependencies] [tool.poetry.dependencies]
python = "^3.10" python = "^3.10"

View File

@@ -1,40 +1,41 @@
import subprocess import subprocess
import sys
from pathlib import Path from pathlib import Path
def ex_dsl(): def ex_dsl():
path = Path.cwd() / "examples" / "dsl" / "." scriptpath = Path.cwd() / "examples" / "dsl" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def ex_events(): def ex_events():
path = Path.cwd() / "examples" / "events" / "." scriptpath = Path.cwd() / "examples" / "events" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def ex_gui(): def ex_gui():
path = Path.cwd() / "examples" / "gui" / "." scriptpath = Path.cwd() / "examples" / "gui" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def ex_levels(): def ex_levels():
path = Path.cwd() / "examples" / "levels" / "." scriptpath = Path.cwd() / "examples" / "levels" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def ex_midi(): def ex_midi():
path = Path.cwd() / "examples" / "midi" / "." scriptpath = Path.cwd() / "examples" / "midi" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def ex_obs(): def ex_obs():
path = Path.cwd() / "examples" / "obs" / "." scriptpath = Path.cwd() / "examples" / "obs" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def ex_observer(): def ex_observer():
path = Path.cwd() / "examples" / "observer" / "." scriptpath = Path.cwd() / "examples" / "observer" / "."
subprocess.run(["py", str(path)]) subprocess.run([sys.executable, str(scriptpath)])
def test(): def test():

View File

@@ -22,7 +22,7 @@ class TestRemoteFactories:
assert len(vm.strip) == 3 assert len(vm.strip) == 3
assert len(vm.bus) == 2 assert len(vm.bus) == 2
assert len(vm.button) == 80 assert len(vm.button) == 80
assert len(vm.vban.instream) == 4 and len(vm.vban.outstream) == 4 assert len(vm.vban.instream) == 6 and len(vm.vban.outstream) == 5
@pytest.mark.skipif( @pytest.mark.skipif(
data.name != "banana", data.name != "banana",
@@ -42,7 +42,7 @@ class TestRemoteFactories:
assert len(vm.strip) == 5 assert len(vm.strip) == 5
assert len(vm.bus) == 5 assert len(vm.bus) == 5
assert len(vm.button) == 80 assert len(vm.button) == 80
assert len(vm.vban.instream) == 8 and len(vm.vban.outstream) == 8 assert len(vm.vban.instream) == 10 and len(vm.vban.outstream) == 9
@pytest.mark.skipif( @pytest.mark.skipif(
data.name != "potato", data.name != "potato",
@@ -63,4 +63,4 @@ class TestRemoteFactories:
assert len(vm.strip) == 8 assert len(vm.strip) == 8
assert len(vm.bus) == 8 assert len(vm.bus) == 8
assert len(vm.button) == 80 assert len(vm.button) == 80
assert len(vm.vban.instream) == 8 and len(vm.vban.outstream) == 8 assert len(vm.vban.instream) == 10 and len(vm.vban.outstream) == 9

View File

@@ -210,7 +210,7 @@ class BusLevel(IRemote):
def fget(x): def fget(x):
return round(20 * log(x, 10), 1) if x > 0 else -200.0 return round(20 * log(x, 10), 1) if x > 0 else -200.0
if self._remote.running and self._remote.event.ldirty: if not self._remote.stopped() and self._remote.event.ldirty:
vals = self._remote.cache["bus_level"][self.range[0] : self.range[-1]] vals = self._remote.cache["bus_level"][self.range[0] : self.range[-1]]
else: else:
vals = [self._remote.get_level(mode, i) for i in range(*self.range)] vals = [self._remote.get_level(mode, i) for i in range(*self.range)]
@@ -232,7 +232,7 @@ class BusLevel(IRemote):
Expected to be used in a callback only. Expected to be used in a callback only.
""" """
if self._remote.running: if not self._remote.stopped():
return any(self._remote._bus_comp[self.range[0] : self.range[-1]]) return any(self._remote._bus_comp[self.range[0] : self.range[-1]])
is_updated = isdirty is_updated = isdirty

View File

@@ -16,110 +16,110 @@ class CBindings(metaclass=ABCMeta):
Maps expected ctype argument and res types for each binding. Maps expected ctype argument and res types for each binding.
""" """
logger_cbindings = logger.getChild("Cbindings") logger_cbindings = logger.getChild("CBindings")
vm_login = libc.VBVMR_Login bind_login = libc.VBVMR_Login
vm_login.restype = LONG bind_login.restype = LONG
vm_login.argtypes = None bind_login.argtypes = None
vm_logout = libc.VBVMR_Logout bind_logout = libc.VBVMR_Logout
vm_logout.restype = LONG bind_logout.restype = LONG
vm_logout.argtypes = None bind_logout.argtypes = None
vm_runvm = libc.VBVMR_RunVoicemeeter bind_run_voicemeeter = libc.VBVMR_RunVoicemeeter
vm_runvm.restype = LONG bind_run_voicemeeter.restype = LONG
vm_runvm.argtypes = [LONG] bind_run_voicemeeter.argtypes = [LONG]
vm_get_type = libc.VBVMR_GetVoicemeeterType bind_get_voicemeeter_type = libc.VBVMR_GetVoicemeeterType
vm_get_type.restype = LONG bind_get_voicemeeter_type.restype = LONG
vm_get_type.argtypes = [ct.POINTER(LONG)] bind_get_voicemeeter_type.argtypes = [ct.POINTER(LONG)]
vm_get_version = libc.VBVMR_GetVoicemeeterVersion bind_get_voicemeeter_version = libc.VBVMR_GetVoicemeeterVersion
vm_get_version.restype = LONG bind_get_voicemeeter_version.restype = LONG
vm_get_version.argtypes = [ct.POINTER(LONG)] bind_get_voicemeeter_version.argtypes = [ct.POINTER(LONG)]
if hasattr(libc, "VBVMR_MacroButton_IsDirty"): if hasattr(libc, "VBVMR_MacroButton_IsDirty"):
vm_mdirty = libc.VBVMR_MacroButton_IsDirty bind_macro_button_is_dirty = libc.VBVMR_MacroButton_IsDirty
vm_mdirty.restype = LONG bind_macro_button_is_dirty.restype = LONG
vm_mdirty.argtypes = None bind_macro_button_is_dirty.argtypes = None
if hasattr(libc, "VBVMR_MacroButton_GetStatus"): if hasattr(libc, "VBVMR_MacroButton_GetStatus"):
vm_get_buttonstatus = libc.VBVMR_MacroButton_GetStatus bind_macro_button_get_status = libc.VBVMR_MacroButton_GetStatus
vm_get_buttonstatus.restype = LONG bind_macro_button_get_status.restype = LONG
vm_get_buttonstatus.argtypes = [LONG, ct.POINTER(FLOAT), LONG] bind_macro_button_get_status.argtypes = [LONG, ct.POINTER(FLOAT), LONG]
if hasattr(libc, "VBVMR_MacroButton_SetStatus"): if hasattr(libc, "VBVMR_MacroButton_SetStatus"):
vm_set_buttonstatus = libc.VBVMR_MacroButton_SetStatus bind_macro_button_set_status = libc.VBVMR_MacroButton_SetStatus
vm_set_buttonstatus.restype = LONG bind_macro_button_set_status.restype = LONG
vm_set_buttonstatus.argtypes = [LONG, FLOAT, LONG] bind_macro_button_set_status.argtypes = [LONG, FLOAT, LONG]
vm_pdirty = libc.VBVMR_IsParametersDirty bind_is_parameters_dirty = libc.VBVMR_IsParametersDirty
vm_pdirty.restype = LONG bind_is_parameters_dirty.restype = LONG
vm_pdirty.argtypes = None bind_is_parameters_dirty.argtypes = None
vm_get_parameter_float = libc.VBVMR_GetParameterFloat bind_get_parameter_float = libc.VBVMR_GetParameterFloat
vm_get_parameter_float.restype = LONG bind_get_parameter_float.restype = LONG
vm_get_parameter_float.argtypes = [ct.POINTER(CHAR), ct.POINTER(FLOAT)] bind_get_parameter_float.argtypes = [ct.POINTER(CHAR), ct.POINTER(FLOAT)]
vm_set_parameter_float = libc.VBVMR_SetParameterFloat bind_set_parameter_float = libc.VBVMR_SetParameterFloat
vm_set_parameter_float.restype = LONG bind_set_parameter_float.restype = LONG
vm_set_parameter_float.argtypes = [ct.POINTER(CHAR), FLOAT] bind_set_parameter_float.argtypes = [ct.POINTER(CHAR), FLOAT]
vm_get_parameter_string = libc.VBVMR_GetParameterStringW bind_get_parameter_string_w = libc.VBVMR_GetParameterStringW
vm_get_parameter_string.restype = LONG bind_get_parameter_string_w.restype = LONG
vm_get_parameter_string.argtypes = [ct.POINTER(CHAR), ct.POINTER(WCHAR * 512)] bind_get_parameter_string_w.argtypes = [ct.POINTER(CHAR), ct.POINTER(WCHAR * 512)]
vm_set_parameter_string = libc.VBVMR_SetParameterStringW bind_set_parameter_string_w = libc.VBVMR_SetParameterStringW
vm_set_parameter_string.restype = LONG bind_set_parameter_string_w.restype = LONG
vm_set_parameter_string.argtypes = [ct.POINTER(CHAR), ct.POINTER(WCHAR)] bind_set_parameter_string_w.argtypes = [ct.POINTER(CHAR), ct.POINTER(WCHAR)]
vm_set_parameter_multi = libc.VBVMR_SetParameters bind_set_parameters = libc.VBVMR_SetParameters
vm_set_parameter_multi.restype = LONG bind_set_parameters.restype = LONG
vm_set_parameter_multi.argtypes = [ct.POINTER(CHAR)] bind_set_parameters.argtypes = [ct.POINTER(CHAR)]
vm_get_level = libc.VBVMR_GetLevel bind_get_level = libc.VBVMR_GetLevel
vm_get_level.restype = LONG bind_get_level.restype = LONG
vm_get_level.argtypes = [LONG, LONG, ct.POINTER(FLOAT)] bind_get_level.argtypes = [LONG, LONG, ct.POINTER(FLOAT)]
vm_get_num_indevices = libc.VBVMR_Input_GetDeviceNumber bind_input_get_device_number = libc.VBVMR_Input_GetDeviceNumber
vm_get_num_indevices.restype = LONG bind_input_get_device_number.restype = LONG
vm_get_num_indevices.argtypes = None bind_input_get_device_number.argtypes = None
vm_get_desc_indevices = libc.VBVMR_Input_GetDeviceDescW bind_input_get_device_desc_w = libc.VBVMR_Input_GetDeviceDescW
vm_get_desc_indevices.restype = LONG bind_input_get_device_desc_w.restype = LONG
vm_get_desc_indevices.argtypes = [ bind_input_get_device_desc_w.argtypes = [
LONG, LONG,
ct.POINTER(LONG), ct.POINTER(LONG),
ct.POINTER(WCHAR * 256), ct.POINTER(WCHAR * 256),
ct.POINTER(WCHAR * 256), ct.POINTER(WCHAR * 256),
] ]
vm_get_num_outdevices = libc.VBVMR_Output_GetDeviceNumber bind_output_get_device_number = libc.VBVMR_Output_GetDeviceNumber
vm_get_num_outdevices.restype = LONG bind_output_get_device_number.restype = LONG
vm_get_num_outdevices.argtypes = None bind_output_get_device_number.argtypes = None
vm_get_desc_outdevices = libc.VBVMR_Output_GetDeviceDescW bind_output_get_device_desc_w = libc.VBVMR_Output_GetDeviceDescW
vm_get_desc_outdevices.restype = LONG bind_output_get_device_desc_w.restype = LONG
vm_get_desc_outdevices.argtypes = [ bind_output_get_device_desc_w.argtypes = [
LONG, LONG,
ct.POINTER(LONG), ct.POINTER(LONG),
ct.POINTER(WCHAR * 256), ct.POINTER(WCHAR * 256),
ct.POINTER(WCHAR * 256), ct.POINTER(WCHAR * 256),
] ]
vm_get_midi_message = libc.VBVMR_GetMidiMessage bind_get_midi_message = libc.VBVMR_GetMidiMessage
vm_get_midi_message.restype = LONG bind_get_midi_message.restype = LONG
vm_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):
try: try:
res = func(*args) res = func(*args)
if ok_exp is None: if ok_exp is None:
if res not in ok: if res not in ok:
raise CAPIError(f"{func.__name__} returned {res}") raise CAPIError(func.__name__, res)
elif not ok_exp(res): elif not ok_exp(res) and res not in ok:
raise CAPIError(f"{func.__name__} returned {res}") raise CAPIError(func.__name__, res)
return res return res
except CAPIError as e: except CAPIError as e:
self.logger_cbindings.exception(f"{type(e).__name__}: {e}") self.logger_cbindings.exception(str(e))
raise raise

View File

@@ -147,8 +147,13 @@ class Loader(metaclass=SingletonType):
self.logger.info( self.logger.info(
f"config file with name {identifier} already in memory, skipping.." f"config file with name {identifier} already in memory, skipping.."
) )
return False return
self.parser = dataextraction_factory(data) try:
self.parser = dataextraction_factory(data)
except tomllib.TOMLDecodeError as e:
ERR_MSG = (str(e), f"When attempting to load {identifier}.toml")
self.logger.error(f"{type(e).__name__}: {' '.join(ERR_MSG)}")
return
return True return True
def register(self, identifier, data=None): def register(self, identifier, data=None):

View File

@@ -1,10 +1,22 @@
class InstallError(Exception): class VMError(Exception):
"""Base VM Exception class. Raised when general errors occur."""
def __init__(self, msg):
self.message = msg
super().__init__(self.message)
def __str__(self):
return f"{type(self).__name__}: {self.message}"
class InstallError(VMError):
"""Exception raised when installation errors occur""" """Exception raised when installation errors occur"""
class CAPIError(Exception): class CAPIError(VMError):
"""Exception raised when the C-API returns error values""" """Exception raised when the C-API returns an error code"""
def __init__(self, fn_name, code, msg=None):
class VMError(Exception): self.fn_name = fn_name
"""Exception raised when general errors occur""" self.code = code
super(CAPIError, self).__init__(msg if msg else f"{fn_name} returned {code}")

View File

@@ -2,7 +2,7 @@ import logging
from abc import abstractmethod 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, NoReturn from typing import Iterable
from . import misc from . import misc
from .bus import request_bus_obj as bus from .bus import request_bus_obj as bus
@@ -51,7 +51,7 @@ class FactoryBuilder:
) )
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
def _pinfo(self, name: str) -> NoReturn: def _pinfo(self, name: str) -> None:
"""prints progress status for each step""" """prints progress status for each step"""
name = name.split("_")[1] name = name.split("_")[1]
self.logger.debug(self._info[int(getattr(self.BuilderProgress, name))]) self.logger.debug(self._info[int(getattr(self.BuilderProgress, name))])

View File

@@ -32,29 +32,37 @@ class KindMapClass(metaclass=SingletonType):
insert: int insert: int
@property @property
def phys_in(self): def phys_in(self) -> int:
return self.ins[0] return self.ins[0]
@property @property
def virt_in(self): def virt_in(self) -> int:
return self.ins[-1] return self.ins[-1]
@property @property
def phys_out(self): def phys_out(self) -> int:
return self.outs[0] return self.outs[0]
@property @property
def virt_out(self): def virt_out(self) -> int:
return self.outs[-1] return self.outs[-1]
@property @property
def num_strip(self): def num_strip(self) -> int:
return sum(self.ins) return sum(self.ins)
@property @property
def num_bus(self): def num_bus(self) -> int:
return sum(self.outs) return sum(self.outs)
@property
def num_strip_levels(self) -> int:
return 2 * self.phys_in + 8 * self.virt_in
@property
def num_bus_levels(self) -> int:
return 8 * (self.phys_out + self.virt_out)
def __str__(self) -> str: def __str__(self) -> str:
return self.name.capitalize() return self.name.capitalize()
@@ -64,7 +72,7 @@ class BasicMap(KindMapClass):
name: str name: str
ins: tuple = (2, 1) ins: tuple = (2, 1)
outs: tuple = (1, 1) outs: tuple = (1, 1)
vban: tuple = (4, 4) vban: tuple = (4, 4, 1, 1)
asio: tuple = (0, 0) asio: tuple = (0, 0)
insert: int = 0 insert: int = 0
@@ -74,7 +82,7 @@ class BananaMap(KindMapClass):
name: str name: str
ins: tuple = (3, 2) ins: tuple = (3, 2)
outs: tuple = (3, 2) outs: tuple = (3, 2)
vban: tuple = (8, 8) vban: tuple = (8, 8, 1, 1)
asio: tuple = (6, 8) asio: tuple = (6, 8)
insert: int = 22 insert: int = 22
@@ -84,7 +92,7 @@ class PotatoMap(KindMapClass):
name: str name: str
ins: tuple = (5, 3) ins: tuple = (5, 3)
outs: tuple = (5, 3) outs: tuple = (5, 3)
vban: tuple = (8, 8) vban: tuple = (8, 8, 1, 1)
asio: tuple = (10, 8) asio: tuple = (10, 8)
insert: int = 34 insert: int = 34

View File

@@ -1,5 +1,13 @@
from enum import IntEnum
from .iremote import IRemote from .iremote import IRemote
ButtonModes = IntEnum(
"ButtonModes",
"state stateonly trigger",
start=1,
)
class Adapter(IRemote): class Adapter(IRemote):
"""Adapter to the common interface.""" """Adapter to the common interface."""
@@ -8,9 +16,13 @@ class Adapter(IRemote):
pass pass
def getter(self, mode): def getter(self, mode):
self.logger.debug(f"getter: button[{self.index}].{ButtonModes(mode).name}")
return self._remote.get_buttonstatus(self.index, mode) return self._remote.get_buttonstatus(self.index, mode)
def setter(self, val, mode): def setter(self, mode, val):
self.logger.debug(
f"setter: button[{self.index}].{ButtonModes(mode).name}={val}"
)
self._remote.set_buttonstatus(self.index, val, mode) self._remote.set_buttonstatus(self.index, val, mode)
@@ -22,24 +34,24 @@ class MacroButton(Adapter):
@property @property
def state(self) -> bool: def state(self) -> bool:
return self.getter(1) == 1 return self.getter(ButtonModes.state) == 1
@state.setter @state.setter
def state(self, val): def state(self, val: bool):
self.setter(1 if val else 0, 1) self.setter(ButtonModes.state, 1 if val else 0)
@property @property
def stateonly(self) -> bool: def stateonly(self) -> bool:
return self.getter(2) == 1 return self.getter(ButtonModes.stateonly) == 1
@stateonly.setter @stateonly.setter
def stateonly(self, val): def stateonly(self, val: bool):
self.setter(1 if val else 0, 2) self.setter(ButtonModes.stateonly, 1 if val else 0)
@property @property
def trigger(self) -> bool: def trigger(self) -> bool:
return self.getter(3) == 1 return self.getter(ButtonModes.trigger) == 1
@trigger.setter @trigger.setter
def trigger(self, val): def trigger(self, val: bool):
self.setter(1 if val else 0, 3) self.setter(ButtonModes.trigger, 1 if val else 0)

View File

@@ -1,6 +1,5 @@
from typing import Optional from typing import Optional
from .error import VMError
from .iremote import IRemote from .iremote import IRemote
from .kinds import kinds_all from .kinds import kinds_all
@@ -196,7 +195,7 @@ class Option(IRemote):
def sr(self, val: int): def sr(self, val: int):
opts = (44100, 48000, 88200, 96000, 176400, 192000) opts = (44100, 48000, 88200, 96000, 176400, 192000)
if val not in opts: if val not in opts:
raise VMError(f"Expected one of: {opts}") self.logger.warning(f"sr got: {val} but expected a value in {opts}")
self.setter("sr", val) self.setter("sr", val)
@property @property

View File

@@ -1,9 +1,10 @@
import ctypes as ct import ctypes as ct
import logging import logging
import threading
import time import time
from abc import abstractmethod from abc import abstractmethod
from queue import Queue from queue import Queue
from typing import Iterable, NoReturn, Optional, Union from typing import Iterable, Optional, Union
from .cbindings import CBindings from .cbindings import CBindings
from .error import CAPIError, VMError from .error import CAPIError, VMError
@@ -13,7 +14,7 @@ from .kinds import KindId
from .misc import Midi, VmGui from .misc import Midi, VmGui
from .subject import Subject from .subject import Subject
from .updater import Producer, Updater from .updater import Producer, Updater
from .util import grouper, polling, script from .util import deep_merge, grouper, polling, script
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -28,11 +29,11 @@ class Remote(CBindings):
self.cache = {} self.cache = {}
self.midi = Midi() self.midi = Midi()
self.subject = self.observer = Subject() self.subject = self.observer = Subject()
self.running = False
self.event = Event( self.event = Event(
{k: kwargs.pop(k) for k in ("pdirty", "mdirty", "midi", "ldirty")} {k: kwargs.pop(k) for k in ("pdirty", "mdirty", "midi", "ldirty")}
) )
self.gui = VmGui() self.gui = VmGui()
self.stop_event = None
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
for attr, val in kwargs.items(): for attr, val in kwargs.items():
@@ -52,49 +53,55 @@ class Remote(CBindings):
def init_thread(self): def init_thread(self):
"""Starts updates thread.""" """Starts updates thread."""
self.running = True
self.event.info() self.event.info()
self.logger.debug("initiating events thread") self.logger.debug("initiating events thread")
self.stop_event = threading.Event()
self.stop_event.clear()
queue = Queue() queue = Queue()
self.updater = Updater(self, queue) self.updater = Updater(self, queue)
self.updater.start() self.updater.start()
self.producer = Producer(self, queue) self.producer = Producer(self, queue, self.stop_event)
self.producer.start() self.producer.start()
def login(self) -> NoReturn: def stopped(self):
return self.stop_event is None or self.stop_event.is_set()
def login(self) -> None:
"""Login to the API, initialize dirty parameters""" """Login to the API, initialize dirty parameters"""
self.gui.launched = self.call(self.vm_login, ok=(0, 1)) == 0 self.gui.launched = self.call(self.bind_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."
) )
self.run_voicemeeter(self.kind.name) self.run_voicemeeter(self.kind.name)
self.logger.info(f"{type(self).__name__}: Successfully logged into {self}") self.logger.info(
f"{type(self).__name__}: Successfully logged into {self} version {self.version}"
)
self.clear_dirty() self.clear_dirty()
def run_voicemeeter(self, kind_id: str) -> NoReturn: def run_voicemeeter(self, kind_id: str) -> None:
if kind_id not in (kind.name.lower() for kind in KindId): if kind_id not in (kind.name.lower() for kind in KindId):
raise VMError(f"Unexpected Voicemeeter type: '{kind_id}'") raise VMError(f"Unexpected Voicemeeter type: '{kind_id}'")
if kind_id == "potato" and bits == 8: if kind_id == "potato" and bits == 8:
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.call(self.vm_runvm, value) self.call(self.bind_run_voicemeeter, 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.call(self.vm_get_type, ct.byref(type_)) self.call(self.bind_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.vm_get_version, ct.byref(ver)) self.call(self.bind_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,
@@ -105,17 +112,21 @@ 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.vm_pdirty, ok=(0, 1)) == 1 return self.call(self.bind_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.vm_mdirty, ok=(0, 1)) == 1 return self.call(self.bind_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}")
ERR_MSG = (
"no bind for VBVMR_MacroButton_IsDirty.",
"are you using an old version of the API?",
)
raise CAPIError( raise CAPIError(
"no bind for VBVMR_MacroButton_IsDirty. are you using an old version of the API?" "VBVMR_MacroButton_IsDirty", -9, msg=" ".join(ERR_MSG)
) from e ) from e
@property @property
@@ -127,12 +138,14 @@ 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) -> NoReturn: def clear_dirty(self) -> None:
try: try:
while self.pdirty or self.mdirty: while self.pdirty or self.mdirty:
pass pass
except CAPIError: except CAPIError as e:
self.logger.error("no bind for mdirty, clearing pdirty only") if not (e.fn_name == "VBVMR_MacroButton_IsDirty" and e.code == -9):
raise
self.logger.error(f"{e} clearing pdirty only.")
while self.pdirty: while self.pdirty:
pass pass
@@ -141,59 +154,74 @@ 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.vm_get_parameter_string, param.encode(), ct.byref(buf)) self.call(self.bind_get_parameter_string_w, param.encode(), ct.byref(buf))
else: else:
buf = ct.c_float() buf = ct.c_float()
self.call(self.vm_get_parameter_float, param.encode(), ct.byref(buf)) self.call(self.bind_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]) -> None:
"""Sets a string or float parameter. Caches value""" """Sets a string or float parameter. Caches value"""
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.vm_set_parameter_string, param.encode(), ct.c_wchar_p(val)) self.call(
self.bind_set_parameter_string_w, param.encode(), ct.c_wchar_p(val)
)
else: else:
self.call( self.call(
self.vm_set_parameter_float, param.encode(), ct.c_float(float(val)) self.bind_set_parameter_float, param.encode(), ct.c_float(float(val))
) )
self.cache[param] = val self.cache[param] = val
@polling @polling
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() c_state = ct.c_float()
try: try:
self.call( self.call(
self.vm_get_buttonstatus, self.bind_macro_button_get_status,
ct.c_long(id), ct.c_long(id_),
ct.byref(state), ct.byref(c_state),
ct.c_long(mode), ct.c_long(mode),
) )
except AttributeError as e: except AttributeError as e:
self.logger.exception(f"{type(e).__name__}: {e}") self.logger.exception(f"{type(e).__name__}: {e}")
ERR_MSG = (
"no bind for VBVMR_MacroButton_GetStatus.",
"are you using an old version of the API?",
)
raise CAPIError( raise CAPIError(
"no bind for VBVMR_MacroButton_GetStatus. are you using an old version of the API?" "VBVMR_MacroButton_GetStatus", -9, msg=" ".join(ERR_MSG)
) from e ) from e
return int(state.value) return int(c_state.value)
def set_buttonstatus(self, id: int, state: int, mode: int) -> NoReturn: def set_buttonstatus(self, id_: int, val: int, mode: int) -> None:
"""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(val))
try: try:
self.call(self.vm_set_buttonstatus, ct.c_long(id), c_state, ct.c_long(mode)) self.call(
self.bind_macro_button_set_status,
ct.c_long(id_),
c_state,
ct.c_long(mode),
)
except AttributeError as e: except AttributeError as e:
self.logger.exception(f"{type(e).__name__}: {e}") self.logger.exception(f"{type(e).__name__}: {e}")
ERR_MSG = (
"no bind for VBVMR_MacroButton_SetStatus.",
"are you using an old version of the API?",
)
raise CAPIError( raise CAPIError(
"no bind for VBVMR_MacroButton_SetStatus. are you using an old version of the API?" "VBVMR_MacroButton_SetStatus", -9, msg=" ".join(ERR_MSG)
) from e ) 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:
"""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"vm_get_num_{direction}devices") func = getattr(self, f"bind_{direction}put_get_device_number")
res = self.call(func, ok_exp=lambda r: r >= 0) res = self.call(func, ok_exp=lambda r: r >= 0)
return res return res
@@ -204,7 +232,7 @@ 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"vm_get_desc_{direction}devices") func = getattr(self, f"bind_{direction}put_get_device_desc_w")
self.call( self.call(
func, func,
ct.c_long(index), ct.c_long(index),
@@ -217,7 +245,9 @@ 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.vm_get_level, ct.c_long(type_), ct.c_long(index), ct.byref(val)) self.call(
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:
@@ -227,18 +257,21 @@ class Remote(CBindings):
return ( return (
tuple( tuple(
self.get_level(self.strip_mode, i) self.get_level(self.strip_mode, i)
for i in range(2 * self.kind.phys_in + 8 * self.kind.virt_in) for i in range(self.kind.num_strip_levels)
),
tuple(
self.get_level(3, i)
for i in range(8 * (self.kind.phys_out + self.kind.virt_out))
), ),
tuple(self.get_level(3, i) for i in range(self.kind.num_bus_levels)),
) )
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, ok_exp=lambda r: r >= 0) res = self.call(
self.bind_get_midi_message,
ct.byref(buf),
n,
ok=(-5, -6), # no data received from midi device
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)))
@@ -256,7 +289,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.vm_set_parameter_multi, script.encode()) self.call(self.bind_set_parameters, script.encode())
time.sleep(self.DELAY * 5) time.sleep(self.DELAY * 5)
def apply(self, data: dict): def apply(self, data: dict):
@@ -279,28 +312,42 @@ class Remote(CBindings):
def apply_config(self, name): def apply_config(self, name):
"""applies a config from memory""" """applies a config from memory"""
error_msg = ( ERR_MSG = (
f"No config with name '{name}' is loaded into memory", f"No config with name '{name}' is loaded into memory",
f"Known configs: {list(self.configs.keys())}", f"Known configs: {list(self.configs.keys())}",
) )
try: try:
self.apply(self.configs[name]) config = self.configs[name]
self.logger.info(f"Profile '{name}' applied!") except KeyError as e:
except KeyError: self.logger.error(("\n").join(ERR_MSG))
self.logger.error(("\n").join(error_msg)) raise VMError(("\n").join(ERR_MSG)) from e
def logout(self) -> NoReturn: if "extends" in config:
"""Wait for dirty parameters to clear, then logout of the API""" extended = config["extends"]
self.clear_dirty() config = {
time.sleep(0.1) k: v
self.call(self.vm_logout) for k, v in deep_merge(self.configs[extended], config)
self.logger.info(f"{type(self).__name__}: Successfully logged out of {self}") if k not in ("extends")
}
self.logger.debug(
f"profile '{name}' extends '{extended}', profiles merged.."
)
self.apply(config)
self.logger.info(f"Profile '{name}' applied!")
def end_thread(self): def end_thread(self):
self.logger.debug("events thread shutdown started") if not self.stopped():
self.running = False self.logger.debug("events thread shutdown started")
self.stop_event.set()
self.producer.join() # wait for producer thread to complete cycle
def __exit__(self, exc_type, exc_value, exc_traceback) -> NoReturn: def logout(self) -> None:
"""Logout of the API"""
time.sleep(0.1)
self.call(self.bind_logout)
self.logger.info(f"{type(self).__name__}: Successfully logged out of {self}")
def __exit__(self, exc_type, exc_value, exc_traceback) -> None:
"""teardown procedures""" """teardown procedures"""
self.end_thread() self.end_thread()
self.logout() self.logout()

View File

@@ -364,7 +364,7 @@ class VirtualStrip(Strip):
self.setter("karaoke", val) self.setter("karaoke", val)
@property @property
def bass(self): def bass(self) -> float:
return round(self.getter("EQGain1"), 1) return round(self.getter("EQGain1"), 1)
@bass.setter @bass.setter
@@ -372,7 +372,7 @@ class VirtualStrip(Strip):
self.setter("EQGain1", val) self.setter("EQGain1", val)
@property @property
def mid(self): def mid(self) -> float:
return round(self.getter("EQGain2"), 1) return round(self.getter("EQGain2"), 1)
@mid.setter @mid.setter
@@ -382,7 +382,7 @@ class VirtualStrip(Strip):
med = mid med = mid
@property @property
def treble(self): def treble(self) -> float:
return round(self.getter("EQGain3"), 1) return round(self.getter("EQGain3"), 1)
high = treble high = treble
@@ -415,7 +415,7 @@ class StripLevel(IRemote):
def fget(x): def fget(x):
return round(20 * log(x, 10), 1) if x > 0 else -200.0 return round(20 * log(x, 10), 1) if x > 0 else -200.0
if self._remote.running and self._remote.event.ldirty: if not self._remote.stopped() and self._remote.event.ldirty:
vals = self._remote.cache["strip_level"][self.range[0] : self.range[-1]] vals = self._remote.cache["strip_level"][self.range[0] : self.range[-1]]
else: else:
vals = [self._remote.get_level(mode, i) for i in range(*self.range)] vals = [self._remote.get_level(mode, i) for i in range(*self.range)]
@@ -448,7 +448,7 @@ class StripLevel(IRemote):
Expected to be used in a callback only. Expected to be used in a callback only.
""" """
if self._remote.running: if not self._remote.stopped():
return any(self._remote._strip_comp[self.range[0] : self.range[-1]]) return any(self._remote._strip_comp[self.range[0] : self.range[-1]])
is_updated = isdirty is_updated = isdirty

View File

@@ -10,14 +10,18 @@ logger = logging.getLogger(__name__)
class Producer(threading.Thread): class Producer(threading.Thread):
"""Continously send job queue to the Updater thread at a rate of self._remote.ratelimit.""" """Continously send job queue to the Updater thread at a rate of self._remote.ratelimit."""
def __init__(self, remote, queue): def __init__(self, remote, queue, stop_event):
super().__init__(name="producer", daemon=True) super().__init__(name="producer", daemon=False)
self._remote = remote self._remote = remote
self.queue = queue self.queue = queue
self.stop_event = stop_event
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
def stopped(self):
return self.stop_event.is_set()
def run(self): def run(self):
while self._remote.running: while not self.stopped():
if self._remote.event.pdirty: if self._remote.event.pdirty:
self.queue.put("pdirty") self.queue.put("pdirty")
if self._remote.event.mdirty: if self._remote.event.mdirty:
@@ -36,10 +40,8 @@ class Updater(threading.Thread):
super().__init__(name="updater", daemon=True) super().__init__(name="updater", daemon=True)
self._remote = remote self._remote = remote
self.queue = queue self.queue = queue
self._remote._strip_comp = [False] * ( self._remote._strip_comp = [False] * (self._remote.kind.num_strip_levels)
2 * self._remote.kind.phys_in + 8 * self._remote.kind.virt_in self._remote._bus_comp = [False] * (self._remote.kind.num_bus_levels)
)
self._remote._bus_comp = [False] * (self._remote.kind.num_bus * 8)
( (
self._remote.cache["strip_level"], self._remote.cache["strip_level"],
self._remote.cache["bus_level"], self._remote.cache["bus_level"],
@@ -58,12 +60,7 @@ class Updater(threading.Thread):
Generate _strip_comp, _bus_comp and update level cache if ldirty. Generate _strip_comp, _bus_comp and update level cache if ldirty.
""" """
while True: while event := self.queue.get():
event = self.queue.get()
if event is None:
self.logger.debug(f"terminating {self.name} thread")
break
if event == "pdirty" and self._remote.pdirty: if event == "pdirty" and self._remote.pdirty:
self._remote.subject.notify(event) self._remote.subject.notify(event)
elif event == "mdirty" and self._remote.mdirty: elif event == "mdirty" and self._remote.mdirty:
@@ -75,3 +72,4 @@ class Updater(threading.Thread):
self._remote.cache["strip_level"] = self._remote._strip_buf self._remote.cache["strip_level"] = self._remote._strip_buf
self._remote.cache["bus_level"] = self._remote._bus_buf self._remote.cache["bus_level"] = self._remote._bus_buf
self._remote.subject.notify(event) self._remote.subject.notify(event)
self.logger.debug(f"terminating {self.name} thread")

View File

@@ -70,3 +70,17 @@ def grouper(n, iterable, fillvalue=None):
""" """
args = [iter(iterable)] * n args = [iter(iterable)] * n
return zip_longest(fillvalue=fillvalue, *args) return zip_longest(fillvalue=fillvalue, *args)
def deep_merge(dict1, dict2):
"""Generator function for deep merging two dicts"""
for k in set(dict1) | set(dict2):
if k in dict1 and k in dict2:
if isinstance(dict1[k], dict) and isinstance(dict2[k], dict):
yield k, dict(deep_merge(dict1[k], dict2[k]))
else:
yield k, dict2[k]
elif k in dict1:
yield k, dict1[k]
else:
yield k, dict2[k]

View File

@@ -1,7 +1,7 @@
from abc import abstractmethod from abc import abstractmethod
from .error import VMError
from .iremote import IRemote from .iremote import IRemote
from .kinds import kinds_all
class VbanStream(IRemote): class VbanStream(IRemote):
@@ -50,7 +50,9 @@ class VbanStream(IRemote):
@port.setter @port.setter
def port(self, val: int): def port(self, val: int):
if not 1024 <= val <= 65535: if not 1024 <= val <= 65535:
raise VMError("Expected value from 1024 to 65535") self.logger.warning(
f"port got: {val} but expected a value from 1024 to 65535"
)
self.setter("port", val) self.setter("port", val)
@property @property
@@ -61,7 +63,7 @@ class VbanStream(IRemote):
def sr(self, val: int): def sr(self, val: int):
opts = (11025, 16000, 22050, 24000, 32000, 44100, 48000, 64000, 88200, 96000) opts = (11025, 16000, 22050, 24000, 32000, 44100, 48000, 64000, 88200, 96000)
if val not in opts: if val not in opts:
raise VMError(f"Expected one of: {opts}") self.logger.warning(f"sr got: {val} but expected a value in {opts}")
self.setter("sr", val) self.setter("sr", val)
@property @property
@@ -71,7 +73,7 @@ class VbanStream(IRemote):
@channel.setter @channel.setter
def channel(self, val: int): def channel(self, val: int):
if not 1 <= val <= 8: if not 1 <= val <= 8:
raise VMError("Expected value from 1 to 8") self.logger.warning(f"channel got: {val} but expected a value from 1 to 8")
self.setter("channel", val) self.setter("channel", val)
@property @property
@@ -81,7 +83,7 @@ class VbanStream(IRemote):
@bit.setter @bit.setter
def bit(self, val: int): def bit(self, val: int):
if val not in (16, 24): if val not in (16, 24):
raise VMError("Expected value 16 or 24") self.logger.warning(f"bit got: {val} but expected value 16 or 24")
self.setter("bit", 1 if (val == 16) else 2) self.setter("bit", 1 if (val == 16) else 2)
@property @property
@@ -91,7 +93,7 @@ class VbanStream(IRemote):
@quality.setter @quality.setter
def quality(self, val: int): def quality(self, val: int):
if not 0 <= val <= 4: if not 0 <= val <= 4:
raise VMError("Expected value from 0 to 4") self.logger.warning(f"quality got: {val} but expected a value from 0 to 4")
self.setter("quality", val) self.setter("quality", val)
@property @property
@@ -101,7 +103,7 @@ class VbanStream(IRemote):
@route.setter @route.setter
def route(self, val: int): def route(self, val: int):
if not 0 <= val <= 8: if not 0 <= val <= 8:
raise VMError("Expected value from 0 to 8") self.logger.warning(f"route got: {val} but expected a value from 0 to 8")
self.setter("route", val) self.setter("route", val)
@@ -132,6 +134,18 @@ class VbanInstream(VbanStream):
return super(VbanInstream, self).bit return super(VbanInstream, self).bit
class VbanAudioInstream(VbanInstream):
"""Represents a VBAN Audio Instream"""
class VbanMidiInstream(VbanInstream):
"""Represents a VBAN Midi Instream"""
class VbanTextInstream(VbanInstream):
"""Represents a VBAN Text Instream"""
class VbanOutstream(VbanStream): class VbanOutstream(VbanStream):
""" """
class representing a vban outstream class representing a vban outstream
@@ -147,6 +161,42 @@ class VbanOutstream(VbanStream):
return "out" return "out"
class VbanAudioOutstream(VbanOutstream):
"""Represents a VBAN Audio Outstream"""
class VbanMidiOutstream(VbanOutstream):
"""Represents a VBAN Midi Outstream"""
def _make_stream_pair(remote, kind):
num_instream, num_outstream, num_midi, num_text = kind.vban
def _make_cls(i, dir):
match dir:
case "in":
if i < num_instream:
return VbanAudioInstream(remote, i)
elif i < num_instream + num_midi:
return VbanMidiInstream(remote, i)
else:
return VbanTextInstream(remote, i)
case "out":
if i < num_outstream:
return VbanAudioOutstream(remote, i)
else:
return VbanMidiOutstream(remote, i)
return (
tuple(_make_cls(i, "in") for i in range(num_instream + num_midi + num_text)),
tuple(_make_cls(i, "out") for i in range(num_outstream + num_midi)),
)
def _make_stream_pairs(remote):
return {kind.name: _make_stream_pair(remote, kind) for kind in kinds_all}
class Vban: class Vban:
""" """
class representing the vban module class representing the vban module
@@ -156,9 +206,7 @@ class Vban:
def __init__(self, remote): def __init__(self, remote):
self.remote = remote self.remote = remote
num_instream, num_outstream = remote.kind.vban self.instream, self.outstream = _make_stream_pairs(remote)[remote.kind.name]
self.instream = tuple(VbanInstream(remote, i) for i in range(num_instream))
self.outstream = tuple(VbanOutstream(remote, i) for i in range(num_outstream))
def enable(self): def enable(self):
self.remote.set("vban.Enable", 1) self.remote.set("vban.Enable", 1)