onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2025-05-21 17:50:22 +01:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: onyx_online:8b912a2d08da5251f1877980575311a7dd4c7d44
Branches Tags
onyx_online:dev
onyx_online:add-event-stop-object
onyx_online:main
..
compare: onyx_online:f6218d20324471673f6c8f2ae623fa37c1bd7a47
Branches Tags
onyx_online:dev
onyx_online:add-event-stop-object
onyx_online:main

No commits in common. "8b912a2d08da5251f1877980575311a7dd4c7d44" and "f6218d20324471673f6c8f2ae623fa37c1bd7a47" have entirely different histories.
8b912a2d08 ... f6218d2032

33 changed files with 323 additions and 1037 deletions
Show Stats Expand all files Collapse all files

4
.gitignore vendored
View File

@ -156,6 +156,4 @@ quick.py
#config
config.toml
vban.toml
.vscode/
vban.toml

30
CHANGELOG.md
View File

@ -11,36 +11,6 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
- [x]
## [2.0.0] - 2023-06-25
This update introduces some breaking changes:
### Changed
- `strip[i].comp` now references StripComp class
- To change the comp knob you should now use the property `strip[i].comp.knob`
- `strip[i].gate` now references StripGate class
- To change the gate knob you should now use the property `strip[i].gate.knob`
- `bus[i].eq` now references BusEQ class
- To set bus[i].{eq,eq_ab} as before you should now use bus[i].eq.on and bus[i].eq.ab
- new error class `VBANCMDConnectionError` raised when a connection fails or times out.
There are other non-breaking changes:
### Changed
- now using a producer thread to send events to the updater thread.
- factory.request_vbancmd_obj simply raises a `VBANCMDError` if passed an incorrect kind.
- module level loggers implemented (with class loggers as child loggers)
### Added
- `strip[i].eq` added to PhysicalStrip
## [1.8.0]
### Added

106
README.md
View File

@ -18,9 +18,9 @@ For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
## Tested against
- Basic 1.0.8.8
- Banana 2.0.6.8
- Potato 3.0.2.8
- Basic 1.0.8.4
- Banana 2.0.6.4
- Potato 3.0.2.4
## Requirements
@ -71,19 +71,19 @@ class ManyThings:
def other_things(self):
self.vban.bus[3].gain = -6.3
self.vban.bus[4].eq.on = True
self.vban.bus[4].eq = True
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.on}",
f"bus 4 eq has been set to {self.vban.bus[4].eq}",
)
print("\n".join(info))
def main():
KIND_ID = "banana"
kind_id = "banana"
with vban_cmd.api(
KIND_ID, ip="gamepc.local", port=6980, streamname="Command1"
kind_id, ip="gamepc.local", port=6980, streamname="Command1"
) as vban:
do = ManyThings(vban)
do.things()
@ -93,7 +93,7 @@ def main():
vban.apply(
{
"strip-2": {"A1": True, "B1": True, "gain": -6.0},
"bus-2": {"mute": True, "eq": {"on": True}},
"bus-2": {"mute": True},
}
)
@ -104,9 +104,9 @@ if __name__ == "__main__":
Otherwise you must remember to call `vban.login()`, `vban.logout()` at the start/end of your code.
## `KIND_ID`
## `kind_id`
Pass the kind of Voicemeeter as an argument. KIND_ID may be:
Pass the kind of Voicemeeter as an argument. kind_id may be:
- `basic`
- `banana`
@ -124,6 +124,8 @@ The following properties are available.
- `label`: string
- `gain`: float, -60 to 12
- `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
example:
@ -150,69 +152,6 @@ vban.strip[5].appmute("Spotify", True)
vban.strip[5].appgain("Spotify", 0.5)
```
##### Strip.Comp
The following properties are available.
- `knob`: float, from 0.0 to 10.0
- `gainin`: float, from -24.0 to 24.0
- `ratio`: float, from 1.0 to 8.0
- `threshold`: float, from -40.0 to -3.0
- `attack`: float, from 0.0 to 200.0
- `release`: float, from 0.0 to 5000.0
- `knee`: float, from 0.0 to 1.0
- `gainout`: float, from -24.0 to 24.0
- `makeup`: boolean
example:
```python
print(vm.strip[4].comp.knob)
```
Strip Comp properties are defined as write only.
`knob` defined for all versions, all other parameters potato only.
##### Strip.Gate
The following properties are available.
- `knob`: float, from 0.0 to 10.0
- `threshold`: float, from -60.0 to -10.0
- `damping`: float, from -60.0 to -10.0
- `bpsidechain`: int, from 100 to 4000
- `attack`: float, from 0.0 to 1000.0
- `hold`: float, from 0.0 to 5000.0
- `release`: float, from 0.0 to 5000.0
example:
```python
vm.strip[2].gate.attack = 300.8
```
Strip Gate properties are defined as write only, potato version only.
`knob` defined for all versions, all other parameters potato only.
##### Strip.Denoiser
The following properties are available.
- `knob`: float, from 0.0 to 10.0
strip.denoiser properties are defined as write only, potato version only.
##### Strip.EQ
The following properties are available.
- `on`: boolean
- `ab`: boolean
Strip EQ properties are defined as write only, potato version only.
##### Gainlayers
- `gain`: float, from -60.0 to 12.0
@ -244,6 +183,8 @@ Level properties will return -200.0 if no audio detected.
The following properties are available.
- `mono`: boolean
- `eq`: boolean
- `eq_ab`: boolean
- `mute`: boolean
- `label`: string
- `gain`: float, -60 to 12
@ -255,13 +196,6 @@ vban.bus[4].eq = true
print(vban.bus[0].label)
```
##### Bus.EQ
The following properties are available.
- `on`: boolean
- `ab`: boolean
##### Modes
The following properties are available.
@ -391,8 +325,9 @@ opts = {
"ip": "<ip address>",
"streamname": "Command1",
"port": 6980,
"subs": {"ldirty": True},
}
with vban_cmd.api('banana', ldirty=True, **opts) as vban:
with vban_cmd.api('banana', **opts) as vban:
...
```
@ -451,15 +386,16 @@ print(vban.event.get())
## VbanCmd class
`vban_cmd.api(kind_id: str, **opts)`
`vban_cmd.api(kind_id: str, **opts: dict)`
You may pass the following optional keyword arguments:
- `ip`: str, ip or hostname of remote machine
- `streamname`: str, name of the stream to connect to.
- `port`: int=6980, vban udp port of remote machine.
- `pdirty`: parameter updates
- `ldirty`: level updates
- `subs`: dict={"pdirty": True, "ldirty": False}, controls which updates to listen for.
- `pdirty`: parameter updates
- `ldirty`: level updates
#### `vban.pdirty`
@ -479,7 +415,7 @@ vban.sendtext("Strip[0].Mute=1;Bus[0].Mono=1")
#### `vban.public_packet`
Returns a `VbanRtPacket`. Designed to be used internally by the interface but available for parsing through this read only property object. States not guaranteed to be current (requires use of dirty parameters to confirm).
Returns a Voicemeeter rt data packet object. Designed to be used internally by the interface but available for parsing through this read only property object. States not guaranteed to be current (requires use of dirty parameters to confirm).
### `Errors`

8
configs/banana/example.toml
View File

@ -2,12 +2,12 @@
label = "PhysStrip0"
A1 = true
gain = -8.8
comp.knob = 3.2
comp = 3.2
[strip-1]
label = "PhysStrip1"
B1 = true
gate.knob = 4.1
gate = 4.1
[strip-2]
label = "PhysStrip2"
@ -31,12 +31,12 @@ mono = true
[bus-2]
label = "PhysBus2"
eq.on = true
eq = true
mode = "composite"
[bus-3]
label = "VirtBus0"
eq.ab = true
eq_ab = true
mode = "upmix61"
[bus-4]

8
configs/potato/example.toml
View File

@ -2,12 +2,12 @@
label = "PhysStrip0"
A1 = true
gain = -8.8
comp.knob = 3.2
comp = 3.2
[strip-1]
label = "PhysStrip1"
B1 = true
gate.knob = 4.1
gate = 4.1
[strip-2]
label = "PhysStrip2"
@ -47,7 +47,7 @@ mono = true
[bus-2]
label = "PhysBus2"
eq.on = true
eq = true
[bus-3]
label = "PhysBus3"
@ -59,7 +59,7 @@ mode = "composite"
[bus-5]
label = "VirtBus0"
eq.ab = true
eq_ab = true
[bus-6]
label = "VirtBus1"

