onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2026-03-03 00:39:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: onyx_online:v2.6.0
Branches Tags
onyx_online:dev
onyx_online:implement-ping
onyx_online:refactor-VbanPacketNBS0
onyx_online:add-packet-nbs1
onyx_online:add-event-stop-object
onyx_online:main
onyx_online:v2.9.2
onyx_online:v2.9.1
onyx_online:v2.9.0
onyx_online:v2.8.1
onyx_online:v2.8.0
onyx_online:v2.7.1
onyx_online:v2.7.0
onyx_online:v2.6.0
...
compare: onyx_online:dev
Branches Tags
onyx_online:dev
onyx_online:implement-ping
onyx_online:refactor-VbanPacketNBS0
onyx_online:add-packet-nbs1
onyx_online:add-event-stop-object
onyx_online:main
onyx_online:v2.9.2
onyx_online:v2.9.1
onyx_online:v2.9.0
onyx_online:v2.8.1
onyx_online:v2.8.0
onyx_online:v2.7.1
onyx_online:v2.7.0
onyx_online:v2.6.0

25 Commits
v2.6.0 ... dev

Author SHA1 Message Date
onyx-and-iris
55b3125e10 fixes regression in apply() 
patch bump
 2026-03-02 23:52:06 +00:00
onyx-and-iris
7b3340042c upd reference to ip 2026-03-02 23:26:38 +00:00
onyx-and-iris
6ea0859180 patch bump 2026-03-02 23:25:03 +00:00
onyx-and-iris
81ed963bea fix references to remote.ip 2026-03-02 23:24:09 +00:00
onyx-and-iris
0b99b6a67f update references to ip kwarg 2026-03-02 23:21:57 +00:00
onyx-and-iris
86d0aa91c3 add a ratelimit decorator to {VbanCmd}.sendtext() 
ip kwarg renamed to host.
 2026-03-02 23:20:45 +00:00
onyx-and-iris
cf66ae252c minor bump 2026-03-02 20:50:12 +00:00
onyx-and-iris
42f6f29d1e add 2.9.0 to CHANGELOG 2026-03-02 20:50:05 +00:00
onyx-and-iris
a210766b7b improve framecounter thread safety 2026-03-02 20:44:08 +00:00
onyx-and-iris
7d741d6e8b {VbanCmd}._ping() now causes login to fail fast if no pong is received. 2026-03-02 20:26:35 +00:00
onyx-and-iris
8be9d3cb7f implement ping/pong dataclasses 2026-03-02 20:25:58 +00:00
onyx-and-iris
23b99cb66b perform some path magic so Voicemeeter receives the entire path 
patch bump
 2026-03-01 21:29:09 +00:00
onyx-and-iris
2fd7b8ad8b minor bump 2026-03-01 21:13:38 +00:00
onyx-and-iris
c851cb5abe add Recorder section to README 2026-03-01 21:11:48 +00:00
onyx-and-iris
dc681f50d0 add Recorder 
add it to banana+potato
 2026-03-01 21:10:10 +00:00
onyx-and-iris
a0ec00652b reduce the level of logging for packet parse errors 
patch bump
 2026-03-01 17:22:06 +00:00
onyx-and-iris
69263c22f2 add 2.7.0 to CHANGELOG 2026-03-01 17:04:37 +00:00
onyx-and-iris
ad2cfeaae6 entry point now accepts a 'matrix' kind although it's main purpose is to disable the rt listener threads. 
{VbanCmd}.sendtext():
- remove the @script decorator which I'm sure nobody has ever used anyway
- if rt listeners are disabled and it's a matrix query request, attempt to read a response.
 2026-03-01 16:21:47 +00:00
onyx-and-iris
1123fe6432 move header validation into class methods 
add _parse_vban_service_header() helper function
 2026-03-01 16:17:03 +00:00
onyx-and-iris
3c3e415d7e add _send_request() helper method. 2026-03-01 11:09:45 +00:00
onyx-and-iris
8cfeb45fcb update imports 2026-03-01 11:09:28 +00:00
onyx-and-iris
10b38b3fcc move packet classes into internal packet module 2026-03-01 11:09:22 +00:00
onyx-and-iris
ff5ac193c8 add ChannelState interface, use it in the meta functions. 
reword busmodes bitwise logic.

comment out ratelimit, this will probably get permanently removed.
 2026-03-01 03:37:57 +00:00
onyx-and-iris
2f3cd0e07f use db levels throughout the package. This is cleaner than converting to db but comparing raw integer values. 2026-03-01 01:08:02 +00:00
onyx-and-iris
d689b3a301 move voicemeetertype(), voicemeeterversion() and samplerate() properties into VbanPacket 
add NamedTuples for Levels, Labels and States.

refactor the levels properties

update the math in util.comp()

StripLevel/BusLevel getters updated according to changes in VbanPacketNBS0

remove {VbanCmd}._get_levels(), it's no longer necessary.
 2026-03-01 00:25:22 +00:00
18 changed files with 1423 additions and 606 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

@ -159,3 +159,5 @@ config.toml
vban.toml vban.toml
.vscode/ .vscode/
PING_FEATURE.md

28
CHANGELOG.md
View File

