diff --git a/README.md b/README.md new file mode 100644 index 0000000..77eb411 --- /dev/null +++ b/README.md @@ -0,0 +1,299 @@ +[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black) +[![Imports: isort](https://img.shields.io/badge/%20imports-isort-%231674b1?style=flat&labelColor=ef8336)](https://pycqa.github.io/isort/) +![Tests Status](./tests/basic.svg?dummy=8484744) +![Tests Status](./tests/banana.svg?dummy=8484744) +![Tests Status](./tests/potato.svg?dummy=8484744) + +# Python Wrapper for Voicemeeter API + +This package offers a Python interface for the Voicemeeter Remote C API. + +For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md) + +## Tested against + +- Basic 1.0.8.2 +- Banana 2.0.6.2 +- Potato 3.0.2.2 + +## Requirements + +- [Voicemeeter](https://voicemeeter.com/) +- Python 3.11 or greater + +## Installation + +### `Pip` + +Install voicemeeter-api package from your console + +`pip install voicemeeter-api` + + +## `Use` + +Simplest use case, use a context manager to request a Remote class of a kind. + +Login and logout are handled for you in this scenario. + +#### `__main__.py` + +```python +import voicemeeterlib + + +class ManyThings: + def __init__(self, vm): + self.vm = vm + + def things(self): + self.vm.strip[0].label = "podmic" + self.vm.strip[0].mute = True + print( + f"strip 0 ({self.vm.strip[0].label}) has been set to {self.vm.strip[0].mute}" + ) + + def other_things(self): + info = ( + f"bus 3 gain has been set to {self.vm.bus[3].gain}", + f"bus 4 eq has been set to {self.vm.bus[4].eq}", + ) + self.vm.bus[3].gain = -6.3 + self.vm.bus[4].eq = True + print("\n".join(info)) + + +def main(): + with voicemeeterlib.api(kind_id) as vm: + do = ManyThings(vm) + do.things() + do.other_things() + + # set many parameters at once + vm.apply( + { + "strip-2": {"A1": True, "B1": True, "gain": -6.0}, + "bus-2": {"mute": True}, + "button-0": {"state": True}, + "vban-in-0": {"on": True}, + "vban-out-1": {"name": "streamname"}, + } + ) + + +if __name__ == "__main__": + kind_id = "banana" + + main() +``` + +Otherwise you must remember to call `vm.login()`, `vm.logout()` at the start/end of your code. + +## `kind_id` + +Pass the kind of Voicemeeter as an argument. kind_id may be: + +- `basic` +- `banana` +- `potato` + +## `Available commands` + +### Channels (strip/bus) + +The following properties exist for audio channels. + +- `mono`: boolean +- `mute`: boolean +- `gain`: float, from -60 to 12 +- `mc`, `k`: boolean +- `comp`, `gate`: float, from 0 to 10 +- `limit`: int, from -40 to 12 +- `A1 - A5`, `B1 - B3`: boolean +- `eq`: boolean +- `label`: string +- `device`: string +- `sr`: int + +example: + +```python +vm.strip[3].gain = 3.7 +print(strip[0].label) + +vm.bus[4].mono = true +``` + +### Macrobuttons + +Three modes defined: state, stateonly and trigger. + +- `state`: boolean +- `stateonly`: boolean +- `trigger`: boolean + +example: + +```python +vm.button[37].state = true +vm.button[55].trigger = false +``` + +### Recorder + +The following methods are Available + +- `play()` +- `stop()` +- `pause()` +- `record()` +- `ff()` +- `rew()` + The following properties accept boolean values. +- `loop`: boolean +- `A1 - A5`: boolean +- `B1 - A3`: boolean + Load accepts a string: +- `load`: string + +example: + +```python +vm.recorder.play() +vm.recorder.stop() + +# Enable loop play +vm.recorder.loop = True + +# Disable recorder out channel B2 +vm.recorder.B2 = False + +# filepath as raw string +vm.recorder.load(r'C:\music\mytune.mp3') +``` + +### VBAN + +- `vm.vban.enable()` `vm.vban.disable()` Turn VBAN on or off + +For each vban in/out stream the following properties are defined: + +- `on`: boolean +- `name`: string +- `ip`: string +- `port`: int, range from 1024 to 65535 +- `sr`: int, (11025, 16000, 22050, 24000, 32000, 44100, 48000, 64000, 88200, 96000) +- `channel`: int, from 1 to 8 +- `bit`: int, 16 or 24 +- `quality`: int, from 0 to 4 +- `route`: int, from 0 to 8 + +SR, channel and bit are defined as readonly for instreams. Attempting to write to those parameters will throw an error. They are read and write for outstreams. + +example: + +```python +# turn VBAN on +vm.vban.enable() + +# turn on vban instream 0 +vm.vban.instream[0].on = True + +# set bit property for outstream 3 to 24 +vm.vban.outstream[3].bit = 24 +``` + +### Command + +Certain 'special' commands are defined by the API as performing actions rather than setting values. The following methods are available: + +- `show()` : Bring Voiceemeter GUI to the front +- `shutdown()` : Shuts down the GUI +- `restart()` : Restart the audio engine + +The following properties are write only and accept boolean values. + +- `showvbanchat`: boolean +- `lock`: boolean + +example: + +```python +vm.command.restart() +vm.command.showvbanchat = true +``` + +### Multiple parameters + +- `apply` + Set many strip/bus/macrobutton/vban parameters at once, for example: + +```python +vm.apply( + { + "strip-2": {"A1": True, "B1": True, "gain": -6.0}, + "bus-2": {"mute": True}, + "button-0": {"state": True}, + "vban-in-0": {"on": True}, + "vban-out-1": {"name": "streamname"}, + } +) +``` + +Or for each class you may do: + +```python +vm.strip[0].apply(mute: true, gain: 3.2, A1: true) +vm.vban.outstream[0].apply(on: true, name: 'streamname', bit: 24) +``` + +## `Base Module` + +### Remote class + +Access to lower level Getters and Setters are provided with these functions: + +- `vm.get(param, is_string=false)`: For getting the value of any parameter. Set string to true if getting a property value expected to return a string. +- `vm.set(param, value)`: For setting the value of any parameter. + +Access to lower level polling functions are provided with these functions: + +- `vm.pdirty()`: Returns true if a parameter has been updated. +- `vm.mdirty()`: Returns true if a macrobutton has been updated. +- `vm.ldirty()`: Returns true if a level has been updated. + +example: + +```python +vm.get('Strip[2].Mute') +vm.set('Strip[4].Label', 'stripname') +vm.set('Strip[0].Gain', -3.6) +``` + +## Config Files + +`vm.apply_config('config')` + +You may load config files in TOML format. +Three example profiles have been included with the package. Remember to save +current settings before loading a profile. To set one you may do: + +```python +import voicemeeterlib +with voicemeeterlib.api('banana') as vm: + vm.apply_profile('config') +``` + +will load a config file at configs/banana/config.toml for Voicemeeter Banana. + +### Run tests + +To run all tests: + +``` +pytest -v +``` + +### Official Documentation + +- [Voicemeeter Remote C API](https://github.com/onyx-and-iris/Voicemeeter-SDK/blob/main/VoicemeeterRemoteAPI.pdf) diff --git a/__main__.py b/__main__.py new file mode 100644 index 0000000..b21dbc0 --- /dev/null +++ b/__main__.py @@ -0,0 +1,46 @@ +import voicemeeterlib + + +class ManyThings: + def __init__(self, vm): + self.vm = vm + + def things(self): + self.vm.strip[0].label = "podmic" + self.vm.strip[0].mute = True + print( + f"strip 0 ({self.vm.strip[0].label}) has been set to {self.vm.strip[0].mute}" + ) + + def other_things(self): + info = ( + f"bus 3 gain has been set to {self.vm.bus[3].gain}", + f"bus 4 eq has been set to {self.vm.bus[4].eq}", + ) + self.vm.bus[3].gain = -6.3 + self.vm.bus[4].eq = True + print("\n".join(info)) + + +def main(): + with voicemeeterlib.api(kind_id) as vm: + do = ManyThings(vm) + do.things() + do.other_things() + + # set many parameters at once + vm.apply( + { + "strip-2": {"A1": True, "B1": True, "gain": -6.0}, + "bus-2": {"mute": True}, + "button-0": {"state": True}, + "vban-in-0": {"on": True}, + "vban-out-1": {"name": "streamname"}, + } + ) + + +if __name__ == "__main__": + kind_id = "banana" + + main() diff --git a/poetry.lock b/poetry.lock new file mode 100644 index 0000000..4e93c34 --- /dev/null +++ b/poetry.lock @@ -0,0 +1,288 @@ +[[package]] +name = "atomicwrites" +version = "1.4.0" +description = "Atomic file writes." +category = "dev" +optional = false +python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*" + +[[package]] +name = "attrs" +version = "21.4.0" +description = "Classes Without Boilerplate" +category = "dev" +optional = false +python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*" + +[package.extras] +dev = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six", "mypy", "pytest-mypy-plugins", "zope.interface", "furo", "sphinx", "sphinx-notfound-page", "pre-commit", "cloudpickle"] +docs = ["furo", "sphinx", "zope.interface", "sphinx-notfound-page"] +tests = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six", "mypy", "pytest-mypy-plugins", "zope.interface", "cloudpickle"] +tests_no_zope = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six", "mypy", "pytest-mypy-plugins", "cloudpickle"] + +[[package]] +name = "black" +version = "22.3.0" +description = "The uncompromising code formatter." +category = "dev" +optional = false +python-versions = ">=3.6.2" + +[package.dependencies] +click = ">=8.0.0" +mypy-extensions = ">=0.4.3" +pathspec = ">=0.9.0" +platformdirs = ">=2" + +[package.extras] +colorama = ["colorama (>=0.4.3)"] +d = ["aiohttp (>=3.7.4)"] +jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"] +uvloop = ["uvloop (>=0.15.2)"] + +[[package]] +name = "click" +version = "8.1.3" +description = "Composable command line interface toolkit" +category = "dev" +optional = false +python-versions = ">=3.7" + +[package.dependencies] +colorama = {version = "*", markers = "platform_system == \"Windows\""} + +[[package]] +name = "colorama" +version = "0.4.4" +description = "Cross-platform colored terminal text." +category = "dev" +optional = false +python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*" + +[[package]] +name = "iniconfig" +version = "1.1.1" +description = "iniconfig: brain-dead simple config-ini parsing" +category = "dev" +optional = false +python-versions = "*" + +[[package]] +name = "isort" +version = "5.10.1" +description = "A Python utility / library to sort Python imports." +category = "dev" +optional = false +python-versions = ">=3.6.1,<4.0" + +[package.extras] +pipfile_deprecated_finder = ["pipreqs", "requirementslib"] +requirements_deprecated_finder = ["pipreqs", "pip-api"] +colors = ["colorama (>=0.4.3,<0.5.0)"] +plugins = ["setuptools"] + +[[package]] +name = "mypy-extensions" +version = "0.4.3" +description = "Experimental type system extensions for programs checked with the mypy typechecker." +category = "dev" +optional = false +python-versions = "*" + +[[package]] +name = "packaging" +version = "21.3" +description = "Core utilities for Python packages" +category = "dev" +optional = false +python-versions = ">=3.6" + +[package.dependencies] +pyparsing = ">=2.0.2,<3.0.5 || >3.0.5" + +[[package]] +name = "pathspec" +version = "0.9.0" +description = "Utility library for gitignore style pattern matching of file paths." +category = "dev" +optional = false +python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,>=2.7" + +[[package]] +name = "platformdirs" +version = "2.5.2" +description = "A small Python module for determining appropriate platform-specific dirs, e.g. a \"user data dir\"." +category = "dev" +optional = false +python-versions = ">=3.7" + +[package.extras] +docs = ["furo (>=2021.7.5b38)", "proselint (>=0.10.2)", "sphinx-autodoc-typehints (>=1.12)", "sphinx (>=4)"] +test = ["appdirs (==1.4.4)", "pytest-cov (>=2.7)", "pytest-mock (>=3.6)", "pytest (>=6)"] + +[[package]] +name = "pluggy" +version = "1.0.0" +description = "plugin and hook calling mechanisms for python" +category = "dev" +optional = false +python-versions = ">=3.6" + +[package.extras] +dev = ["pre-commit", "tox"] +testing = ["pytest", "pytest-benchmark"] + +[[package]] +name = "py" +version = "1.11.0" +description = "library with cross-python path, ini-parsing, io, code, log facilities" +category = "dev" +optional = false +python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*" + +[[package]] +name = "pyparsing" +version = "3.0.9" +description = "pyparsing module - Classes and methods to define and execute parsing grammars" +category = "dev" +optional = false +python-versions = ">=3.6.8" + +[package.extras] +diagrams = ["railroad-diagrams", "jinja2"] + +[[package]] +name = "pytest" +version = "7.1.2" +description = "pytest: simple powerful testing with Python" +category = "dev" +optional = false +python-versions = ">=3.7" + +[package.dependencies] +atomicwrites = {version = ">=1.0", markers = "sys_platform == \"win32\""} +attrs = ">=19.2.0" +colorama = {version = "*", markers = "sys_platform == \"win32\""} +iniconfig = "*" +packaging = "*" +pluggy = ">=0.12,<2.0" +py = ">=1.8.2" +tomli = ">=1.0.0" + +[package.extras] +testing = ["argcomplete", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "xmlschema"] + +[[package]] +name = "pytest-randomly" +version = "3.12.0" +description = "Pytest plugin to randomly order tests and control random.seed." +category = "dev" +optional = false +python-versions = ">=3.7" + +[package.dependencies] +pytest = "*" + +[[package]] +name = "tomli" +version = "2.0.1" +description = "A lil' TOML parser" +category = "dev" +optional = false +python-versions = ">=3.7" + +[metadata] +lock-version = "1.1" +python-versions = "^3.11" +content-hash = "2db696ec0337e9c38835928d3f15cd36c4dc2c9baa7d77e725b25e9ce6cc4539" + +[metadata.files] +atomicwrites = [ + {file = "atomicwrites-1.4.0-py2.py3-none-any.whl", hash = "sha256:6d1784dea7c0c8d4a5172b6c620f40b6e4cbfdf96d783691f2e1302a7b88e197"}, + {file = "atomicwrites-1.4.0.tar.gz", hash = "sha256:ae70396ad1a434f9c7046fd2dd196fc04b12f9e91ffb859164193be8b6168a7a"}, +] +attrs = [ + {file = "attrs-21.4.0-py2.py3-none-any.whl", hash = "sha256:2d27e3784d7a565d36ab851fe94887c5eccd6a463168875832a1be79c82828b4"}, + {file = "attrs-21.4.0.tar.gz", hash = "sha256:626ba8234211db98e869df76230a137c4c40a12d72445c45d5f5b716f076e2fd"}, +] +black = [ + {file = "black-22.3.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:2497f9c2386572e28921fa8bec7be3e51de6801f7459dffd6e62492531c47e09"}, + {file = "black-22.3.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:5795a0375eb87bfe902e80e0c8cfaedf8af4d49694d69161e5bd3206c18618bb"}, + {file = "black-22.3.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:e3556168e2e5c49629f7b0f377070240bd5511e45e25a4497bb0073d9dda776a"}, + {file = "black-22.3.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:67c8301ec94e3bcc8906740fe071391bce40a862b7be0b86fb5382beefecd968"}, + {file = "black-22.3.0-cp310-cp310-win_amd64.whl", hash = "sha256:fd57160949179ec517d32ac2ac898b5f20d68ed1a9c977346efbac9c2f1e779d"}, + {file = "black-22.3.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:cc1e1de68c8e5444e8f94c3670bb48a2beef0e91dddfd4fcc29595ebd90bb9ce"}, + {file = "black-22.3.0-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6d2fc92002d44746d3e7db7cf9313cf4452f43e9ea77a2c939defce3b10b5c82"}, + {file = "black-22.3.0-cp36-cp36m-win_amd64.whl", hash = "sha256:a6342964b43a99dbc72f72812bf88cad8f0217ae9acb47c0d4f141a6416d2d7b"}, + {file = "black-22.3.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:328efc0cc70ccb23429d6be184a15ce613f676bdfc85e5fe8ea2a9354b4e9015"}, + {file = "black-22.3.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:06f9d8846f2340dfac80ceb20200ea5d1b3f181dd0556b47af4e8e0b24fa0a6b"}, + {file = "black-22.3.0-cp37-cp37m-win_amd64.whl", hash = "sha256:ad4efa5fad66b903b4a5f96d91461d90b9507a812b3c5de657d544215bb7877a"}, + {file = "black-22.3.0-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:e8477ec6bbfe0312c128e74644ac8a02ca06bcdb8982d4ee06f209be28cdf163"}, + {file = "black-22.3.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:637a4014c63fbf42a692d22b55d8ad6968a946b4a6ebc385c5505d9625b6a464"}, + {file = "black-22.3.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:863714200ada56cbc366dc9ae5291ceb936573155f8bf8e9de92aef51f3ad0f0"}, + {file = "black-22.3.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:10dbe6e6d2988049b4655b2b739f98785a884d4d6b85bc35133a8fb9a2233176"}, + {file = "black-22.3.0-cp38-cp38-win_amd64.whl", hash = "sha256:cee3e11161dde1b2a33a904b850b0899e0424cc331b7295f2a9698e79f9a69a0"}, + {file = "black-22.3.0-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:5891ef8abc06576985de8fa88e95ab70641de6c1fca97e2a15820a9b69e51b20"}, + {file = "black-22.3.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:30d78ba6bf080eeaf0b7b875d924b15cd46fec5fd044ddfbad38c8ea9171043a"}, + {file = "black-22.3.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:ee8f1f7228cce7dffc2b464f07ce769f478968bfb3dd1254a4c2eeed84928aad"}, + {file = "black-22.3.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6ee227b696ca60dd1c507be80a6bc849a5a6ab57ac7352aad1ffec9e8b805f21"}, + {file = "black-22.3.0-cp39-cp39-win_amd64.whl", hash = "sha256:9b542ced1ec0ceeff5b37d69838106a6348e60db7b8fdd245294dc1d26136265"}, + {file = "black-22.3.0-py3-none-any.whl", hash = "sha256:bc58025940a896d7e5356952228b68f793cf5fcb342be703c3a2669a1488cb72"}, + {file = "black-22.3.0.tar.gz", hash = "sha256:35020b8886c022ced9282b51b5a875b6d1ab0c387b31a065b84db7c33085ca79"}, +] +click = [ + {file = "click-8.1.3-py3-none-any.whl", hash = "sha256:bb4d8133cb15a609f44e8213d9b391b0809795062913b383c62be0ee95b1db48"}, + {file = "click-8.1.3.tar.gz", hash = "sha256:7682dc8afb30297001674575ea00d1814d808d6a36af415a82bd481d37ba7b8e"}, +] +colorama = [ + {file = "colorama-0.4.4-py2.py3-none-any.whl", hash = "sha256:9f47eda37229f68eee03b24b9748937c7dc3868f906e8ba69fbcbdd3bc5dc3e2"}, + {file = "colorama-0.4.4.tar.gz", hash = "sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b"}, +] +iniconfig = [ + {file = "iniconfig-1.1.1-py2.py3-none-any.whl", hash = "sha256:011e24c64b7f47f6ebd835bb12a743f2fbe9a26d4cecaa7f53bc4f35ee9da8b3"}, + {file = "iniconfig-1.1.1.tar.gz", hash = "sha256:bc3af051d7d14b2ee5ef9969666def0cd1a000e121eaea580d4a313df4b37f32"}, +] +isort = [ + {file = "isort-5.10.1-py3-none-any.whl", hash = "sha256:6f62d78e2f89b4500b080fe3a81690850cd254227f27f75c3a0c491a1f351ba7"}, + {file = "isort-5.10.1.tar.gz", hash = "sha256:e8443a5e7a020e9d7f97f1d7d9cd17c88bcb3bc7e218bf9cf5095fe550be2951"}, +] +mypy-extensions = [ + {file = "mypy_extensions-0.4.3-py2.py3-none-any.whl", hash = "sha256:090fedd75945a69ae91ce1303b5824f428daf5a028d2f6ab8a299250a846f15d"}, + {file = "mypy_extensions-0.4.3.tar.gz", hash = "sha256:2d82818f5bb3e369420cb3c4060a7970edba416647068eb4c5343488a6c604a8"}, +] +packaging = [ + {file = "packaging-21.3-py3-none-any.whl", hash = "sha256:ef103e05f519cdc783ae24ea4e2e0f508a9c99b2d4969652eed6a2e1ea5bd522"}, + {file = "packaging-21.3.tar.gz", hash = "sha256:dd47c42927d89ab911e606518907cc2d3a1f38bbd026385970643f9c5b8ecfeb"}, +] +pathspec = [ + {file = "pathspec-0.9.0-py2.py3-none-any.whl", hash = "sha256:7d15c4ddb0b5c802d161efc417ec1a2558ea2653c2e8ad9c19098201dc1c993a"}, + {file = "pathspec-0.9.0.tar.gz", hash = "sha256:e564499435a2673d586f6b2130bb5b95f04a3ba06f81b8f895b651a3c76aabb1"}, +] +platformdirs = [ + {file = "platformdirs-2.5.2-py3-none-any.whl", hash = "sha256:027d8e83a2d7de06bbac4e5ef7e023c02b863d7ea5d079477e722bb41ab25788"}, + {file = "platformdirs-2.5.2.tar.gz", hash = "sha256:58c8abb07dcb441e6ee4b11d8df0ac856038f944ab98b7be6b27b2a3c7feef19"}, +] +pluggy = [ + {file = "pluggy-1.0.0-py2.py3-none-any.whl", hash = "sha256:74134bbf457f031a36d68416e1509f34bd5ccc019f0bcc952c7b909d06b37bd3"}, + {file = "pluggy-1.0.0.tar.gz", hash = "sha256:4224373bacce55f955a878bf9cfa763c1e360858e330072059e10bad68531159"}, +] +py = [ + {file = "py-1.11.0-py2.py3-none-any.whl", hash = "sha256:607c53218732647dff4acdfcd50cb62615cedf612e72d1724fb1a0cc6405b378"}, + {file = "py-1.11.0.tar.gz", hash = "sha256:51c75c4126074b472f746a24399ad32f6053d1b34b68d2fa41e558e6f4a98719"}, +] +pyparsing = [ + {file = "pyparsing-3.0.9-py3-none-any.whl", hash = "sha256:5026bae9a10eeaefb61dab2f09052b9f4307d44aee4eda64b309723d8d206bbc"}, + {file = "pyparsing-3.0.9.tar.gz", hash = "sha256:2b020ecf7d21b687f219b71ecad3631f644a47f01403fa1d1036b0c6416d70fb"}, +] +pytest = [ + {file = "pytest-7.1.2-py3-none-any.whl", hash = "sha256:13d0e3ccfc2b6e26be000cb6568c832ba67ba32e719443bfe725814d3c42433c"}, + {file = "pytest-7.1.2.tar.gz", hash = "sha256:a06a0425453864a270bc45e71f783330a7428defb4230fb5e6a731fde06ecd45"}, +] +pytest-randomly = [ + {file = "pytest-randomly-3.12.0.tar.gz", hash = "sha256:d60c2db71ac319aee0fc6c4110a7597d611a8b94a5590918bfa8583f00caccb2"}, + {file = "pytest_randomly-3.12.0-py3-none-any.whl", hash = "sha256:f4f2e803daf5d1ba036cc22bf4fe9dbbf99389ec56b00e5cba732fb5c1d07fdd"}, +] +tomli = [ + {file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"}, + {file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"}, +] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..544d6e5 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[tool.poetry] +name = "voicemeeter-api" +version = "0.1.0" +description = "A Python wrapper for the Voiceemeter API" +authors = ["onyx-and-iris "] +packages = [ + { include = "voicemeeterlib" }, +] + +[tool.poetry.dependencies] +python = "^3.11" + +[tool.poetry.dev-dependencies] +pytest = "^7.1.2" +pytest-randomly = "^3.12.0" +black = "^22.3.0" +isort = "^5.10.1" + +[build-system] +requires = ["poetry-core>=1.0.0"] +build-backend = "poetry.core.masonry.api" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..71f98a0 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,41 @@ +import random +import sys +from dataclasses import dataclass + +import voicemeeterlib +from voicemeeterlib.kinds import KindId, kinds_all +from voicemeeterlib.kinds import request_kind_map as kindmap + +# let's keep things random +kind_id = random.choice(tuple(kind_id.name.lower() for kind_id in KindId)) + +vmrs = {kind.name: voicemeeterlib.api(kind.name) for kind in kinds_all} +tests = vmrs[kind_id] +kind = kindmap(kind_id) + + +@dataclass +class Data: + """bounds data to map tests to a kind""" + + name: str = kind.name + phys_in: int = kind.ins[0] - 1 + virt_in: int = kind.ins[0] + kind.ins[1] - 1 + phys_out: int = kind.outs[0] - 1 + virt_out: int = kind.outs[0] + kind.outs[1] - 1 + vban_in: int = kind.vban[0] - 1 + vban_out: int = kind.vban[1] - 1 + button_lower: int = 0 + button_upper: int = 79 + + +data = Data() + + +def setup_module(): + print(f"\nRunning tests for kind [{data.name}]\n", file=sys.stdout) + tests.login() + + +def teardown_module(): + tests.logout() diff --git a/tests/banana.svg b/tests/banana.svg new file mode 100644 index 0000000..62ae6fd --- /dev/null +++ b/tests/banana.svg @@ -0,0 +1 @@ +tests: 103tests103 \ No newline at end of file diff --git a/tests/basic.svg b/tests/basic.svg new file mode 100644 index 0000000..fe77673 --- /dev/null +++ b/tests/basic.svg @@ -0,0 +1 @@ +tests: 99tests99 \ No newline at end of file diff --git a/tests/potato.svg b/tests/potato.svg new file mode 100644 index 0000000..ebdffb3 --- /dev/null +++ b/tests/potato.svg @@ -0,0 +1 @@ +tests: 107tests107 \ No newline at end of file diff --git a/tests/test_higher.py b/tests/test_higher.py new file mode 100644 index 0000000..15f3893 --- /dev/null +++ b/tests/test_higher.py @@ -0,0 +1,298 @@ +import pytest + +from tests import data, tests + + +@pytest.mark.parametrize("value", [False, True]) +class TestSetAndGetBoolHigher: + __test__ = True + + """strip tests, physical and virtual""" + + @pytest.mark.parametrize( + "index,param", + [ + (data.phys_in, "mute"), + (data.phys_in, "mono"), + (data.virt_in, "mc"), + (data.virt_in, "mono"), + ], + ) + def test_it_sets_and_gets_strip_bool_params(self, index, param, value): + setattr(tests.strip[index], param, value) + assert getattr(tests.strip[index], param) == value + + """ bus tests, physical and virtual """ + + @pytest.mark.parametrize( + "index,param", + [ + (data.phys_out, "eq"), + (data.phys_out, "mute"), + (data.virt_out, "eq_ab"), + (data.virt_out, "sel"), + ], + ) + def test_it_sets_and_gets_bus_bool_params(self, index, param, value): + setattr(tests.bus[index], param, value) + assert getattr(tests.bus[index], param) == value + + """ bus modes tests, physical and virtual """ + + @pytest.mark.parametrize( + "index,param", + [ + (data.phys_out, "normal"), + (data.phys_out, "amix"), + (data.phys_out, "rearonly"), + (data.virt_out, "normal"), + (data.virt_out, "upmix41"), + (data.virt_out, "composite"), + ], + ) + def test_it_sets_and_gets_bus_bool_params(self, index, param, value): + setattr(tests.bus[index].mode, param, value) + assert getattr(tests.bus[index].mode, param) == value + + """ macrobutton tests """ + + @pytest.mark.parametrize( + "index,param", + [(data.button_lower, "state"), (data.button_upper, "trigger")], + ) + def test_it_sets_and_gets_macrobutton_bool_params(self, index, param, value): + setattr(tests.button[index], param, value) + assert getattr(tests.button[index], param) == value + + """ vban instream tests """ + + @pytest.mark.parametrize( + "index,param", + [(data.vban_in, "on")], + ) + def test_it_sets_and_gets_vban_instream_bool_params(self, index, param, value): + setattr(tests.vban.instream[index], param, value) + assert getattr(tests.vban.instream[index], param) == value + + """ vban outstream tests """ + + @pytest.mark.parametrize( + "index,param", + [(data.vban_out, "on")], + ) + def test_it_sets_and_gets_vban_outstream_bool_params(self, index, param, value): + setattr(tests.vban.outstream[index], param, value) + assert getattr(tests.vban.outstream[index], param) == value + + """ command tests """ + + @pytest.mark.parametrize( + "param", + [("lock")], + ) + def test_it_sets_command_bool_params(self, param, value): + setattr(tests.command, param, value) + + """ recorder tests """ + + @pytest.mark.skipif( + data.name == "basic", + reason="Skip test if kind is basic", + ) + @pytest.mark.parametrize( + "param", + [("A1"), ("B2")], + ) + def test_it_sets_and_gets_recorder_bool_params(self, param, value): + setattr(tests.recorder, param, value) + assert getattr(tests.recorder, param) == value + + @pytest.mark.skipif( + data.name == "basic", + reason="Skip test if kind is basic", + ) + @pytest.mark.parametrize( + "param", + [("loop")], + ) + def test_it_sets_recorder_bool_params(self, param, value): + setattr(tests.recorder, param, value) + + +class TestSetAndGetIntHigher: + __test__ = True + + """strip tests, physical and virtual""" + + @pytest.mark.parametrize( + "index,param,value", + [ + (data.phys_in, "limit", -40), + (data.phys_in, "limit", 12), + (data.virt_in, "k", 0), + (data.virt_in, "k", 4), + ], + ) + def test_it_sets_and_gets_strip_bool_params(self, index, param, value): + setattr(tests.strip[index], param, value) + assert getattr(tests.strip[index], param) == value + + """ vban outstream tests """ + + @pytest.mark.parametrize( + "index,param,value", + [(data.vban_out, "sr", 48000)], + ) + def test_it_sets_and_gets_vban_outstream_bool_params(self, index, param, value): + setattr(tests.vban.outstream[index], param, value) + assert getattr(tests.vban.outstream[index], param) == value + + +class TestSetAndGetFloatHigher: + __test__ = True + + """strip tests, physical and virtual""" + + @pytest.mark.parametrize( + "index,param,value", + [ + (data.phys_in, "gain", -3.6), + (data.virt_in, "gain", 5.8), + (data.phys_in, "comp", 0.0), + (data.virt_in, "comp", 8.2), + (data.phys_in, "gate", 2.3), + (data.virt_in, "gate", 6.7), + ], + ) + def test_it_sets_and_gets_strip_float_params(self, index, param, value): + setattr(tests.strip[index], param, value) + assert getattr(tests.strip[index], param) == value + + @pytest.mark.parametrize( + "index,value", + [(data.phys_in, 2), (data.phys_in, 2), (data.virt_in, 8), (data.virt_in, 8)], + ) + def test_it_gets_prefader_levels_and_compares_length_of_array(self, index, value): + assert len(tests.strip[index].levels.prefader) == value + + @pytest.mark.parametrize( + "index,value", + [(data.phys_in, 2), (data.phys_in, 2), (data.virt_in, 8), (data.virt_in, 8)], + ) + def test_it_gets_postmute_levels_and_compares_length_of_array(self, index, value): + assert len(tests.strip[index].levels.postmute) == value + + @pytest.mark.skipif( + data.name != "potato", + reason="Only test if logged into Potato version", + ) + @pytest.mark.parametrize( + "index, j, value", + [ + (data.phys_in, 0, -20.7), + (data.virt_in, 3, -60), + (data.virt_in, 4, 3.6), + (data.phys_in, 4, -12.7), + ], + ) + def test_it_sets_and_gets_strip_gainlayer_values(self, index, j, value): + tests.strip[index].gainlayer[j].gain = value + assert tests.strip[index].gainlayer[j].gain == value + + """ strip tests, virtual """ + + @pytest.mark.parametrize( + "index, param, value", + [ + (data.virt_in, "treble", -1.6), + (data.virt_in, "mid", 5.8), + (data.virt_in, "bass", -8.1), + ], + ) + def test_it_sets_and_gets_strip_eq_params(self, index, param, value): + setattr(tests.strip[index], param, value) + assert getattr(tests.strip[index], param) == value + + """ bus tests, physical and virtual """ + + @pytest.mark.parametrize( + "index, param, value", + [(data.phys_out, "gain", -3.6), (data.virt_out, "gain", 5.8)], + ) + def test_it_sets_and_gets_bus_float_params(self, index, param, value): + setattr(tests.bus[index], param, value) + assert getattr(tests.bus[index], param) == value + + @pytest.mark.parametrize( + "index,value", + [(data.phys_out, 8), (data.virt_out, 8)], + ) + def test_it_gets_prefader_levels_and_compares_length_of_array(self, index, value): + assert len(tests.bus[index].levels.all) == value + + +@pytest.mark.parametrize("value", ["test0", "test1"]) +class TestSetAndGetStringHigher: + __test__ = True + + """strip tests, physical and virtual""" + + @pytest.mark.parametrize( + "index, param", + [(data.phys_in, "label"), (data.virt_in, "label")], + ) + def test_it_sets_and_gets_strip_string_params(self, index, param, value): + setattr(tests.strip[index], param, value) + assert getattr(tests.strip[index], param) == value + + """ bus tests, physical and virtual """ + + @pytest.mark.parametrize( + "index, param", + [(data.phys_out, "label"), (data.virt_out, "label")], + ) + def test_it_sets_and_gets_bus_string_params(self, index, param, value): + setattr(tests.bus[index], param, value) + assert getattr(tests.bus[index], param) == value + + """ vban instream tests """ + + @pytest.mark.parametrize( + "index, param", + [(data.vban_in, "name")], + ) + def test_it_sets_and_gets_vban_instream_string_params(self, index, param, value): + setattr(tests.vban.instream[index], param, value) + assert getattr(tests.vban.instream[index], param) == value + + """ vban outstream tests """ + + @pytest.mark.parametrize( + "index, param", + [(data.vban_out, "name")], + ) + def test_it_sets_and_gets_vban_outstream_string_params(self, index, param, value): + setattr(tests.vban.outstream[index], param, value) + assert getattr(tests.vban.outstream[index], param) == value + + +@pytest.mark.parametrize("value", [False, True]) +class TestSetAndGetMacroButtonHigher: + __test__ = True + + """macrobutton tests""" + + @pytest.mark.parametrize( + "index, param", + [ + (0, "state"), + (39, "stateonly"), + (69, "trigger"), + (22, "stateonly"), + (45, "state"), + (65, "trigger"), + ], + ) + def test_it_sets_and_gets_macrobutton_params(self, index, param, value): + setattr(tests.button[index], param, value) + assert getattr(tests.button[index], param) == value diff --git a/tests/test_lower.py b/tests/test_lower.py new file mode 100644 index 0000000..b6a4072 --- /dev/null +++ b/tests/test_lower.py @@ -0,0 +1,78 @@ +import pytest + +from tests import data, tests + + +class TestSetAndGetFloatLower: + __test__ = True + + """VBVMR_SetParameterFloat, VBVMR_GetParameterFloat""" + + @pytest.mark.parametrize( + "param,value", + [ + (f"Strip[{data.phys_in}].Mute", 1), + (f"Bus[{data.virt_out}].Eq.on", 1), + (f"Strip[{data.phys_in}].Mute", 0), + (f"Bus[{data.virt_out}].Eq.on", 0), + ], + ) + def test_it_sets_and_gets_mute_eq_float_params(self, param, value): + tests.set(param, value) + assert (round(tests.get(param))) == value + + @pytest.mark.parametrize( + "param,value", + [ + (f"Strip[{data.phys_in}].Comp", 5.3), + (f"Strip[{data.virt_in}].Gain", -37.5), + (f"Bus[{data.virt_out}].Gain", -22.7), + ], + ) + def test_it_sets_and_gets_comp_gain_float_params(self, param, value): + tests.set(param, value) + assert (round(tests.get(param), 1)) == value + + +@pytest.mark.parametrize("value", ["test0", "test1"]) +class TestSetAndGetStringLower: + __test__ = True + + """VBVMR_SetParameterStringW, VBVMR_GetParameterStringW""" + + @pytest.mark.parametrize( + "param", + [(f"Strip[{data.phys_out}].label"), (f"Bus[{data.virt_out}].label")], + ) + def test_it_sets_and_gets_string_params(self, param, value): + tests.set(param, value) + assert tests.get(param, string=True) == value + + +@pytest.mark.parametrize("value", [0, 1]) +class TestMacroButtonsLower: + """VBVMR_MacroButton_SetStatus, VBVMR_MacroButton_GetStatus""" + + @pytest.mark.parametrize( + "index, mode", + [(33, 1), (49, 1)], + ) + def test_it_sets_and_gets_macrobuttons_state(self, index, mode, value): + tests.set_buttonstatus(index, value, mode) + assert tests.get_buttonstatus(index, mode) == value + + @pytest.mark.parametrize( + "index, mode", + [(14, 2), (12, 2)], + ) + def test_it_sets_and_gets_macrobuttons_stateonly(self, index, mode, value): + tests.set_buttonstatus(index, value, mode) + assert tests.get_buttonstatus(index, mode) == value + + @pytest.mark.parametrize( + "index, mode", + [(50, 3), (65, 3)], + ) + def test_it_sets_and_gets_macrobuttons_trigger(self, index, mode, value): + tests.set_buttonstatus(index, value, mode) + assert tests.get_buttonstatus(index, mode) == value diff --git a/voicemeeterlib/__init__.py b/voicemeeterlib/__init__.py new file mode 100644 index 0000000..68322ae --- /dev/null +++ b/voicemeeterlib/__init__.py @@ -0,0 +1,3 @@ +from .factory import request_remote_obj as api + +__ALL__ = ["api"] diff --git a/voicemeeterlib/base.py b/voicemeeterlib/base.py new file mode 100644 index 0000000..630933e --- /dev/null +++ b/voicemeeterlib/base.py @@ -0,0 +1,284 @@ +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() diff --git a/voicemeeterlib/bus.py b/voicemeeterlib/bus.py new file mode 100644 index 0000000..38f7054 --- /dev/null +++ b/voicemeeterlib/bus.py @@ -0,0 +1,207 @@ +import time +from abc import abstractmethod +from math import log +from typing import Union + +from .error import VMError +from .iremote import IRemote +from .meta import bus_mode_prop + + +class Bus(IRemote): + """ + Implements the common interface + + Defines concrete implementation for bus + """ + + @abstractmethod + def __str__(self): + pass + + @property + def identifier(self) -> str: + return f"bus[{self.index}]" + + @property + def mute(self) -> bool: + return self.getter("mute") == 1 + + @mute.setter + def mute(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("mute is a boolean parameter") + self.setter("mute", 1 if val else 0) + + @property + def mono(self) -> bool: + return self.getter("mono") == 1 + + @mono.setter + def mono(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("mono is a boolean parameter") + self.setter("mono", 1 if val else 0) + + @property + def eq(self) -> bool: + return self.getter("eq.On") == 1 + + @eq.setter + def eq(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("eq is a boolean parameter") + self.setter("eq.On", 1 if val else 0) + + @property + def eq_ab(self) -> bool: + return self.getter("eq.ab") == 1 + + @eq_ab.setter + def eq_ab(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("eq_ab is a boolean parameter") + self.setter("eq.ab", 1 if val else 0) + + @property + def sel(self) -> bool: + return self.getter("sel") == 1 + + @sel.setter + def sel(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("sel is a boolean parameter") + self.setter("sel", 1 if val else 0) + + @property + def label(self) -> str: + return self.getter("Label", is_string=True) + + @label.setter + def label(self, val: str): + if not isinstance(val, str): + raise VMError("label is a string parameter") + self.setter("Label", val) + + @property + def gain(self) -> float: + return round(self.getter("gain"), 1) + + @gain.setter + def gain(self, val: float): + self.setter("gain", val) + + def fadeto(self, target: float, time_: int): + self.setter("FadeTo", f"({target}, {time_})") + time.sleep(self._remote.DELAY) + + def fadeby(self, change: float, time_: int): + self.setter("FadeBy", f"({change}, {time_})") + time.sleep(self._remote.DELAY) + + +class PhysicalBus(Bus): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + @property + def device(self) -> str: + return self.getter("device.name", is_string=True) + + @property + def sr(self) -> int: + return int(self.getter("device.sr")) + + +class VirtualBus(Bus): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + +class BusLevel(IRemote): + def __init__(self, remote, index): + super().__init__(remote, index) + self.level_map = tuple( + (i, i + 8) + for i in range(0, (remote.kind.phys_out + remote.kind.virt_out) * 8, 8) + ) + + def getter(self): + """Returns a tuple of level values for the channel.""" + + def fget(i): + return round(20 * log(i, 10), 1) if i > 0 else -200.0 + + range_ = self.level_map[self.index] + return tuple(fget(i) for i in self._remote._bus_levels[range_[0] : range_[-1]]) + + @property + def identifier(self) -> str: + return f"Bus[{self.index}]" + + @property + def all(self) -> tuple: + return self.getter() + + @property + def updated(self) -> tuple: + return self._remote._bus_comp + + +def _make_bus_mode_mixin(): + """Creates a mixin of Bus Modes.""" + + def identifier(self) -> str: + return f"Bus[{self.index}].mode" + + return type( + "BusModeMixin", + (IRemote,), + { + "identifier": property(identifier), + **{ + mode: bus_mode_prop(mode) + for mode in [ + "normal", + "amix", + "bmix", + "repeat", + "composite", + "tvmix", + "upmix21", + "upmix41", + "upmix61", + "centeronly", + "lfeonly", + "rearonly", + ] + }, + }, + ) + + +def bus_factory(phys_bus, remote, i) -> Union[PhysicalBus, VirtualBus]: + """ + Factory method for buses + + Returns a physical or virtual bus subclass + """ + BUS_cls = PhysicalBus if phys_bus else VirtualBus + BUSMODEMIXIN_cls = _make_bus_mode_mixin() + return type( + f"{BUS_cls.__name__}{remote.kind}", + (BUS_cls,), + { + "levels": BusLevel(remote, i), + "mode": BUSMODEMIXIN_cls(remote, i), + }, + )(remote, i) + + +def request_bus_obj(phys_bus, remote, i) -> Bus: + """ + Bus entry point. Wraps factory method. + + Returns a reference to a bus subclass of a kind + """ + return bus_factory(phys_bus, remote, i) diff --git a/voicemeeterlib/cbindings.py b/voicemeeterlib/cbindings.py new file mode 100644 index 0000000..998d90a --- /dev/null +++ b/voicemeeterlib/cbindings.py @@ -0,0 +1,105 @@ +import ctypes as ct +from abc import ABCMeta +from ctypes.wintypes import CHAR, FLOAT, LONG, WCHAR + +from .error import CAPIError +from .inst import libc + + +class CBindings(metaclass=ABCMeta): + """ + C bindings defined here. + + Maps expected ctype argument and res types for each binding. + """ + + vm_login = libc.VBVMR_Login + vm_login.restype = LONG + vm_login.argtypes = None + + vm_logout = libc.VBVMR_Logout + vm_logout.restype = LONG + vm_logout.argtypes = None + + vm_runvm = libc.VBVMR_RunVoicemeeter + vm_runvm.restype = LONG + vm_runvm.argtypes = [LONG] + + vm_get_type = libc.VBVMR_GetVoicemeeterType + vm_get_type.restype = LONG + vm_get_type.argtypes = [ct.POINTER(LONG)] + + vm_get_version = libc.VBVMR_GetVoicemeeterVersion + vm_get_version.restype = LONG + vm_get_version.argtypes = [ct.POINTER(LONG)] + + vm_mdirty = libc.VBVMR_MacroButton_IsDirty + vm_mdirty.restype = LONG + vm_mdirty.argtypes = None + + vm_get_buttonstatus = libc.VBVMR_MacroButton_GetStatus + vm_get_buttonstatus.restype = LONG + vm_get_buttonstatus.argtypes = [LONG, ct.POINTER(FLOAT), LONG] + + vm_set_buttonstatus = libc.VBVMR_MacroButton_SetStatus + vm_set_buttonstatus.restype = LONG + vm_set_buttonstatus.argtypes = [LONG, FLOAT, LONG] + + vm_pdirty = libc.VBVMR_IsParametersDirty + vm_pdirty.restype = LONG + vm_pdirty.argtypes = None + + vm_get_parameter_float = libc.VBVMR_GetParameterFloat + vm_get_parameter_float.restype = LONG + vm_get_parameter_float.argtypes = [ct.POINTER(CHAR), ct.POINTER(FLOAT)] + + vm_set_parameter_float = libc.VBVMR_SetParameterFloat + vm_set_parameter_float.restype = LONG + vm_set_parameter_float.argtypes = [ct.POINTER(CHAR), FLOAT] + + vm_get_parameter_string = libc.VBVMR_GetParameterStringW + vm_get_parameter_string.restype = LONG + vm_get_parameter_string.argtypes = [ct.POINTER(CHAR), ct.POINTER(WCHAR * 512)] + + vm_set_parameter_string = libc.VBVMR_SetParameterStringW + vm_set_parameter_string.restype = LONG + vm_set_parameter_string.argtypes = [ct.POINTER(CHAR), ct.POINTER(WCHAR)] + + vm_set_parameter_multi = libc.VBVMR_SetParameters + vm_set_parameter_multi.restype = LONG + vm_set_parameter_multi.argtypes = [ct.POINTER(CHAR), ct.POINTER(CHAR)] + + vm_get_level = libc.VBVMR_GetLevel + vm_get_level.restype = LONG + vm_get_level.argtypes = [LONG, LONG, ct.POINTER(FLOAT)] + + vm_get_num_indevices = libc.VBVMR_Input_GetDeviceNumber + vm_get_num_indevices.restype = LONG + vm_get_num_indevices.argtypes = None + + vm_get_desc_indevices = libc.VBVMR_Input_GetDeviceDescW + vm_get_desc_indevices.restype = LONG + vm_get_desc_indevices.argtypes = [ + LONG, + ct.POINTER(LONG), + ct.POINTER(WCHAR * 256), + ct.POINTER(WCHAR * 256), + ] + + vm_get_num_outdevices = libc.VBVMR_Output_GetDeviceNumber + vm_get_num_outdevices.restype = LONG + vm_get_num_outdevices.argtypes = None + + vm_get_desc_outdevices = libc.VBVMR_Output_GetDeviceDescW + vm_get_desc_outdevices.restype = LONG + vm_get_desc_outdevices.argtypes = [ + LONG, + ct.POINTER(LONG), + ct.POINTER(WCHAR * 256), + ct.POINTER(WCHAR * 256), + ] + + def call(self, func): + res = func() + if res != 0: + raise CAPIError(f"Function {func.func.__name__} returned {res}") diff --git a/voicemeeterlib/command.py b/voicemeeterlib/command.py new file mode 100644 index 0000000..52c1424 --- /dev/null +++ b/voicemeeterlib/command.py @@ -0,0 +1,55 @@ +from .error import VMError +from .iremote import IRemote +from .meta import action_prop + + +class Command(IRemote): + """ + Implements the common interface + + Defines concrete implementation for command + """ + + @classmethod + def make(cls, remote): + """ + Factory function for command class. + + Returns a Command class of a kind. + """ + CMD_cls = type( + f"Command{remote.kind}", + (cls,), + { + **{ + param: action_prop(param) + for param in ["show", "shutdown", "restart"] + }, + "hide": action_prop("show", val=0), + }, + ) + return CMD_cls(remote) + + def __str__(self): + return f"{type(self).__name__}" + + @property + def identifier(self) -> str: + return "Command" + + def set_showvbanchat(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("showvbanchat is a boolean parameter") + self.setter("DialogShow.VBANCHAT", 1 if val else 0) + + showvbanchat = property(fset=set_showvbanchat) + + def set_lock(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("lock is a boolean parameter") + self.setter("lock", 1 if val else 0) + + lock = property(fset=set_lock) + + def reset(self): + self._remote.apply_config("reset") diff --git a/voicemeeterlib/config.py b/voicemeeterlib/config.py new file mode 100644 index 0000000..678c288 --- /dev/null +++ b/voicemeeterlib/config.py @@ -0,0 +1,191 @@ +import itertools +from pathlib import Path + +import tomllib + +from .kinds import request_kind_map as kindmap + + +class TOMLStrBuilder: + """builds a config profile, as a string, for the toml parser""" + + def __init__(self, kind): + self.kind = kind + self.phys_in, self.virt_in = kind.ins + self.phys_out, self.virt_out = kind.outs + + self.higher = itertools.chain( + [f"strip-{i}" for i in range(kind.num_strip)], + [f"bus-{i}" for i in range(kind.num_bus)], + ) + + def init_config(self, profile=None): + self.virt_strip_params = ( + [ + "mute = false", + "mono = false", + "solo = false", + "gain = 0.0", + ] + + [f"A{i} = false" for i in range(1, self.phys_out + 1)] + + [f"B{i} = false" for i in range(1, self.virt_out + 1)] + ) + self.phys_strip_params = self.virt_strip_params + [ + "comp = 0.0", + "gate = 0.0", + ] + self.bus_bool = ["mono = false", "eq = false", "mute = false"] + + if profile == "reset": + self.reset_config() + + def reset_config(self): + self.phys_strip_params = list( + map(lambda x: x.replace("B1 = false", "B1 = true"), self.phys_strip_params) + ) + self.virt_strip_params = list( + map(lambda x: x.replace("A1 = false", "A1 = true"), self.virt_strip_params) + ) + + def build(self, profile="reset"): + self.init_config(profile) + toml_str = str() + for eachclass in self.higher: + toml_str += f"[{eachclass}]\n" + toml_str = self.join(eachclass, toml_str) + return toml_str + + def join(self, eachclass, toml_str): + kls, index = eachclass.split("-") + match kls: + case "strip": + toml_str += ("\n").join( + self.phys_strip_params + if int(index) < self.phys_in + else self.virt_strip_params + ) + case "bus": + toml_str += ("\n").join(self.bus_bool) + case _: + pass + return toml_str + "\n" + + +class TOMLDataExtractor: + def __init__(self, file): + self._data = dict() + with open(file, "rb") as f: + self._data = tomllib.load(f) + + @property + def data(self): + return self._data + + @data.setter + def data(self, value): + self._data = value + + +def dataextraction_factory(file): + """ + factory function for parser + + this opens the possibility for other parsers to be added + """ + if file.suffix == ".toml": + extractor = TOMLDataExtractor + else: + raise ValueError("Cannot extract data from {}".format(file)) + return extractor(file) + + +class SingletonType(type): + """ensure only a single instance of Loader object""" + + _instances = {} + + def __call__(cls, *args, **kwargs): + if cls not in cls._instances: + cls._instances[cls] = super(SingletonType, cls).__call__(*args, **kwargs) + return cls._instances[cls] + + +class Loader(metaclass=SingletonType): + """ + invokes the parser + + checks if config already in memory + + loads data into memory if not found + """ + + def __init__(self, kind): + self._kind = kind + self._configs = dict() + self.defaults(kind) + self.parser = None + + def defaults(self, kind): + self.builder = TOMLStrBuilder(kind) + toml_str = self.builder.build() + self.register("reset", tomllib.loads(toml_str)) + + def parse(self, identifier, data): + if identifier in self._configs: + print(f"config file with name {identifier} already in memory, skipping..") + return False + self.parser = dataextraction_factory(data) + return True + + def register(self, identifier, data=None): + self._configs[identifier] = data if data else self.parser.data + print(f"config {self.name}/{identifier} loaded into memory") + + def deregister(self): + self._configs.clear() + self.defaults(self._kind) + + @property + def configs(self): + return self._configs + + @property + def name(self): + return self._kind.name + + +def loader(kind): + """ + traverses defined paths for config files + + directs the loader + + returns configs loaded into memory + """ + loader = Loader(kind) + + for path in ( + Path.cwd() / "configs" / kind.name, + Path(__file__).parent / "configs" / kind.name, + Path.home() / "Documents/Voicemeeter" / "configs" / kind.name, + ): + if path.is_dir(): + print(f"Checking [{path}] for TOML config files:") + for file in path.glob("*.toml"): + identifier = file.with_suffix("").stem + if loader.parse(identifier, file): + loader.register(identifier) + return loader.configs + + +def request_config(kind_id: str): + """ + config entry point. + + Returns all configs loaded into memory for a kind + """ + try: + configs = loader(kindmap(kind_id)) + except KeyError as e: + print(f"Unknown Voicemeeter kind '{kind_id}'") + return configs diff --git a/voicemeeterlib/device.py b/voicemeeterlib/device.py new file mode 100644 index 0000000..07e33c8 --- /dev/null +++ b/voicemeeterlib/device.py @@ -0,0 +1,72 @@ +from abc import abstractmethod +from typing import Union + +from .iremote import IRemote + + +class Adapter(IRemote): + """Adapter to the common interface.""" + + @abstractmethod + def ins(self): + pass + + @abstractmethod + def outs(self): + pass + + @abstractmethod + def input(self): + pass + + @abstractmethod + def output(self): + pass + + def identifier(self): + pass + + def getter(self, index: int = None, direction: str = None) -> Union[int, dict]: + if index is None: + return self._remote.get_num_devices(direction) + + vals = self._remote.get_device_description(index, direction) + types = {1: "mme", 3: "wdm", 4: "ks", 5: "asio"} + return {"name": vals[0], "type": types[vals[1]], "id": vals[2]} + + +class Device(Adapter): + """Defines concrete implementation for device""" + + @classmethod + def make(cls, remote): + """ + Factory function for device. + + Returns a Device class of a kind. + """ + + def num_ins(cls) -> int: + return cls.getter(direction="in") + + def num_outs(cls) -> int: + return cls.getter(direction="out") + + DEVICE_cls = type( + f"Device{remote.kind}", + (cls,), + { + "ins": property(num_ins), + "outs": property(num_outs), + }, + ) + return DEVICE_cls(remote) + + def __str__(self): + return f"{type(self).__name__}" + + def input(self, index: int) -> dict: + return self.getter(index=index, direction="in") + + def output(self, index: int) -> dict: + return self.getter(index=index, direction="out") diff --git a/voicemeeterlib/error.py b/voicemeeterlib/error.py new file mode 100644 index 0000000..4230d24 --- /dev/null +++ b/voicemeeterlib/error.py @@ -0,0 +1,16 @@ +class InstallError(Exception): + """errors related to installation""" + + pass + + +class CAPIError(Exception): + """errors related to low-level C API calls""" + + pass + + +class VMError(Exception): + """general errors""" + + pass diff --git a/voicemeeterlib/factory.py b/voicemeeterlib/factory.py new file mode 100644 index 0000000..83ead9a --- /dev/null +++ b/voicemeeterlib/factory.py @@ -0,0 +1,211 @@ +from abc import abstractmethod +from enum import IntEnum +from functools import cached_property +from typing import Iterable, NoReturn, Self + +from .base import Remote +from .bus import request_bus_obj as bus +from .command import Command +from .config import request_config as configs +from .device import Device +from .kinds import KindMapClass +from .kinds import request_kind_map as kindmap +from .macrobutton import MacroButton +from .recorder import Recorder +from .strip import request_strip_obj as strip +from .vban import request_vban_obj as vban + + +class FactoryBuilder: + """ + Builder class for factories. + + Separates construction from representation. + """ + + BuilderProgress = IntEnum( + "BuilderProgress", "strip bus command macrobutton vban device recorder", start=0 + ) + + def __init__(self, factory, kind: KindMapClass): + self._factory = factory + self.kind = kind + self._info = ( + f"Finished building strips for {self._factory}", + f"Finished building buses for {self._factory}", + f"Finished building commands for {self._factory}", + f"Finished building macrobuttons for {self._factory}", + f"Finished building vban in/out streams for {self._factory}", + f"Finished building device for {self._factory}", + f"Finished building recorder for {self._factory}", + ) + + def _pinfo(self, name: str) -> NoReturn: + """prints progress status for each step""" + name = name.split("_")[1] + print(self._info[int(getattr(self.BuilderProgress, name))]) + + def make_strip(self) -> Self: + self._factory.strip = tuple( + strip(self.kind.phys_in < i, self._factory, i) + for i in range(self.kind.num_strip) + ) + return self + + def make_bus(self) -> Self: + self._factory.bus = tuple( + bus(self.kind.phys_out < i, self._factory, i) + for i in range(self.kind.num_bus) + ) + return self + + def make_command(self) -> Self: + self._factory.command = Command.make(self._factory) + return self + + def make_macrobutton(self) -> Self: + self._factory.button = tuple(MacroButton(self._factory, i) for i in range(80)) + return self + + def make_vban(self) -> Self: + self._factory.vban = vban(self._factory) + return self + + def make_device(self) -> Self: + self._factory.device = Device.make(self._factory) + return self + + def make_recorder(self) -> Self: + self._factory.recorder = Recorder.make(self._factory) + return self + + +class FactoryBase(Remote): + """Base class for factories, subclasses Remote.""" + + def __init__(self, kind_id: str, **kwargs): + defaultkwargs = {"sync": False, "ratelimit": 0.033} + kwargs = defaultkwargs | kwargs + self.kind = kindmap(kind_id) + super().__init__(**kwargs) + self.builder = FactoryBuilder(self, self.kind) + self._steps = ( + self.builder.make_strip, + self.builder.make_bus, + self.builder.make_command, + self.builder.make_macrobutton, + self.builder.make_vban, + self.builder.make_device, + ) + self._configs = None + + def __str__(self) -> str: + return f"Voicemeeter {self.kind}" + + @property + @abstractmethod + def steps(self): + pass + + @cached_property + def configs(self): + self._configs = configs(self.kind.name) + return self._configs + + +class BasicFactory(FactoryBase): + """ + Represents a Basic Remote subclass + + Responsible for directing the builder class + """ + + def __new__(cls, *args, **kwargs): + if cls is BasicFactory: + raise TypeError(f"'{cls.__name__}' does not support direct instantiation") + return object.__new__(cls) + + def __init__(self, kind_id, **kwargs): + super().__init__(kind_id, **kwargs) + [step()._pinfo(step.__name__) for step in self.steps] + + @property + def steps(self) -> Iterable: + """steps required to build the interface for a kind""" + return self._steps + + +class BananaFactory(FactoryBase): + """ + Represents a Banana Remote subclass + + Responsible for directing the builder class + """ + + def __new__(cls, *args, **kwargs): + if cls is BananaFactory: + raise TypeError(f"'{cls.__name__}' does not support direct instantiation") + return object.__new__(cls) + + def __init__(self, kind_id, **kwargs): + super().__init__(kind_id, **kwargs) + [step()._pinfo(step.__name__) for step in self.steps] + + @property + def steps(self) -> Iterable: + """steps required to build the interface for a kind""" + return self._steps + (self.builder.make_recorder,) + + +class PotatoFactory(FactoryBase): + """ + Represents a Potato Remote subclass + + Responsible for directing the builder class + """ + + def __new__(cls, *args, **kwargs): + if cls is PotatoFactory: + raise TypeError(f"'{cls.__name__}' does not support direct instantiation") + return object.__new__(cls) + + def __init__(self, kind_id: str, **kwargs): + super().__init__(kind_id, **kwargs) + [step()._pinfo(step.__name__) for step in self.steps] + + @property + def steps(self) -> Iterable: + """steps required to build the interface for a kind""" + return self._steps + (self.builder.make_recorder,) + + +def remote_factory(kind_id: str, **kwargs) -> Remote: + """ + Factory method, invokes a factory creation class of a kind + + Returns a Remote class of a kind + """ + match kind_id: + case "basic": + _factory = BasicFactory + case "banana": + _factory = BananaFactory + case "potato": + _factory = PotatoFactory + case _: + raise ValueError(f"Unknown Voicemeeter kind '{kind_id}'") + return type(f"Remote{kind_id.capitalize()}", (_factory,), {})(kind_id, **kwargs) + + +def request_remote_obj(kind_id: str, **kwargs) -> Remote: + """ + Interface entry point. Wraps factory method and handles errors + + Returns a reference to a Remote class of a kind + """ + REMOTE_obj = None + try: + REMOTE_obj = remote_factory(kind_id, **kwargs) + except (ValueError, TypeError) as e: + raise SystemExit(e) + return REMOTE_obj diff --git a/voicemeeterlib/inst.py b/voicemeeterlib/inst.py new file mode 100644 index 0000000..4f7882e --- /dev/null +++ b/voicemeeterlib/inst.py @@ -0,0 +1,41 @@ +import ctypes as ct +import platform +import winreg +from pathlib import Path + +from .error import InstallError + +bits = 64 if ct.sizeof(ct.c_voidp) == 8 else 32 + +if platform.system() != "Windows": + raise InstallError("Only Windows OS supported") + + +VM_KEY = "VB:Voicemeeter {17359A74-1236-5467}" +REG_KEY = "".join( + [ + "SOFTWARE", + ("\\WOW6432Node" if bits == 64 else ""), + "\\Microsoft\\Windows\\CurrentVersion\\Uninstall", + ] +) + + +def get_vmpath(): + with winreg.OpenKey( + winreg.HKEY_LOCAL_MACHINE, r"{}".format(REG_KEY + "\\" + VM_KEY) + ) as vm_key: + path = winreg.QueryValueEx(vm_key, r"UninstallString")[0] + return path + + +vm_path = Path(get_vmpath()) +vm_parent = vm_path.parent + +DLL_NAME = f'VoicemeeterRemote{"64" if bits == 64 else ""}.dll' + +dll_path = vm_parent.joinpath(DLL_NAME) +if not dll_path.is_file(): + raise InstallError(f"Could not find {DLL_NAME}") + +libc = ct.CDLL(str(dll_path)) diff --git a/voicemeeterlib/iremote.py b/voicemeeterlib/iremote.py new file mode 100644 index 0000000..7b6765f --- /dev/null +++ b/voicemeeterlib/iremote.py @@ -0,0 +1,36 @@ +import time +from abc import ABCMeta, abstractmethod +from typing import Self + + +class IRemote(metaclass=ABCMeta): + """ + Common interface between base class and extended (higher) classes + + Provides some default implementation + """ + + def __init__(self, remote, index=None): + self._remote = remote + self.index = index + + def getter(self, param, **kwargs): + """Gets a parameter value""" + return self._remote.get(f"{self.identifier}.{param}", **kwargs) + + def setter(self, param, val): + """Sets a parameter value""" + self._remote.set(f"{self.identifier}.{param}", val) + + @abstractmethod + def identifier(self): + pass + + def apply(self, data: dict) -> Self: + for attr, val in data.items(): + if hasattr(self, attr): + setattr(self, attr, val) + return self + + def then_wait(self): + time.sleep(self._remote.DELAY) diff --git a/voicemeeterlib/kinds.py b/voicemeeterlib/kinds.py new file mode 100644 index 0000000..0ca4003 --- /dev/null +++ b/voicemeeterlib/kinds.py @@ -0,0 +1,104 @@ +from dataclasses import dataclass +from enum import Enum, unique + + +@unique +class KindId(Enum): + BASIC = 1 + BANANA = 2 + POTATO = 3 + + +class SingletonType(type): + """ensure only a single instance of a kind map object""" + + _instances = {} + + def __call__(cls, *args, **kwargs): + if cls not in cls._instances: + cls._instances[cls] = super(SingletonType, cls).__call__(*args, **kwargs) + return cls._instances[cls] + + +@dataclass +class KindMapClass(metaclass=SingletonType): + name: str + ins: tuple + outs: tuple + vban: tuple + + @property + def phys_in(self): + return self.ins[0] + + @property + def virt_in(self): + return self.ins[-1] + + @property + def phys_out(self): + return self.outs[0] + + @property + def virt_out(self): + return self.outs[-1] + + @property + def num_strip(self): + return sum(self.ins) + + @property + def num_bus(self): + return sum(self.outs) + + def __str__(self) -> str: + return self.name.capitalize() + + +@dataclass +class BasicMap(KindMapClass): + name: str + ins: tuple = (2, 1) + outs: tuple = (1, 1) + vban: tuple = (4, 4) + + +@dataclass +class BananaMap(KindMapClass): + name: str + ins: tuple = (3, 2) + outs: tuple = (3, 2) + vban: tuple = (8, 8) + + +@dataclass +class PotatoMap(KindMapClass): + name: str + ins: tuple = (5, 3) + outs: tuple = (5, 3) + vban: tuple = (8, 8) + + +def kind_factory(kind_id): + match kind_id: + case "basic": + _kind_map = BasicMap + case "banana": + _kind_map = BananaMap + case "potato": + _kind_map = PotatoMap + case _: + raise ValueError(f"Unknown Voicemeeter kind {kind_id}") + return _kind_map(name=kind_id) + + +def request_kind_map(kind_id): + KIND_obj = None + try: + KIND_obj = kind_factory(kind_id) + except ValueError as e: + print(e) + return KIND_obj + + +kinds_all = list(request_kind_map(kind_id.name.lower()) for kind_id in KindId) diff --git a/voicemeeterlib/macrobutton.py b/voicemeeterlib/macrobutton.py new file mode 100644 index 0000000..6faf9c3 --- /dev/null +++ b/voicemeeterlib/macrobutton.py @@ -0,0 +1,52 @@ +from .error import VMError +from .iremote import IRemote + + +class Adapter(IRemote): + """Adapter to the common interface.""" + + def identifier(self): + pass + + def getter(self, id, mode): + return self._remote.get_buttonstatus(id, mode) + + def setter(self, id, val, mode): + self._remote.set_buttonstatus(id, val, mode) + + +class MacroButton(Adapter): + """Defines concrete implementation for macrobutton""" + + def __str__(self): + return f"{type(self).__name__}{self._remote.kind}{self.index}" + + @property + def state(self) -> bool: + return self.getter(self.index, 1) == 1 + + @state.setter + def state(self, val): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("state is a boolean parameter") + self.setter(self.index, 1 if val else 0, 1) + + @property + def stateonly(self) -> bool: + return self.getter(self.index, 2) == 1 + + @stateonly.setter + def stateonly(self, val): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("stateonly is a boolean parameter") + self.setter(self.index, 1 if val else 0, 2) + + @property + def trigger(self) -> bool: + return self.getter(self.index, 3) == 1 + + @trigger.setter + def trigger(self, val): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("trigger is a boolean parameter") + self.setter(self.index, 1 if val else 0, 3) diff --git a/voicemeeterlib/meta.py b/voicemeeterlib/meta.py new file mode 100644 index 0000000..e45ccb0 --- /dev/null +++ b/voicemeeterlib/meta.py @@ -0,0 +1,51 @@ +from .error import VMError + + +def bool_prop(param): + """meta function for boolean parameters""" + + def fget(self) -> bool: + return self.getter(param) == 1 + + def fset(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError(f"{param} is a boolean parameter") + self.setter(param, 1 if val else 0) + + return property(fget, fset) + + +def float_prop(param): + """meta function for float parameters""" + + def fget(self): + return self.getter(param) + + def fset(self, val): + self.setter(param, val) + + return property(fget, fset) + + +def action_prop(param, val: int = 1): + """A param that performs an action""" + + def fdo(self): + self.setter(param, val) + + return fdo + + +def bus_mode_prop(param): + """meta function for bus mode parameters""" + + def fget(self) -> bool: + self._remote.clear_dirty() + return self.getter(param) == 1 + + def fset(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError(f"{param} is a boolean parameter") + self.setter(param, 1 if val else 0) + + return property(fget, fset) diff --git a/voicemeeterlib/recorder.py b/voicemeeterlib/recorder.py new file mode 100644 index 0000000..3d0dcff --- /dev/null +++ b/voicemeeterlib/recorder.py @@ -0,0 +1,76 @@ +from .error import VMError +from .iremote import IRemote +from .kinds import kinds_all +from .meta import action_prop, bool_prop + + +class Recorder(IRemote): + """ + Implements the common interface + + Defines concrete implementation for recorder + """ + + @classmethod + def make(cls, remote): + """ + Factory function for recorder. + + Returns a Recorder class of a kind. + """ + ChannelMixin = _channel_mixins[remote.kind.name] + REC_cls = type( + f"Recorder{remote.kind}", + (cls, ChannelMixin), + { + **{ + param: action_prop(param) + for param in [ + "play", + "stop", + "pause", + "replay", + "record", + "ff", + "rw", + ] + }, + }, + ) + return REC_cls(remote) + + def __str__(self): + return f"{type(self).__name__}" + + @property + def identifier(self) -> str: + return "recorder" + + def load(self, file: str): + try: + self.setter("load", file) + except UnicodeError: + raise VMError("File full directory must be a raw string") + + def set_loop(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("Error True or False expected") + self.setter("mode.loop", 1 if val else 0) + + loop = property(fset=set_loop) + + +def _make_channel_mixin(kind): + """Creates a channel out property mixin""" + num_A, num_B = kind.outs + return type( + f"ChannelMixin{kind.name}", + (), + { + **{f"A{i}": bool_prop(f"A{i}") for i in range(1, num_A + 1)}, + **{f"B{i}": bool_prop(f"B{i}") for i in range(1, num_B + 1)}, + }, + ) + + +_channel_mixins = {kind.name: _make_channel_mixin(kind) for kind in kinds_all} diff --git a/voicemeeterlib/strip.py b/voicemeeterlib/strip.py new file mode 100644 index 0000000..8ac723c --- /dev/null +++ b/voicemeeterlib/strip.py @@ -0,0 +1,320 @@ +import time +from abc import abstractmethod +from math import log +from typing import Union + +from .error import VMError +from .iremote import IRemote +from .kinds import kinds_all +from .meta import bool_prop + + +class Strip(IRemote): + """ + Implements the common interface + + Defines concrete implementation for strip + """ + + @abstractmethod + def __str__(self): + pass + + @property + def identifier(self) -> str: + return f"strip[{self.index}]" + + @property + def mono(self) -> bool: + return self.getter("mono") == 1 + + @mono.setter + def mono(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("mono is a boolean parameter") + self.setter("mono", 1 if val else 0) + + @property + def solo(self) -> bool: + return self.getter("solo") == 1 + + @solo.setter + def solo(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("solo is a boolean parameter") + self.setter("solo", 1 if val else 0) + + @property + def mute(self) -> bool: + return self.getter("mute") == 1 + + @mute.setter + def mute(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("mute is a boolean parameter") + self.setter("mute", 1 if val else 0) + + @property + def limit(self) -> int: + return int(self.getter("limit")) + + @limit.setter + def limit(self, val: int): + self.setter("limit", val) + + @property + def label(self) -> str: + return self.getter("Label", is_string=True) + + @label.setter + def label(self, val: str): + if not isinstance(val, str): + raise VMError("label is a string parameter") + self.setter("Label", val) + + @property + def gain(self) -> float: + return round(self.getter("gain"), 1) + + @gain.setter + def gain(self, val: float): + self.setter("gain", val) + + def fadeto(self, target: float, time_: int): + self.setter("FadeTo", f"({target}, {time_})") + time.sleep(self.remote.delay) + + def fadeby(self, change: float, time_: int): + self.setter("FadeBy", f"({change}, {time_})") + time.sleep(self.remote.delay) + + +class PhysicalStrip(Strip): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + @property + def comp(self) -> float: + return round(self.getter("Comp"), 1) + + @comp.setter + def comp(self, val: float): + self.setter("Comp", val) + + @property + def gate(self) -> float: + return round(self.getter("Gate"), 1) + + @gate.setter + def gate(self, val: float): + self.setter("Gate", val) + + @property + def audibility(self) -> float: + return round(self.getter("audibility"), 1) + + @audibility.setter + def audibility(self, val: float): + self.setter("audibility", val) + + @property + def device(self): + return self.getter("device.name", is_string=True) + + @property + def sr(self): + return int(self.getter("device.sr")) + + +class VirtualStrip(Strip): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + @property + def mc(self) -> bool: + return self.getter("mc") == 1 + + @mc.setter + def mc(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("mc is a boolean parameter") + self.setter("mc", 1 if val else 0) + + mono = mc + + @property + def k(self) -> int: + return int(self.getter("karaoke")) + + @k.setter + def k(self, val: int): + self.setter("karaoke", val) + + @property + def bass(self): + return round(self.getter("EQGain1"), 1) + + @bass.setter + def bass(self, val: float): + self.setter("EQGain1", val) + + @property + def mid(self): + return round(self.getter("EQGain2"), 1) + + @mid.setter + def mid(self, val: float): + self.setter("EQGain2", val) + + med = mid + + @property + def treble(self): + return round(self.getter("EQGain3"), 1) + + @treble.setter + def treble(self, val: float): + self.setter("EQGain3", val) + + def appgain(self, name: str, gain: float): + self.setter("AppGain", f'("{name}", {gain})') + + def appmute(self, name: str, mute: bool = None): + if not isinstance(mute, bool) and mute not in (0, 1): + raise VMError("appmute is a boolean parameter") + self.setter("AppMute", f'("{name}", {1 if mute else 0})') + + +class StripLevel(IRemote): + def __init__(self, remote, index): + super().__init__(remote, index) + phys_map = tuple((i, i + 2) for i in range(0, remote.kind.phys_in * 2, 2)) + virt_map = tuple( + (i, i + 8) + for i in range( + remote.kind.phys_in * 2, + remote.kind.phys_in * 2 + remote.kind.virt_in * 8, + 8, + ) + ) + self.level_map = phys_map + virt_map + + def getter(self, mode): + """Returns a tuple of level values for the channel.""" + + def fget(i): + res = self._remote.get_level(mode, i) + return round(20 * log(res, 10), 1) if res > 0 else -200.0 + + range_ = self.level_map[self.index] + return tuple(fget(i) for i in range(*range_)) + + def getter_prefader(self): + def fget(i): + return round(20 * log(i, 10), 1) if i > 0 else -200.0 + + range_ = self.level_map[self.index] + return tuple( + fget(i) for i in self._remote._strip_levels[range_[0] : range_[-1]] + ) + + @property + def identifier(self) -> str: + return f"Strip[{self.index}]" + + @property + def prefader(self) -> tuple: + return self.getter_prefader() + + @property + def postfader(self) -> tuple: + return self.getter(1) + + @property + def postmute(self) -> tuple: + return self.getter(2) + + @property + def updated(self) -> tuple: + return self._remote._strip_comp + + +class GainLayer(IRemote): + def __init__(self, remote, index, i): + super().__init__(remote, index) + self._i = i + + @property + def identifier(self) -> str: + return f"Strip[{self.index}]" + + @property + def gain(self): + return self.getter(f"GainLayer[{self._i}]") + + @gain.setter + def gain(self, val): + self.setter(f"GainLayer[{self._i}]", val) + + +def _make_gainlayer_mixin(remote, index): + """Creates a GainLayer mixin""" + return type( + f"GainlayerMixin", + (), + { + "gainlayer": tuple( + GainLayer(remote, index, i) for i in range(remote.kind.num_bus) + ) + }, + ) + + +def _make_channelout_mixin(kind): + """Creates a channel out property mixin""" + return type( + f"ChannelOutMixin{kind}", + (), + { + **{f"A{i}": bool_prop(f"A{i}") for i in range(1, kind.phys_out + 1)}, + **{f"B{i}": bool_prop(f"B{i}") for i in range(1, kind.virt_out + 1)}, + }, + ) + + +__make_channelout_mixins = { + kind.name: _make_channelout_mixin(kind) for kind in kinds_all +} + + +def strip_factory(is_phys_strip, remote, i) -> Union[PhysicalStrip, VirtualStrip]: + """ + Factory method for strips + + Mixes in required classes + + Returns a physical or virtual strip subclass + """ + STRIP_cls = PhysicalStrip if is_phys_strip else VirtualStrip + CHANNELOUTMIXIN_cls = __make_channelout_mixins[remote.kind.name] + + _kls = (STRIP_cls, CHANNELOUTMIXIN_cls) + if remote.kind.name == "potato": + GAINLAYERMIXIN_cls = _make_gainlayer_mixin(remote, i) + _kls += (GAINLAYERMIXIN_cls,) + return type( + f"{STRIP_cls.__name__}{remote.kind}", + _kls, + { + "levels": StripLevel(remote, i), + }, + )(remote, i) + + +def request_strip_obj(is_phys_strip, remote, i) -> Strip: + """ + Strip entry point. Wraps factory method. + + Returns a reference to a strip subclass of a kind + """ + return strip_factory(is_phys_strip, remote, i) diff --git a/voicemeeterlib/subject.py b/voicemeeterlib/subject.py new file mode 100644 index 0000000..643549d --- /dev/null +++ b/voicemeeterlib/subject.py @@ -0,0 +1,39 @@ +class Subject: + """Adds support for observers""" + + def __init__(self): + """list of current observers""" + + self._observers = list() + + @property + def observers(self) -> list: + """returns the current observers""" + + return self._observers + + def notify(self, modifier=None, data=None): + """run callbacks on update""" + + [o.on_update(modifier, data) for o in self._observers] + + def add(self, observer): + """adds an observer to _observers""" + + if observer not in self._observers: + self._observers.append(observer) + else: + print(f"Failed to add: {observer}") + + def remove(self, observer): + """removes an observer from _observers""" + + try: + self._observers.remove(observer) + except ValueError: + print(f"Failed to remove: {observer}") + + def clear(self): + """clears the _observers list""" + + self._observers.clear() diff --git a/voicemeeterlib/util.py b/voicemeeterlib/util.py new file mode 100644 index 0000000..d1c2b5f --- /dev/null +++ b/voicemeeterlib/util.py @@ -0,0 +1,52 @@ +import functools + + +def polling(func): + """ + Offers memoization for a set into get operation. + + If sync clear dirty parameters before fetching new value. + + Useful for loop getting if not running callbacks + """ + + @functools.wraps(func) + def wrapper(*args, **kwargs): + get = func.__name__ == "get" + mb_get = func.__name__ == "get_buttonstatus" + remote, *remaining = args + + if get: + param, *rem = remaining + elif mb_get: + id, mode, *rem = remaining + param = f"mb_{id}_{mode}" + + if param in remote.cache: + return remote.cache.pop(param) + if remote.sync: + remote.clear_dirty() + return func(*args, **kwargs) + + return wrapper + + +def script(func): + """Convert dictionary to script""" + + def wrapper(*args): + remote, script = args + if isinstance(script, dict): + params = "" + for key, val in script.items(): + obj, m2, *rem = key.split("-") + index = int(m2) if m2.isnumeric() else int(*rem) + params += ";".join( + f"{obj}{f'.{m2}stream' if not m2.isnumeric() else ''}[{index}].{k}={int(v) if isinstance(v, bool) else v}" + for k, v in val.items() + ) + params += ";" + script = params + return func(remote, script) + + return wrapper diff --git a/voicemeeterlib/vban.py b/voicemeeterlib/vban.py new file mode 100644 index 0000000..ab0e046 --- /dev/null +++ b/voicemeeterlib/vban.py @@ -0,0 +1,188 @@ +from abc import abstractmethod + +from .error import VMError +from .iremote import IRemote + + +class VbanStream(IRemote): + """ + Implements the common interface + + Defines concrete implementation for vban stream + """ + + @abstractmethod + def __str__(self): + pass + + @property + def identifier(self) -> str: + return f"vban.{self.direction}stream[{self.index}]" + + @property + def on(self) -> bool: + return self.getter("on") == 1 + + @on.setter + def on(self, val: bool): + if not isinstance(val, bool) and val not in (0, 1): + raise VMError("True or False expected") + self.setter("on", 1 if val else 0) + + @property + def name(self) -> str: + return self.getter("name", is_string=True) + + @name.setter + def name(self, val: str): + self.setter("name", val) + + @property + def ip(self) -> str: + return self.getter("ip", is_string=True) + + @ip.setter + def ip(self, val: str): + self.setter("ip", val) + + @property + def port(self) -> int: + return int(self.getter("port")) + + @port.setter + def port(self, val: int): + if val not in range(1024, 65536): + raise VMError("Expected value from 1024 to 65535") + self.setter("port", val) + + @property + def sr(self) -> int: + return int(self.getter("sr")) + + @sr.setter + def sr(self, val: int): + opts = (11025, 16000, 22050, 24000, 32000, 44100, 48000, 64000, 88200, 96000) + if val not in opts: + raise VMError("Expected one of: {opts}") + self.setter("sr", val) + + @property + def channel(self) -> int: + return int(self.getter("channel")) + + @channel.setter + def channel(self, val: int): + if val not in range(1, 9): + raise VMError("Expected value from 1 to 8") + self.setter("channel", val) + + @property + def bit(self) -> int: + return 16 if (int(self.getter("bit") == 1)) else 24 + + @bit.setter + def bit(self, val: int): + if val not in (16, 24): + raise VMError("Expected value 16 or 24") + self.setter("bit", 1 if (val == 16) else 2) + + @property + def quality(self) -> int: + return int(self.getter("quality")) + + @quality.setter + def quality(self, val: int): + if val not in range(5): + raise VMError("Expected value from 0 to 4") + self.setter("quality", val) + + @property + def route(self) -> int: + return int(self.getter("route")) + + @route.setter + def route(self, val: int): + if val not in range(9): + raise VMError("Expected value from 0 to 8") + self.setter("route", val) + + +class VbanInstream(VbanStream): + """ + class representing a vban instream + + subclasses VbanStream + """ + + def __str__(self): + return f"{type(self).__name__}{self._remote.kind}{self.index}" + + @property + def direction(self) -> str: + return "in" + + @property + def sr(self) -> int: + return super(VbanInstream, self).sr + + @property + def channel(self) -> int: + return super(VbanInstream, self).channel + + @property + def bit(self) -> int: + return super(VbanInstream, self).bit + + +class VbanOutstream(VbanStream): + """ + class representing a vban outstream + + Subclasses VbanStream + """ + + def __str__(self): + return f"{type(self).__name__}{self._remote.kind}{self.index}" + + @property + def direction(self) -> str: + return "out" + + +class Vban: + """ + class representing the vban module + + Contains two tuples, one for each stream type + """ + + def __init__(self, remote): + self.remote = remote + num_instream, num_outstream = remote.kind.vban + 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): + self.remote.set("vban.Enable", 1) + + def disable(self): + self.remote.set("vban.Enable", 0) + + +def vban_factory(remote) -> Vban: + """ + Factory method for vban + + Returns a class that represents the VBAN module. + """ + VBAN_cls = Vban + return type(f"{VBAN_cls.__name__}", (VBAN_cls,), {})(remote) + + +def request_vban_obj(remote) -> Vban: + """ + Vban entry point. + + Returns a reference to a Vban class of a kind + """ + return vban_factory(remote)