13
examples/gui/README.md
View File

@ -1,13 +0,0 @@
## About
A single channel GUI demonstrating controls for the first virtual strip if Voicemeeter Banana.
This example demonstrates (to an extent) two way communication.
- Sending parameters values to the Voicemeeter driver.
- Receiving level updates
Parameter updates (pdirty) events are not being received so changing a UI element on the main Voicemeeter app will not be reflected in the example GUI.
## Use
Simply run the script and try the controls.

100
examples/gui/__main__.py
View File

@ -1,100 +0,0 @@
import logging
import vban_cmd
logging.basicConfig(level=logging.DEBUG)
import tkinter as tk
from tkinter import ttk
class App(tk.Tk):
def __init__(self, vban):
super().__init__()
self.vban = vban
self.title(f"{vban} - version {vban.version}")
self.vban.observer.add(self.on_ldirty)
# create widget variables
self.button_var = tk.BooleanVar(value=vban.strip[3].mute)
self.slider_var = tk.DoubleVar(value=vban.strip[3].gain)
self.meter_var = tk.DoubleVar(value=self._get_level())
self.gainlabel_var = tk.StringVar(value=self.slider_var.get())
# initialize style table
self.style = ttk.Style()
self.style.theme_use("clam")
self.style.configure(
"Mute.TButton", foreground="#cd5c5c" if vban.strip[3].mute else "#5a5a5a"
)
# create labelframe and grid it onto the mainframe
self.labelframe = tk.LabelFrame(text=self.vban.strip[3].label)
self.labelframe.grid(padx=1)
# create slider and grid it onto the labelframe
slider = ttk.Scale(
self.labelframe,
from_=12,
to_=-60,
orient="vertical",
variable=self.slider_var,
command=lambda arg: self.on_slider_move(arg),
)
slider.grid(
column=0,
row=0,
)
# create level meter and grid it onto the labelframe
level_meter = ttk.Progressbar(
self.labelframe,
orient="vertical",
variable=self.meter_var,
maximum=72,
mode="determinate",
)
level_meter.grid(column=1, row=0)
# create gainlabel and grid it onto the labelframe
gainlabel = ttk.Label(self.labelframe, textvariable=self.gainlabel_var)
gainlabel.grid(column=0, row=1, columnspan=2)
# create button and grid it onto the labelframe
button = ttk.Button(
self.labelframe,
text="Mute",
style="Mute.TButton",
command=lambda: self.on_button_press(),
)
button.grid(column=0, row=2, columnspan=2, padx=1, pady=2)
# define callbacks
def on_slider_move(self, *args):
val = round(self.slider_var.get(), 1)
self.vban.strip[3].gain = val
self.gainlabel_var.set(val)
def on_button_press(self):
self.button_var.set(not self.button_var.get())
self.vban.strip[3].mute = self.button_var.get()
self.style.configure(
"Mute.TButton", foreground="#cd5c5c" if self.button_var.get() else "#5a5a5a"
)
def _get_level(self):
val = max(self.vban.strip[3].levels.prefader)
return 0 if self.button_var.get() else 72 + val - 12 + self.slider_var.get()
def on_ldirty(self):
self.meter_var.set(self._get_level())
def main():
with vban_cmd.api("banana", ldirty=True) as vban:
app = App(vban)
app.mainloop()
if __name__ == "__main__":
main()

4
examples/obs/README.md
View File

@ -40,12 +40,10 @@ Make sure you have established a working connection to OBS and the remote Voicem
Run the script, change OBS scenes and watch Voicemeeter parameters change.
Closing OBS will end the script.
Pressing `<Enter>` will exit.
## Notes
All but `vban_cmd.iremote` logs are filtered out. Log in DEBUG mode.
This script can be run from a Linux host since the vban-cmd interface relies on UDP packets and obsws-python runs over websockets.
You could for example, set this up to run in the background on a home server such as a Raspberry Pi.

48
examples/obs/__main__.py
View File

@ -1,41 +1,16 @@
import time
from logging import config
import obsws_python as obsws
import logging
import obsws_python as obs
import vban_cmd
config.dictConfig(
{
"version": 1,
"formatters": {
"standard": {
"format": "%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s"
}
},
"handlers": {
"stream": {
"level": "DEBUG",
"class": "logging.StreamHandler",
"formatter": "standard",
}
},
"loggers": {"vban_cmd.iremote": {"handlers": ["stream"], "level": "DEBUG"}},
}
)
logging.basicConfig(level=logging.INFO)
class Observer:
def __init__(self, vban):
self.vban = vban
self.client = obsws.EventClient()
self.client.callback.register(
(
self.on_current_program_scene_changed,
self.on_exit_started,
)
)
self.is_running = True
self.client = obs.EventClient()
self.client.callback.register(self.on_current_program_scene_changed)
def on_start(self):
self.vban.strip[0].mute = True
@ -75,16 +50,13 @@ class Observer:
if fn := fget(scene):
fn()
def on_exit_started(self, _):
self.client.unsubscribe()
self.is_running = False
def main():
with vban_cmd.api("potato") as vban:
observer = Observer(vban)
while observer.is_running:
time.sleep(0.1)
with vban_cmd.api("potato", sync=True) as vban:
obs = Observer(vban)
while cmd := input("<Enter> to exit\n"):
if not cmd:
break
if __name__ == "__main__":

7
examples/obs/setup.py
View File

@ -1,7 +0,0 @@
from setuptools import setup
setup(
name="obs",
description="OBS Example",
install_requires=["obsws-python"],
)

21
examples/observer/__main__.py
View File

@ -5,30 +5,33 @@ import vban_cmd
logging.basicConfig(level=logging.INFO)
class App:
class Observer:
def __init__(self, vban):
self.vban = vban
# register your app as event observer
self.vban.observer.add(self)
self.vban.subject.add(self)
# enable level updates, since they are disabled by default.
self.vban.event.ldirty = True
# define an 'on_update' callback function to receive event updates
def on_update(self, event):
if event == "pdirty":
def on_update(self, subject):
if subject == "pdirty":
print("pdirty!")
elif event == "ldirty":
elif subject == "ldirty":
for bus in self.vban.bus:
if bus.levels.isdirty:
print(bus, bus.levels.all)
def main():
KIND_ID = "banana"
kind_id = "potato"
with vban_cmd.api(KIND_ID, pdirty=True, ldirty=True) as vban:
App(vban)
with vban_cmd.api(kind_id) as vban:
Observer(vban)
while cmd := input("Press <Enter> to exit\n"):
pass
if not cmd:
break
if __name__ == "__main__":

125
poetry.lock generated
View File