@ -11,6 +11,34 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
- [x] - [x]
## [2.9.0] - 2026-03-02
### Added
- Recorder class, see [Recorder](https://github.com/onyx-and-iris/vban-cmd-python?tab=readme-ov-file#recorder) in README.
- Ping/pong implemented. If a pong is not received {VbanCmd}.login() will fail fast. This prevents the rt listener threads from starting up.
- It has the added benefit of automatically detecting the type of VBAN server (Voicemeeter or Matrix).
- A thread lock around the framecounter to improve thread safety since it can be accessed by both the main thread and the Producer thread.
## [2.7.0] - 2026-03-01
### Added
- new kind `matrix` has been added, it does two things:
- scales the interface according to `potato` kind, in practice this has no affect but it's required by the builder classes.
- disables the rt listener threads since we aren't expecting to receive any from a Matrix VBAN server.
- however, matrix responses may still be received with the {VbanCmd}.sendtext() method.
### Changed
- `outbound` kwarg has been renamed to `disable_rt_listeners`. Since it's job is to disable the listener threads for incoming RT packets this new name is more descriptive.
- dataclasses representing packet headers and packets with ident:0 and ident:1 have been moved into an internal packet module.
### Removed
- {VbanCmd}.sendtext() @script decorator removed. It's purpose was to attempt to convert a dictionary to a script but it was poorly implemented and there exists the {VbanCmd}.apply() method already.
## [2.6.0] - 2026-02-26 ## [2.6.0] - 2026-02-26
### Added ### Added

78
README.md
View File

@ -8,19 +8,19 @@
# VBAN CMD # VBAN CMD
This python interface allows you to transmit Voicemeeter parameters over a network. This python interface allows you to send Voicemeeter/Matrix commands over a network.
It may be used standalone or to extend the [Voicemeeter Remote Python API](https://github.com/onyx-and-iris/voicemeeter-api-python) It offers the same public API as [Voicemeeter Remote Python API](https://github.com/onyx-and-iris/voicemeeter-api-python).
There is no support for audio transfer in this package, only parameters. Only the VBAN SERVICE/TEXT subprotocols are supported, there is no support for AUDIO or MIDI in this package.
For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md) For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
## Tested against ## Tested against
- Basic 1.0.8.8 - Basic 1.1.2.2
- Banana 2.0.6.8 - Banana 2.1.2.2
- Potato 3.0.2.8 - Potato 3.1.2.2
## Requirements ## Requirements
@ -29,7 +29,9 @@ For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
## Installation ## Installation
`pip install vban-cmd` ```console
pip install vban-cmd
```
## `Use` ## `Use`
@ -39,14 +41,14 @@ Load VBAN connection info from toml config. A valid `vban.toml` might look like
```toml ```toml
[connection] [connection]
ip = "gamepc.local" host = "localhost"
port = 6980 port = 6980
streamname = "Command1" streamname = "Command1"
``` ```
It should be placed in \<user home directory\> / "Documents" / "Voicemeeter" / "configs" It should be placed in \<user home directory\> / "Documents" / "Voicemeeter" / "configs"
Alternatively you may pass `ip`, `port`, `streamname` as keyword arguments. Alternatively you may pass `host`, `port`, `streamname` as keyword arguments.
#### `__main__.py` #### `__main__.py`
@ -83,7 +85,7 @@ def main():
KIND_ID = 'banana' KIND_ID = 'banana'
with vban_cmd.api( with vban_cmd.api(
KIND_ID, ip='gamepc.local', port=6980, streamname='Command1' KIND_ID, host='localhost', port=6980, streamname='Command1'
) as vban: ) as vban:
do = ManyThings(vban) do = ManyThings(vban)
do.things() do.things()
@ -113,6 +115,8 @@ Pass the kind of Voicemeeter as an argument. KIND_ID may be:
- `banana` - `banana`
- `potato` - `potato`
A fourth kind `matrix` has been added, if you pass it as a KIND_ID you are expected to use the [{VbanCmd}.sendtext()](https://github.com/onyx-and-iris/vban-cmd-python?tab=readme-ov-file#vbansendtextscript) method for sending text requests.
## `Available commands` ## `Available commands`
### Strip ### Strip
@ -345,6 +349,40 @@ vban.strip[0].fadeto(-10.3, 1000)
vban.bus[3].fadeby(-5.6, 500) vban.bus[3].fadeby(-5.6, 500)
``` ```
### Recorder
The following methods are available
- `play()`
- `stop()`
- `pause()`
- `record()`
- `ff()`
- `rew()`
- `load(filepath)`: raw string
- `goto(time_string)`: time string in format `hh:mm:ss`
The following properties are available
- `samplerate`: int, (22050, 24000, 32000, 44100, 48000, 88200, 96000, 176400, 192000)
- `bitresolution`: int, (8, 16, 24, 32)
- `channel`: int, from 1 to 8
- `kbps`: int, (32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320)
- `gain`: float, from -60.0 to 12.0
example:
```python
vban.recorder.play()
vban.recorder.stop()
# filepath as raw string
vban.recorder.load(r'C:\music\mytune.mp3')
# set the goto time to 1m 30s
vban.recorder.goto('00:01:30')
```
### Command ### Command
Certain 'special' commands are defined by the API as performing actions rather than setting values. The following methods are available: Certain 'special' commands are defined by the API as performing actions rather than setting values. The following methods are available:
@ -436,7 +474,7 @@ example:
import vban_cmd import vban_cmd
opts = { opts = {
'ip': '<ip address>', 'host': '<ip address>',
'streamname': 'Command1', 'streamname': 'Command1',
'port': 6980, 'port': 6980,
} }
@ -503,15 +541,17 @@ print(vban.event.get())
You may pass the following optional keyword arguments: You may pass the following optional keyword arguments:
- `ip`: str='localhost', ip or hostname of remote machine - `host`: str='localhost', ip or hostname of remote machine
- `port`: int=6980, vban udp port of remote machine. - `port`: int=6980, vban udp port of remote machine.
- `streamname`: str='Command1', name of the stream to connect to. - `streamname`: str='Command1', name of the stream to connect to.
- `bps`: int=256000, bps rate of the stream. - `bps`: int=256000, bps rate of the stream.
- `channel`: int=0, channel on which to send the UDP requests. - `channel`: int=0, channel on which to send the UDP requests.
- `pdirty`: boolean=False, parameter updates - `pdirty`: boolean=False, parameter updates
- `ldirty`: boolean=False, level updates - `ldirty`: boolean=False, level updates
- `timeout`: int=5, amount of time (seconds) to wait for an incoming RT data packet (parameter states). - `script_ratelimit`: float=0.05, default to 20 script requests per second. This affects vban.sendtext() specifically.
- `outbound`: boolean=False, set `True` if you are only interested in sending commands. (no rt packets will be received) - `timeout`: int=5, timeout for socket operations.
- `disable_rt_listeners`: boolean=False, set `True` if you don't wish to receive RT packets.
- You can still send Matrix string requests ending with `?` and receive a response.
#### `vban.pdirty` #### `vban.pdirty`
@ -529,6 +569,14 @@ Sends a script block as a string request, for example:
vban.sendtext('Strip[0].Mute=1;Bus[0].Mono=1') vban.sendtext('Strip[0].Mute=1;Bus[0].Mono=1')
``` ```
You can even use it to send matrix commands:
```python
vban.sendtext('Point(ASIO128.IN[1..4],ASIO128.OUT[1]).dBGain = -3.0')
vban.sendtext('Command.Version = ?')
```
## Errors ## Errors
- `errors.VBANCMDError`: Base VBANCMD Exception class. - `errors.VBANCMDError`: Base VBANCMD Exception class.
@ -544,7 +592,7 @@ import vban_cmd
logging.basicConfig(level=logging.DEBUG) logging.basicConfig(level=logging.DEBUG)
opts = {'ip': 'ip.local', 'port': 6980, 'streamname': 'Command1'} opts = {'host': 'localhost', 'port': 6980, 'streamname': 'Command1'}
with vban_cmd.api('banana', **opts) as vban: with vban_cmd.api('banana', **opts) as vban:
... ...
``` ```

4
pyproject.toml
View File

@ -1,6 +1,6 @@
[project] [project]
name = "vban-cmd" name = "vban-cmd"
version = "2.6.0" version = "2.9.2"
description = "Python interface for the VBAN RT Packet Service (Sendtext)" description = "Python interface for the VBAN RT Packet Service (Sendtext)"
authors = [{ name = "Onyx and Iris", email = "code@onyxandiris.online" }] authors = [{ name = "Onyx and Iris", email = "code@onyxandiris.online" }]
license = { text = "MIT" } license = { text = "MIT" }
@ -9,7 +9,7 @@ requires-python = ">=3.10"
dependencies = ["tomli (>=2.0.1,<3.0) ; python_version < '3.11'"] dependencies = ["tomli (>=2.0.1,<3.0) ; python_version < '3.11'"]
[tool.poetry.requires-plugins] [tool.poetry.requires-plugins]
poethepoet = "^0.35.0" poethepoet = ">=0.42.0"
[tool.poetry.group.dev.dependencies] [tool.poetry.group.dev.dependencies]
pytest = "^8.3.4" pytest = "^8.3.4"

68
uv.lock generated Normal file
View File

@ -0,0 +1,68 @@
version = 1
revision = 3
requires-python = ">=3.10"
[[package]]
name = "tomli"
version = "2.4.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/82/30/31573e9457673ab10aa432461bee537ce6cef177667deca369efb79df071/tomli-2.4.0.tar.gz", hash = "sha256:aa89c3f6c277dd275d8e243ad24f3b5e701491a860d5121f2cdd399fbb31fc9c", size = 17477, upload-time = "2026-01-11T11:22:38.165Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3c/d9/3dc2289e1f3b32eb19b9785b6a006b28ee99acb37d1d47f78d4c10e28bf8/tomli-2.4.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:b5ef256a3fd497d4973c11bf142e9ed78b150d36f5773f1ca6088c230ffc5867", size = 153663, upload-time = "2026-01-11T11:21:45.27Z" },
{ url = "https://files.pythonhosted.org/packages/51/32/ef9f6845e6b9ca392cd3f64f9ec185cc6f09f0a2df3db08cbe8809d1d435/tomli-2.4.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:5572e41282d5268eb09a697c89a7bee84fae66511f87533a6f88bd2f7b652da9", size = 148469, upload-time = "2026-01-11T11:21:46.873Z" },
{ url = "https://files.pythonhosted.org/packages/d6/c2/506e44cce89a8b1b1e047d64bd495c22c9f71f21e05f380f1a950dd9c217/tomli-2.4.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:551e321c6ba03b55676970b47cb1b73f14a0a4dce6a3e1a9458fd6d921d72e95", size = 236039, upload-time = "2026-01-11T11:21:48.503Z" },
{ url = "https://files.pythonhosted.org/packages/b3/40/e1b65986dbc861b7e986e8ec394598187fa8aee85b1650b01dd925ca0be8/tomli-2.4.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5e3f639a7a8f10069d0e15408c0b96a2a828cfdec6fca05296ebcdcc28ca7c76", size = 243007, upload-time = "2026-01-11T11:21:49.456Z" },
{ url = "https://files.pythonhosted.org/packages/9c/6f/6e39ce66b58a5b7ae572a0f4352ff40c71e8573633deda43f6a379d56b3e/tomli-2.4.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:1b168f2731796b045128c45982d3a4874057626da0e2ef1fdd722848b741361d", size = 240875, upload-time = "2026-01-11T11:21:50.755Z" },
{ url = "https://files.pythonhosted.org/packages/aa/ad/cb089cb190487caa80204d503c7fd0f4d443f90b95cf4ef5cf5aa0f439b0/tomli-2.4.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:133e93646ec4300d651839d382d63edff11d8978be23da4cc106f5a18b7d0576", size = 246271, upload-time = "2026-01-11T11:21:51.81Z" },
{ url = "https://files.pythonhosted.org/packages/0b/63/69125220e47fd7a3a27fd0de0c6398c89432fec41bc739823bcc66506af6/tomli-2.4.0-cp311-cp311-win32.whl", hash = "sha256:b6c78bdf37764092d369722d9946cb65b8767bfa4110f902a1b2542d8d173c8a", size = 96770, upload-time = "2026-01-11T11:21:52.647Z" },
{ url = "https://files.pythonhosted.org/packages/1e/0d/a22bb6c83f83386b0008425a6cd1fa1c14b5f3dd4bad05e98cf3dbbf4a64/tomli-2.4.0-cp311-cp311-win_amd64.whl", hash = "sha256:d3d1654e11d724760cdb37a3d7691f0be9db5fbdaef59c9f532aabf87006dbaa", size = 107626, upload-time = "2026-01-11T11:21:53.459Z" },
{ url = "https://files.pythonhosted.org/packages/2f/6d/77be674a3485e75cacbf2ddba2b146911477bd887dda9d8c9dfb2f15e871/tomli-2.4.0-cp311-cp311-win_arm64.whl", hash = "sha256:cae9c19ed12d4e8f3ebf46d1a75090e4c0dc16271c5bce1c833ac168f08fb614", size = 94842, upload-time = "2026-01-11T11:21:54.831Z" },
{ url = "https://files.pythonhosted.org/packages/3c/43/7389a1869f2f26dba52404e1ef13b4784b6b37dac93bac53457e3ff24ca3/tomli-2.4.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:920b1de295e72887bafa3ad9f7a792f811847d57ea6b1215154030cf131f16b1", size = 154894, upload-time = "2026-01-11T11:21:56.07Z" },
{ url = "https://files.pythonhosted.org/packages/e9/05/2f9bf110b5294132b2edf13fe6ca6ae456204f3d749f623307cbb7a946f2/tomli-2.4.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7d6d9a4aee98fac3eab4952ad1d73aee87359452d1c086b5ceb43ed02ddb16b8", size = 149053, upload-time = "2026-01-11T11:21:57.467Z" },
{ url = "https://files.pythonhosted.org/packages/e8/41/1eda3ca1abc6f6154a8db4d714a4d35c4ad90adc0bcf700657291593fbf3/tomli-2.4.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:36b9d05b51e65b254ea6c2585b59d2c4cb91c8a3d91d0ed0f17591a29aaea54a", size = 243481, upload-time = "2026-01-11T11:21:58.661Z" },
{ url = "https://files.pythonhosted.org/packages/d2/6d/02ff5ab6c8868b41e7d4b987ce2b5f6a51d3335a70aa144edd999e055a01/tomli-2.4.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1c8a885b370751837c029ef9bc014f27d80840e48bac415f3412e6593bbc18c1", size = 251720, upload-time = "2026-01-11T11:22:00.178Z" },
{ url = "https://files.pythonhosted.org/packages/7b/57/0405c59a909c45d5b6f146107c6d997825aa87568b042042f7a9c0afed34/tomli-2.4.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:8768715ffc41f0008abe25d808c20c3d990f42b6e2e58305d5da280ae7d1fa3b", size = 247014, upload-time = "2026-01-11T11:22:01.238Z" },
{ url = "https://files.pythonhosted.org/packages/2c/0e/2e37568edd944b4165735687cbaf2fe3648129e440c26d02223672ee0630/tomli-2.4.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:7b438885858efd5be02a9a133caf5812b8776ee0c969fea02c45e8e3f296ba51", size = 251820, upload-time = "2026-01-11T11:22:02.727Z" },
{ url = "https://files.pythonhosted.org/packages/5a/1c/ee3b707fdac82aeeb92d1a113f803cf6d0f37bdca0849cb489553e1f417a/tomli-2.4.0-cp312-cp312-win32.whl", hash = "sha256:0408e3de5ec77cc7f81960c362543cbbd91ef883e3138e81b729fc3eea5b9729", size = 97712, upload-time = "2026-01-11T11:22:03.777Z" },
{ url = "https://files.pythonhosted.org/packages/69/13/c07a9177d0b3bab7913299b9278845fc6eaaca14a02667c6be0b0a2270c8/tomli-2.4.0-cp312-cp312-win_amd64.whl", hash = "sha256:685306e2cc7da35be4ee914fd34ab801a6acacb061b6a7abca922aaf9ad368da", size = 108296, upload-time = "2026-01-11T11:22:04.86Z" },
{ url = "https://files.pythonhosted.org/packages/18/27/e267a60bbeeee343bcc279bb9e8fbed0cbe224bc7b2a3dc2975f22809a09/tomli-2.4.0-cp312-cp312-win_arm64.whl", hash = "sha256:5aa48d7c2356055feef06a43611fc401a07337d5b006be13a30f6c58f869e3c3", size = 94553, upload-time = "2026-01-11T11:22:05.854Z" },
{ url = "https://files.pythonhosted.org/packages/34/91/7f65f9809f2936e1f4ce6268ae1903074563603b2a2bd969ebbda802744f/tomli-2.4.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:84d081fbc252d1b6a982e1870660e7330fb8f90f676f6e78b052ad4e64714bf0", size = 154915, upload-time = "2026-01-11T11:22:06.703Z" },
{ url = "https://files.pythonhosted.org/packages/20/aa/64dd73a5a849c2e8f216b755599c511badde80e91e9bc2271baa7b2cdbb1/tomli-2.4.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:9a08144fa4cba33db5255f9b74f0b89888622109bd2776148f2597447f92a94e", size = 149038, upload-time = "2026-01-11T11:22:07.56Z" },
{ url = "https://files.pythonhosted.org/packages/9e/8a/6d38870bd3d52c8d1505ce054469a73f73a0fe62c0eaf5dddf61447e32fa/tomli-2.4.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c73add4bb52a206fd0c0723432db123c0c75c280cbd67174dd9d2db228ebb1b4", size = 242245, upload-time = "2026-01-11T11:22:08.344Z" },
{ url = "https://files.pythonhosted.org/packages/59/bb/8002fadefb64ab2669e5b977df3f5e444febea60e717e755b38bb7c41029/tomli-2.4.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1fb2945cbe303b1419e2706e711b7113da57b7db31ee378d08712d678a34e51e", size = 250335, upload-time = "2026-01-11T11:22:09.951Z" },
{ url = "https://files.pythonhosted.org/packages/a5/3d/4cdb6f791682b2ea916af2de96121b3cb1284d7c203d97d92d6003e91c8d/tomli-2.4.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:bbb1b10aa643d973366dc2cb1ad94f99c1726a02343d43cbc011edbfac579e7c", size = 245962, upload-time = "2026-01-11T11:22:11.27Z" },
{ url = "https://files.pythonhosted.org/packages/f2/4a/5f25789f9a460bd858ba9756ff52d0830d825b458e13f754952dd15fb7bb/tomli-2.4.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:4cbcb367d44a1f0c2be408758b43e1ffb5308abe0ea222897d6bfc8e8281ef2f", size = 250396, upload-time = "2026-01-11T11:22:12.325Z" },
{ url = "https://files.pythonhosted.org/packages/aa/2f/b73a36fea58dfa08e8b3a268750e6853a6aac2a349241a905ebd86f3047a/tomli-2.4.0-cp313-cp313-win32.whl", hash = "sha256:7d49c66a7d5e56ac959cb6fc583aff0651094ec071ba9ad43df785abc2320d86", size = 97530, upload-time = "2026-01-11T11:22:13.865Z" },
{ url = "https://files.pythonhosted.org/packages/3b/af/ca18c134b5d75de7e8dc551c5234eaba2e8e951f6b30139599b53de9c187/tomli-2.4.0-cp313-cp313-win_amd64.whl", hash = "sha256:3cf226acb51d8f1c394c1b310e0e0e61fecdd7adcb78d01e294ac297dd2e7f87", size = 108227, upload-time = "2026-01-11T11:22:15.224Z" },
{ url = "https://files.pythonhosted.org/packages/22/c3/b386b832f209fee8073c8138ec50f27b4460db2fdae9ffe022df89a57f9b/tomli-2.4.0-cp313-cp313-win_arm64.whl", hash = "sha256:d20b797a5c1ad80c516e41bc1fb0443ddb5006e9aaa7bda2d71978346aeb9132", size = 94748, upload-time = "2026-01-11T11:22:16.009Z" },
{ url = "https://files.pythonhosted.org/packages/f3/c4/84047a97eb1004418bc10bdbcfebda209fca6338002eba2dc27cc6d13563/tomli-2.4.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:26ab906a1eb794cd4e103691daa23d95c6919cc2fa9160000ac02370cc9dd3f6", size = 154725, upload-time = "2026-01-11T11:22:17.269Z" },
{ url = "https://files.pythonhosted.org/packages/a8/5d/d39038e646060b9d76274078cddf146ced86dc2b9e8bbf737ad5983609a0/tomli-2.4.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:20cedb4ee43278bc4f2fee6cb50daec836959aadaf948db5172e776dd3d993fc", size = 148901, upload-time = "2026-01-11T11:22:18.287Z" },
{ url = "https://files.pythonhosted.org/packages/73/e5/383be1724cb30f4ce44983d249645684a48c435e1cd4f8b5cded8a816d3c/tomli-2.4.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:39b0b5d1b6dd03684b3fb276407ebed7090bbec989fa55838c98560c01113b66", size = 243375, upload-time = "2026-01-11T11:22:19.154Z" },
{ url = "https://files.pythonhosted.org/packages/31/f0/bea80c17971c8d16d3cc109dc3585b0f2ce1036b5f4a8a183789023574f2/tomli-2.4.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a26d7ff68dfdb9f87a016ecfd1e1c2bacbe3108f4e0f8bcd2228ef9a766c787d", size = 250639, upload-time = "2026-01-11T11:22:20.168Z" },
{ url = "https://files.pythonhosted.org/packages/2c/8f/2853c36abbb7608e3f945d8a74e32ed3a74ee3a1f468f1ffc7d1cb3abba6/tomli-2.4.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:20ffd184fb1df76a66e34bd1b36b4a4641bd2b82954befa32fe8163e79f1a702", size = 246897, upload-time = "2026-01-11T11:22:21.544Z" },
{ url = "https://files.pythonhosted.org/packages/49/f0/6c05e3196ed5337b9fe7ea003e95fd3819a840b7a0f2bf5a408ef1dad8ed/tomli-2.4.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:75c2f8bbddf170e8effc98f5e9084a8751f8174ea6ccf4fca5398436e0320bc8", size = 254697, upload-time = "2026-01-11T11:22:23.058Z" },
{ url = "https://files.pythonhosted.org/packages/f3/f5/2922ef29c9f2951883525def7429967fc4d8208494e5ab524234f06b688b/tomli-2.4.0-cp314-cp314-win32.whl", hash = "sha256:31d556d079d72db7c584c0627ff3a24c5d3fb4f730221d3444f3efb1b2514776", size = 98567, upload-time = "2026-01-11T11:22:24.033Z" },
{ url = "https://files.pythonhosted.org/packages/7b/31/22b52e2e06dd2a5fdbc3ee73226d763b184ff21fc24e20316a44ccc4d96b/tomli-2.4.0-cp314-cp314-win_amd64.whl", hash = "sha256:43e685b9b2341681907759cf3a04e14d7104b3580f808cfde1dfdb60ada85475", size = 108556, upload-time = "2026-01-11T11:22:25.378Z" },
{ url = "https://files.pythonhosted.org/packages/48/3d/5058dff3255a3d01b705413f64f4306a141a8fd7a251e5a495e3f192a998/tomli-2.4.0-cp314-cp314-win_arm64.whl", hash = "sha256:3d895d56bd3f82ddd6faaff993c275efc2ff38e52322ea264122d72729dca2b2", size = 96014, upload-time = "2026-01-11T11:22:26.138Z" },
{ url = "https://files.pythonhosted.org/packages/b8/4e/75dab8586e268424202d3a1997ef6014919c941b50642a1682df43204c22/tomli-2.4.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:5b5807f3999fb66776dbce568cc9a828544244a8eb84b84b9bafc080c99597b9", size = 163339, upload-time = "2026-01-11T11:22:27.143Z" },
{ url = "https://files.pythonhosted.org/packages/06/e3/b904d9ab1016829a776d97f163f183a48be6a4deb87304d1e0116a349519/tomli-2.4.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:c084ad935abe686bd9c898e62a02a19abfc9760b5a79bc29644463eaf2840cb0", size = 159490, upload-time = "2026-01-11T11:22:28.399Z" },
{ url = "https://files.pythonhosted.org/packages/e3/5a/fc3622c8b1ad823e8ea98a35e3c632ee316d48f66f80f9708ceb4f2a0322/tomli-2.4.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0f2e3955efea4d1cfbcb87bc321e00dc08d2bcb737fd1d5e398af111d86db5df", size = 269398, upload-time = "2026-01-11T11:22:29.345Z" },
{ url = "https://files.pythonhosted.org/packages/fd/33/62bd6152c8bdd4c305ad9faca48f51d3acb2df1f8791b1477d46ff86e7f8/tomli-2.4.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0e0fe8a0b8312acf3a88077a0802565cb09ee34107813bba1c7cd591fa6cfc8d", size = 276515, upload-time = "2026-01-11T11:22:30.327Z" },
{ url = "https://files.pythonhosted.org/packages/4b/ff/ae53619499f5235ee4211e62a8d7982ba9e439a0fb4f2f351a93d67c1dd2/tomli-2.4.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:413540dce94673591859c4c6f794dfeaa845e98bf35d72ed59636f869ef9f86f", size = 273806, upload-time = "2026-01-11T11:22:32.56Z" },
{ url = "https://files.pythonhosted.org/packages/47/71/cbca7787fa68d4d0a9f7072821980b39fbb1b6faeb5f5cf02f4a5559fa28/tomli-2.4.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:0dc56fef0e2c1c470aeac5b6ca8cc7b640bb93e92d9803ddaf9ea03e198f5b0b", size = 281340, upload-time = "2026-01-11T11:22:33.505Z" },
{ url = "https://files.pythonhosted.org/packages/f5/00/d595c120963ad42474cf6ee7771ad0d0e8a49d0f01e29576ee9195d9ecdf/tomli-2.4.0-cp314-cp314t-win32.whl", hash = "sha256:d878f2a6707cc9d53a1be1414bbb419e629c3d6e67f69230217bb663e76b5087", size = 108106, upload-time = "2026-01-11T11:22:34.451Z" },
{ url = "https://files.pythonhosted.org/packages/de/69/9aa0c6a505c2f80e519b43764f8b4ba93b5a0bbd2d9a9de6e2b24271b9a5/tomli-2.4.0-cp314-cp314t-win_amd64.whl", hash = "sha256:2add28aacc7425117ff6364fe9e06a183bb0251b03f986df0e78e974047571fd", size = 120504, upload-time = "2026-01-11T11:22:35.764Z" },
{ url = "https://files.pythonhosted.org/packages/b3/9f/f1668c281c58cfae01482f7114a4b88d345e4c140386241a1a24dcc9e7bc/tomli-2.4.0-cp314-cp314t-win_arm64.whl", hash = "sha256:2b1e3b80e1d5e52e40e9b924ec43d81570f0e7d09d11081b797bc4692765a3d4", size = 99561, upload-time = "2026-01-11T11:22:36.624Z" },
{ url = "https://files.pythonhosted.org/packages/23/d1/136eb2cb77520a31e1f64cbae9d33ec6df0d78bdf4160398e86eec8a8754/tomli-2.4.0-py3-none-any.whl", hash = "sha256:1f776e7d669ebceb01dee46484485f43a4048746235e683bcdffacdf1fb4785a", size = 14477, upload-time = "2026-01-11T11:22:37.446Z" },
]
[[package]]
name = "vban-cmd"
version = "2.9.1"
source = { editable = "." }
dependencies = [
{ name = "tomli", marker = "python_full_version < '3.11'" },
]
[package.metadata]
requires-dist = [{ name = "tomli", marker = "python_full_version < '3.11'", specifier = ">=2.0.1,<3.0" }]

