diff --git a/README.md b/README.md index 1c470b8..b27a50f 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,6 @@ [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/onyx-and-iris/vban-cmd-python/blob/dev/LICENSE) [![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) @@ -22,252 +23,157 @@ For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md) - Banana 2.0.6.2 - Potato 3.0.2.2 -## Prerequisites +## Requirements - [Voicemeeter](https://voicemeeter.com/) -- Python 3.9+ +- Python 3.11 or greater ## Installation -``` -git clone https://github.com/onyx-and-iris/vban-cmd-python -cd vban-cmd-python -``` +### `Pip` -Just the interface: +Install voicemeeter-api package from your console -``` -pip install . -``` +`pip install vban-cmd` -With development dependencies: -``` -pip install -e .['development'] -``` +## `Use` -## Usage +Simplest use case, use a context manager to request a VbanCmdvban_cmd class of a kind. -#### Use with a context manager: +Login and logout are handled for you in this scenario. -Parameter coverage is not as extensive for this interface as with the Remote API. - -### Example 1 +#### `__main__.py` ```python -import vbancmd +import vban_cmd + class ManyThings: def __init__(self, vban): self.vban = vban def things(self): - # Set the mapping of the second input strip - self.vban.strip[1].A3 = True - print(f'Output A3 of Strip {self.vban.strip[1].label}: {self.vban.strip[1].A3}') + self.vban.strip[0].label = "podmic" + self.vban.strip[0].mute = True + print( + f"strip 0 ({self.vban.strip[0].label}) has been set to {self.vban.strip[0].mute}" + ) def other_things(self): - # Toggle mute for the leftmost output bus - self.vban.bus[0].mute = not self.vban.bus[0].mute + info = ( + f"bus 3 gain has been set to {self.vban.bus[3].gain}", + f"bus 4 eq has been set to {self.vban.bus[4].eq}", + ) + self.vban.bus[3].gain = -6.3 + self.vban.bus[4].eq = True + print("\n".join(info)) def main(): - with vbancmd.connect(kind_id, ip=ip) as vban: + with vban_cmd.api(kind_id) as vban: do = ManyThings(vban) do.things() do.other_things() -if __name__ == '__main__': - kind_id = 'potato' - ip = '' + # set many parameters at once + vban.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" + ip = "" main() ``` -#### Or perform setup/teardown independently: +Otherwise you must remember to call `vban.login()`, `vban.logout()` at the start/end of your code. -### Example 2 +## `kind_id` -```python -import vbancmd - -kind_id = 'potato' -ip = '' - -vban = vbancmd.connect(kind_id, ip=ip) - -# call login() at the start of your code -vban.login() - -# Toggle mute for leftmost input strip -vban.strip[0].mute = not vban.strip[0].mute - -# Toggle eq for leftmost output bus -vban.bus[0].eq = not vban.bus[0].eq - -# call logout() at the end of your code -vban.logout() -``` - -## Profiles - -Profiles through config files are supported. - -Three example profiles are provided with the package, one for each kind of Voicemeeter. -To test one first rename \_profiles directory to profiles. -They will be loaded into memory but not applied. To apply one you may do: -`vmr.apply_profile('config')`, but remember to save your current settings first. - -profiles directory can be safely deleted if you don't wish to load them each time. - -A config can contain any key that `connect.apply()` would accept. Additionally, `extends` can be provided to inherit from another profile. Two profiles are available by default: - -- `blank`, all strip off and all sliders to `0.0`. mono, solo, mute, eq all disabled. -- `base`, all physical strip to `A1`, all virtual strip to `B1`, all sliders to `0.0`. - -Sample `mySetup.toml` - -```toml -extends = 'base' -[strip-0] -mute = 1 - -[strip-5] -A1 = 0 -A2 = 1 -A4 = 1 -gain = 0.0 - -[strip-6] -A1 = 0 -A2 = 1 -A4 = 1 -gain = 0.0 -``` - -## API - -### Kinds - -A _kind_ specifies a major Voicemeeter version. Currently this encompasses +Pass the kind of Voicemeeter as an argument. kind_id may be: - `basic` - `banana` - `potato` -#### `vbancmd.connect(kind_id, **kwargs) -> '(VbanCmd)'` +## `Available commands` -Factory function for remotes. Keyword arguments include: +### Channels (strip/bus) -- `ip`: remote pc you wish to send requests to. -- `streamname`: default 'Command1' -- `port`: default 6990 -- `channel`: from 0 to 255 -- `bps`: bitrate of stream, default 0 should be safe for most cases. - -### `VbanCmd` (higher level) - -#### `vban.type` - -The kind of the Voicemeeter instance. - -#### `vban.version` - -A tuple of the form `(v1, v2, v3, v4)`. - -#### `vban.strip` - -An `InputStrip` tuple, containing both physical and virtual. - -#### `vban.bus` - -An `OutputBus` tuple, containing both physical and virtual. - -#### `vban.show()` - -Shows Voicemeeter if it's hidden. No effect otherwise. - -#### `vban.hide()` - -Hides Voicemeeter if it's shown. No effect otherwise. - -#### `vban.shutdown()` - -Closes Voicemeeter. - -#### `vban.restart()` - -Restarts Voicemeeter's audio engine. - -#### `vban.apply(mapping)` - -Updates values through a dict. -Example: - -```python -vban.apply({ - 'strip-2': dict(A1=True, B1=True, gain=-6.0), - 'bus-2': dict(mute=True), -}) -``` - -### `Strip` - -The following properties are gettable and settable: +The following properties exist for audio channels. - `mono`: boolean - `solo`: boolean - `mute`: boolean - `label`: string - `gain`: float, -60 to 12 -- Output mapping (e.g. `A1`, `B3`, etc.): boolean, depends on the Voicemeeter kind - -The following properties are settable: - +- `A1 - A5`, `B1 - B3`: boolean - `comp`: float, from 0.0 to 10.0 - `gate`: float, from 0.0 to 10.0 - `limit`: int, from -40 to 12 -#### `gainlayer` - -- `gainlayer[j].gain`: float, -60 to 12 - -for example: +example: ```python -# set and get the value of the second input strip, fourth gainlayer -vban.strip[1].gainlayer[3].gain = -6.3 -print(vban.strip[1].gainlayer[3].gain) +vban.strip[3].gain = 3.7 +print(strip[0].label) + +vban.bus[4].mono = true ``` -Gainlayers defined for Potato version only. +### Command -### `Bus` +Certain 'special' commands are defined by the API as performing actions rather than setting values. The following methods are available: -The following properties are gettable and settable: +- `show()` : Bring Voiceemeter GUI to the front +- `shutdown()` : Shuts down the GUI +- `restart()` : Restart the audio engine -- `mute`: boolean -- `mono`: boolean -- `eq`: boolean -- `eq_ab`: boolean -- `label`: string -- `gain`: float, -60 to 12 +The following properties are write only and accept boolean values. -#### `mode` +- `showvbanchat`: boolean +- `lock`: boolean -Bus modes are gettable and settable - -- `normal`, `amix`, `bmix`, `repeat`, `composite`, `tvmix`, `upmix21`, -- `upmix41`, `upmix61`, `centeronly`, `lfeonly`, `rearonly` - -for example: +example: ```python -# set leftmost bus mode to tvmix -vban.bus[0].mode.tvmix = True +vban.command.restart() +vban.command.showvbanchat = true ``` -### `VbanCmd` (lower level) +### Multiple parameters + +- `apply` + Set many strip/bus parameters at once, for example: + +```python +vban.apply( + { + "strip-2": {"A1": True, "B1": True, "gain": -6.0}, + "bus-2": {"mute": True}, + } +) +``` + +Or for each class you may do: + +```python +vban.strip[0].apply(mute: true, gain: 3.2, A1: true) +vban.vban.outstream[0].apply(on: true, name: 'streamname', bit: 24) +``` + +## `Base Module` + +### VbanCmd class #### `vban.pdirty` diff --git a/__main__.py b/__main__.py index 9d11065..275527b 100644 --- a/__main__.py +++ b/__main__.py @@ -1,27 +1,47 @@ -import vbancmd +import vban_cmd + class ManyThings: def __init__(self, vban): self.vban = vban def things(self): - # Set the mapping of the second input strip - self.vban.strip[1].A3 = True - print(f'Output A3 of Strip {self.vban.strip[1].label}: {self.vban.strip[1].A3}') + self.vban.strip[0].label = "podmic" + self.vban.strip[0].mute = True + print( + f"strip 0 ({self.vban.strip[0].label}) has been set to {self.vban.strip[0].mute}" + ) def other_things(self): - # Toggle mute for the leftmost output bus - self.vban.bus[0].mute = not self.vban.bus[0].mute + info = ( + f"bus 3 gain has been set to {self.vban.bus[3].gain}", + f"bus 4 eq has been set to {self.vban.bus[4].eq}", + ) + self.vban.bus[3].gain = -6.3 + self.vban.bus[4].eq = True + print("\n".join(info)) def main(): - with vbancmd.connect(kind_id, ip=ip) as vban: + with vban_cmd.api(kind_id) as vban: do = ManyThings(vban) do.things() do.other_things() -if __name__ == '__main__': - kind_id = 'potato' - ip = '' + # set many parameters at once + vban.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" + ip = "" main() diff --git a/_profiles/banana/config.toml b/configs/banana/config.toml similarity index 100% rename from _profiles/banana/config.toml rename to configs/banana/config.toml diff --git a/_profiles/basic/config.toml b/configs/basic/config.toml similarity index 100% rename from _profiles/basic/config.toml rename to configs/basic/config.toml diff --git a/_profiles/potato/config.toml b/configs/potato/config.toml similarity index 100% rename from _profiles/potato/config.toml rename to configs/potato/config.toml diff --git a/examples/gui/__main__.py b/examples/gui/__main__.py index 57c8b38..342f72f 100644 --- a/examples/gui/__main__.py +++ b/examples/gui/__main__.py @@ -1,16 +1,19 @@ import tkinter as tk -from tkinter import ttk from functools import partial +from tkinter import ttk from typing import NamedTuple -import vbancmd -from vbancmd import kinds +import vban_cmd +from vban_cmd import kinds + class ExampleAppErrors(Exception): pass + class App(tk.Tk): - """ Topmost Level of App """ + """Topmost Level of App""" + @classmethod def make(cls, kind: NamedTuple): """ @@ -18,39 +21,50 @@ class App(tk.Tk): Returns an App class of a kind """ - APP_cls = type(f'App{kind.name}', (cls,), { - 'name': kind.name, - 'ins': kind.ins, - 'outs': kind.outs, - } + APP_cls = type( + f"App{kind.name}", + (cls,), + { + "name": kind.name, + "ins": kind.ins, + "outs": kind.outs, + }, ) return APP_cls def __init__(self): super().__init__() - self.title(f'Voicemeeter{self.name} Example Program') + self.title(f"Voicemeeter{self.name} Example Program") self.phys_in, self.virt_in = self.ins self.col = self.phys_in + self.virt_in self.row = 3 - self.w = {'Basic': 300, 'Banana': 600, 'Potato': 800} + self.w = {"Basic": 300, "Banana": 600, "Potato": 800} self.h = 150 self.defaultsizes = { - 'Basic': f'{self.w[self.name]}x{self.h}', - 'Banana': f'{self.w[self.name]}x{self.h}', - 'Potato': f'{self.w[self.name]}x{self.h}', + "Basic": f"{self.w[self.name]}x{self.h}", + "Banana": f"{self.w[self.name]}x{self.h}", + "Potato": f"{self.w[self.name]}x{self.h}", } self.geometry(self.defaultsizes[self.name]) """ create tkinter variables, generate widgets and configure rows/cols """ self.gains = { - 'strip': [tk.DoubleVar() for i in range(self.phys_in + self.virt_in)], + "strip": [tk.DoubleVar() for i in range(self.phys_in + self.virt_in)], } self.levels = { - 'strip': [tk.DoubleVar() for i in range(self.phys_in + self.virt_in)], + "strip": [tk.DoubleVar() for i in range(self.phys_in + self.virt_in)], } - [self._make_single_channel(i, j) for i, j in enumerate(i for i in range(0, self.col*2, 2))] - scales = [widget for widget in self.winfo_children() if isinstance(widget, tk.Scale)] - [scale.bind('', partial(self.reset_gain, index=i)) for i, scale in enumerate(scales)] + [ + self._make_single_channel(i, j) + for i, j in enumerate(i for i in range(0, self.col * 2, 2)) + ] + scales = [ + widget for widget in self.winfo_children() if isinstance(widget, tk.Scale) + ] + [ + scale.bind("", partial(self.reset_gain, index=i)) + for i, scale in enumerate(scales) + ] """ configure grid """ self.col_row_configure() @@ -60,68 +74,86 @@ class App(tk.Tk): @property def id_(self): - return 'strip' + return "strip" def _make_single_channel(self, i, j): """ Creates a label, progressbar, scale, and mute """ - ttk.Label(self, text=f'{vban.strip[i].label}').grid(column=j, row=0, columnspan=2) + ttk.Label(self, text=f"{vban.strip[i].label}").grid( + column=j, row=0, columnspan=2 + ) - ttk.Progressbar(self, maximum=72, orient='vertical', mode='determinate', variable=self.levels[self.id_][i]).grid(column=j, row=1) - ttk.Scale(self, from_=12.0, to=-60.0, orient='vertical', variable=self.gains[self.id_][i], - command=partial(self.scale_callback, index=i)).grid(column=j+1, row=1) + ttk.Progressbar( + self, + maximum=72, + orient="vertical", + mode="determinate", + variable=self.levels[self.id_][i], + ).grid(column=j, row=1) + ttk.Scale( + self, + from_=12.0, + to=-60.0, + orient="vertical", + variable=self.gains[self.id_][i], + command=partial(self.scale_callback, index=i), + ).grid(column=j + 1, row=1) - ttk.Button(self, text='MUTE', - command=partial(self.toggle, 'mute', i), style=f'Mute{i}.TButton').grid(column=j, row=2, columnspan=2, sticky=(tk.W, tk.E)) + ttk.Button( + self, + text="MUTE", + command=partial(self.toggle, "mute", i), + style=f"Mute{i}.TButton", + ).grid(column=j, row=2, columnspan=2, sticky=(tk.W, tk.E)) def scale_callback(self, *args, index=None): - """ callback function for scale widgets """ + """callback function for scale widgets""" vban.strip[index].gain = self.gains[self.id_][index].get() def reset_gain(self, *args, index=None): - """ reset gain to 0 when double click mouse """ + """reset gain to 0 when double click mouse""" vban.strip[index].gain = 0 self.gains[self.id_][index].set(0) def toggle(self, param, index): - """ toggles a strip parameter """ + """toggles a strip parameter""" setattr(vban.strip[index], param, not getattr(vban.strip[index], param)) def col_row_configure(self): - [self.columnconfigure(i, weight=1) for i in range(self.col*2)] - [child.grid_configure(padx=1, pady=1) - for child in self.winfo_children()] + [self.columnconfigure(i, weight=1) for i in range(self.col * 2)] + [child.grid_configure(padx=1, pady=1) for child in self.winfo_children()] def watch_levels(self, i): self.after(1, self.watch_levels_step, i) def watch_levels_step(self, i): val = vban.strip[i].levels.prefader[0] + vban.strip[i].gain - self.levels[self.id_][i].set((0 if vban.strip[i].mute else 100 + (val-30))) + self.levels[self.id_][i].set((0 if vban.strip[i].mute else 100 + (val - 30))) self.after(20, self.watch_levels_step, i) -_apps = {kind.id: App.make(kind) for kind in kinds.all} +_apps = {kind.id: App.make(kind) for kind in kinds.all} + def connect(kind_id: str) -> App: - """ return App of the kind requested """ + """return App of the kind requested""" try: APP_cls = _apps[kind_id] return APP_cls() except KeyError: - raise ExampleAppErrors(f'Invalid kind: {kind_id}') + raise ExampleAppErrors(f"Invalid kind: {kind_id}") if __name__ == "__main__": - kind_id = 'potato' + kind_id = "potato" opts = { # make sure VBAN is configured on remote machine then set IP accordingly - 'ip': 'ws.local', - 'streamname': 'testing', - 'port': 6990, + "ip": "ws.local", + "streamname": "testing", + "port": 6990, } - with vbancmd.connect(kind_id, **opts) as vban: + with vban_cmd.connect(kind_id, **opts) as vban: app = connect(kind_id) app.mainloop() diff --git a/poetry.lock b/poetry.lock new file mode 100644 index 0000000..094e0c4 --- /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.5" +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.5-py2.py3-none-any.whl", hash = "sha256:854bf444933e37f5824ae7bfc1e98d5bce2ebe4160d46b5edf346a89358e99da"}, + {file = "colorama-0.4.5.tar.gz", hash = "sha256:e6c6b4334fc50988a639d9b98aa429a0b57da6e17b9a44f0451f930b6967b7a4"}, +] +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..800064b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[tool.poetry] +name = "vban-cmd" +version = "1.0.1" +description = "Python interface for the VBAN RT Packet Service (Sendtext)" +authors = ["onyx-and-iris "] +license = "MIT" +readme = "README.md" +repository = "https://github.com/onyx-and-iris/vban-cmd-python" + +[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/setup.py b/setup.py deleted file mode 100644 index 4149104..0000000 --- a/setup.py +++ /dev/null @@ -1,10 +0,0 @@ -from setuptools import setup - -setup( - name="vbancmd", - version="0.4.0", - description="VBAN CMD Python API", - packages=["vbancmd"], - install_requires=["toml"], - extras_require={"development": ["pytest", "pytest-randomly", "genbadge[tests]"]}, -) diff --git a/tests/__init__.py b/tests/__init__.py index bafbf16..4c10f5e 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,30 +1,32 @@ -from dataclasses import dataclass -import vbancmd -from vbancmd import kinds import random import sys +from dataclasses import dataclass + +import vban_cmd +from vban_cmd.kinds import KindId, kinds_all +from vban_cmd.kinds import request_kind_map as kindmap # let's keep things random -kind_id = random.choice(tuple(kind.id for kind in kinds.all)) +kind_id = random.choice(tuple(kind_id.name.lower() for kind_id in KindId)) opts = { - "ip": "codey.local", - "streamname": "codey", + "ip": "ws.local", + "streamname": "workstation", "port": 6990, "bps": 0, "sync": True, } -vbans = {kind.id: vbancmd.connect(kind_id, **opts) for kind in kinds.all} +vbans = {kind.name: vban_cmd.api(kind.name, **opts) for kind in kinds_all} tests = vbans[kind_id] -kind = kinds.get(kind_id) +kind = kindmap(kind_id) @dataclass class Data: """bounds data to map tests to a kind""" - name: str = kind.id + 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 @@ -41,7 +43,6 @@ data = Data() def setup_module(): print(f"\nRunning tests for kind [{data.name}]\n", file=sys.stdout) tests.login() - tests.apply_profile("blank") def teardown_module(): diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..92df958 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,7 @@ +def pytest_addoption(parser): + parser.addoption( + "--run-slow", + action="store_true", + default=False, + help="Run slow tests", + ) diff --git a/tests/pre-commit.ps1 b/tests/pre-commit.ps1 deleted file mode 100644 index 53f5bbd..0000000 --- a/tests/pre-commit.ps1 +++ /dev/null @@ -1,31 +0,0 @@ -Function RunTests { - $coverage = "./tests/pytest_coverage.log" - $run_tests = "pytest -v --capture=tee-sys --junitxml=./tests/.coverage.xml" - $match_pattern = "^=|^\s*$|^Running|^Using|^plugins|^collecting|^tests" - - if ( Test-Path $coverage ) { Clear-Content $coverage } - - ForEach ($line in $(Invoke-Expression $run_tests)) { - If ( $line -Match $match_pattern ) { - if ( $line -Match "^Running tests for kind \[(\w+)\]" ) { $kind = $Matches[1] } - $line | Tee-Object -FilePath $coverage -Append - } - } - Write-Output "$(Get-TimeStamp)" | Out-file $coverage -Append - - Invoke-Expression "genbadge tests -t 90 -i ./tests/.coverage.xml -o ./tests/$kind.svg" -} - -Function Get-TimeStamp { - - return "[{0:MM/dd/yy} {0:HH:mm:ss}]" -f (Get-Date) - -} - -if ($MyInvocation.InvocationName -ne ".") { - Invoke-Expression ".\venv\Scripts\Activate.ps1" - - RunTests - - Invoke-Expression "deactivate" -} diff --git a/tests/test_higher.py b/tests/test_higher.py index 2d7c121..398cbbb 100644 --- a/tests/test_higher.py +++ b/tests/test_higher.py @@ -1,5 +1,6 @@ import pytest -from tests import tests, data + +from tests import data, tests @pytest.mark.parametrize("value", [False, True]) @@ -12,15 +13,27 @@ class TestSetAndGetBoolHigher: "index,param", [ (data.phys_in, "mute"), - (data.phys_in, "mono"), - (data.virt_in, "mc"), - (data.virt_in, "mono"), + (data.virt_in, "solo"), ], ) 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 + @pytest.mark.skipif( + data.name == "banana", + reason="Only test if logged into Basic or Potato version", + ) + @pytest.mark.parametrize( + "index,param", + [ + (data.phys_in, "mc"), + ], + ) + def test_it_sets_and_gets_strip_bool_params_mc(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( diff --git a/tests/test_lower.py b/tests/test_lower.py index 43d0f39..06962d4 100644 --- a/tests/test_lower.py +++ b/tests/test_lower.py @@ -1,7 +1,9 @@ +import time + import pytest -from tests import tests, data -from vbancmd import kinds -import re +from vban_cmd import kinds + +from tests import data, tests class TestPublicPacketLower: @@ -10,9 +12,15 @@ class TestPublicPacketLower: """Tests for a valid rt data packet""" def test_it_gets_an_rt_data_packet(self): - assert tests.public_packet.voicemeetertype in (kind.id for kind in kinds.all) + assert tests.public_packet.voicemeetertype in ( + kind.name for kind in kinds.kinds_all + ) +@pytest.mark.skipif( + "not config.getoption('--run-slow')", + reason="Only run when --run-slow is given", +) @pytest.mark.parametrize("value", [0, 1]) class TestSetRT: __test__ = True @@ -26,7 +34,8 @@ class TestSetRT: ("bus", data.virt_out, "mono"), ], ) - def test_it_gets_an_rt_data_packet(self, kls, index, param, value): - tests.set_rt(f"{kls}[{index}]", param, value) + def test_it_sends_a_text_request(self, kls, index, param, value): + tests._set_rt(f"{kls}[{index}]", param, value) + time.sleep(0.1) target = getattr(tests, kls)[index] assert getattr(target, param) == bool(value) diff --git a/vban_cmd/__init__.py b/vban_cmd/__init__.py new file mode 100644 index 0000000..ed7e87a --- /dev/null +++ b/vban_cmd/__init__.py @@ -0,0 +1,3 @@ +from .factory import request_vbancmd_obj as api + +__ALL__ = ["api"] diff --git a/vban_cmd/base.py b/vban_cmd/base.py new file mode 100644 index 0000000..2dd967c --- /dev/null +++ b/vban_cmd/base.py @@ -0,0 +1,269 @@ +import socket +import time +from abc import ABCMeta, abstractmethod +from enum import IntEnum +from threading import Thread +from typing import NoReturn, Optional, Union + +from .packet import ( + HEADER_SIZE, + RegisterRTHeader, + TextRequestHeader, + VBAN_VMRT_Packet_Data, + VBAN_VMRT_Packet_Header, +) +from .subject import Subject +from .util import script + +Socket = IntEnum("Socket", "register request response", start=0) + + +class VbanCmd(metaclass=ABCMeta): + """Base class responsible for communicating over VBAN RT Service""" + + DELAY = 0.001 + # fmt: off + BPS_OPTS = [ + 0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250, + 38400, 57600, 115200, 128000, 230400, 250000, 256000, 460800, 921600, + 1000000, 1500000, 2000000, 3000000, + ] + # fmt: on + + def __init__(self, **kwargs): + for attr, val in kwargs.items(): + setattr(self, attr, val) + + self.text_header = TextRequestHeader( + name=self.streamname, + bps_index=self.BPS_OPTS.index(self.bps), + channel=self.channel, + ) + self.register_header = RegisterRTHeader() + self.expected_packet = VBAN_VMRT_Packet_Header() + + self.socks = tuple( + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + for _, _ in enumerate(Socket) + ) + self.running = True + self.subject = Subject() + self.cache = {} + + @abstractmethod + def __str__(self): + """Ensure subclasses override str magic method""" + pass + + def __enter__(self): + self.login() + return self + + def login(self): + """Start listening for RT Packets""" + + self.socks[Socket.response.value].bind( + (socket.gethostbyname(socket.gethostname()), self.port) + ) + worker = Thread(target=self._send_register_rt, daemon=True) + worker.start() + self._public_packet = self._get_rt() + worker2 = Thread(target=self._updates, daemon=True) + worker2.start() + + def _send_register_rt(self): + """Fires a subscription packet every 10 seconds""" + + while self.running: + self.socks[Socket.register.value].sendto( + self.register_header.header, + (socket.gethostbyname(self.ip), self.port), + ) + count = int.from_bytes(self.register_header.framecounter, "little") + 1 + self.register_header.framecounter = count.to_bytes(4, "little") + time.sleep(10) + + def _fetch_rt_packet(self) -> Optional[VBAN_VMRT_Packet_Data]: + """Returns a valid RT Data Packet or None""" + data, _ = self.socks[Socket.response.value].recvfrom(2048) + # check for packet data + if len(data) > HEADER_SIZE: + # check if packet is of type VBAN + if self.expected_packet.header == data[: HEADER_SIZE - 4]: + # check if packet is of type vmrt_data + if int.from_bytes(data[4:5]) == int(0x60): + return VBAN_VMRT_Packet_Data( + _voicemeeterType=data[28:29], + _reserved=data[29:30], + _buffersize=data[30:32], + _voicemeeterVersion=data[32:36], + _optionBits=data[36:40], + _samplerate=data[40:44], + _inputLeveldB100=data[44:112], + _outputLeveldB100=data[112:240], + _TransportBit=data[240:244], + _stripState=data[244:276], + _busState=data[276:308], + _stripGaindB100Layer1=data[308:324], + _stripGaindB100Layer2=data[324:340], + _stripGaindB100Layer3=data[340:356], + _stripGaindB100Layer4=data[356:372], + _stripGaindB100Layer5=data[372:388], + _stripGaindB100Layer6=data[388:404], + _stripGaindB100Layer7=data[404:420], + _stripGaindB100Layer8=data[420:436], + _busGaindB100=data[436:452], + _stripLabelUTF8c60=data[452:932], + _busLabelUTF8c60=data[932:1412], + ) + + def _get_rt(self) -> VBAN_VMRT_Packet_Data: + """Attempt to fetch data packet until a valid one found""" + + def fget(): + data = False + while not data: + data = self._fetch_rt_packet() + time.sleep(self.DELAY) + return data + + return fget() + + def _set_rt( + self, + id_: str, + param: Optional[str] = None, + val: Optional[Union[int, float]] = None, + ): + """Sends a string request command over a network.""" + cmd = id_ if not param else f"{id_}.{param}={val}" + self.socks[Socket.request.value].sendto( + self.text_header.header + cmd.encode(), + (socket.gethostbyname(self.ip), self.port), + ) + count = int.from_bytes(self.text_header.framecounter, "little") + 1 + self.text_header.framecounter = count.to_bytes(4, "little") + if param: + self.cache[f"{id_}.{param}"] = val + + @script + def sendtext(self, cmd): + """Sends a multiple parameter string over a network.""" + self._set_rt(cmd) + time.sleep(self.DELAY) + + @property + def type(self) -> str: + """Returns the type of Voicemeeter installation.""" + return self.public_packet.voicemeetertype + + @property + def version(self) -> str: + """Returns Voicemeeter's version as a tuple""" + v1, v2, v3, v4 = self.public_packet.voicemeeterversion + return f"{v1}.{v2}.{v3}.{v4}" + + @property + def pdirty(self): + """True iff a parameter has changed""" + return self._pdirty + + @property + def ldirty(self): + """True iff a level value has changed.""" + return self._ldirty + + @property + def public_packet(self): + return self._public_packet + + def clear_dirty(self): + while self.pdirty: + pass + + def _updates(self) -> NoReturn: + while self.running: + private_packet = self._get_rt() + strip_comp, bus_comp = ( + tuple( + not a == b + for a, b in zip( + private_packet.inputlevels, self.public_packet.inputlevels + ) + ), + tuple( + not a == b + for a, b in zip( + private_packet.outputlevels, self.public_packet.outputlevels + ) + ), + ) + + if self._public_packet != private_packet: + self._public_packet = private_packet + if private_packet.pdirty(self.public_packet): + self.subject.notify("pdirty") + if any(any(list_) for list_ in (strip_comp, bus_comp)): + self.subject.notify( + "ldirty", + ( + self.public_packet.inputlevels, + strip_comp, + self.public_packet.outputlevels, + bus_comp, + ), + ) + time.sleep(self.ratelimit) + + @property + def strip_levels(self): + """Returns the full strip level array for a kind, PREFADER mode, before math conversion""" + return tuple( + list(filter(lambda x: x != ((1 << 16) - 1), self.public_packet.inputlevels)) + ) + + @property + def bus_levels(self): + """Returns the full bus level array for a kind, before math conversion""" + return tuple( + list( + filter(lambda x: x != ((1 << 16) - 1), self.public_packet.outputlevels) + ) + ) + + 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"): + return getattr(self, obj)[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): + self.running = False + time.sleep(0.2) + [sock.close() for sock in self.socks] + + def __exit__(self, exc_type, exc_value, exc_traceback): + self.logout() diff --git a/vban_cmd/bus.py b/vban_cmd/bus.py new file mode 100644 index 0000000..bdf3310 --- /dev/null +++ b/vban_cmd/bus.py @@ -0,0 +1,151 @@ +from abc import abstractmethod +from typing import Union + +from .iremote import IRemote +from .meta import bus_mode_prop, channel_bool_prop, channel_label_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 gain(self) -> float: + def fget(): + val = self.public_packet.busgain[self.index] + if val < 10000: + return -val + elif val == ((1 << 16) - 1): + return 0 + else: + return ((1 << 16) - 1) - val + + val = self.getter("gain") + if val is None: + val = fget() * 0.01 + return round(val, 1) + + @gain.setter + def gain(self, val: float): + self.setter("gain", val) + + +class PhysicalBus(Bus): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + @property + def device(self) -> str: + return + + @property + def sr(self) -> int: + return + + +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.""" + + range_ = self.level_map[self.index] + return tuple( + round(-i * 0.01, 1) 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), + **{param: channel_bool_prop(param) for param in ["mute", "mono"]}, + "eq": channel_bool_prop("eq.On"), + "eq_ab": channel_bool_prop("eq.ab"), + "label": channel_label_prop(), + }, + )(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/vbancmd/command.py b/vban_cmd/command.py similarity index 70% rename from vbancmd/command.py rename to vban_cmd/command.py index 67a3c34..96af0ad 100644 --- a/vbancmd/command.py +++ b/vban_cmd/command.py @@ -1,25 +1,14 @@ -import abc -from .errors import VMCMDErrors +from .error import VMCMDErrors +from .iremote import IRemote from .meta import action_prop -class ICommand(abc.ABC): - """Command Base Class""" +class Command(IRemote): + """ + Implements the common interface - def __init__(self, remote): - self._remote = remote - - def setter(self, param, val): - """Sends a string request RT packet.""" - self._remote.set_rt(f"{self.identifier}", param, val) - - @abc.abstractmethod - def identifier(self): - pass - - -class Command(ICommand): - """Command Concrete Class""" + Defines concrete implementation for command + """ @classmethod def make(cls, remote): @@ -29,7 +18,7 @@ class Command(ICommand): Returns a Command class of a kind. """ CMD_cls = type( - f"Command{remote.kind.name}", + f"Command{remote.kind}", (cls,), { **{ diff --git a/vban_cmd/config.py b/vban_cmd/config.py new file mode 100644 index 0000000..678c288 --- /dev/null +++ b/vban_cmd/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/vbancmd/errors.py b/vban_cmd/error.py similarity index 60% rename from vbancmd/errors.py rename to vban_cmd/error.py index 43e9b48..909fe12 100644 --- a/vbancmd/errors.py +++ b/vban_cmd/error.py @@ -1,2 +1,4 @@ class VMCMDErrors(Exception): + """general errors""" + pass diff --git a/vban_cmd/factory.py b/vban_cmd/factory.py new file mode 100644 index 0000000..f4339ce --- /dev/null +++ b/vban_cmd/factory.py @@ -0,0 +1,190 @@ +from abc import abstractmethod +from enum import IntEnum +from functools import cached_property +from typing import Iterable, NoReturn, Self + +from .base import VbanCmd +from .bus import request_bus_obj as bus +from .command import Command +from .config import request_config as configs +from .kinds import KindMapClass +from .kinds import request_kind_map as kindmap +from .strip import request_strip_obj as strip + + +class FactoryBuilder: + """ + Builder class for factories. + + Separates construction from representation. + """ + + BuilderProgress = IntEnum("BuilderProgress", "strip bus command", 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}", + ) + + 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 + + +class FactoryBase(VbanCmd): + """Base class for factories, subclasses VbanCmd.""" + + def __init__(self, kind_id: str, **kwargs): + defaultkwargs = { + "ip": None, + "port": 6980, + "streamname": "Command1", + "bps": 0, + "channel": 0, + "ratelimit": 0, + "sync": False, + } + 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._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 VbanCmd 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 VbanCmd 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 + + +class PotatoFactory(FactoryBase): + """ + Represents a Potato VbanCmd 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 + + +def vbancmd_factory(kind_id: str, **kwargs) -> VbanCmd: + """ + Factory method, invokes a factory creation class of a kind + + Returns a VbanCmd 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"VbanCmd{kind_id.capitalize()}", (_factory,), {})(kind_id, **kwargs) + + +def request_vbancmd_obj(kind_id: str, **kwargs) -> VbanCmd: + """ + Interface entry point. Wraps factory method and handles errors + + Returns a reference to a VbanCmd class of a kind + """ + VBANCMD_obj = None + try: + VBANCMD_obj = vbancmd_factory(kind_id, **kwargs) + except (ValueError, TypeError) as e: + raise SystemExit(e) + return VBANCMD_obj diff --git a/vbancmd/channel.py b/vban_cmd/iremote.py similarity index 75% rename from vbancmd/channel.py rename to vban_cmd/iremote.py index d4cb008..6de79f6 100644 --- a/vbancmd/channel.py +++ b/vban_cmd/iremote.py @@ -1,7 +1,6 @@ -import abc -from .errors import VMCMDErrors +import time +from abc import ABCMeta, abstractmethod from dataclasses import dataclass -from time import sleep @dataclass @@ -76,24 +75,28 @@ class Modes: ) -class IChannel(abc.ABC): - """Base class for InputStrip and OutputBus.""" +class IRemote(metaclass=ABCMeta): + """ + Common interface between base class and extended (higher) classes - def __init__(self, remote, index): + Provides some default implementation + """ + + def __init__(self, remote, index=None): self._remote = remote self.index = index self._modes = Modes() def getter(self, param): - cmd = f"{self.identifier}[{self.index}].{param}" + cmd = f"{self.identifier}.{param}" if cmd in self._remote.cache: return self._remote.cache.pop(cmd) def setter(self, param, val): """Sends a string request RT packet.""" - self._remote.set_rt(f"{self.identifier}[{self.index}]", param, val) + self._remote._set_rt(f"{self.identifier}", param, val) - @abc.abstractmethod + @abstractmethod def identifier(self): pass @@ -102,13 +105,16 @@ class IChannel(abc.ABC): """Returns an RT data packet.""" return self._remote.public_packet - def apply(self, mapping): + def apply(self, data): """Sets all parameters of a dict for the channel.""" script = "" - for key, val in mapping.items(): - if not hasattr(self, key): - raise VMCMDErrors(f"Invalid {self.identifier} attribute: {key}") - self._remote.cache[f"{self.identifier}[{self.index}].{key}"] = val - script += f"{self.identifier}[{self.index}].{key}={val};" + for attr, val in data.items(): + if hasattr(self, attr): + self._remote.cache[f"{self.identifier}[{self.index}].{attr}"] = val + script += f"{self.identifier}[{self.index}].{attr}={val};" self._remote.sendtext(script) + return self + + def then_wait(self): + time.sleep(self._remote.DELAY) diff --git a/vban_cmd/kinds.py b/vban_cmd/kinds.py new file mode 100644 index 0000000..0ca4003 --- /dev/null +++ b/vban_cmd/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/vbancmd/meta.py b/vban_cmd/meta.py similarity index 82% rename from vbancmd/meta.py rename to vban_cmd/meta.py index 4830e3e..cf500e6 100644 --- a/vbancmd/meta.py +++ b/vban_cmd/meta.py @@ -1,17 +1,20 @@ -from .util import cache_bool, cache_string -from .errors import VMCMDErrors - from functools import partial +from .error import VMCMDErrors +from .util import cache_bool, cache_string + def channel_bool_prop(param): - """A channel bool prop. (strip|bus)""" + """meta function for channel boolean parameters""" @partial(cache_bool, param=param) def fget(self): return ( not int.from_bytes( - getattr(self.public_packet, f"{self.identifier}state")[self.index], + getattr( + self.public_packet, + f"{'strip' if 'strip' in type(self).__name__.lower() else 'bus'}state", + )[self.index], "little", ) & getattr(self._modes, f'_{param.replace(".", "_").lower()}') @@ -27,11 +30,14 @@ def channel_bool_prop(param): def channel_label_prop(): - """A channel label prop. (strip|bus)""" + """meta function for channel label parameters""" @partial(cache_string, param="label") def fget(self) -> str: - return getattr(self.public_packet, f"{self.identifier}labels")[self.index] + return getattr( + self.public_packet, + f"{'strip' if 'strip' in type(self).__name__.lower() else 'bus'}labels", + )[self.index] def fset(self, val: str): if not isinstance(val, str): @@ -42,7 +48,7 @@ def channel_label_prop(): def strip_output_prop(param): - """A strip output prop. (A1-A5, B1-B3)""" + """meta function for strip output parameters. (A1-A5, B1-B3)""" @partial(cache_bool, param=param) def fget(self): @@ -61,7 +67,7 @@ def strip_output_prop(param): def bus_mode_prop(param): - """A bus mode prop.""" + """meta function for bus mode parameters""" @partial(cache_bool, param=param) def fget(self): diff --git a/vbancmd/dataclass.py b/vban_cmd/packet.py similarity index 97% rename from vbancmd/dataclass.py rename to vban_cmd/packet.py index 36f3328..1f22b53 100644 --- a/vbancmd/dataclass.py +++ b/vban_cmd/packet.py @@ -8,7 +8,7 @@ HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16 + 4 @dataclass class VBAN_VMRT_Packet_Data: - """RT Packet Data""" + """Represents the structure of a VMRT data packet""" _voicemeeterType: bytes _reserved: bytes @@ -198,7 +198,7 @@ class VBAN_VMRT_Packet_Data: @dataclass class VBAN_VMRT_Packet_Header: - """RT PACKET header (expected from Voicemeeter server)""" + """Represents a RESPONSE RT PACKET header""" name = "Voicemeeter-RTP" vban: bytes = "VBAN".encode() @@ -220,36 +220,9 @@ class VBAN_VMRT_Packet_Header: return header -@dataclass -class RegisterRTHeader: - """REGISTER RT PACKET header""" - - name = "Register RTP" - timeout = 15 - vban: bytes = "VBAN".encode() - format_sr: bytes = (0x60).to_bytes(1, "little") - format_nbs: bytes = (0).to_bytes(1, "little") - format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, "little") - format_bit: bytes = (timeout & 0x000000FF).to_bytes(1, "little") # timeout - streamname: bytes = name.encode("ascii") + bytes(16 - len(name)) - framecounter: bytes = (0).to_bytes(4, "little") - - @property - def header(self): - header = self.vban - header += self.format_sr - header += self.format_nbs - header += self.format_nbc - header += self.format_bit - header += self.streamname - header += self.framecounter - assert len(header) == HEADER_SIZE, f"Header expected {HEADER_SIZE} bytes" - return header - - @dataclass class TextRequestHeader: - """VBAN-TEXT request header""" + """Represents a REQUEST RT PACKET header""" name: str bps_index: int @@ -282,3 +255,30 @@ class TextRequestHeader: header += self.framecounter assert len(header) == HEADER_SIZE, f"Header expected {HEADER_SIZE} bytes" return header + + +@dataclass +class RegisterRTHeader: + """Represents a REGISTER RT PACKET header""" + + name = "Register RTP" + timeout = 15 + vban: bytes = "VBAN".encode() + format_sr: bytes = (0x60).to_bytes(1, "little") + format_nbs: bytes = (0).to_bytes(1, "little") + format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, "little") + format_bit: bytes = (timeout & 0x000000FF).to_bytes(1, "little") # timeout + streamname: bytes = name.encode("ascii") + bytes(16 - len(name)) + framecounter: bytes = (0).to_bytes(4, "little") + + @property + def header(self): + header = self.vban + header += self.format_sr + header += self.format_nbs + header += self.format_nbc + header += self.format_bit + header += self.streamname + header += self.framecounter + assert len(header) == HEADER_SIZE, f"Header expected {HEADER_SIZE} bytes" + return header diff --git a/vban_cmd/strip.py b/vban_cmd/strip.py new file mode 100644 index 0000000..813e05e --- /dev/null +++ b/vban_cmd/strip.py @@ -0,0 +1,225 @@ +from abc import abstractmethod +from typing import Union + +from .iremote import IRemote +from .kinds import kinds_all +from .meta import channel_bool_prop, channel_label_prop, strip_output_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 limit(self) -> int: + return + + @limit.setter + def limit(self, val: int): + self.setter("limit", val) + + @property + def gain(self) -> float: + val = self.getter("gain") + if val is None: + val = self.gainlayer[0].gain + return round(val, 1) + + @gain.setter + def gain(self, val: float): + self.setter("gain", val) + + +class PhysicalStrip(Strip): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + @property + def comp(self) -> float: + return + + @comp.setter + def comp(self, val: float): + self.setter("Comp", val) + + @property + def gate(self) -> float: + return + + @gate.setter + def gate(self, val: float): + self.setter("gate", val) + + @property + def device(self): + return + + @property + def sr(self): + return + + +class VirtualStrip(Strip): + def __str__(self): + return f"{type(self).__name__}{self.index}" + + mc = channel_bool_prop("mc") + + mono = mc + + @property + def k(self) -> int: + return + + @k.setter + def k(self, val: int): + self.setter("karaoke", val) + + +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_prefader(self): + range_ = self.level_map[self.index] + return tuple( + round(-i * 0.01, 1) + 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 + + @property + def postmute(self) -> tuple: + return + + @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) -> float: + def fget(): + val = getattr(self.public_packet, f"stripgainlayer{self._i+1}")[self.index] + if val < 10000: + return -val + elif val == ((1 << 16) - 1): + return 0 + else: + return ((1 << 16) - 1) - val + + val = self.getter(f"GainLayer[{self._i}]") + if val is None: + val = fget() * 0.01 + return round(val, 1) + + @gain.setter + def gain(self, val: float): + 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}": strip_output_prop(f"A{i}") for i in range(1, kind.phys_out + 1) + }, + **{ + f"B{i}": strip_output_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] + GAINLAYERMIXIN_cls = _make_gainlayer_mixin(remote, i) + + return type( + f"{STRIP_cls.__name__}{remote.kind}", + (STRIP_cls, CHANNELOUTMIXIN_cls, GAINLAYERMIXIN_cls), + { + "levels": StripLevel(remote, i), + **{param: channel_bool_prop(param) for param in ["mono", "solo", "mute"]}, + "label": channel_label_prop(), + }, + )(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/vban_cmd/subject.py b/vban_cmd/subject.py new file mode 100644 index 0000000..643549d --- /dev/null +++ b/vban_cmd/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/vbancmd/util.py b/vban_cmd/util.py similarity index 85% rename from vbancmd/util.py rename to vban_cmd/util.py index db7236d..9f23661 100644 --- a/vbancmd/util.py +++ b/vban_cmd/util.py @@ -1,19 +1,9 @@ -from pathlib import Path - - -PROJECT_DIR = str(Path(__file__).parents[1]) - - -def project_path(): - return PROJECT_DIR - - def cache_bool(func, param): """Check cache for a bool prop""" def wrapper(*args, **kwargs): self, *rem = args - cmd = f"{self.identifier}[{self.index}].{param}" + cmd = f"{self.identifier}.{param}" if cmd in self._remote.cache: return self._remote.cache.pop(cmd) == 1 return func(*args, **kwargs) @@ -26,7 +16,7 @@ def cache_string(func, param): def wrapper(*args, **kwargs): self, *rem = args - cmd = f"{self.identifier}[{self.index}].{param}" + cmd = f"{self.identifier}.{param}" if cmd in self._remote.cache: return self._remote.cache.pop(cmd) return func(*args, **kwargs) diff --git a/vbancmd/__init__.py b/vbancmd/__init__.py deleted file mode 100644 index 878a80e..0000000 --- a/vbancmd/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .vbancmd import connect - -__ALL__ = ["connect"] diff --git a/vbancmd/bus.py b/vbancmd/bus.py deleted file mode 100644 index ed4f9c3..0000000 --- a/vbancmd/bus.py +++ /dev/null @@ -1,132 +0,0 @@ -from .errors import VMCMDErrors -from .channel import IChannel -from . import kinds -from .meta import bus_mode_prop, channel_bool_prop, channel_label_prop - - -class OutputBus(IChannel): - """Base class for output buses.""" - - @classmethod - def make(cls, is_physical, remote, index, *args, **kwargs): - """ - Factory function for output busses. - Returns a physical/virtual bus of a kind. - """ - BusModeMixin = _make_bus_mode_mixin(IChannel) - OutputBus = PhysicalOutputBus if is_physical else VirtualOutputBus - OB_cls = type( - f"Bus{remote.kind.name}", - (OutputBus,), - { - "levels": BusLevel(remote, index), - "mode": BusModeMixin(remote, index), - **{param: channel_bool_prop(param) for param in ["mute", "mono"]}, - "eq": channel_bool_prop("eq.On"), - "eq_ab": channel_bool_prop("eq.ab"), - "label": channel_label_prop(), - }, - ) - return OB_cls(remote, index, *args, **kwargs) - - @property - def identifier(self): - return "bus" - - @property - def gain(self) -> float: - def fget(): - val = self.public_packet.busgain[self.index] - if val < 10000: - return -val - elif val == ((1 << 16) - 1): - return 0 - else: - return ((1 << 16) - 1) - val - - val = self.getter("gain") - if val is None: - val = fget() * 0.01 - return round(val, 1) - - @gain.setter - def gain(self, val: float): - self.setter("gain", val) - - -class PhysicalOutputBus(OutputBus): - @property - def device(self) -> str: - return - - @property - def sr(self) -> int: - return - - -class VirtualOutputBus(OutputBus): - pass - - -class BusLevel(IChannel): - def __init__(self, remote, index): - super().__init__(remote, index) - self.level_map = _bus_maps[remote.kind.id] - - @property - def identifier(self) -> str: - return f"Bus[{self.index}]" - - def getter_level(self, mode=None): - def fget(i, data): - val = data.outputlevels[i] - return -val * 0.01 - - range_ = self.level_map[self.index] - data = self.public_packet - levels = tuple(round(fget(i, data), 1) for i in range(*range_)) - return levels - - @property - def all(self) -> tuple: - return self.getter_level() - - -def _make_bus_level_map(kind): - phys_out, virt_out = kind.outs - return tuple((i, i + 8) for i in range(0, (phys_out + virt_out) * 8, 8)) - - -_bus_maps = {kind.id: _make_bus_level_map(kind) for kind in kinds.all} - - -def _make_bus_mode_mixin(kls): - """Creates a mixin of Bus Modes.""" - - def identifier(self) -> str: - return f"Bus[{self.index}].mode" - - return type( - "BusModeMixin", - (kls,), - { - "identifier": property(identifier), - **{ - mode: bus_mode_prop(mode) - for mode in [ - "normal", - "amix", - "bmix", - "repeat", - "composite", - "tvmix", - "upmix21", - "upmix41", - "upmix61", - "centeronly", - "lfeonly", - "rearonly", - ] - }, - }, - ) diff --git a/vbancmd/kinds.py b/vbancmd/kinds.py deleted file mode 100644 index a9e56ab..0000000 --- a/vbancmd/kinds.py +++ /dev/null @@ -1,30 +0,0 @@ -import sys -import platform -from collections import namedtuple -from .errors import VMCMDErrors - -""" -Represents a major version of Voicemeeter and describes -its strip layout. -""" -VMKind = namedtuple("VMKind", ["id", "name", "ins", "outs", "executable", "vban"]) - -bits = 64 if sys.maxsize > 2**32 else 32 -os = platform.system() -# fmt: off -_kind_map = { - "basic": VMKind("basic", "Basic", (2, 1), (1, 1), "voicemeeter.exe", (4, 4)), - "banana": VMKind("banana", "Banana", (3, 2), (3, 2), "voicemeeterpro.exe", (8, 8)), - "potato": VMKind("potato", "Potato", (5, 3), (5, 3), f'voicemeeter8{"x64" if bits == 64 else ""}.exe', (8, 8),), -} -# fmt: on - - -def get(kind_id): - try: - return _kind_map[kind_id] - except KeyError: - raise VMCMDErrors(f"Invalid Voicemeeter kind: {kind_id}") - - -all = list(_kind_map.values()) diff --git a/vbancmd/profiles.py b/vbancmd/profiles.py deleted file mode 100644 index 97d4a42..0000000 --- a/vbancmd/profiles.py +++ /dev/null @@ -1,80 +0,0 @@ -import toml -from . import kinds -from .util import project_path -from pathlib import Path - -profiles = {} - - -def _make_blank_profile(kind): - phys_in, virt_in = kind.ins - phys_out, virt_out = kind.outs - all_input_strip_config = { - "gain": 0.0, - "solo": False, - "mute": False, - "mono": False, - **{f"A{i}": False for i in range(1, phys_out + 1)}, - **{f"B{i}": False for i in range(1, virt_out + 1)}, - } - phys_input_strip_config = { - "comp": 0.0, - "gate": 0.0, - } - output_bus_config = { - "gain": 0.0, - "eq": False, - "mute": False, - "mono": False, - } - all_ = {f"strip-{i}": all_input_strip_config for i in range(phys_in + virt_in)} - phys = {f"strip-{i}": phys_input_strip_config for i in range(phys_in)} - abc = all_ - for i in phys.keys(): - abc[i] = all_[i] | phys[i] - return { - **abc, - **{f"bus-{i}": output_bus_config for i in range(phys_out + virt_out)}, - } - - -def _make_base_profile(kind): - phys_in, virt_in = kind.ins - blank = _make_blank_profile(kind) - overrides = { - **{f"strip-{i}": dict(B1=True) for i in range(phys_in)}, - **{f"strip-{i}": dict(A1=True) for i in range(phys_in, phys_in + virt_in)}, - } - base = blank - for i in overrides.keys(): - base[i] = blank[i] | overrides[i] - return base - - -for kind in kinds.all: - profiles[kind.id] = { - "blank": _make_blank_profile(kind), - "base": _make_base_profile(kind), - } - -# Load profiles from config files in profiles//.toml -for kind in kinds.all: - profiles_paths = [ - Path(project_path()) / "profiles" / kind.id, - Path.cwd() / "profiles" / kind.id, - Path.home() / "Documents/Voicemeeter" / "profiles" / kind.id, - ] - for path in profiles_paths: - if path.is_dir(): - filenames = list(path.glob("*.toml")) - configs = {} - for filename in filenames: - name = filename.with_suffix("").stem - try: - configs[name] = toml.load(filename) - except toml.TomlDecodeError: - print(f"Invalid TOML profile: {kind.id}/{filename.stem}") - - for name, cfg in configs.items(): - print(f"Loaded profile {kind.id}/{name}") - profiles[kind.id][name] = cfg diff --git a/vbancmd/strip.py b/vbancmd/strip.py deleted file mode 100644 index 9849e30..0000000 --- a/vbancmd/strip.py +++ /dev/null @@ -1,200 +0,0 @@ -from .errors import VMCMDErrors -from .channel import IChannel -from . import kinds -from .meta import strip_output_prop, channel_bool_prop, channel_label_prop - - -class InputStrip(IChannel): - """Base class for input strips.""" - - @classmethod - def make(cls, is_physical, remote, index, **kwargs): - """ - Factory function for input strips. - Returns a physical/virtual strip of a kind. - """ - PhysStrip, VirtStrip = _strip_pairs[remote.kind.id] - InputStrip = PhysStrip if is_physical else VirtStrip - GainLayerMixin = _make_gainlayer_mixin(remote, index) - IS_cls = type( - f"Strip{remote.kind.name}", - (InputStrip, GainLayerMixin), - { - "levels": StripLevel(remote, index), - **{ - param: channel_bool_prop(param) - for param in ["mono", "solo", "mute"] - }, - "label": channel_label_prop(), - }, - ) - return IS_cls(remote, index, **kwargs) - - @property - def identifier(self): - return "strip" - - @property - def limit(self) -> int: - return - - @limit.setter - def limit(self, val: int): - self.setter("limit", val) - - @property - def gain(self) -> float: - val = self.getter("gain") - if val is None: - val = self.gainlayer[0].gain - return round(val, 1) - - @gain.setter - def gain(self, val: float): - self.setter("gain", val) - - -class PhysicalInputStrip(InputStrip): - @property - def comp(self) -> float: - return - - @comp.setter - def comp(self, val: float): - self.setter("Comp", val) - - @property - def gate(self) -> float: - return - - @gate.setter - def gate(self, val: float): - self.setter("gate", val) - - @property - def device(self): - return - - @property - def sr(self): - return - - -class VirtualInputStrip(InputStrip): - mc = channel_bool_prop("mc") - - mono = mc - - @property - def k(self) -> int: - return - - @k.setter - def k(self, val: int): - self.setter("karaoke", val) - - -class StripLevel(InputStrip): - def __init__(self, remote, index): - super().__init__(remote, index) - self.level_map = _strip_maps[remote.kind.id] - - def getter_level(self, mode=None): - def fget(i, data): - val = data.inputlevels[i] - return -val * 0.01 - - range_ = self.level_map[self.index] - data = self.public_packet - levels = tuple(round(fget(i, data), 1) for i in range(*range_)) - return levels - - @property - def prefader(self) -> tuple: - return self.getter_level() - - @property - def postfader(self) -> tuple: - return - - @property - def postmute(self) -> tuple: - return - - -class GainLayer(InputStrip): - def __init__(self, remote, index, i): - super().__init__(remote, index) - self._i = i - - @property - def gain(self) -> float: - def fget(): - val = getattr(self.public_packet, f"stripgainlayer{self._i+1}")[self.index] - if val < 10000: - return -val - elif val == ((1 << 16) - 1): - return 0 - else: - return ((1 << 16) - 1) - val - - val = self.getter(f"GainLayer[{self._i}]") - if val is None: - val = fget() * 0.01 - return round(val, 1) - - @gain.setter - def gain(self, val: float): - 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(8))}, - ) - - -def _make_strip_mixin(kind): - """Creates a mixin with the kind's strip layout set as class variables.""" - num_A, num_B = kind.outs - return type( - f"StripMixin{kind.name}", - (), - { - **{f"A{i}": strip_output_prop(f"A{i}") for i in range(1, num_A + 1)}, - **{f"B{i}": strip_output_prop(f"B{i}") for i in range(1, num_B + 1)}, - }, - ) - - -_strip_mixins = {kind.id: _make_strip_mixin(kind) for kind in kinds.all} - - -def _make_strip_pair(kind): - """Creates a PhysicalInputStrip and a VirtualInputStrip of a kind.""" - StripMixin = _strip_mixins[kind.id] - PhysStrip = type( - f"PhysicalInputStrip{kind.name}", (PhysicalInputStrip, StripMixin), {} - ) - VirtStrip = type( - f"VirtualInputStrip{kind.name}", (VirtualInputStrip, StripMixin), {} - ) - return (PhysStrip, VirtStrip) - - -_strip_pairs = {kind.id: _make_strip_pair(kind) for kind in kinds.all} - - -def _make_strip_level_map(kind): - phys_in, virt_in = kind.ins - phys_map = tuple((i, i + 2) for i in range(0, phys_in * 2, 2)) - virt_map = tuple( - (i, i + 8) for i in range(phys_in * 2, phys_in * 2 + virt_in * 8, 8) - ) - return phys_map + virt_map - - -_strip_maps = {kind.id: _make_strip_level_map(kind) for kind in kinds.all} diff --git a/vbancmd/subject.py b/vbancmd/subject.py deleted file mode 100644 index a3252fe..0000000 --- a/vbancmd/subject.py +++ /dev/null @@ -1,35 +0,0 @@ -class Subject: - def __init__(self): - """list of current observers""" - - self._observables = [] - - def notify(self, modifier=None, data=None): - """run callbacks on update""" - - for observer in self._observables: - observer.on_update(modifier, data) - - def add(self, observer): - """adds an observer to observables""" - - if observer not in self._observables: - self._observables.append(observer) - - def remove(self, observer): - """removes an observer from observables""" - - try: - self._observables.remove(observer) - except ValueError: - pass - - def get(self) -> list: - """returns the current observables""" - - return self._observables - - def clear(self): - """clears the observables list""" - - self._observables.clear() diff --git a/vbancmd/vbancmd.py b/vbancmd/vbancmd.py deleted file mode 100644 index cad184b..0000000 --- a/vbancmd/vbancmd.py +++ /dev/null @@ -1,393 +0,0 @@ -import abc -import select -import socket -from time import sleep -from threading import Thread -from typing import NamedTuple, NoReturn, Optional, Union - -from .errors import VMCMDErrors -from . import kinds -from . import profiles -from .dataclass import ( - HEADER_SIZE, - VBAN_VMRT_Packet_Data, - VBAN_VMRT_Packet_Header, - RegisterRTHeader, - TextRequestHeader, -) -from .strip import InputStrip -from .bus import OutputBus -from .command import Command -from .util import script -from .subject import Subject - - -class VbanCmd(abc.ABC): - def __init__(self, **kwargs): - self._ip = kwargs["ip"] - self._port = kwargs["port"] - self._streamname = kwargs["streamname"] - self._bps = kwargs["bps"] - self._channel = kwargs["channel"] - self._delay = kwargs["delay"] - self._sync = kwargs["sync"] - # fmt: off - self._bps_opts = [ - 0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250, - 38400, 57600, 115200, 128000, 230400, 250000, 256000, 460800, 921600, - 1000000, 1500000, 2000000, 3000000, - ] - # fmt: on - if self._channel not in range(256): - raise VMCMDErrors("Channel must be in range 0 to 255") - self._text_header = TextRequestHeader( - name=self._streamname, - bps_index=self._bps_opts.index(self._bps), - channel=self._channel, - ) - self._register_rt_header = RegisterRTHeader() - self.expected_packet = VBAN_VMRT_Packet_Header() - - self._rt_register_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - self._rt_packet_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - self._sendrequest_string_socket = socket.socket( - socket.AF_INET, socket.SOCK_DGRAM - ) - - is_readable = [] - is_writable = [ - self._rt_register_socket, - self._rt_packet_socket, - self._sendrequest_string_socket, - ] - is_error = [] - self.ready_to_read, self.ready_to_write, in_error = select.select( - is_readable, is_writable, is_error, 60 - ) - self._public_packet = None - self.running = True - self._pdirty = False - self._ldirty = False - self.subject = Subject() - self.cache = {} - - def __enter__(self): - self.login() - return self - - def login(self): - """ - Start listening for RT Packets - - Start background threads: - - Register to RT service - Keep public packet updated. - """ - self._rt_packet_socket.bind( - (socket.gethostbyname(socket.gethostname()), self._port) - ) - worker = Thread(target=self._send_register_rt, daemon=True) - worker.start() - self._public_packet = self._get_rt() - worker2 = Thread(target=self._updates, daemon=True) - worker2.start() - self.clear_dirty() - - def _send_register_rt(self): - """ - Continuously register to the RT Packet Service - - This function to be run in its own thread. - """ - while self.running: - if self._rt_register_socket in self.ready_to_write: - self._rt_register_socket.sendto( - self._register_rt_header.header, - (socket.gethostbyname(self._ip), self._port), - ) - count = ( - int.from_bytes(self._register_rt_header.framecounter, "little") + 1 - ) - self._register_rt_header.framecounter = count.to_bytes(4, "little") - sleep(10) - - def _fetch_rt_packet(self) -> Optional[VBAN_VMRT_Packet_Data]: - """Returns a valid RT Data Packet or None""" - if self._rt_packet_socket in self.ready_to_write: - data, _ = self._rt_packet_socket.recvfrom(1024 * 2) - # check for packet data - if len(data) > HEADER_SIZE: - # check if packet is of type rt service - if self.expected_packet.header == data[: HEADER_SIZE - 4]: - return VBAN_VMRT_Packet_Data( - _voicemeeterType=data[28:29], - _reserved=data[29:30], - _buffersize=data[30:32], - _voicemeeterVersion=data[32:36], - _optionBits=data[36:40], - _samplerate=data[40:44], - _inputLeveldB100=data[44:112], - _outputLeveldB100=data[112:240], - _TransportBit=data[240:244], - _stripState=data[244:276], - _busState=data[276:308], - _stripGaindB100Layer1=data[308:324], - _stripGaindB100Layer2=data[324:340], - _stripGaindB100Layer3=data[340:356], - _stripGaindB100Layer4=data[356:372], - _stripGaindB100Layer5=data[372:388], - _stripGaindB100Layer6=data[388:404], - _stripGaindB100Layer7=data[404:420], - _stripGaindB100Layer8=data[420:436], - _busGaindB100=data[436:452], - _stripLabelUTF8c60=data[452:932], - _busLabelUTF8c60=data[932:1412], - ) - - @property - def pdirty(self): - """True iff a parameter has changed""" - return self._pdirty - - @property - def ldirty(self): - """True iff a level value has changed.""" - return self._ldirty - - @property - def public_packet(self): - return self._public_packet - - def clear_dirty(self): - while self.pdirty: - pass - - def _updates(self) -> NoReturn: - """ - Continously update public packet in background. - - Set parameter and level dirty flags. - - Update public packet only if new private packet is found. - - Then notify observers of updates to states. - - This function to be run in its own thread. - """ - while self.running: - private_packet = self._get_rt() - - private_input_levels = private_packet.inputlevels - public_input_levels = self.public_packet.inputlevels - strip_comp = [ - not a == b - for a, b in zip( - private_input_levels, - public_input_levels, - ) - ] - private_output_levels = private_packet.outputlevels - public_output_levels = self.public_packet.outputlevels - bus_comp = [ - not a == b - for a, b in zip( - private_output_levels, - public_output_levels, - ) - ] - - self._pdirty = private_packet.pdirty(self.public_packet) - self._ldirty = any(any(list_) for list_ in [strip_comp, bus_comp]) - - if self._public_packet != private_packet: - self._public_packet = private_packet - if self.pdirty: - self.subject.notify("pdirty") - if self.ldirty: - self.subject.notify( - "ldirty", - [ - public_input_levels, - strip_comp, - public_output_levels, - bus_comp, - ], - ) - sleep(self._delay) - - def _get_rt(self) -> VBAN_VMRT_Packet_Data: - """Attempt to fetch data packet until a valid one found""" - - def fget(): - data = False - while not data: - data = self._fetch_rt_packet() - return data - - return fget() - - def set_rt( - self, - id_: str, - param: Optional[str] = None, - val: Optional[Union[int, float]] = None, - ): - """Sends a string request command over a network.""" - cmd = id_ if not param else f"{id_}.{param}={val}" - if self._sendrequest_string_socket in self.ready_to_write: - self._sendrequest_string_socket.sendto( - self._text_header.header + cmd.encode(), - (socket.gethostbyname(self._ip), self._port), - ) - count = int.from_bytes(self._text_header.framecounter, "little") + 1 - self._text_header.framecounter = count.to_bytes(4, "little") - if param: - self.cache[f"{id_}.{param}"] = val - - @script - def sendtext(self, cmd): - """Sends a multiple parameter string over a network.""" - self.set_rt(cmd) - sleep(self._delay) - - @property - def type(self): - """Returns the type of Voicemeeter installation.""" - return self.public_packet.voicemeetertype - - @property - def version(self): - """Returns Voicemeeter's version as a tuple""" - return self.public_packet.voicemeeterversion - - def show(self) -> NoReturn: - """Shows Voicemeeter if it's hidden.""" - self.command.show() - - def hide(self) -> NoReturn: - """Hides Voicemeeter if it's shown.""" - self.command.hide() - - def shutdown(self) -> NoReturn: - """Closes Voicemeeter.""" - self.command.shutdown() - - def restart(self) -> NoReturn: - """Restarts Voicemeeter's audio engine.""" - self.command.restart() - - def apply(self, mapping: dict): - """Sets all parameters of a di""" - for key, submapping in mapping.items(): - obj, index = key.split("-") - - if obj in ("strip"): - target = self.strip[int(index)] - elif obj in ("bus"): - target = self.bus[int(index)] - else: - raise ValueError(obj) - target.apply(submapping) - - def apply_profile(self, name: str): - try: - profile = self.profiles[name] - if "extends" in profile: - base = self.profiles[profile["extends"]] - del profile["extends"] - for key in profile.keys(): - if key in base: - base[key] = base[key] | profile[key] - else: - base[key] = profile[key] - profile = base - self.apply(profile) - except KeyError: - raise VMCMDErrors(f"Unknown profile: {self.kind.id}/{name}") - - def reset(self) -> NoReturn: - self.apply_profile("base") - - @property - def strip_levels(self): - """Returns the full strip level array for a kind, PREFADER mode, before math conversion""" - return tuple( - list(filter(lambda x: x != ((1 << 16) - 1), self.public_packet.inputlevels)) - ) - - @property - def bus_levels(self): - """Returns the full bus level array for a kind, before math conversion""" - return tuple( - list( - filter(lambda x: x != ((1 << 16) - 1), self.public_packet.outputlevels) - ) - ) - - def logout(self): - self.running = False - sleep(0.2) - self._rt_register_socket.close() - self._sendrequest_string_socket.close() - self._rt_packet_socket.close() - - def __exit__(self, exc_type, exc_value, exc_traceback): - self.logout() - - -def _make_remote(kind: NamedTuple) -> VbanCmd: - """ - Creates a new remote class and sets its number of inputs - and outputs for a VM kind. - - The returned class will subclass VbanCmd. - """ - - def init(self, **kwargs): - defaultkwargs = { - "ip": None, - "port": 6990, - "streamname": "Command1", - "bps": 0, - "channel": 0, - "delay": 0.001, - "sync": False, - } - kwargs = defaultkwargs | kwargs - VbanCmd.__init__(self, **kwargs) - self.kind = kind - self.phys_in, self.virt_in = kind.ins - self.phys_out, self.virt_out = kind.outs - self.strip_comp = [False for _ in range(2 * self.phys_in + 8 * self.virt_in)] - self.bus_comp = [False for _ in range(8 * (self.phys_out + self.virt_out))] - self.strip = tuple( - InputStrip.make((i < self.phys_in), self, i) - for i in range(self.phys_in + self.virt_in) - ) - self.bus = tuple( - OutputBus.make((i < self.phys_out), self, i) - for i in range(self.phys_out + self.virt_out) - ) - self.command = Command.make(self) - - def get_profiles(self): - return profiles.profiles[kind.id] - - return type( - f"VbanCmd{kind.name}", - (VbanCmd,), - {"__init__": init, "profiles": property(get_profiles)}, - ) - - -_remotes = {kind.id: _make_remote(kind) for kind in kinds.all} - - -def connect(kind_id: str, **kwargs): - """Connect to Voicemeeter and sets its strip layout.""" - try: - VBANCMD_cls = _remotes[kind_id] - return VBANCMD_cls(**kwargs) - except KeyError as err: - raise VMCMDErrors(f"Invalid Voicemeeter kind: {kind_id}")