@ -33,22 +33,6 @@ d = ["aiohttp (>=3.7.4)"]
jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"]
uvloop = ["uvloop (>=0.15.2)"]
[[package]]
name = "cachetools"
version = "5.3.1"
description = "Extensible memoizing collections and decorators"
category = "dev"
optional = false
python-versions = ">=3.7"
[[package]]
name = "chardet"
version = "5.1.0"
description = "Universal encoding detector for Python 3"
category = "dev"
optional = false
python-versions = ">=3.7"
[[package]]
name = "click"
version = "8.1.3"
@ -62,31 +46,11 @@ colorama = {version = "*", markers = "platform_system == \"Windows\""}
[[package]]
name = "colorama"
version = "0.4.6"
version = "0.4.5"
description = "Cross-platform colored terminal text."
category = "dev"
optional = false
python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7"
[[package]]
name = "distlib"
version = "0.3.6"
description = "Distribution utilities"
category = "dev"
optional = false
python-versions = "*"
[[package]]
name = "filelock"
version = "3.12.2"
description = "A platform independent file lock."
category = "dev"
optional = false
python-versions = ">=3.7"
[package.extras]
docs = ["furo (>=2023.5.20)", "sphinx-autodoc-typehints (>=1.23,!=1.23.4)", "sphinx (>=7.0.1)"]
testing = ["covdefaults (>=2.3)", "coverage (>=7.2.7)", "diff-cover (>=7.5)", "pytest-cov (>=4.1)", "pytest-mock (>=3.10)", "pytest-timeout (>=2.1)", "pytest (>=7.3.1)"]
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[[package]]
name = "iniconfig"
@ -120,11 +84,14 @@ python-versions = "*"
[[package]]
name = "packaging"
version = "23.1"
version = "21.3"
description = "Core utilities for Python packages"
category = "dev"
optional = false
python-versions = ">=3.7"
python-versions = ">=3.6"
[package.dependencies]
pyparsing = ">=2.0.2,<3.0.5 || >3.0.5"
[[package]]
name = "pathspec"
@ -136,15 +103,15 @@ python-versions = ">=3.7"
[[package]]
name = "platformdirs"
version = "3.7.0"
description = "A small Python package for determining appropriate platform-specific dirs, e.g. a \"user data dir\"."
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 (>=2023.5.20)", "proselint (>=0.13)", "sphinx-autodoc-typehints (>=1.23,!=1.23.4)", "sphinx (>=7.0.1)"]
test = ["appdirs (==1.4.4)", "covdefaults (>=2.3)", "pytest-cov (>=4.1)", "pytest-mock (>=3.10)", "pytest (>=7.3.1)"]
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"
@ -155,8 +122,8 @@ optional = false
python-versions = ">=3.6"
[package.extras]
dev = ["pre-commit", "tox"]
testing = ["pytest", "pytest-benchmark"]
testing = ["pytest-benchmark", "pytest"]
dev = ["tox", "pre-commit"]
[[package]]
name = "py"
@ -167,20 +134,15 @@ optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[[package]]
name = "pyproject-api"
version = "1.5.2"
description = "API to interact with the python pyproject.toml based projects"
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.7"
[package.dependencies]
packaging = ">=23.1"
tomli = {version = ">=2.0.1", markers = "python_version < \"3.11\""}
python-versions = ">=3.6.8"
[package.extras]
docs = ["furo (>=2023.5.20)", "sphinx-autodoc-typehints (>=1.23,!=1.23.4)", "sphinx (>=7.0.1)"]
testing = ["covdefaults (>=2.3)", "importlib-metadata (>=6.6)", "pytest-cov (>=4.1)", "pytest-mock (>=3.10)", "pytest (>=7.3.1)", "setuptools (>=67.8)", "wheel (>=0.40)"]
diagrams = ["railroad-diagrams", "jinja2"]
[[package]]
name = "pytest"
@ -232,61 +194,16 @@ category = "main"
optional = false
python-versions = ">=3.7"
[[package]]
name = "tox"
version = "4.6.3"
description = "tox is a generic virtualenv management and test command line tool"
category = "dev"
optional = false
python-versions = ">=3.7"
[package.dependencies]
cachetools = ">=5.3.1"
chardet = ">=5.1"
colorama = ">=0.4.6"
filelock = ">=3.12.2"
packaging = ">=23.1"
platformdirs = ">=3.5.3"
pluggy = ">=1"
pyproject-api = ">=1.5.2"
tomli = {version = ">=2.0.1", markers = "python_version < \"3.11\""}
virtualenv = ">=20.23.1"
[package.extras]
docs = ["furo (>=2023.5.20)", "sphinx-argparse-cli (>=1.11.1)", "sphinx-autodoc-typehints (>=1.23.2,!=1.23.4)", "sphinx-copybutton (>=0.5.2)", "sphinx-inline-tabs (>=2023.4.21)", "sphinx (>=7.0.1)", "sphinxcontrib-towncrier (>=0.2.1a0)", "towncrier (>=23.6)"]
testing = ["build[virtualenv] (>=0.10)", "covdefaults (>=2.3)", "detect-test-pollution (>=1.1.1)", "devpi-process (>=0.3.1)", "diff-cover (>=7.6)", "distlib (>=0.3.6)", "flaky (>=3.7)", "hatch-vcs (>=0.3)", "hatchling (>=1.17.1)", "psutil (>=5.9.5)", "pytest-cov (>=4.1)", "pytest-mock (>=3.11.1)", "pytest-xdist (>=3.3.1)", "pytest (>=7.3.2)", "re-assert (>=1.1)", "time-machine (>=2.10)", "wheel (>=0.40)"]
[[package]]
name = "virtualenv"
version = "20.23.1"
description = "Virtual Python Environment builder"
category = "dev"
optional = false
python-versions = ">=3.7"
[package.dependencies]
distlib = ">=0.3.6,<1"
filelock = ">=3.12,<4"
platformdirs = ">=3.5.1,<4"
[package.extras]
docs = ["furo (>=2023.5.20)", "proselint (>=0.13)", "sphinx-argparse (>=0.4)", "sphinx (>=7.0.1)", "sphinxcontrib-towncrier (>=0.2.1a0)", "towncrier (>=23.6)"]
test = ["covdefaults (>=2.3)", "coverage-enable-subprocess (>=1)", "coverage (>=7.2.7)", "flaky (>=3.7)", "packaging (>=23.1)", "pytest-env (>=0.8.1)", "pytest-freezer (>=0.4.6)", "pytest-mock (>=3.10)", "pytest-randomly (>=3.12)", "pytest-timeout (>=2.1)", "pytest (>=7.3.1)", "setuptools (>=67.8)", "time-machine (>=2.9)"]
[metadata]
lock-version = "1.1"
python-versions = "^3.10"
content-hash = "5d0edd070ea010edb4e2ade88dc37324b8b4b04f22db78e49db161185365849b"
content-hash = "9f887ae517ade09119bf1f2cf77261d2445ae95857b69470ce1707f9791ce080"
[metadata.files]
attrs = []
black = []
cachetools = []
chardet = []
click = []
colorama = []
distlib = []
filelock = []
iniconfig = []
isort = []
mypy-extensions = []
@ -295,10 +212,8 @@ pathspec = []
platformdirs = []
pluggy = []
py = []
pyproject-api = []
pyparsing = []
pytest = []
pytest-randomly = []
pytest-repeat = []
tomli = []
tox = []
virtualenv = []

17
pyproject.toml
View File

@ -1,6 +1,6 @@
[tool.poetry]
name = "vban-cmd"
version = "2.0.0"
version = "1.8.1"
description = "Python interface for the VBAN RT Packet Service (Sendtext)"
authors = ["onyx-and-iris <code@onyxandiris.online>"]
license = "MIT"
@ -18,26 +18,11 @@ pytest-randomly = "^3.12.0"
pytest-repeat = "^0.9.1"
black = "^22.3.0"
isort = "^5.10.1"
tox = "^4.6.3"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.poetry.scripts]
gui = "scripts:ex_gui"
obs = "scripts:ex_obs"
observer = "scripts:ex_observer"
test = "scripts:test"
[tool.tox]
legacy_tox_ini = """
[tox]
envlist = py310,py311
[testenv]
allowlist_externals = poetry
commands =
poetry install -v
poetry run pytest tests/
"""

9
scripts.py
View File

@ -2,11 +2,6 @@ import subprocess
from pathlib import Path
def ex_gui():
path = Path.cwd() / "examples" / "gui" / "."
subprocess.run(["py", str(path)])
def ex_obs():
path = Path.cwd() / "examples" / "obs" / "."
subprocess.run(["py", str(path)])
@ -15,7 +10,3 @@ def ex_obs():
def ex_observer():
path = Path.cwd() / "examples" / "observer" / "."
subprocess.run(["py", str(path)])
def test():
subprocess.run(["tox"])

20
tests/__init__.py
View File

@ -3,21 +3,23 @@ import sys
from dataclasses import dataclass
import vban_cmd
from vban_cmd.kinds import KindId
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.name.lower() for kind_id in KindId))
kind_id = random.choice(tuple(kind_id.name.lower() for kind_id in KindId))
opts = {
"ip": "testing.local",
"streamname": "testing",
"ip": "ws.local",
"streamname": "workstation",
"port": 6990,
"bps": 0,
"sync": True,
}
vban = vban_cmd.api(KIND_ID, **opts)
kind = kindmap(KIND_ID)
vbans = {kind.name: vban_cmd.api(kind.name, **opts) for kind in kinds_all}
tests = vbans[kind_id]
kind = kindmap(kind_id)
@dataclass
@ -40,9 +42,9 @@ data = Data()
def setup_module():
print(f"\nRunning tests for kind [{data.name}]\n", file=sys.stdout)
vban.login()
vban.command.reset()
tests.login()
tests.command.reset()
def teardown_module():
vban.logout()
tests.logout()

14
tests/test_configs.py
View File