78
vban_cmd/bus.py
View File

@ -4,7 +4,7 @@ from typing import Union
from .enums import NBS, BusModes from .enums import NBS, BusModes
from .iremote import IRemote from .iremote import IRemote
from .meta import bus_mode_prop, channel_bool_prop, channel_label_prop from .meta import bus_mode_prop, channel_bool_prop, channel_int_prop, channel_label_prop
class Bus(IRemote): class Bus(IRemote):
@ -90,20 +90,9 @@ class BusLevel(IRemote):
def getter(self): def getter(self):
"""Returns a tuple of level values for the channel.""" """Returns a tuple of level values for the channel."""
def fget(i):
return round((((1 << 16) - 1) - i) * -0.01, 1)
if not self._remote.stopped() and self._remote.event.ldirty: if not self._remote.stopped() and self._remote.event.ldirty:
return tuple( return self._remote.cache['bus_level'][self.range[0] : self.range[-1]]
fget(i) return self.public_packets[NBS.zero].levels.bus[self.range[0] : self.range[-1]]
for i in self._remote.cache['bus_level'][self.range[0] : self.range[-1]]
)
return tuple(
fget(i)
for i in self._remote._get_levels(self.public_packets[NBS.zero])[1][
self.range[0] : self.range[-1]
]
)
@property @property
def identifier(self) -> str: def identifier(self) -> str:
@ -128,45 +117,49 @@ class BusLevel(IRemote):
def _make_bus_mode_mixin(): def _make_bus_mode_mixin():
"""Creates a mixin of Bus Modes.""" """Creates a mixin of Bus Modes."""
modestates = { mode_names = [
'normal': [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], 'normal',
'amix': [1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1], 'amix',
'repeat': [0, 2, 2, 0, 0, 2, 2, 0, 0, 2, 2], 'repeat',
'bmix': [1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3], 'bmix',
'composite': [0, 0, 0, 4, 4, 4, 4, 0, 0, 0, 0], 'composite',
'tvmix': [1, 0, 1, 4, 5, 4, 5, 0, 1, 0, 1], 'tvmix',
'upmix21': [0, 2, 2, 4, 4, 6, 6, 0, 0, 2, 2], 'upmix21',
'upmix41': [1, 2, 3, 4, 5, 6, 7, 0, 1, 2, 3], 'upmix41',
'upmix61': [0, 0, 0, 0, 0, 0, 0, 8, 8, 8, 8], 'upmix61',
'centeronly': [1, 0, 1, 0, 1, 0, 1, 8, 9, 8, 9], 'centeronly',
'lfeonly': [0, 2, 2, 0, 0, 2, 2, 8, 8, 10, 10], 'lfeonly',
'rearonly': [1, 2, 3, 0, 1, 2, 3, 8, 9, 10, 11], 'rearonly',
} ]
def identifier(self) -> str: def identifier(self) -> str:
return f'bus[{self.index}].mode' return f'bus[{self.index}].mode'
def get(self): def get(self):
states = [ """Get current bus mode using ChannelState for clean bit extraction."""
( mode_cache_items = [
int.from_bytes( (k, v)
self.public_packets[NBS.zero].busstate[self.index], 'little' for k, v in self._remote.cache.items()
) if k.startswith(f'{self.identifier}.') and v == 1
& val
)
>> 4
for val in self._modes.modevals
] ]
for k, v in modestates.items():
if states == v: if mode_cache_items:
return k latest_cached = mode_cache_items[-1][0]
mode_name = latest_cached.split('.')[-1]
return mode_name
bus_state = self.public_packets[NBS.zero].states.bus[self.index]
# Extract bus mode from bits 4-7 (mask 0xF0, shift right by 4)
mode_value = (bus_state._state & 0x000000F0) >> 4
return mode_names[mode_value] if mode_value < len(mode_names) else 'normal'
return type( return type(
'BusModeMixin', 'BusModeMixin',
(IRemote,), (IRemote,),
{ {
'identifier': property(identifier), 'identifier': property(identifier),
'modestates': modestates,
**{mode.name: bus_mode_prop(mode.name) for mode in BusModes}, **{mode.name: bus_mode_prop(mode.name) for mode in BusModes},
'get': get, 'get': get,
}, },
@ -188,7 +181,8 @@ def bus_factory(phys_bus, remote, i) -> Union[PhysicalBus, VirtualBus]:
'eq': BusEQ.make(remote, i), 'eq': BusEQ.make(remote, i),
'levels': BusLevel(remote, i), 'levels': BusLevel(remote, i),
'mode': BUSMODEMIXIN_cls(remote, i), 'mode': BUSMODEMIXIN_cls(remote, i),
**{param: channel_bool_prop(param) for param in ['mute', 'mono']}, **{param: channel_bool_prop(param) for param in ('mute',)},
**{param: channel_int_prop(param) for param in ('mono',)},
'label': channel_label_prop(), 'label': channel_label_prop(),
}, },
)(remote, i) )(remote, i)

30
vban_cmd/factory.py
View File

