mirror of
https://github.com/onyx-and-iris/vban-cmd-python.git
synced 2024-11-15 17:10:46 +00:00
major version bump due to dependency change.
now packaged with poetry. added to pypi. major version bump due to dependency change. interface reworked to match the remote-api interface. readme updated with changes to installation pre-commit hook temporarily removed
This commit is contained in:
parent
933d182f60
commit
b9db01c8f4
268
README.md
268
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 = '<ip address>'
|
||||
# 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 = "<ip address>"
|
||||
|
||||
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 = '<ip address>'
|
||||
|
||||
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`
|
||||
|
||||
|
40
__main__.py
40
__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 = '<ip address>'
|
||||
# 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 = "<ip address>"
|
||||
|
||||
main()
|
||||
|
@ -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('<Double-Button-1>', 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("<Double-Button-1>", 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()
|
||||
|
288
poetry.lock
generated
Normal file
288
poetry.lock
generated
Normal file
@ -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"},
|
||||
]
|
22
pyproject.toml
Normal file
22
pyproject.toml
Normal file
@ -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 <code@onyxandiris.online>"]
|
||||
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"
|
10
setup.py
10
setup.py
@ -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]"]},
|
||||
)
|
@ -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():
|
||||
|
7
tests/conftest.py
Normal file
7
tests/conftest.py
Normal file
@ -0,0 +1,7 @@
|
||||
def pytest_addoption(parser):
|
||||
parser.addoption(
|
||||
"--run-slow",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="Run slow tests",
|
||||
)
|
@ -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"
|
||||
}
|
@ -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(
|
||||
|
@ -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)
|
||||
|
3
vban_cmd/__init__.py
Normal file
3
vban_cmd/__init__.py
Normal file
@ -0,0 +1,3 @@
|
||||
from .factory import request_vbancmd_obj as api
|
||||
|
||||
__ALL__ = ["api"]
|
269
vban_cmd/base.py
Normal file
269
vban_cmd/base.py
Normal file
@ -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()
|
151
vban_cmd/bus.py
Normal file
151
vban_cmd/bus.py
Normal file
@ -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)
|
@ -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,),
|
||||
{
|
||||
**{
|
191
vban_cmd/config.py
Normal file
191
vban_cmd/config.py
Normal file
@ -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
|
@ -1,2 +1,4 @@
|
||||
class VMCMDErrors(Exception):
|
||||
"""general errors"""
|
||||
|
||||
pass
|
190
vban_cmd/factory.py
Normal file
190
vban_cmd/factory.py
Normal file
@ -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
|
@ -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)
|
104
vban_cmd/kinds.py
Normal file
104
vban_cmd/kinds.py
Normal file
@ -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)
|
@ -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):
|
@ -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
|
225
vban_cmd/strip.py
Normal file
225
vban_cmd/strip.py
Normal file
@ -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)
|
39
vban_cmd/subject.py
Normal file
39
vban_cmd/subject.py
Normal file
@ -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()
|
@ -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)
|
@ -1,3 +0,0 @@
|
||||
from .vbancmd import connect
|
||||
|
||||
__ALL__ = ["connect"]
|
132
vbancmd/bus.py
132
vbancmd/bus.py
@ -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",
|
||||
]
|
||||
},
|
||||
},
|
||||
)
|
@ -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())
|
@ -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/<kind_id>/<profile>.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
|
200
vbancmd/strip.py
200
vbancmd/strip.py
@ -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}
|
@ -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()
|
@ -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}")
|
Loading…
Reference in New Issue
Block a user