@ -1,6 +1,8 @@
import time
import pytest
from tests import data, vban
from tests import data, tests
class TestSetAndGetBoolHigher:
@ -10,18 +12,18 @@ class TestSetAndGetBoolHigher:
@classmethod
def setup_class(cls):
vban.apply_config("example")
tests.apply_config("example")
def test_it_tests_config_string(self):
assert "PhysStrip" in vban.strip[data.phys_in].label
assert "VirtStrip" in vban.strip[data.virt_in].label
assert "PhysStrip" in tests.strip[data.phys_in].label
assert "VirtStrip" in tests.strip[data.virt_in].label
def test_it_tests_config_bool(self):
assert vban.strip[0].A1 == True
assert tests.strip[0].A1 == True
@pytest.mark.skipif(
"not config.getoption('--run-slow')",
reason="Only run when --run-slow is given",
)
def test_it_tests_config_busmode(self):
assert vban.bus[data.phys_out].mode.get() == "composite"
assert tests.bus[data.phys_out].mode.get() == "composite"

32
tests/test_factory.py
View File

@ -1,6 +1,6 @@
import pytest
from tests import data, vban
from tests import data, tests
class TestRemoteFactories:
@ -11,33 +11,33 @@ class TestRemoteFactories:
reason="Skip test if kind is not basic",
)
def test_it_tests_remote_attrs_for_basic(self):
assert hasattr(vban, "strip")
assert hasattr(vban, "bus")
assert hasattr(vban, "command")
assert hasattr(tests, "strip")
assert hasattr(tests, "bus")
assert hasattr(tests, "command")
assert len(vban.strip) == 3
assert len(vban.bus) == 2
assert len(tests.strip) == 3
assert len(tests.bus) == 2
@pytest.mark.skipif(
data.name != "banana",
reason="Skip test if kind is not basic",
)
def test_it_tests_remote_attrs_for_banana(self):
assert hasattr(vban, "strip")
assert hasattr(vban, "bus")
assert hasattr(vban, "command")
assert hasattr(tests, "strip")
assert hasattr(tests, "bus")
assert hasattr(tests, "command")
assert len(vban.strip) == 5
assert len(vban.bus) == 5
assert len(tests.strip) == 5
assert len(tests.bus) == 5
@pytest.mark.skipif(
data.name != "potato",
reason="Skip test if kind is not basic",
)
def test_it_tests_remote_attrs_for_potato(self):
assert hasattr(vban, "strip")
assert hasattr(vban, "bus")
assert hasattr(vban, "command")
assert hasattr(tests, "strip")
assert hasattr(tests, "bus")
assert hasattr(tests, "command")
assert len(vban.strip) == 8
assert len(vban.bus) == 8
assert len(tests.strip) == 8
assert len(tests.bus) == 8

89
tests/test_higher.py
View File

@ -1,6 +1,6 @@
import pytest
from tests import data, vban
from tests import data, tests
@pytest.mark.parametrize("value", [False, True])
@ -17,8 +17,8 @@ class TestSetAndGetBoolHigher:
],
)
def test_it_sets_and_gets_strip_bool_params(self, index, param, value):
setattr(vban.strip[index], param, value)
assert getattr(vban.strip[index], param) == value
setattr(tests.strip[index], param, value)
assert getattr(tests.strip[index], param) == value
@pytest.mark.skipif(
data.name == "banana",
@ -31,22 +31,23 @@ class TestSetAndGetBoolHigher:
],
)
def test_it_sets_and_gets_strip_bool_params_mc(self, index, param, value):
setattr(vban.strip[index], param, value)
assert getattr(vban.strip[index], param) == value
setattr(tests.strip[index], param, value)
assert getattr(tests.strip[index], param) == value
""" bus tests, physical and virtual """
@pytest.mark.parametrize(
"index,param",
[
(data.phys_out, "eq"),
(data.phys_out, "mute"),
(data.virt_out, "eq_ab"),
(data.virt_out, "sel"),
],
)
def test_it_sets_and_gets_bus_bool_params(self, index, param, value):
assert hasattr(vban.bus[index], param)
setattr(vban.bus[index], param, value)
assert getattr(vban.bus[index], param) == value
setattr(tests.bus[index], param, value)
assert getattr(tests.bus[index], param) == value
""" bus modes tests, physical and virtual """
@ -65,8 +66,8 @@ class TestSetAndGetBoolHigher:
# here it only makes sense to set/get bus modes as True
if not value:
value = True
setattr(vban.bus[index].mode, param, value)
assert getattr(vban.bus[index].mode, param) == value
setattr(tests.bus[index].mode, param, value)
assert getattr(tests.bus[index].mode, param) == value
""" command tests """
@ -75,7 +76,7 @@ class TestSetAndGetBoolHigher:
[("lock")],
)
def test_it_sets_command_bool_params(self, param, value):
setattr(vban.command, param, value)
setattr(tests.command, param, value)
class TestSetAndGetIntHigher:
@ -93,8 +94,8 @@ class TestSetAndGetIntHigher:
],
)
def test_it_sets_and_gets_strip_bool_params(self, index, param, value):
setattr(vban.strip[index], param, value)
assert getattr(vban.strip[index], param) == value
setattr(tests.strip[index], param, value)
assert getattr(tests.strip[index], param) == value
class TestSetAndGetFloatHigher:
@ -112,15 +113,15 @@ class TestSetAndGetFloatHigher:
],
)
def test_it_sets_and_gets_strip_float_params(self, index, param, value):
setattr(vban.strip[index], param, value)
assert getattr(vban.strip[index], param) == value
setattr(tests.strip[index], param, value)
assert getattr(tests.strip[index], param) == value
@pytest.mark.parametrize(
"index,value",
[(data.phys_in, 2), (data.phys_in, 2), (data.virt_in, 8), (data.virt_in, 8)],
)
def test_it_gets_prefader_levels_and_compares_length_of_array(self, index, value):
assert len(vban.strip[index].levels.prefader) == value
assert len(tests.strip[index].levels.prefader) == value
@pytest.mark.skipif(
data.name != "potato",
@ -136,42 +137,8 @@ class TestSetAndGetFloatHigher:
],
)
def test_it_sets_and_gets_strip_gainlayer_values(self, index, j, value):
vban.strip[index].gainlayer[j].gain = value
assert vban.strip[index].gainlayer[j].gain == value
""" strip tests, physical """
@pytest.mark.skipif(
data.name != "potato",
reason="Only test if logged into Potato version",
)
@pytest.mark.parametrize(
"index, param, value",
[
(data.phys_in, "gainin", -8.6),
(data.phys_in, "knee", 0.24),
],
)
def test_it_sets_strip_comp_params(self, index, param, value):
assert hasattr(vban.strip[index].comp, param)
setattr(vban.strip[index].comp, param, value)
# we can set but not get this value. Not in RT Packet.
@pytest.mark.skipif(
data.name != "potato",
reason="Only test if logged into Potato version",
)
@pytest.mark.parametrize(
"index, param, value",
[
(data.phys_in, "bpsidechain", 120),
(data.phys_in, "hold", 3000),
],
)
def test_it_sets_and_gets_strip_gate_params(self, index, param, value):
assert hasattr(vban.strip[index].gate, param)
setattr(vban.strip[index].gate, param, value)
# we can set but not get this value. Not in RT Packet.
tests.strip[index].gainlayer[j].gain = value
assert tests.strip[index].gainlayer[j].gain == value
""" strip tests, virtual """
@ -184,8 +151,8 @@ class TestSetAndGetFloatHigher:
],
)
def test_it_sets_and_gets_strip_eq_params(self, index, param, value):
setattr(vban.strip[index], param, value)
assert getattr(vban.strip[index], param) == value
setattr(tests.strip[index], param, value)
assert getattr(tests.strip[index], param) == value
""" bus tests, physical and virtual """
@ -194,15 +161,15 @@ class TestSetAndGetFloatHigher:
[(data.phys_out, "gain", -3.6), (data.virt_out, "gain", 5.8)],
)
def test_it_sets_and_gets_bus_float_params(self, index, param, value):
setattr(vban.bus[index], param, value)
assert getattr(vban.bus[index], param) == value
setattr(tests.bus[index], param, value)
assert getattr(tests.bus[index], param) == value
@pytest.mark.parametrize(
"index,value",
[(data.phys_out, 8), (data.virt_out, 8)],
)
def test_it_gets_prefader_levels_and_compares_length_of_array(self, index, value):
assert len(vban.bus[index].levels.all) == value
assert len(tests.bus[index].levels.all) == value
@pytest.mark.parametrize("value", ["test0", "test1"])
@ -216,8 +183,8 @@ class TestSetAndGetStringHigher:
[(data.phys_in, "label"), (data.virt_in, "label")],
)
def test_it_sets_and_gets_strip_string_params(self, index, param, value):
setattr(vban.strip[index], param, value)
assert getattr(vban.strip[index], param) == value
setattr(tests.strip[index], param, value)
assert getattr(tests.strip[index], param) == value
""" bus tests, physical and virtual """
@ -226,5 +193,5 @@ class TestSetAndGetStringHigher:
[(data.phys_out, "label"), (data.virt_out, "label")],
)
def test_it_sets_and_gets_bus_string_params(self, index, param, value):
setattr(vban.bus[index], param, value)
assert getattr(vban.bus[index], param) == value
setattr(tests.bus[index], param, value)
assert getattr(tests.bus[index], param) == value