@ -11,6 +11,7 @@ from .error import VBANCMDError
from .kinds import KindMapClass from .kinds import KindMapClass
from .kinds import request_kind_map as kindmap from .kinds import request_kind_map as kindmap
from .macrobutton import MacroButton from .macrobutton import MacroButton
from .recorder import Recorder
from .strip import request_strip_obj as strip from .strip import request_strip_obj as strip
from .vban import request_vban_obj as vban from .vban import request_vban_obj as vban
from .vbancmd import VbanCmd from .vbancmd import VbanCmd
@ -26,7 +27,7 @@ class FactoryBuilder:
""" """
BuilderProgress = IntEnum( BuilderProgress = IntEnum(
'BuilderProgress', 'strip bus command macrobutton vban', start=0 'BuilderProgress', 'strip bus command macrobutton vban recorder', start=0
) )
def __init__(self, factory, kind: KindMapClass): def __init__(self, factory, kind: KindMapClass):
@ -38,6 +39,7 @@ class FactoryBuilder:
f'Finished building commands for {self._factory}', f'Finished building commands for {self._factory}',
f'Finished building macrobuttons for {self._factory}', f'Finished building macrobuttons for {self._factory}',
f'Finished building vban in/out streams for {self._factory}', f'Finished building vban in/out streams for {self._factory}',
f'Finished building recorder for {self._factory}',
) )
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
@ -72,24 +74,30 @@ class FactoryBuilder:
self._factory.vban = vban(self._factory) self._factory.vban = vban(self._factory)
return self return self
def make_recorder(self):
self._factory.recorder = Recorder.make(self._factory)
return self
class FactoryBase(VbanCmd): class FactoryBase(VbanCmd):
"""Base class for factories, subclasses VbanCmd.""" """Base class for factories, subclasses VbanCmd."""
def __init__(self, kind_id: str, **kwargs): def __init__(self, kind_id: str, **kwargs):
defaultkwargs = { defaultkwargs = {
'ip': 'localhost', 'host': 'localhost',
'port': 6980, 'port': 6980,
'streamname': 'Command1', 'streamname': 'Command1',
'bps': 256000, 'bps': 256000,
'channel': 0, 'channel': 0,
'ratelimit': 0.01, 'script_ratelimit': 0.05, # 20 commands per second, to avoid overloading Voicemeeter
'timeout': 5, 'timeout': 5, # timeout on socket operations, in seconds
'outbound': False, 'disable_rt_listeners': False,
'sync': False, 'sync': False,
'pdirty': False, 'pdirty': False,
'ldirty': False, 'ldirty': False,
} }
if 'ip' in kwargs:
defaultkwargs['host'] = kwargs.pop('ip') # for backwards compatibility
if 'subs' in kwargs: if 'subs' in kwargs:
defaultkwargs |= kwargs.pop('subs') # for backwards compatibility defaultkwargs |= kwargs.pop('subs') # for backwards compatibility
kwargs = defaultkwargs | kwargs kwargs = defaultkwargs | kwargs
@ -166,7 +174,7 @@ class BananaFactory(FactoryBase):
@property @property
def steps(self) -> Iterable: def steps(self) -> Iterable:
"""steps required to build the interface for a kind""" """steps required to build the interface for a kind"""
return self._steps return self._steps + (self.builder.make_recorder,)
class PotatoFactory(FactoryBase): class PotatoFactory(FactoryBase):
@ -188,7 +196,7 @@ class PotatoFactory(FactoryBase):
@property @property
def steps(self) -> Iterable: def steps(self) -> Iterable:
"""steps required to build the interface for a kind""" """steps required to build the interface for a kind"""
return self._steps return self._steps + (self.builder.make_recorder,)
def vbancmd_factory(kind_id: str, **kwargs) -> VbanCmd: def vbancmd_factory(kind_id: str, **kwargs) -> VbanCmd:
@ -202,7 +210,13 @@ def vbancmd_factory(kind_id: str, **kwargs) -> VbanCmd:
_factory = BasicFactory _factory = BasicFactory
case 'banana': case 'banana':
_factory = BananaFactory _factory = BananaFactory
case 'potato': case 'potato' | 'matrix':
# matrix is a special kind where:
# - we don't need to scale the interface with the builder (in other words kind is arbitrary).
# - we don't ever need to use real-time listeners, so we disable them to avoid confusion
if kind_id == 'matrix':
kwargs['disable_rt_listeners'] = True
kind_id = 'potato'
_factory = PotatoFactory _factory = PotatoFactory
case _: case _:
raise ValueError(f"Unknown Voicemeeter kind '{kind_id}'") raise ValueError(f"Unknown Voicemeeter kind '{kind_id}'")

12
vban_cmd/iremote.py
View File

@ -1,6 +1,5 @@
import abc import abc
import logging import logging
import time
from dataclasses import dataclass from dataclasses import dataclass
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -123,6 +122,8 @@ class IRemote(abc.ABC):
def apply(self, data): def apply(self, data):
"""Sets all parameters of a dict for the channel.""" """Sets all parameters of a dict for the channel."""
script = ''
def fget(attr, val): def fget(attr, val):
if attr == 'mode': if attr == 'mode':
return (f'mode.{val}', 1) return (f'mode.{val}', 1)
@ -138,14 +139,9 @@ class IRemote(abc.ABC):
val = 1 if val else 0 val = 1 if val else 0
self._remote.cache[self._cmd(attr)] = val self._remote.cache[self._cmd(attr)] = val
self._remote._script += f'{self._cmd(attr)}={val};' script += f'{self._cmd(attr)}={val};'
else: else:
target = getattr(self, attr) target = getattr(self, attr)
target.apply(val) target.apply(val)
self._remote.sendtext(self._remote._script) self._remote.sendtext(script)
return self
def then_wait(self):
self._remote._script = str()
time.sleep(self._remote.DELAY)

98
vban_cmd/meta.py
View File

@ -1,7 +1,7 @@
from functools import partial from functools import partial
from .enums import NBS from .enums import NBS, BusModes
from .util import cache_bool, cache_float, cache_string from .util import cache_bool, cache_float, cache_int, cache_string
def channel_bool_prop(param): def channel_bool_prop(param):
@ -11,17 +11,23 @@ def channel_bool_prop(param):
def fget(self): def fget(self):
cmd = self._cmd(param) cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}') self.logger.debug(f'getter: {cmd}')
return (
not int.from_bytes( states = self.public_packets[NBS.zero].states
getattr( channel_states = (
self.public_packets[NBS.zero], states.strip if 'strip' in type(self).__name__.lower() else states.bus
f'{"strip" if "strip" in type(self).__name__.lower() else "bus"}state',
)[self.index],
'little',
)
& getattr(self._modes, f'_{param.lower()}')
== 0
) )
channel_state = channel_states[self.index]
if param.lower() == 'mute':
return channel_state.mute
elif param.lower() == 'solo':
return channel_state.solo
elif param.lower() == 'mono':
return channel_state.mono
elif param.lower() == 'mc':
return channel_state.mc
else:
return channel_state.get_mode(getattr(self._modes, f'_{param.lower()}'))
def fset(self, val): def fset(self, val):
self.setter(param, 1 if val else 0) self.setter(param, 1 if val else 0)
@ -29,18 +35,46 @@ def channel_bool_prop(param):
return property(fget, fset) return property(fget, fset)
def channel_int_prop(param):
"""meta function for channel integer parameters"""
@partial(cache_int, param=param)
def fget(self):
cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}')
states = self.public_packets[NBS.zero].states
channel_states = (
states.strip if 'strip' in type(self).__name__.lower() else states.bus
)
channel_state = channel_states[self.index]
# Special case: bus mono is an integer (0-2) encoded using bits 2 and 9
if param.lower() == 'mono' and 'bus' in type(self).__name__.lower():
bit_2 = (channel_state._state >> 2) & 1
bit_9 = (channel_state._state >> 9) & 1
return (bit_9 << 1) | bit_2
else:
return channel_state.get_mode_int(getattr(self._modes, f'_{param.lower()}'))
def fset(self, val):
self.setter(param, val)
return property(fget, fset)
def channel_label_prop(): def channel_label_prop():
"""meta function for channel label parameters""" """meta function for channel label parameters"""
@partial(cache_string, param='label') @partial(cache_string, param='label')
def fget(self) -> str: def fget(self) -> str:
return getattr( if 'strip' in type(self).__name__.lower():
self.public_packets[NBS.zero], return self.public_packets[NBS.zero].labels.strip[self.index]
f'{"strip" if "strip" in type(self).__name__.lower() else "bus"}labels', else:
)[self.index] return self.public_packets[NBS.zero].labels.bus[self.index]
def fset(self, val: str): def fset(self, val: str):
self.setter('label', str(val)) self.setter('label', f'"{val}"')
return property(fget, fset) return property(fget, fset)
@ -52,13 +86,10 @@ def strip_output_prop(param):
def fget(self): def fget(self):
cmd = self._cmd(param) cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}') self.logger.debug(f'getter: {cmd}')
return (
not int.from_bytes( strip_state = self.public_packets[NBS.zero].states.strip[self.index]
self.public_packets[NBS.zero].stripstate[self.index], 'little'
) return strip_state.get_mode(getattr(self._modes, f'_bus{param.lower()}'))
& getattr(self._modes, f'_bus{param.lower()}')
== 0
)
def fset(self, val): def fset(self, val):
self.setter(param, 1 if val else 0) self.setter(param, 1 if val else 0)
@ -73,16 +104,15 @@ def bus_mode_prop(param):
def fget(self): def fget(self):
cmd = self._cmd(param) cmd = self._cmd(param)
self.logger.debug(f'getter: {cmd}') self.logger.debug(f'getter: {cmd}')
return [
( bus_state = self.public_packets[NBS.zero].states.bus[self.index]
int.from_bytes(
self.public_packets[NBS.zero].busstate[self.index], 'little' # Extract current bus mode from bits 4-7
) current_mode = (bus_state._state & 0x000000F0) >> 4
& val
) expected_mode = getattr(BusModes, param.lower())
>> 4
for val in self._modes.modevals return current_mode == expected_mode
] == self.modestates[param]
def fset(self, val): def fset(self, val):
self.setter(param, 1 if val else 0) self.setter(param, 1 if val else 0)

356
vban_cmd/packet/headers.py Normal file
View File