10
tests/test_lower.py
View File

@ -1,10 +1,10 @@
import time
import pytest
from tests import data, vban
from vban_cmd import kinds
from tests import data, tests
class TestPublicPacketLower:
__test__ = True
@ -12,7 +12,7 @@ class TestPublicPacketLower:
"""Tests for a valid rt data packet"""
def test_it_gets_an_rt_data_packet(self):
assert vban.public_packet.voicemeetertype in (
assert tests.public_packet.voicemeetertype in (
kind.name for kind in kinds.kinds_all
)
@ -35,7 +35,7 @@ class TestSetRT:
],
)
def test_it_sends_a_text_request(self, kls, index, param, value):
vban._set_rt(f"{kls}[{index}]", param, value)
tests._set_rt(f"{kls}[{index}]", param, value)
time.sleep(0.02)
target = getattr(vban, kls)[index]
target = getattr(tests, kls)[index]
assert getattr(target, param) == bool(value)

20
vban_cmd/bus.py
View File

@ -52,23 +52,6 @@ class Bus(IRemote):
time.sleep(self._remote.DELAY)
class BusEQ(IRemote):
@classmethod
def make(cls, remote, index):
BUSEQ_cls = type(
f"BusEQ{remote.kind}",
(cls,),
{
**{param: channel_bool_prop(param) for param in ["on", "ab"]},
},
)
return BUSEQ_cls(remote, index)
@property
def identifier(self) -> str:
return f"Bus[{self.index}].eq"
class PhysicalBus(Bus):
def __str__(self):
return f"{type(self).__name__}{self.index}"
@ -184,10 +167,11 @@ def bus_factory(phys_bus, remote, i) -> Union[PhysicalBus, VirtualBus]:
f"{BUS_cls.__name__}{remote.kind}",
(BUS_cls,),
{
"eq": BusEQ.make(remote, i),
"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)

7
vban_cmd/command.py
View File

@ -1,5 +1,5 @@
from .iremote import IRemote
from .meta import action_fn
from .meta import action_prop
class Command(IRemote):
@ -21,9 +21,10 @@ class Command(IRemote):
(cls,),
{
**{
param: action_fn(param) for param in ["show", "shutdown", "restart"]
param: action_prop(param)
for param in ["show", "shutdown", "restart"]
},
"hide": action_fn("show", val=0),
"hide": action_prop("show", val=0),
},
)
return CMD_cls(remote)

35
vban_cmd/config.py
View File

@ -2,8 +2,6 @@ import itertools
import logging
from pathlib import Path
from .error import VBANCMDError
try:
import tomllib
except ModuleNotFoundError:
@ -11,8 +9,6 @@ except ModuleNotFoundError:
from .kinds import request_kind_map as kindmap
logger = logging.getLogger(__name__)
class TOMLStrBuilder:
"""builds a config profile, as a string, for the toml parser"""
@ -36,18 +32,10 @@ class TOMLStrBuilder:
+ [f"B{i} = false" for i in range(1, self.kind.virt_out + 1)]
)
self.phys_strip_params = self.virt_strip_params + [
"comp.knob = 0.0",
"gate.knob = 0.0",
"denoiser.knob = 0.0",
"eq.on = false",
]
self.bus_float = ["gain = 0.0"]
self.bus_params = [
"mono = false",
"eq.on = false",
"mute = false",
"gain = 0.0",
"comp = 0.0",
"gate = 0.0",
]
self.bus_bool = ["mono = false", "eq = false", "mute = false"]
if profile == "reset":
self.reset_config()
@ -78,7 +66,7 @@ class TOMLStrBuilder:
else self.virt_strip_params
)
case "bus":
toml_str += ("\n").join(self.bus_params)
toml_str += ("\n").join(self.bus_bool)
case _:
pass
return toml_str + "\n"
@ -131,9 +119,10 @@ class Loader(metaclass=SingletonType):
loads data into memory if not found
"""
logger = logging.getLogger("config.Loader")
def __init__(self, kind):
self._kind = kind
self.logger = logger.getChild(self.__class__.__name__)
self._configs = dict()
self.defaults(kind)
self.parser = None
@ -177,16 +166,16 @@ def loader(kind):
returns configs loaded into memory
"""
logger_loader = logger.getChild("loader")
logger = logging.getLogger("config.loader")
loader = Loader(kind)
for path in (
Path.cwd() / "configs" / kind.name,
Path.home() / ".config" / "vban-cmd" / kind.name,
Path.home() / "Documents" / "Voicemeeter" / "configs" / kind.name,
Path(__file__).parent / "configs" / kind.name,
Path.home() / "Documents/Voicemeeter" / "configs" / kind.name,
):
if path.is_dir():
logger_loader.info(f"Checking [{path}] for TOML config files:")
logger.info(f"Checking [{path}] for TOML config files:")
for file in path.glob("*.toml"):
identifier = file.with_suffix("").stem
if loader.parse(identifier, file):
@ -202,6 +191,6 @@ def request_config(kind_id: str):
"""
try:
configs = loader(kindmap(kind_id))
except KeyError:
raise VBANCMDError(f"Unknown Voicemeeter kind {kind_id}")
except KeyError as e:
print(f"Unknown Voicemeeter kind '{kind_id}'")
return configs

6
vban_cmd/error.py
View File

@ -1,6 +1,4 @@
class VBANCMDError(Exception):
"""Exception raised when general errors occur"""
"""general errors"""
class VBANCMDConnectionError(Exception):
"""Exception raised when connection/timeout errors occur"""
pass

5
vban_cmd/event.py
View File

@ -1,15 +1,14 @@
import logging
from typing import Iterable, Union
logger = logging.getLogger(__name__)
class Event:
"""Keeps track of event subscriptions"""
logger = logging.getLogger("event.event")
def __init__(self, subs: dict):
self.subs = subs
self.logger = logger.getChild(self.__class__.__name__)
def info(self, msg=None):
info = (f"{msg} events",) if msg else ()

19
vban_cmd/factory.py
View File

@ -7,14 +7,11 @@ from typing import Iterable, NoReturn
from .bus import request_bus_obj as bus
from .command import Command
from .config import request_config as configs
from .error import VBANCMDError
from .kinds import KindMapClass
from .kinds import request_kind_map as kindmap
from .strip import request_strip_obj as strip
from .vbancmd import VbanCmd
logger = logging.getLogger(__name__)
class FactoryBuilder:
"""
@ -23,6 +20,7 @@ class FactoryBuilder:
Separates construction from representation.
"""
logger = logging.getLogger("vbancmd.factorybuilder")
BuilderProgress = IntEnum("BuilderProgress", "strip bus command", start=0)
def __init__(self, factory, kind: KindMapClass):
@ -33,7 +31,6 @@ class FactoryBuilder:
f"Finished building buses for {self._factory}",
f"Finished building commands for {self._factory}",
)
self.logger = logger.getChild(self.__class__.__name__)
def _pinfo(self, name: str) -> NoReturn:
"""prints progress status for each step"""
@ -63,6 +60,9 @@ class FactoryBase(VbanCmd):
"""Base class for factories, subclasses VbanCmd."""
def __init__(self, kind_id: str, **kwargs):
defaultsubs = {"pdirty": True, "ldirty": False}
if "subs" in kwargs:
defaultsubs = defaultsubs | kwargs.pop("subs")
defaultkwargs = {
"ip": None,
"port": 6980,
@ -70,13 +70,9 @@ class FactoryBase(VbanCmd):
"bps": 0,
"channel": 0,
"ratelimit": 0.01,
"timeout": 5,
"sync": False,
"pdirty": False,
"ldirty": False,
"subs": defaultsubs,
}
if "subs" in kwargs:
defaultkwargs |= kwargs.pop("subs") # for backwards compatibility
kwargs = defaultkwargs | kwargs
self.kind = kindmap(kind_id)
super().__init__(**kwargs)
@ -192,12 +188,9 @@ def request_vbancmd_obj(kind_id: str, **kwargs) -> VbanCmd:
Returns a reference to a VbanCmd class of a kind
"""
logger_entry = logger.getChild("factory.request_vbancmd_obj")
VBANCMD_obj = None
try:
VBANCMD_obj = vbancmd_factory(kind_id, **kwargs)
except (ValueError, TypeError) as e:
logger_entry.exception(f"{type(e).__name__}: {e}")
raise VBANCMDError(str(e)) from e
raise SystemExit(e)
return VBANCMD_obj

44
vban_cmd/iremote.py
View File

@ -1,10 +1,7 @@
import logging
import time
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class Modes:
@ -29,9 +26,9 @@ class Modes:
_mask: hex = 0x000000F0
_on: hex = 0x00000100 # eq.on
_eq_on: hex = 0x00000100
_cross: hex = 0x00000200
_ab: hex = 0x00000800 # eq.ab
_eq_ab: hex = 0x00000800
_busa: hex = 0x00001000
_busa1: hex = 0x00001000
@ -88,12 +85,10 @@ class IRemote(metaclass=ABCMeta):
def __init__(self, remote, index=None):
self._remote = remote
self.index = index
self.logger = logger.getChild(self.__class__.__name__)
self._modes = Modes()
def getter(self, param):
cmd = self._cmd(param)
self.logger.debug(f"getter: {cmd}")
cmd = f"{self.identifier}.{param}"
if cmd in self._remote.cache:
return self._remote.cache.pop(cmd)
if self._remote.sync:
@ -101,14 +96,7 @@ class IRemote(metaclass=ABCMeta):
def setter(self, param, val):
"""Sends a string request RT packet."""
self.logger.debug(f"setter: {self._cmd(param)}={val}")
self._remote._set_rt(self.identifier, param, val)
def _cmd(self, param):
cmd = (self.identifier,)
if param:
cmd += (f".{param}",)
return "".join(cmd)
self._remote._set_rt(f"{self.identifier}", param, val)
@abstractmethod
def identifier(self):
@ -125,26 +113,20 @@ class IRemote(metaclass=ABCMeta):
def fget(attr, val):
if attr == "mode":
return (f"mode.{val}", 1)
elif attr == "knob":
return ("", val)
return (attr, val)
script = str()
for attr, val in data.items():
if not isinstance(val, dict):
if attr in dir(self): # avoid calling getattr (with hasattr)
attr, val = fget(attr, val)
if isinstance(val, bool):
val = 1 if val else 0
if hasattr(self, attr):
attr, val = fget(attr, val)
if isinstance(val, bool):
val = 1 if val else 0
self._remote.cache[self._cmd(attr)] = val
self._remote._script += f"{self._cmd(attr)}={val};"
else:
target = getattr(self, attr)
target.apply(val)
self._remote.cache[f"{self.identifier}.{attr}"] = val
script += f"{self.identifier}.{attr}={val};"
self._remote.sendtext(script)
return self
def then_wait(self):
self.logger.debug(self._remote._script)
self._remote.sendtext(self._remote._script)
self._remote._script = str()
time.sleep(self._remote.DELAY)

4
vban_cmd/kinds.py
View File

@ -1,8 +1,6 @@
from dataclasses import dataclass
from enum import Enum, unique
from .error import VBANCMDError
@unique
class KindId(Enum):
@ -99,7 +97,7 @@ def request_kind_map(kind_id):
try:
KIND_obj = kind_factory(kind_id)
except ValueError as e:
raise VBANCMDError(str(e)) from e
print(e)
return KIND_obj

6
vban_cmd/meta.py
View File

@ -16,7 +16,7 @@ def channel_bool_prop(param):
)[self.index],
"little",
)
& getattr(self._modes, f"_{param.lower()}")
& getattr(self._modes, f'_{param.replace(".", "_").lower()}')
== 0
)
@ -91,8 +91,8 @@ def bus_mode_prop(param):
return property(fget, fset)
def action_fn(param, val=1):
"""A function that performs an action"""
def action_prop(param, val=1):
"""A param that performs an action"""
def fdo(self):
self.setter(param, val)

46
vban_cmd/packet.py
View File

@ -1,6 +1,6 @@
from dataclasses import dataclass
from typing import Generator
from .kinds import KindMapClass
from .util import comp
VBAN_SERVICE_RTPACKETREGISTER = 32
@ -13,29 +13,11 @@ HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16 + 4
class VbanRtPacket:
"""Represents the body of a VBAN RT data packet"""
_kind: KindMapClass
_voicemeeterType: bytes
_reserved: bytes
_buffersize: bytes
_voicemeeterVersion: bytes
_optionBits: bytes
_samplerate: bytes
_inputLeveldB100: bytes
_outputLeveldB100: bytes
_TransportBit: bytes
_stripState: bytes
_busState: bytes
_stripGaindB100Layer1: bytes
_stripGaindB100Layer2: bytes
_stripGaindB100Layer3: bytes
_stripGaindB100Layer4: bytes
_stripGaindB100Layer5: bytes
_stripGaindB100Layer6: bytes
_stripGaindB100Layer7: bytes
_stripGaindB100Layer8: bytes
_busGaindB100: bytes
_stripLabelUTF8c60: bytes
_busLabelUTF8c60: bytes
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
self._strip_level = self._generate_levels(self._inputLeveldB100)
self._bus_level = self._generate_levels(self._outputLeveldB100)
def _generate_levels(self, levelarray) -> tuple:
return tuple(
@ -43,14 +25,6 @@ class VbanRtPacket:
for i in range(0, len(levelarray), 2)
)
@property
def strip_levels(self):
return self._generate_levels(self._inputLeveldB100)
@property
def bus_levels(self):
return self._generate_levels(self._outputLeveldB100)
def pdirty(self, other) -> bool:
"""True iff any defined parameter has changed"""
@ -72,8 +46,8 @@ class VbanRtPacket:
def ldirty(self, strip_cache, bus_cache) -> bool:
self._strip_comp, self._bus_comp = (
tuple(not val for val in comp(strip_cache, self.strip_levels)),
tuple(not val for val in comp(bus_cache, self.bus_levels)),
tuple(not val for val in comp(strip_cache, self._strip_level)),
tuple(not val for val in comp(bus_cache, self._bus_level)),
)
return any(any(l) for l in (self._strip_comp, self._bus_comp))
@ -103,12 +77,12 @@ class VbanRtPacket:
@property
def inputlevels(self) -> tuple:
"""returns the entire level array across all inputs for a kind"""
return self.strip_levels[0 : (2 * self._kind.phys_in + 8 * self._kind.virt_in)]
return self._strip_level[0 : (2 * self._kind.phys_in + 8 * self._kind.virt_in)]
@property
def outputlevels(self) -> tuple:
"""returns the entire level array across all outputs for a kind"""
return self.bus_levels[0 : 8 * self._kind.num_bus]
return self._bus_level[0 : 8 * self._kind.num_bus]
@property
def stripstate(self) -> tuple:

207
vban_cmd/strip.py
View File