@ -0,0 +1,356 @@
from dataclasses import dataclass
from vban_cmd.enums import NBS
from vban_cmd.kinds import KindMapClass
VBAN_PROTOCOL_TXT = 0x40
VBAN_PROTOCOL_SERVICE = 0x60
VBAN_SERVICE_RTPACKETREGISTER = 32
VBAN_SERVICE_RTPACKET = 33
VBAN_SERVICE_PING = 0
VBAN_SERVICE_PONG = 0 # PONG uses same service type as PING
VBAN_SERVICE_MASK = 0xE0
VBAN_PROTOCOL_MASK = 0xE0
VBAN_SERVICE_REQUESTREPLY = 0x02
VBAN_SERVICE_FNCT_REPLY = 0x02
PINGPONG_PACKET_SIZE = 704 # Size of the PING/PONG header + payload in bytes
MAX_PACKET_SIZE = 1436
HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16
@dataclass
class VbanPacket:
"""Represents the header of an incoming VBAN data packet"""
nbs: NBS
_kind: KindMapClass
_voicemeeterType: bytes
_reserved: bytes
_buffersize: bytes
_voicemeeterVersion: bytes
_optionBits: bytes
_samplerate: bytes
@property
def voicemeetertype(self) -> str:
"""returns voicemeeter type as a string"""
return ['', 'basic', 'banana', 'potato'][
int.from_bytes(self._voicemeeterType, 'little')
]
@property
def voicemeeterversion(self) -> tuple:
"""returns voicemeeter version as a tuple"""
return tuple(self._voicemeeterVersion[i] for i in range(3, -1, -1))
@property
def samplerate(self) -> int:
"""returns samplerate as an int"""
return int.from_bytes(self._samplerate, 'little')
@dataclass
class VbanSubscribeHeader:
"""Represents the header of a subscription packet"""
nbs: NBS = NBS.zero
name: str = 'Register-RTP'
timeout: int = 15
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def format_sr(self) -> bytes:
return VBAN_PROTOCOL_SERVICE.to_bytes(1, 'little')
@property
def format_nbs(self) -> bytes:
return (self.nbs.value & 0xFF).to_bytes(1, 'little')
@property
def format_nbc(self) -> bytes:
return VBAN_SERVICE_RTPACKETREGISTER.to_bytes(1, 'little')
@property
def format_bit(self) -> bytes:
return (self.timeout & 0xFF).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(nbs=nbs)
data = bytearray()
data.extend(header.vban)
data.extend(header.format_sr)
data.extend(header.format_nbs)
data.extend(header.format_nbc)
data.extend(header.format_bit)
data.extend(header.streamname)
data.extend(framecounter.to_bytes(4, 'little'))
return bytes(data)
def _parse_vban_service_header(data: bytes) -> dict:
"""Common parsing and validation for VBAN service protocol headers."""
if len(data) < HEADER_SIZE:
raise ValueError('Data is too short to be a valid VBAN header')
if data[:4] != b'VBAN':
raise ValueError('Invalid VBAN magic bytes')
format_sr = data[4]
format_nbs = data[5]
format_nbc = data[6]
format_bit = data[7]
# Verify this is a service protocol packet
protocol = format_sr & VBAN_PROTOCOL_MASK
if protocol != VBAN_PROTOCOL_SERVICE:
raise ValueError(f'Not a service protocol packet: {protocol:02x}')
# Extract stream name and frame counter
name = data[8:24].rstrip(b'\x00').decode('utf-8', errors='ignore')
framecounter = int.from_bytes(data[24:28], 'little')
return {
'format_sr': format_sr,
'format_nbs': format_nbs,
'format_nbc': format_nbc,
'format_bit': format_bit,
'name': name,
'framecounter': framecounter,
}
@dataclass
class VbanResponseHeader:
"""Represents the header of a response packet"""
name: str = 'Voicemeeter-RTP'
format_sr: int = VBAN_PROTOCOL_SERVICE
format_nbs: int = 0
format_nbc: int = VBAN_SERVICE_RTPACKET
format_bit: int = 0
framecounter: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod
def from_bytes(cls, data: bytes):
"""Parse a VbanResponseHeader from bytes."""
parsed = _parse_vban_service_header(data)
# Validate this is an RTPacket response
if parsed['format_nbc'] != VBAN_SERVICE_RTPACKET:
raise ValueError(
f'Not an RTPacket response packet: {parsed["format_nbc"]:02x}'
)
return cls(**parsed)
@dataclass
class VbanMatrixResponseHeader:
"""Represents the header of a matrix response packet"""
name: str = 'Request Reply'
format_sr: int = VBAN_PROTOCOL_SERVICE
format_nbs: int = VBAN_SERVICE_FNCT_REPLY
format_nbc: int = VBAN_SERVICE_REQUESTREPLY
format_bit: int = 0
framecounter: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii')[:16].ljust(16, b'\x00')
@classmethod
def from_bytes(cls, data: bytes):
"""Parse a matrix response packet from bytes."""
parsed = _parse_vban_service_header(data)
# Validate this is a service reply packet
if parsed['format_nbs'] != VBAN_SERVICE_FNCT_REPLY:
raise ValueError(f'Not a service reply packet: {parsed["format_nbs"]:02x}')
return cls(**parsed)
@classmethod
def extract_payload(cls, data: bytes) -> str:
"""Extract the text payload from a matrix response packet."""
if len(data) <= HEADER_SIZE:
return ''
payload_bytes = data[HEADER_SIZE:]
return payload_bytes.rstrip(b'\x00').decode('utf-8', errors='ignore')
@classmethod
def parse_response(cls, data: bytes) -> tuple['VbanMatrixResponseHeader', str]:
"""Parse a complete matrix response packet returning header and payload."""
header = cls.from_bytes(data)
payload = cls.extract_payload(data)
return header, payload
@dataclass
class VbanPingHeader:
"""Represents the header of a PING packet"""
name: str = 'PING0'
format_sr: int = VBAN_PROTOCOL_SERVICE
format_nbs: int = 0
format_nbc: int = VBAN_SERVICE_PING
format_bit: int = 0
framecounter: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii')[:16].ljust(16, b'\x00')
@classmethod
def to_bytes(cls, framecounter: int = 0) -> bytes:
"""Creates the PING header bytes only."""
header = cls(framecounter=framecounter)
data = bytearray()
data.extend(header.vban)
data.extend(header.format_sr.to_bytes(1, 'little'))
data.extend(header.format_nbs.to_bytes(1, 'little'))
data.extend(header.format_nbc.to_bytes(1, 'little'))
data.extend(header.format_bit.to_bytes(1, 'little'))
data.extend(header.streamname)
data.extend(header.framecounter.to_bytes(4, 'little'))
return bytes(data)
@dataclass
class VbanPongHeader:
"""Represents the header of a PONG response packet"""
name: str = 'PING0'
format_sr: int = VBAN_PROTOCOL_SERVICE
format_nbs: int = 0
format_nbc: int = VBAN_SERVICE_PONG
format_bit: int = 0
framecounter: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii')[:16].ljust(16, b'\x00')
@classmethod
def from_bytes(cls, data: bytes):
"""Parse a PONG response packet from bytes."""
parsed = _parse_vban_service_header(data)
# PONG responses use the same service type as PING (0x00)
# and are identified by having payload data
if parsed['format_nbc'] != VBAN_SERVICE_PONG:
raise ValueError(f'Not a PONG response packet: {parsed["format_nbc"]:02x}')
return cls(**parsed)
@classmethod
def is_pong_response(cls, data: bytes) -> bool:
"""Check if packet is a PONG response by analyzing the actual response format."""
try:
parsed = _parse_vban_service_header(data)
# Validate this is a service protocol packet with PING/PONG service type
if parsed['format_nbc'] != VBAN_SERVICE_PONG:
return False
if parsed['name'] not in ['PING0', 'VBAN Service']:
return False
# PONG should have payload data (same size as PING)
return len(data) >= PINGPONG_PACKET_SIZE
except (ValueError, Exception):
return False
@dataclass
class VbanRequestHeader:
"""Represents the header of a request packet"""
name: str
bps_index: int
channel: int
framecounter: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def sr(self) -> bytes:
return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
@property
def nbs(self) -> bytes:
return (0).to_bytes(1, 'little')
@property
def nbc(self) -> bytes:
return (self.channel).to_bytes(1, 'little')
@property
def bit(self) -> bytes:
return (0x10).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
return self.name.encode()[:16].ljust(16, b'\x00')
@classmethod
def to_bytes(
cls, name: str, bps_index: int, channel: int, framecounter: int
) -> bytes:
header = cls(
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
)
data = bytearray()
data.extend(header.vban)
data.extend(header.sr)
data.extend(header.nbs)
data.extend(header.nbc)
data.extend(header.bit)
data.extend(header.streamname)
data.extend(header.framecounter.to_bytes(4, 'little'))
return bytes(data)
@classmethod
def encode_with_payload(
cls, name: str, bps_index: int, channel: int, framecounter: int, payload: str
) -> bytes:
"""Creates the complete packet with header and payload."""
return cls.to_bytes(name, bps_index, channel, framecounter) + payload.encode()

288
vban_cmd/packet/nbs0.py Normal file
View File

@ -0,0 +1,288 @@
from dataclasses import dataclass
from typing import NamedTuple
from vban_cmd.enums import NBS
from vban_cmd.kinds import KindMapClass
from vban_cmd.util import comp
from .headers import VbanPacket
class Levels(NamedTuple):
strip: tuple[float, ...]
bus: tuple[float, ...]
class ChannelState:
"""Represents the processed state of a single strip or bus channel"""
def __init__(self, state_bytes: bytes):
# Convert 4-byte state to integer once for efficient lookups
self._state = int.from_bytes(state_bytes, 'little')
def get_mode(self, mode_value: int) -> bool:
"""Get boolean state for a specific mode"""
return (self._state & mode_value) != 0
def get_mode_int(self, mode_value: int) -> int:
"""Get integer state for a specific mode"""
return self._state & mode_value
# Common boolean modes
@property
def mute(self) -> bool:
return (self._state & 0x00000001) != 0
@property
def solo(self) -> bool:
return (self._state & 0x00000002) != 0
@property
def mono(self) -> bool:
return (self._state & 0x00000004) != 0
@property
def mc(self) -> bool:
return (self._state & 0x00000008) != 0
# EQ modes
@property
def eq_on(self) -> bool:
return (self._state & 0x00000100) != 0
@property
def eq_ab(self) -> bool:
return (self._state & 0x00000800) != 0
# Bus assignments (strip to bus routing)
@property
def busa1(self) -> bool:
return (self._state & 0x00001000) != 0
@property
def busa2(self) -> bool:
return (self._state & 0x00002000) != 0
@property
def busa3(self) -> bool:
return (self._state & 0x00004000) != 0
@property
def busa4(self) -> bool:
return (self._state & 0x00008000) != 0
@property
def busb1(self) -> bool:
return (self._state & 0x00010000) != 0
@property
def busb2(self) -> bool:
return (self._state & 0x00020000) != 0
@property
def busb3(self) -> bool:
return (self._state & 0x00040000) != 0
class States(NamedTuple):
strip: tuple[ChannelState, ...]
bus: tuple[ChannelState, ...]
class Labels(NamedTuple):
strip: tuple[str, ...]
bus: tuple[str, ...]
@dataclass
class VbanPacketNBS0(VbanPacket):
"""Represents the body of a VBAN data packet with ident:0"""
_inputLeveldB100: bytes
_outputLeveldB100: bytes
_TransportBit: bytes
_stripState: bytes
_busState: bytes
_stripGaindB100Layer1: bytes
_stripGaindB100Layer2: bytes
_stripGaindB100Layer3: bytes
_stripGaindB100Layer4: bytes
_stripGaindB100Layer5: bytes
_stripGaindB100Layer6: bytes
_stripGaindB100Layer7: bytes
_stripGaindB100Layer8: bytes
_busGaindB100: bytes
_stripLabelUTF8c60: bytes
_busLabelUTF8c60: bytes
@classmethod
def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes):
return cls(
nbs=nbs,
_kind=kind,
_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 pdirty(self, other) -> bool:
"""True iff any defined parameter has changed"""
self_gains = (
self._stripGaindB100Layer1
+ self._stripGaindB100Layer2
+ self._stripGaindB100Layer3
+ self._stripGaindB100Layer4
+ self._stripGaindB100Layer5
+ self._stripGaindB100Layer6
+ self._stripGaindB100Layer7
+ self._stripGaindB100Layer8
)
other_gains = (
other._stripGaindB100Layer1
+ other._stripGaindB100Layer2
+ other._stripGaindB100Layer3
+ other._stripGaindB100Layer4
+ other._stripGaindB100Layer5
+ other._stripGaindB100Layer6
+ other._stripGaindB100Layer7
+ other._stripGaindB100Layer8
)
return (
self._stripState != other._stripState
or self._busState != other._busState
or self_gains != other_gains
or self._busGaindB100 != other._busGaindB100
or self._stripLabelUTF8c60 != other._stripLabelUTF8c60
or self._busLabelUTF8c60 != other._busLabelUTF8c60
)
def ldirty(self, strip_cache, bus_cache) -> bool:
"""True iff any level has changed, ignoring changes when levels are very quiet"""
self._strip_comp, self._bus_comp = (
tuple(not val for val in comp(strip_cache, self.strip_levels)),
tuple(not val for val in comp(bus_cache, self.bus_levels)),
)
return any(self._strip_comp) or any(self._bus_comp)
@property
def strip_levels(self) -> tuple[float, ...]:
"""Returns strip levels in dB"""
return tuple(
round(
int.from_bytes(self._inputLeveldB100[i : i + 2], 'little', signed=True)
* 0.01,
1,
)
for i in range(0, len(self._inputLeveldB100), 2)
)[: self._kind.num_strip_levels]
@property
def bus_levels(self) -> tuple[float, ...]:
"""Returns bus levels in dB"""
return tuple(
round(
int.from_bytes(self._outputLeveldB100[i : i + 2], 'little', signed=True)
* 0.01,
1,
)
for i in range(0, len(self._outputLeveldB100), 2)
)[: self._kind.num_bus_levels]
@property
def levels(self) -> Levels:
"""Returns strip and bus levels as a namedtuple"""
return Levels(strip=self.strip_levels, bus=self.bus_levels)
@property
def states(self) -> States:
"""returns States object with processed strip and bus channel states"""
return States(
strip=tuple(
ChannelState(self._stripState[i : i + 4]) for i in range(0, 32, 4)
),
bus=tuple(ChannelState(self._busState[i : i + 4]) for i in range(0, 32, 4)),
)
@property
def gainlayers(self) -> tuple:
"""returns tuple of all strip gain layers as tuples"""
return tuple(
tuple(
round(
int.from_bytes(
getattr(self, f'_stripGaindB100Layer{layer}')[i : i + 2],
'little',
signed=True,
)
* 0.01,
2,
)
for i in range(0, 16, 2)
)
for layer in range(1, 9)
)
@property
def busgain(self) -> tuple:
"""returns tuple of bus gains"""
return tuple(
round(
int.from_bytes(self._busGaindB100[i : i + 2], 'little', signed=True)
* 0.01,
2,
)
for i in range(0, 16, 2)
)
@property
def labels(self) -> Labels:
"""returns Labels namedtuple of strip and bus labels"""
def _extract_labels_from_bytes(label_bytes: bytes) -> tuple[str, ...]:
"""Extract null-terminated UTF-8 labels from 60-byte chunks"""
labels = []
for i in range(0, len(label_bytes), 60):
chunk = label_bytes[i : i + 60]
null_pos = chunk.find(b'\x00')
if null_pos == -1:
try:
label = chunk.decode('utf-8', errors='replace').rstrip('\x00')
except UnicodeDecodeError:
label = ''
else:
try:
label = (
chunk[:null_pos].decode('utf-8', errors='replace')
if null_pos > 0
else ''
)
except UnicodeDecodeError:
label = ''
labels.append(label)
return tuple(labels)
return Labels(
strip=_extract_labels_from_bytes(self._stripLabelUTF8c60),
bus=_extract_labels_from_bytes(self._busLabelUTF8c60),
)

350
vban_cmd/packet.py → vban_cmd/packet/nbs1.py
View File

@ -2,222 +2,14 @@ import struct
from dataclasses import dataclass from dataclasses import dataclass
from typing import NamedTuple from typing import NamedTuple
from .enums import NBS from vban_cmd.enums import NBS
from .kinds import KindMapClass from vban_cmd.kinds import KindMapClass
from .util import comp
VBAN_PROTOCOL_TXT = 0x40 from .headers import VbanPacket
VBAN_PROTOCOL_SERVICE = 0x60
VBAN_SERVICE_RTPACKETREGISTER = 32
VBAN_SERVICE_RTPACKET = 33
VBAN_SERVICE_MASK = 0xE0
MAX_PACKET_SIZE = 1436
HEADER_SIZE = 4 + 1 + 1 + 1 + 1 + 16
VMPARAMSTRIP_SIZE = 174 VMPARAMSTRIP_SIZE = 174
@dataclass
class VbanRtPacket:
"""Represents the body of a VBAN RT data packet"""
nbs: NBS
_kind: KindMapClass
_voicemeeterType: bytes
_reserved: bytes
_buffersize: bytes
_voicemeeterVersion: bytes
_optionBits: bytes
_samplerate: bytes
@dataclass
class VbanRtPacketNBS0(VbanRtPacket):
"""Represents the body of a VBAN RT data packet with NBS 0"""
_inputLeveldB100: bytes
_outputLeveldB100: bytes
_TransportBit: bytes
_stripState: bytes
_busState: bytes
_stripGaindB100Layer1: bytes
_stripGaindB100Layer2: bytes
_stripGaindB100Layer3: bytes
_stripGaindB100Layer4: bytes
_stripGaindB100Layer5: bytes
_stripGaindB100Layer6: bytes
_stripGaindB100Layer7: bytes
_stripGaindB100Layer8: bytes
_busGaindB100: bytes
_stripLabelUTF8c60: bytes
_busLabelUTF8c60: bytes
@classmethod
def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes):
return cls(
nbs=nbs,
_kind=kind,
_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 _generate_levels(self, levelarray) -> tuple:
return tuple(
int.from_bytes(levelarray[i : i + 2], 'little')
for i in range(0, len(levelarray), 2)
)
@property
def strip_levels(self):
return self._generate_levels(self._inputLeveldB100)
@property
def bus_levels(self):
return self._generate_levels(self._outputLeveldB100)
def pdirty(self, other) -> bool:
"""True iff any defined parameter has changed"""
return not (
self._stripState == other._stripState
and self._busState == other._busState
and self._stripGaindB100Layer1 == other._stripGaindB100Layer1
and self._stripGaindB100Layer2 == other._stripGaindB100Layer2
and self._stripGaindB100Layer3 == other._stripGaindB100Layer3
and self._stripGaindB100Layer4 == other._stripGaindB100Layer4
and self._stripGaindB100Layer5 == other._stripGaindB100Layer5
and self._stripGaindB100Layer6 == other._stripGaindB100Layer6
and self._stripGaindB100Layer7 == other._stripGaindB100Layer7
and self._stripGaindB100Layer8 == other._stripGaindB100Layer8
and self._busGaindB100 == other._busGaindB100
and self._stripLabelUTF8c60 == other._stripLabelUTF8c60
and self._busLabelUTF8c60 == other._busLabelUTF8c60
)
def ldirty(self, strip_cache, bus_cache) -> bool:
self._strip_comp, self._bus_comp = (
tuple(not val for val in comp(strip_cache, self.strip_levels)),
tuple(not val for val in comp(bus_cache, self.bus_levels)),
)
return any(any(li) for li in (self._strip_comp, self._bus_comp))
@property
def voicemeetertype(self) -> str:
"""returns voicemeeter type as a string"""
type_ = ('basic', 'banana', 'potato')
return type_[int.from_bytes(self._voicemeeterType, 'little') - 1]
@property
def voicemeeterversion(self) -> tuple:
"""returns voicemeeter version as a tuple"""
return tuple(
reversed(
tuple(
int.from_bytes(self._voicemeeterVersion[i : i + 1], 'little')
for i in range(4)
)
)
)
@property
def samplerate(self) -> int:
"""returns samplerate as an int"""
return int.from_bytes(self._samplerate, 'little')
@property
def inputlevels(self) -> tuple:
"""returns the entire level array across all inputs for a kind"""
return self.strip_levels[0 : self._kind.num_strip_levels]
@property
def outputlevels(self) -> tuple:
"""returns the entire level array across all outputs for a kind"""
return self.bus_levels[0 : self._kind.num_bus_levels]
@property
def stripstate(self) -> tuple:
"""returns tuple of strip states accessable through bit modes"""
return tuple(self._stripState[i : i + 4] for i in range(0, 32, 4))
@property
def busstate(self) -> tuple:
"""returns tuple of bus states accessable through bit modes"""
return tuple(self._busState[i : i + 4] for i in range(0, 32, 4))
"""
these functions return an array of gainlayers[i] across all strips
ie stripgainlayer1 = [strip[0].gainlayer[0], strip[1].gainlayer[0], strip[2].gainlayer[0]...]
"""
@property
def gainlayers(self) -> tuple:
"""returns tuple of all strip gain layers as tuples"""
return tuple(
tuple(
round(
int.from_bytes(
getattr(self, f'_stripGaindB100Layer{layer}')[i : i + 2],
'little',
signed=True,
)
* 0.01,
2,
)
for i in range(0, 16, 2)
)
for layer in range(1, 9)
)
@property
def busgain(self) -> tuple:
"""returns tuple of bus gains"""
return tuple(
round(
int.from_bytes(self._busGaindB100[i : i + 2], 'little', signed=True)
* 0.01,
2,
)
for i in range(0, 16, 2)
)
@property
def striplabels(self) -> tuple:
"""returns tuple of strip labels"""
return tuple(
self._stripLabelUTF8c60[i : i + 60].decode().split('\x00')[0]
for i in range(0, 480, 60)
)
@property
def buslabels(self) -> tuple:
"""returns tuple of bus labels"""
return tuple(
self._busLabelUTF8c60[i : i + 60].decode().split('\x00')[0]
for i in range(0, 480, 60)
)
class Audibility(NamedTuple): class Audibility(NamedTuple):
knob: float knob: float
comp: float comp: float
@ -535,8 +327,8 @@ class VbanVMParamStrip:
@dataclass @dataclass
class VbanRtPacketNBS1(VbanRtPacket): class VbanPacketNBS1(VbanPacket):
"""Represents the body of a VBAN RT data packet with NBS 1""" """Represents the body of a VBAN data packet with ident:1"""
strips: tuple[VbanVMParamStrip, ...] strips: tuple[VbanVMParamStrip, ...]
@ -563,135 +355,3 @@ class VbanRtPacketNBS1(VbanRtPacket):
for i in range(16) for i in range(16)
), ),
) )
@dataclass
class SubscribeHeader:
"""Represents the header of an RT subscription packet"""
nbs: NBS = NBS.zero
name: str = 'Register-RTP'
timeout: int = 15
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def format_sr(self) -> bytes:
return VBAN_PROTOCOL_SERVICE.to_bytes(1, 'little')
@property
def format_nbs(self) -> bytes:
return (self.nbs.value & 0xFF).to_bytes(1, 'little')
@property
def format_nbc(self) -> bytes:
return VBAN_SERVICE_RTPACKETREGISTER.to_bytes(1, 'little')
@property
def format_bit(self) -> bytes:
return (self.timeout & 0xFF).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
header = cls(nbs=nbs)
data = bytearray()
data.extend(header.vban)
data.extend(header.format_sr)
data.extend(header.format_nbs)
data.extend(header.format_nbc)
data.extend(header.format_bit)
data.extend(header.streamname)
data.extend(framecounter.to_bytes(4, 'little'))
return bytes(data)
@dataclass
class VbanRtPacketHeader:
"""Represents the header of an RT response packet"""
name: str = 'Voicemeeter-RTP'
format_sr: int = VBAN_PROTOCOL_SERVICE
format_nbs: int = 0
format_nbc: int = VBAN_SERVICE_RTPACKET
format_bit: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def streamname(self) -> bytes:
return self.name.encode('ascii') + bytes(16 - len(self.name))
@classmethod
def from_bytes(cls, data: bytes):
if len(data) < HEADER_SIZE:
raise ValueError('Data is too short to be a valid VbanRTPPacketHeader')
name = data[8:24].rstrip(b'\x00').decode('utf-8')
return cls(
name=name,
format_sr=data[4] & VBAN_SERVICE_MASK,
format_nbs=data[5],
format_nbc=data[6],
format_bit=data[7],
)
@dataclass
class RequestHeader:
"""Represents the header of an RT request packet"""
name: str
bps_index: int
channel: int
framecounter: int = 0
@property
def vban(self) -> bytes:
return b'VBAN'
@property
def sr(self) -> bytes:
return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
@property
def nbs(self) -> bytes:
return (0).to_bytes(1, 'little')
@property
def nbc(self) -> bytes:
return (self.channel).to_bytes(1, 'little')
@property
def bit(self) -> bytes:
return (0x10).to_bytes(1, 'little')
@property
def streamname(self) -> bytes:
return self.name.encode() + bytes(16 - len(self.name))
@classmethod
def to_bytes(
cls, name: str, bps_index: int, channel: int, framecounter: int
) -> bytes:
header = cls(
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
)
data = bytearray()
data.extend(header.vban)
data.extend(header.sr)
data.extend(header.nbs)
data.extend(header.nbc)
data.extend(header.bit)
data.extend(header.streamname)
data.extend(header.framecounter.to_bytes(4, 'little'))
return bytes(data)

124
vban_cmd/packet/ping0.py Normal file
View File