@ -51,22 +51,25 @@ class Strip(IRemote):
class PhysicalStrip(Strip):
@classmethod
def make(cls, remote, index):
return type(
f"PhysicalStrip{remote.kind}",
(cls,),
{
"comp": StripComp(remote, index),
"gate": StripGate(remote, index),
"denoiser": StripDenoiser(remote, index),
"eq": StripEQ(remote, index),
},
)
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
@ -76,182 +79,6 @@ class PhysicalStrip(Strip):
return
class StripComp(IRemote):
@property
def identifier(self) -> str:
return f"Strip[{self.index}].comp"
@property
def knob(self) -> float:
return
@knob.setter
def knob(self, val: float):
self.setter("", val)
@property
def gainin(self) -> float:
return
@gainin.setter
def gainin(self, val: float):
self.setter("GainIn", val)
@property
def ratio(self) -> float:
return
@ratio.setter
def ratio(self, val: float):
self.setter("Ratio", val)
@property
def threshold(self) -> float:
return
@threshold.setter
def threshold(self, val: float):
self.setter("Threshold", val)
@property
def attack(self) -> float:
return
@attack.setter
def attack(self, val: float):
self.setter("Attack", val)
@property
def release(self) -> float:
return
@release.setter
def release(self, val: float):
self.setter("Release", val)
@property
def knee(self) -> float:
return
@knee.setter
def knee(self, val: float):
self.setter("Knee", val)
@property
def gainout(self) -> float:
return
@gainout.setter
def gainout(self, val: float):
self.setter("GainOut", val)
@property
def makeup(self) -> bool:
return
@makeup.setter
def makeup(self, val: bool):
self.setter("makeup", 1 if val else 0)
class StripGate(IRemote):
@property
def identifier(self) -> str:
return f"Strip[{self.index}].gate"
@property
def knob(self) -> float:
return
@knob.setter
def knob(self, val: float):
self.setter("", val)
@property
def threshold(self) -> float:
return
@threshold.setter
def threshold(self, val: float):
self.setter("Threshold", val)
@property
def damping(self) -> float:
return
@damping.setter
def damping(self, val: float):
self.setter("Damping", val)
@property
def bpsidechain(self) -> int:
return
@bpsidechain.setter
def bpsidechain(self, val: int):
self.setter("BPSidechain", val)
@property
def attack(self) -> float:
return
@attack.setter
def attack(self, val: float):
self.setter("Attack", val)
@property
def hold(self) -> float:
return
@hold.setter
def hold(self, val: float):
self.setter("Hold", val)
@property
def release(self) -> float:
return
@release.setter
def release(self, val: float):
self.setter("Release", val)
class StripDenoiser(IRemote):
@property
def identifier(self) -> str:
return f"Strip[{self.index}].denoiser"
@property
def knob(self) -> float:
return
@knob.setter
def knob(self, val: float):
self.setter("", val)
class StripEQ(IRemote):
@property
def identifier(self) -> str:
return f"Strip[{self.index}].eq"
@property
def on(self):
return
@on.setter
def on(self, val: bool):
self.setter("on", 1 if val else 0)
@property
def ab(self):
return
@ab.setter
def ab(self, val: bool):
self.setter("ab", 1 if val else 0)
class VirtualStrip(Strip):
def __str__(self):
return f"{type(self).__name__}{self.index}"
@ -405,7 +232,7 @@ def strip_factory(is_phys_strip, remote, i) -> Union[PhysicalStrip, VirtualStrip
Returns a physical or virtual strip subclass
"""
STRIP_cls = PhysicalStrip.make(remote, i) if is_phys_strip else VirtualStrip
STRIP_cls = PhysicalStrip if is_phys_strip else VirtualStrip
CHANNELOUTMIXIN_cls = _make_channelout_mixins[remote.kind.name]
GAINLAYERMIXIN_cls = _make_gainlayer_mixin(remote, i)

64
vban_cmd/subject.py
View File

@ -1,14 +1,15 @@
import logging
logger = logging.getLogger(__name__)
class Subject:
"""Adds support for observers"""
logger = logging.getLogger("subject.subject")
def __init__(self):
"""Adds support for observers and callbacks"""
"""list of current observers"""
self._observers = list()
self.logger = logger.getChild(self.__class__.__name__)
@property
def observers(self) -> list:
@ -16,57 +17,38 @@ class Subject:
return self._observers
def notify(self, event):
def notify(self, modifier=None):
"""run callbacks on update"""
for o in self._observers:
if hasattr(o, "on_update"):
o.on_update(event)
else:
if o.__name__ == f"on_{event}":
o()
[o.on_update(modifier) for o in self._observers]
def add(self, observer):
"""adds an observer to observers"""
"""adds an observer to _observers"""
try:
iterator = iter(observer)
for o in iterator:
if o not in self._observers:
self._observers.append(o)
self.logger.info(f"{o} added to event observers")
else:
self.logger.error(f"Failed to add {o} to event observers")
except TypeError:
if observer not in self._observers:
self._observers.append(observer)
self.logger.info(f"{observer} added to event observers")
else:
self.logger.error(f"Failed to add {observer} to event observers")
if observer not in self._observers:
self._observers.append(observer)
self.logger.info(f"{type(observer).__name__} added to event observers")
else:
self.logger.error(
f"Failed to add {type(observer).__name__} to event observers"
)
register = add
def remove(self, observer):
"""removes an observer from observers"""
"""removes an observer from _observers"""
try:
iterator = iter(observer)
for o in iterator:
try:
self._observers.remove(o)
self.logger.info(f"{o} removed from event observers")
except ValueError:
self.logger.error(f"Failed to remove {o} from event observers")
except TypeError:
try:
self._observers.remove(observer)
self.logger.info(f"{observer} removed from event observers")
except ValueError:
self.logger.error(f"Failed to remove {observer} from event observers")
self._observers.remove(observer)
self.logger.info(f"{type(observer).__name__} removed from event observers")
except ValueError:
self.logger.error(
f"Failed to remove {type(observer).__name__} from event observers"
)
deregister = remove
def clear(self):
"""clears the observers list"""
"""clears the _observers list"""
self._observers.clear()

76
vban_cmd/vbancmd.py
View File

@ -3,17 +3,18 @@ import socket
import time
from abc import ABCMeta, abstractmethod
from pathlib import Path
from queue import Queue
from typing import Iterable, Optional, Union
from .error import VBANCMDError
try:
import tomllib
except ModuleNotFoundError:
import tomli as tomllib
from .event import Event
from .packet import RequestHeader
from .subject import Subject
from .util import Socket, script
from .worker import Producer, Subscriber, Updater
logger = logging.getLogger(__name__)
from .worker import Subscriber, Updater
class VbanCmd(metaclass=ABCMeta):
@ -27,14 +28,15 @@ class VbanCmd(metaclass=ABCMeta):
1000000, 1500000, 2000000, 3000000,
]
# fmt: on
logger = logging.getLogger("vbancmd.vbancmd")
def __init__(self, **kwargs):
self.logger = logger.getChild(self.__class__.__name__)
self.event = Event({k: kwargs.pop(k) for k in ("pdirty", "ldirty")})
if not kwargs["ip"]:
kwargs |= self._conn_from_toml()
for attr, val in kwargs.items():
setattr(self, attr, val)
if self.ip is None:
conn = self._conn_from_toml()
for attr, val in conn.items():
setattr(self, attr, val)
self.packet_request = RequestHeader(
name=self.streamname,
@ -44,8 +46,9 @@ class VbanCmd(metaclass=ABCMeta):
self.socks = tuple(
socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for _ in Socket
)
self.subject = self.observer = Subject()
self.subject = Subject()
self.cache = {}
self.event = Event(self.subs)
self._pdirty = False
self._ldirty = False
@ -54,31 +57,11 @@ class VbanCmd(metaclass=ABCMeta):
"""Ensure subclasses override str magic method"""
pass
def _conn_from_toml(self) -> dict:
try:
import tomllib
except ModuleNotFoundError:
import tomli as tomllib
def get_filepath():
filepaths = [
Path.cwd() / "vban.toml",
Path.home() / ".config" / "vban-cmd" / "vban.toml",
Path.home() / "Documents" / "Voicemeeter" / "vban.toml",
]
for filepath in filepaths:
if filepath.exists():
return filepath
if filepath := get_filepath():
with open(filepath, "rb") as f:
conn = tomllib.load(f)
assert (
"ip" in conn["connection"]
), "please provide ip, by kwarg or config"
return conn["connection"]
else:
raise VBANCMDError("no ip provided and no vban.toml located.")
def _conn_from_toml(self) -> str:
filepath = Path.cwd() / "vban.toml"
with open(filepath, "rb") as f:
conn = tomllib.load(f)
return conn["connection"]
def __enter__(self):
self.login()
@ -92,11 +75,8 @@ class VbanCmd(metaclass=ABCMeta):
self.subscriber = Subscriber(self)
self.subscriber.start()
queue = Queue()
self.updater = Updater(self, queue)
self.updater = Updater(self)
self.updater.start()
self.producer = Producer(self, queue)
self.producer.start()
self.logger.info(f"{type(self).__name__}: Successfully logged into {self}")
@ -107,27 +87,20 @@ class VbanCmd(metaclass=ABCMeta):
val: Optional[Union[int, float]] = None,
):
"""Sends a string request command over a network."""
cmd = f"{id_}={val};" if not param else f"{id_}.{param}={val};"
cmd = id_ if not param else f"{id_}.{param}={val};"
self.socks[Socket.request].sendto(
self.packet_request.header + cmd.encode(),
(socket.gethostbyname(self.ip), self.port),
)
self.packet_request.framecounter = (
int.from_bytes(self.packet_request.framecounter, "little") + 1
).to_bytes(4, "little")
count = int.from_bytes(self.packet_request.framecounter, "little") + 1
self.packet_request.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.socks[Socket.request].sendto(
self.packet_request.header + cmd.encode(),
(socket.gethostbyname(self.ip), self.port),
)
self.packet_request.framecounter = (
int.from_bytes(self.packet_request.framecounter, "little") + 1
).to_bytes(4, "little")
self._set_rt(cmd)
time.sleep(self.DELAY)
@property
@ -184,7 +157,6 @@ class VbanCmd(metaclass=ABCMeta):
else:
raise ValueError(obj)
self._script = str()
[param(key).apply(datum).then_wait() for key, datum in data.items()]
def apply_config(self, name):
@ -196,7 +168,7 @@ class VbanCmd(metaclass=ABCMeta):
try:
self.apply(self.configs[name])
self.logger.info(f"Profile '{name}' applied!")
except KeyError:
except KeyError as e:
self.logger.error(("\n").join(error_msg))
def logout(self):

156
vban_cmd/worker.py
View File

@ -4,66 +4,60 @@ import threading
import time
from typing import Optional
from .error import VBANCMDConnectionError
from .error import VBANCMDError
from .packet import HEADER_SIZE, SubscribeHeader, VbanRtPacket, VbanRtPacketHeader
from .util import Socket
logger = logging.getLogger(__name__)
from .util import Socket, comp
class Subscriber(threading.Thread):
"""fire a subscription packet every 10 seconds"""
def __init__(self, remote):
super().__init__(name="subscriber", daemon=True)
super().__init__(name="subscriber", target=self.subscribe, daemon=True)
self._remote = remote
self.logger = logger.getChild(self.__class__.__name__)
self.packet = SubscribeHeader()
def run(self):
def subscribe(self):
while self._remote.running:
try:
self._remote.socks[Socket.register].sendto(
self.packet.header,
(socket.gethostbyname(self._remote.ip), self._remote.port),
)
self.packet.framecounter = (
int.from_bytes(self.packet.framecounter, "little") + 1
).to_bytes(4, "little")
count = int.from_bytes(self.packet.framecounter, "little") + 1
self.packet.framecounter = count.to_bytes(4, "little")
time.sleep(10)
except socket.gaierror as e:
self.logger.exception(f"{type(e).__name__}: {e}")
raise VBANCMDConnectionError(
f"unable to resolve hostname {self._remote.ip}"
) from e
except socket.gaierror:
err_msg = f"Unable to resolve hostname {self._remote.ip}"
print(err_msg)
raise VBANCMDError(err_msg)
class Producer(threading.Thread):
"""Continously send job queue to the Updater thread at a rate of self._remote.ratelimit."""
class Updater(threading.Thread):
"""
continously updates the public packet
def __init__(self, remote, queue):
super().__init__(name="producer", daemon=True)
notifies observers of event updates
"""
logger = logging.getLogger("worker.updater")
def __init__(self, remote):
super().__init__(name="updater", target=self.update, daemon=True)
self._remote = remote
self.queue = queue
self.logger = logger.getChild(self.__class__.__name__)
self._remote.socks[Socket.response].settimeout(5)
self._remote.socks[Socket.response].bind(
(socket.gethostbyname(socket.gethostname()), self._remote.port)
)
self.packet_expected = VbanRtPacketHeader()
self._remote._public_packet = self._get_rt()
(
self._remote.cache["strip_level"],
self._remote.cache["bus_level"],
) = self._remote._get_levels(self._remote.public_packet)
def _get_rt(self) -> VbanRtPacket:
"""Attempt to fetch data packet until a valid one found"""
def fget():
data = None
while not data:
data = self._fetch_rt_packet()
time.sleep(self._remote.DELAY)
return data
return fget()
p_in, v_in = self._remote.kind.ins
self._remote._strip_comp = [False] * (2 * p_in + 8 * v_in)
self._remote._bus_comp = [False] * (self._remote.kind.num_bus * 8)
def _fetch_rt_packet(self) -> Optional[VbanRtPacket]:
try:
@ -72,6 +66,7 @@ class Producer(threading.Thread):
if len(data) > HEADER_SIZE:
# check if packet is of type rt packet response
if self.packet_expected.header == data[: HEADER_SIZE - 4]:
self.logger.debug("valid packet received")
return VbanRtPacket(
_kind=self._remote.kind,
_voicemeeterType=data[28:29],
@ -97,14 +92,26 @@ class Producer(threading.Thread):
_stripLabelUTF8c60=data[452:932],
_busLabelUTF8c60=data[932:1412],
)
except TimeoutError as e:
self.logger.exception(f"{type(e).__name__}: {e}")
raise VBANCMDConnectionError(
f"timeout waiting for RtPacket from {self._remote.ip}"
) from e
except TimeoutError:
err_msg = f"Unable to establish connection with {self._remote.ip}"
print(err_msg)
raise VBANCMDError(err_msg)
def run(self):
def _get_rt(self) -> VbanRtPacket:
"""Attempt to fetch data packet until a valid one found"""
def fget():
data = None
while not data:
data = self._fetch_rt_packet()
time.sleep(self._remote.DELAY)
return data
return fget()
def update(self):
while self._remote.running:
start = time.time()
_pp = self._get_rt()
pdirty = _pp.pdirty(self._remote.public_packet)
ldirty = _pp.ldirty(
@ -112,63 +119,24 @@ class Producer(threading.Thread):
)
if pdirty or ldirty:
self.logger.debug("dirty state, updating public packet")
self._remote._public_packet = _pp
self._remote._pdirty = pdirty
self._remote._ldirty = ldirty
self._remote._pdirty = pdirty
self._remote._ldirty = ldirty
if self._remote.event.pdirty:
self.queue.put("pdirty")
if self._remote.event.ldirty:
self.queue.put("ldirty")
time.sleep(self._remote.ratelimit)
self.logger.debug(f"terminating {self.name} thread")
self.queue.put(None)
class Updater(threading.Thread):
"""
continously updates the public packet
notifies observers of event updates
"""
def __init__(self, remote, queue):
super().__init__(name="updater", daemon=True)
self._remote = remote
self.queue = queue
self.logger = logger.getChild(self.__class__.__name__)
self._remote.socks[Socket.response].settimeout(self._remote.timeout)
self._remote.socks[Socket.response].bind(
(socket.gethostbyname(socket.gethostname()), self._remote.port)
)
p_in, v_in = self._remote.kind.ins
self._remote._strip_comp = [False] * (2 * p_in + 8 * v_in)
self._remote._bus_comp = [False] * (self._remote.kind.num_bus * 8)
def run(self):
"""
Continously update observers of dirty states.
Generate _strip_comp, _bus_comp and update level cache if ldirty.
"""
while True:
event = self.queue.get()
if event is None:
self.logger.debug(f"terminating {self.name} thread")
break
if event == "pdirty" and self._remote.pdirty:
self._remote.subject.notify(event)
elif event == "ldirty" and self._remote.ldirty:
if self._remote.event.pdirty and self._remote.pdirty:
self._remote.subject.notify("pdirty")
if self._remote.event.ldirty and self._remote.ldirty:
self._remote._strip_comp, self._remote._bus_comp = (
self._remote._public_packet._strip_comp,
self._remote._public_packet._bus_comp,
_pp._strip_comp,
_pp._bus_comp,
)
(
self._remote.cache["strip_level"],
self._remote.cache["bus_level"],
) = (
self._remote._public_packet.inputlevels,
self._remote._public_packet.outputlevels,
self._remote.cache["strip_level"], self._remote.cache["bus_level"] = (
_pp.inputlevels,
_pp.outputlevels,
)
self._remote.subject.notify(event)
self._remote.subject.notify("ldirty")
elapsed = time.time() - start
if self._remote.ratelimit - elapsed > 0:
time.sleep(self._remote.ratelimit - elapsed)