@ -0,0 +1,124 @@
from dataclasses import dataclass
from enum import Enum
from .headers import VbanPingHeader
# VBAN PING bitType constants
VBANPING_TYPE_RECEPTOR = 0x00000001 # Simple receptor
VBANPING_TYPE_TRANSMITTER = 0x00000002 # Simple Transmitter
VBANPING_TYPE_RECEPTORSPOT = 0x00000004 # SPOT receptor
VBANPING_TYPE_TRANSMITTERSPOT = 0x00000008 # SPOT transmitter
VBANPING_TYPE_VIRTUALDEVICE = 0x00000010 # Virtual Device
VBANPING_TYPE_VIRTUALMIXER = 0x00000020 # Virtual Mixer
VBANPING_TYPE_MATRIX = 0x00000040 # MATRIX
VBANPING_TYPE_DAW = 0x00000080 # Workstation
VBANPING_TYPE_SERVER = 0x01000000 # VBAN SERVER
# VBAN PING bitfeature constants
VBANPING_FEATURE_AUDIO = 0x00000001
VBANPING_FEATURE_AOIP = 0x00000002
VBANPING_FEATURE_VOIP = 0x00000004
VBANPING_FEATURE_SERIAL = 0x00000100
VBANPING_FEATURE_MIDI = 0x00000300
VBANPING_FEATURE_FRAME = 0x00001000
VBANPING_FEATURE_TXT = 0x00010000
class VbanServerType(Enum):
"""VBAN server types detected from PONG responses"""
UNKNOWN = 0
VOICEMEETER = VBANPING_TYPE_VIRTUALMIXER
MATRIX = VBANPING_TYPE_MATRIX
@dataclass
class VbanPing0Payload:
"""Represents the VBAN PING0 payload structure as defined in the VBAN protocol documentation."""
def __init__(self):
self.bit_type = VBANPING_TYPE_RECEPTOR
self.bit_feature = VBANPING_FEATURE_TXT
self.bit_feature_ex = 0x00000000
self.preferred_rate = 48000
self.min_rate = 8000
self.max_rate = 192000
self.color_rgb = 0x00FF0000
self.version = b'\x01\x02\x03\x04'
self.gps_position = b'\x00' * 8
self.user_position = b'\x00' * 8
self.lang_code = b'EN\x00\x00\x00\x00\x00\x00'
self.reserved = b'\x00' * 8
self.reserved_ex = b'\x00' * 64
self.distant_ip = b'\x00' * 32
self.distant_port = 0
self.distant_reserved = 0
self.device_name = b'VBAN-CMD-Python\x00'.ljust(64, b'\x00')
self.manufacturer_name = b'Python-VBAN\x00'.ljust(64, b'\x00')
self.application_name = b'vban-cmd\x00'.ljust(64, b'\x00')
self.host_name = b'localhost\x00'.ljust(64, b'\x00')
self.user_name = b'Python User\x00'.ljust(128, b'\x00')
self.user_comment = b'VBAN CMD Python Client\x00'.ljust(128, b'\x00')
@classmethod
def to_bytes(cls) -> bytes:
"""Convert payload to bytes"""
payload = cls()
data = bytearray()
data.extend(payload.bit_type.to_bytes(4, 'little'))
data.extend(payload.bit_feature.to_bytes(4, 'little'))
data.extend(payload.bit_feature_ex.to_bytes(4, 'little'))
data.extend(payload.preferred_rate.to_bytes(4, 'little'))
data.extend(payload.min_rate.to_bytes(4, 'little'))
data.extend(payload.max_rate.to_bytes(4, 'little'))
data.extend(payload.color_rgb.to_bytes(4, 'little'))
data.extend(payload.version)
data.extend(payload.gps_position)
data.extend(payload.user_position)
data.extend(payload.lang_code)
data.extend(payload.reserved)
data.extend(payload.reserved_ex)
data.extend(payload.distant_ip)
data.extend(payload.distant_port.to_bytes(2, 'little'))
data.extend(payload.distant_reserved.to_bytes(2, 'little'))
data.extend(payload.device_name)
data.extend(payload.manufacturer_name)
data.extend(payload.application_name)
data.extend(payload.host_name)
data.extend(payload.user_name)
data.extend(payload.user_comment)
return bytes(data)
@classmethod
def create_packet(cls, framecounter: int) -> bytes:
"""Creates a complete PING packet with header and payload."""
data = bytearray()
data.extend(VbanPingHeader.to_bytes(framecounter))
data.extend(cls.to_bytes())
return bytes(data)
@staticmethod
def detect_server_type(pong_data: bytes) -> VbanServerType:
"""Detect server type from PONG response packet.
Args:
pong_data: Raw bytes from PONG response packet
Returns:
VbanServerType enum indicating the detected server type
"""
try:
if len(pong_data) >= 32:
frame_counter_bytes = pong_data[28:32]
frame_counter = int.from_bytes(frame_counter_bytes, 'little')
if frame_counter == VbanServerType.MATRIX.value:
return VbanServerType.MATRIX
elif frame_counter == VbanServerType.VOICEMEETER.value:
return VbanServerType.VOICEMEETER
return VbanServerType.UNKNOWN
except Exception:
return VbanServerType.UNKNOWN

138
vban_cmd/recorder.py Normal file
View File

@ -0,0 +1,138 @@
import os
import re
from .error import VBANCMDError
from .iremote import IRemote
from .meta import action_fn
class Recorder(IRemote):
"""
Implements the common interface
Defines concrete implementation for recorder
"""
@classmethod
def make(cls, remote):
"""
Factory function for recorder class.
Returns a Recorder class of a kind.
"""
Recorder_cls = type(
f'Recorder{remote.kind}',
(cls,),
{
**{
param: action_fn(param)
for param in [
'play',
'stop',
'pause',
'replay',
'record',
'ff',
'rew',
]
},
},
)
return Recorder_cls(remote)
def __str__(self):
return f'{type(self).__name__}'
@property
def identifier(self) -> str:
return 'recorder'
@property
def samplerate(self) -> int:
return
@samplerate.setter
def samplerate(self, val: int):
opts = (22050, 24000, 32000, 44100, 48000, 88200, 96000, 176400, 192000)
if val not in opts:
self.logger.warning(f'samplerate got: {val} but expected a value in {opts}')
self.setter('samplerate', val)
@property
def bitresolution(self) -> int:
return
@bitresolution.setter
def bitresolution(self, val: int):
opts = (8, 16, 24, 32)
if val not in opts:
self.logger.warning(
f'bitresolution got: {val} but expected a value in {opts}'
)
self.setter('bitresolution', val)
@property
def channel(self) -> int:
return
@channel.setter
def channel(self, val: int):
if not 1 <= val <= 8:
self.logger.warning(f'channel got: {val} but expected a value from 1 to 8')
self.setter('channel', val)
@property
def kbps(self):
return
@kbps.setter
def kbps(self, val: int):
opts = (32, 40, 48, 56, 64, 80, 96, 112, 128, 160, 192, 224, 256, 320)
if val not in opts:
self.logger.warning(f'kbps got: {val} but expected a value in {opts}')
self.setter('kbps', val)
@property
def gain(self) -> float:
return
@gain.setter
def gain(self, val: float):
self.setter('gain', val)
def load(self, file: os.PathLike):
try:
# Convert to string, use forward slashes, and wrap in quotes for spaces
file_path = f'"{os.fspath(file).replace(chr(92), "/")}"'
self.setter('load', file_path)
except UnicodeError:
raise VBANCMDError('File full directory must be a raw string')
def goto(self, time_str):
def get_sec():
"""Get seconds from time string"""
h, m, s = time_str.split(':')
return int(h) * 3600 + int(m) * 60 + int(s)
time_str = str(time_str) # coerce the type
if (
re.match(
r'^(?:[01]\d|2[0123]):(?:[012345]\d):(?:[012345]\d)$',
time_str,
)
is not None
):
self.setter('goto', get_sec())
else:
self.logger.warning(
"goto expects a string that matches the format 'hh:mm:ss'"
)
def filetype(self, val: str):
opts = {'wav': 1, 'aiff': 2, 'bwf': 3, 'mp3': 100}
try:
self.setter('filetype', opts[val.lower()])
except KeyError:
self.logger.warning(
f'filetype got: {val} but expected a value in {list(opts.keys())}'
)

19
vban_cmd/strip.py
View File

@ -540,22 +540,11 @@ class StripLevel(IRemote):
def getter(self): def getter(self):
"""Returns a tuple of level values for the channel.""" """Returns a tuple of level values for the channel."""
def fget(i):
return round((((1 << 16) - 1) - i) * -0.01, 1)
if not self._remote.stopped() and self._remote.event.ldirty: if not self._remote.stopped() and self._remote.event.ldirty:
return tuple( return self._remote.cache['strip_level'][self.range[0] : self.range[-1]]
fget(i) return self.public_packets[NBS.zero].levels.strip[
for i in self._remote.cache['strip_level'][ self.range[0] : self.range[-1]
self.range[0] : self.range[-1] ]
]
)
return tuple(
fget(i)
for i in self._remote._get_levels(self.public_packets[NBS.zero])[0][
self.range[0] : self.range[-1]
]
)
@property @property
def identifier(self) -> str: def identifier(self) -> str:

64
vban_cmd/util.py
View File

@ -1,6 +1,23 @@
import time
from typing import Iterator from typing import Iterator
def ratelimit(func):
"""ratelimit decorator for {VbanCmd}.sendtext, to prevent flooding the network with script requests."""
def wrapper(*args, **kwargs):
self, *rem = args
if self.script_ratelimit > 0:
now = time.time()
elapsed = now - self._last_script_request_time
if elapsed < self.script_ratelimit:
time.sleep(self.script_ratelimit - elapsed)
self._last_script_request_time = time.time()
return func(*args, **kwargs)
return wrapper
def cache_bool(func, param): def cache_bool(func, param):
"""Check cache for a bool prop""" """Check cache for a bool prop"""
@ -15,13 +32,27 @@ def cache_bool(func, param):
return wrapper return wrapper
def cache_int(func, param):
"""Check cache for an int prop"""
def wrapper(*args, **kwargs):
self, *rem = args
if self._cmd(param) in self._remote.cache:
return self._remote.cache.pop(self._cmd(param))
if self._remote.sync:
self._remote.clear_dirty()
return func(*args, **kwargs)
return wrapper
def cache_string(func, param): def cache_string(func, param):
"""Check cache for a string prop""" """Check cache for a string prop"""
def wrapper(*args, **kwargs): def wrapper(*args, **kwargs):
self, *rem = args self, *rem = args
if self._cmd(param) in self._remote.cache: if self._cmd(param) in self._remote.cache:
return self._remote.cache.pop(self._cmd(param)) return self._remote.cache.pop(self._cmd(param)).strip('"')
if self._remote.sync: if self._remote.sync:
self._remote.clear_dirty() self._remote.clear_dirty()
return func(*args, **kwargs) return func(*args, **kwargs)
@ -49,39 +80,20 @@ def depth(d):
return 0 return 0
def script(func):
"""Convert dictionary to script"""
def wrapper(*args):
remote, script = args
if isinstance(script, dict):
params = ''
for key, val in script.items():
obj, m2, *rem = key.split('-')
index = int(m2) if m2.isnumeric() else int(*rem)
params += ';'.join(
f'{obj}{f".{m2}stream" if not m2.isnumeric() else ""}[{index}].{k}={int(v) if isinstance(v, bool) else v}'
for k, v in val.items()
)
params += ';'
script = params
return func(remote, script)
return wrapper
def comp(t0: tuple, t1: tuple) -> Iterator[bool]: def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
""" """
Generator function, accepts two tuples. Generator function, accepts two tuples of dB values.
Evaluates equality of each member in both tuples. Evaluates equality of each member in both tuples.
Only ignores changes when levels are very quiet (below -72 dB).
""" """
for a, b in zip(t0, t1): for a, b in zip(t0, t1):
if ((1 << 16) - 1) - b <= 7200: # If both values are very quiet (below -72dB), ignore small changes
yield a == b if a <= -72.0 and b <= -72.0:
yield a == b # Both quiet, check if they're equal
else: else:
yield True yield a != b # At least one has significant level, detect changes
def deep_merge(dict1, dict2): def deep_merge(dict1, dict2):

200
vban_cmd/vbancmd.py
View File

@ -5,14 +5,19 @@ import threading
import time import time
from pathlib import Path from pathlib import Path
from queue import Queue from queue import Queue
from typing import Iterable, Union from typing import Mapping, Union
from .enums import NBS from .enums import NBS
from .error import VBANCMDError from .error import VBANCMDConnectionError, VBANCMDError
from .event import Event from .event import Event
from .packet import RequestHeader from .packet.headers import (
VbanMatrixResponseHeader,
VbanPongHeader,
VbanRequestHeader,
)
from .packet.ping0 import VbanPing0Payload, VbanServerType
from .subject import Subject from .subject import Subject
from .util import bump_framecounter, deep_merge, script from .util import bump_framecounter, deep_merge, ratelimit
from .worker import Producer, Subscriber, Updater from .worker import Producer, Subscriber, Updater
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -33,21 +38,23 @@ class VbanCmd(abc.ABC):
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
self.event = Event({k: kwargs.pop(k) for k in ('pdirty', 'ldirty')}) self.event = Event({k: kwargs.pop(k) for k in ('pdirty', 'ldirty')})
if not kwargs['ip']: if not kwargs['host']:
kwargs |= self._conn_from_toml() kwargs |= self._conn_from_toml()
for attr, val in kwargs.items(): for attr, val in kwargs.items():
setattr(self, attr, val) setattr(self, attr, val)
self._framecounter = 0 self._framecounter = 0
self._framecounter_lock = threading.Lock()
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.settimeout(self.timeout)
self.subject = self.observer = Subject() self.subject = self.observer = Subject()
self.cache = {} self.cache = {}
self._pdirty = False self._pdirty = False
self._ldirty = False self._ldirty = False
self._script = str()
self.stop_event = None self.stop_event = None
self.producer = None self.producer = None
self._last_script_request_time = 0
@abc.abstractmethod @abc.abstractmethod
def __str__(self): def __str__(self):
@ -86,8 +93,12 @@ class VbanCmd(abc.ABC):
self.logout() self.logout()
def login(self) -> None: def login(self) -> None:
"""Starts the subscriber and updater threads (unless in outbound mode)""" """Sends a PING packet to the VBAN server to verify connectivity and detect server type.
if not self.outbound: If the server is detected as Matrix, RT listeners will be disabled for compatibility.
"""
self._ping()
if not self.disable_rt_listeners:
self.event.info() self.event.info()
self.stop_event = threading.Event() self.stop_event = threading.Event()
@ -102,7 +113,7 @@ class VbanCmd(abc.ABC):
self.producer.start() self.producer.start()
self.logger.info( self.logger.info(
"Successfully logged into VBANCMD {kind} with ip='{ip}', port={port}, streamname='{streamname}'".format( "Successfully logged into VBANCMD {kind} with host='{host}', port={port}, streamname='{streamname}'".format(
**self.__dict__ **self.__dict__
) )
) )
@ -120,39 +131,133 @@ class VbanCmd(abc.ABC):
def stopped(self): def stopped(self):
return self.stop_event is None or self.stop_event.is_set() return self.stop_event is None or self.stop_event.is_set()
def _get_next_framecounter(self) -> int:
"""Thread-safe method to get and increment framecounter."""
with self._framecounter_lock:
current = self._framecounter
self._framecounter = bump_framecounter(self._framecounter)
return current
def _ping(self, timeout: float = None) -> None:
"""Send a PING packet and wait for PONG response to verify connectivity."""
if timeout is None:
timeout = min(self.timeout, 3.0)
ping_packet = VbanPing0Payload.create_packet(self._get_next_framecounter())
original_timeout = self.sock.gettimeout()
self.sock.settimeout(0.5)
try:
self.sock.sendto(ping_packet, (socket.gethostbyname(self.host), self.port))
self.logger.debug(f'PING sent to {self.host}:{self.port}')
start_time = time.time()
response_count = 0
while time.time() - start_time < timeout:
try:
data, addr = self.sock.recvfrom(2048)
response_count += 1
self.logger.debug(
f'Received packet #{response_count} from {addr}: {len(data)} bytes'
)
self.logger.debug(
f'Response header: {data[: min(32, len(data))].hex()}'
)
if VbanPongHeader.is_pong_response(data):
self.logger.debug(
f'PONG received from {addr}, connectivity confirmed'
)
server_type = VbanPing0Payload.detect_server_type(data)
self._handle_server_type(server_type)
return # Exit after successful PONG response
else:
if len(data) >= 8:
if data[:4] == b'VBAN':
protocol = data[4] & 0xE0
nbc = data[6]
self.logger.debug(
f'Non-PONG VBAN packet: protocol=0x{protocol:02x}, nbc=0x{nbc:02x}'
)
else:
self.logger.debug('Non-VBAN packet received')
except socket.timeout:
continue
self.logger.debug(
f'PING timeout after {timeout}s, received {response_count} non-PONG packets'
)
raise VBANCMDConnectionError(
f'PING timeout: No response from {self.host}:{self.port} after {timeout}s'
)
except socket.gaierror as e:
raise VBANCMDConnectionError(
f'Unable to resolve hostname {self.host}'
) from e
except Exception as e:
raise VBANCMDConnectionError(f'PING failed: {e}') from e
finally:
self.sock.settimeout(original_timeout)
def _handle_server_type(self, server_type: VbanServerType) -> None:
"""Handle the detected server type by adjusting settings accordingly."""
match server_type:
case VbanServerType.VOICEMEETER:
self.logger.debug(
'Detected Voicemeeter VBAN server - RT listeners supported'
)
case VbanServerType.MATRIX:
self.logger.info(
'Detected Matrix VBAN server - disabling RT listeners for compatibility'
)
self.disable_rt_listeners = True
case _:
self.logger.debug(
f'Unknown server type ({server_type}) - using default settings'
)
def _send_request(self, payload: str) -> None:
"""Sends a request packet over the network and bumps the framecounter."""
self.sock.sendto(
VbanRequestHeader.encode_with_payload(
name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel,
framecounter=self._get_next_framecounter(),
payload=payload,
),
(socket.gethostbyname(self.host), self.port),
)
def _set_rt(self, cmd: str, val: Union[str, float]): def _set_rt(self, cmd: str, val: Union[str, float]):
"""Sends a string request command over a network.""" """Sends a string request command over a network."""
req_packet = RequestHeader.to_bytes( self._send_request(f'{cmd}={val};')
name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel,
framecounter=self._framecounter,
)
self.sock.sendto(
req_packet + f'{cmd}={val};'.encode(),
(socket.gethostbyname(self.ip), self.port),
)
self._framecounter = bump_framecounter(self._framecounter)
self.cache[cmd] = val self.cache[cmd] = val
@script @ratelimit
def sendtext(self, script): def sendtext(self, script) -> str | None:
"""Sends a multiple parameter string over a network.""" """Sends a multiple parameter string over a network."""
req_packet = RequestHeader.to_bytes( self._send_request(script)
name=self.streamname,
bps_index=self.BPS_OPTS.index(self.bps),
channel=self.channel,
framecounter=self._framecounter,
)
self.sock.sendto(
req_packet + script.encode(),
(socket.gethostbyname(self.ip), self.port),
)
self._framecounter = bump_framecounter(self._framecounter)
self.logger.debug(f'sendtext: {script}') self.logger.debug(f'sendtext: {script}')
time.sleep(self.DELAY)
if self.disable_rt_listeners and script.endswith(('?', '?;')):
try:
data, _ = self.sock.recvfrom(2048)
payload = VbanMatrixResponseHeader.extract_payload(data)
except ValueError as e:
self.logger.warning(f'Error extracting matrix response: {e}')
except TimeoutError as e:
self.logger.exception(f'Timeout waiting for matrix response: {e}')
raise VBANCMDConnectionError(
f'Timeout waiting for response from {self.host}:{self.port}'
) from e
return payload
@property @property
def type(self) -> str: def type(self) -> str:
@ -184,23 +289,8 @@ class VbanCmd(abc.ABC):
while self.pdirty: while self.pdirty:
time.sleep(self.DELAY) time.sleep(self.DELAY)
def _get_levels(self, packet) -> Iterable: def apply(self, data: Mapping):
""" """Set all parameters of a dict"""
returns both level arrays (strip_levels, bus_levels) BEFORE math conversion
strip levels in PREFADER mode.
"""
return (
packet.inputlevels,
packet.outputlevels,
)
def apply(self, data: dict):
"""
Sets all parameters of a dict
minor delay between each recursion
"""
def target(key): def target(key):
match key.split('-'): match key.split('-'):
@ -220,7 +310,9 @@ class VbanCmd(abc.ABC):
raise ValueError(ERR_MSG) raise ValueError(ERR_MSG)
return target[int(index)] return target[int(index)]
[target(key).apply(di).then_wait() for key, di in data.items()] for key, di in data.items():
target(key).apply(di)
time.sleep(self.DELAY)
def apply_config(self, name): def apply_config(self, name):
"""applies a config from memory""" """applies a config from memory"""

90
vban_cmd/worker.py
View File

@ -1,21 +1,17 @@
import logging import logging
import socket
import threading import threading
import time import time
from .enums import NBS from .enums import NBS
from .error import VBANCMDConnectionError from .error import VBANCMDConnectionError
from .packet import ( from .packet.headers import (
HEADER_SIZE, HEADER_SIZE,
VBAN_PROTOCOL_SERVICE, VbanPacket,
VBAN_SERVICE_RTPACKET, VbanResponseHeader,
SubscribeHeader, VbanSubscribeHeader,
VbanRtPacket,
VbanRtPacketHeader,
VbanRtPacketNBS0,
VbanRtPacketNBS1,
) )
from .util import bump_framecounter from .packet.nbs0 import VbanPacketNBS0
from .packet.nbs1 import VbanPacketNBS1
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -28,24 +24,18 @@ class Subscriber(threading.Thread):
self._remote = remote self._remote = remote
self.stop_event = stop_event self.stop_event = stop_event
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
self._framecounter = 0
def run(self): def run(self):
while not self.stopped(): while not self.stopped():
try: for nbs in NBS:
for nbs in NBS: sub_packet = VbanSubscribeHeader().to_bytes(
sub_packet = SubscribeHeader().to_bytes(nbs, self._framecounter) nbs, self._remote._get_next_framecounter()
self._remote.sock.sendto( )
sub_packet, (self._remote.ip, self._remote.port) self._remote.sock.sendto(
) sub_packet, (self._remote.host, self._remote.port)
self._framecounter = bump_framecounter(self._framecounter) )
self.wait_until_stopped(10) self.wait_until_stopped(10)
except socket.gaierror as e:
self.logger.exception(f'{type(e).__name__}: {e}')
raise VBANCMDConnectionError(
f'unable to resolve hostname {self._remote.ip}'
) from e
self.logger.debug(f'terminating {self.name} thread') self.logger.debug(f'terminating {self.name} thread')
def stopped(self): def stopped(self):
@ -68,51 +58,43 @@ class Producer(threading.Thread):
self.queue = queue self.queue = queue
self.stop_event = stop_event self.stop_event = stop_event
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
self._remote.sock.settimeout(self._remote.timeout)
self._remote._public_packets = [None] * (max(NBS) + 1) self._remote._public_packets = [None] * (max(NBS) + 1)
_pp = self._get_rt() _pp = self._get_rt()
self._remote._public_packets[_pp.nbs] = _pp self._remote._public_packets[_pp.nbs] = _pp
( (
self._remote.cache['strip_level'], self._remote.cache['strip_level'],
self._remote.cache['bus_level'], self._remote.cache['bus_level'],
) = self._remote._get_levels(self._remote.public_packets[NBS.zero]) ) = self._remote.public_packets[NBS.zero].levels
def _get_rt(self) -> VbanRtPacket: def _get_rt(self) -> VbanPacket:
"""Attempt to fetch data packet until a valid one found""" """Attempt to fetch data packet until a valid one found"""
while True: while True:
if resp := self._fetch_rt_packet(): try:
return resp data, _ = self._remote.sock.recvfrom(2048)
if len(data) < HEADER_SIZE:
continue
except TimeoutError as e:
self.logger.exception(f'{type(e).__name__}: {e}')
raise VBANCMDConnectionError(
f'timeout waiting for response from {self._remote.host}:{self._remote.port}'
) from e
def _fetch_rt_packet(self) -> VbanRtPacket | None: try:
try: header = VbanResponseHeader.from_bytes(data[:HEADER_SIZE])
data, _ = self._remote.sock.recvfrom(2048) except ValueError as e:
if len(data) < HEADER_SIZE: self.logger.debug(f'Error parsing response packet: {e}')
return continue
response_header = VbanRtPacketHeader.from_bytes(data[:HEADER_SIZE]) match header.format_nbs:
if (
response_header.format_sr != VBAN_PROTOCOL_SERVICE
or response_header.format_nbc != VBAN_SERVICE_RTPACKET
):
return
match response_header.format_nbs:
case NBS.zero: case NBS.zero:
return VbanRtPacketNBS0.from_bytes( return VbanPacketNBS0.from_bytes(
nbs=NBS.zero, kind=self._remote.kind, data=data nbs=NBS.zero, kind=self._remote.kind, data=data
) )
case NBS.one: case NBS.one:
return VbanRtPacketNBS1.from_bytes( return VbanPacketNBS1.from_bytes(
nbs=NBS.one, kind=self._remote.kind, data=data nbs=NBS.one, kind=self._remote.kind, data=data
) )
return None
except TimeoutError as e:
self.logger.exception(f'{type(e).__name__}: {e}')
raise VBANCMDConnectionError(
f'timeout waiting for RtPacket from {self._remote.ip}'
) from e
def stopped(self): def stopped(self):
return self.stop_event.is_set() return self.stop_event.is_set()
@ -140,7 +122,6 @@ class Producer(threading.Thread):
self.queue.put('pdirty') self.queue.put('pdirty')
if self._remote.event.ldirty: if self._remote.event.ldirty:
self.queue.put('ldirty') self.queue.put('ldirty')
time.sleep(self._remote.ratelimit)
self.logger.debug(f'terminating {self.name} thread') self.logger.debug(f'terminating {self.name} thread')
self.queue.put(None) self.queue.put(None)
@ -177,9 +158,6 @@ class Updater(threading.Thread):
( (
self._remote.cache['strip_level'], self._remote.cache['strip_level'],
self._remote.cache['bus_level'], self._remote.cache['bus_level'],
) = ( ) = self._remote.public_packets[NBS.zero].levels
self._remote._public_packets[NBS.zero].inputlevels,
self._remote._public_packets[NBS.zero].outputlevels,
)
self._remote.subject.notify(event) self._remote.subject.notify(event)
self.logger.debug(f'terminating {self.name} thread') self.logger.debug(f'terminating {self.name} thread')