mirror of
https://github.com/onyx-and-iris/vban-cmd-python.git
synced 2026-04-19 21:33:29 +00:00
Compare commits
17 Commits
ed8e281f7f
...
v2.6.0
| Author | SHA1 | Date | |
|---|---|---|---|
| a8ef82166c | |||
| 79f06ecc79 | |||
| b291c3a477 | |||
| c335d35b9f | |||
| 911d2f64a6 | |||
| e58d6c7242 | |||
| 870a95b41e | |||
| 59880bf582 | |||
| cc58d1f081 | |||
| e37dea38b3 | |||
| 7f3b0ac7c9 | |||
| 0512fac710 | |||
| d439da725c | |||
| 45ffed9f63 | |||
| 14f79d1388 | |||
| b45bd38706 | |||
| 312b5c5842 |
53
.github/workflows/publish.yml
vendored
Normal file
53
.github/workflows/publish.yml
vendored
Normal file
@@ -0,0 +1,53 @@
|
|||||||
|
name: Publish to PyPI
|
||||||
|
|
||||||
|
on:
|
||||||
|
release:
|
||||||
|
types: [published]
|
||||||
|
push:
|
||||||
|
tags:
|
||||||
|
- 'v*.*.*'
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
build:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- uses: actions/setup-python@v5
|
||||||
|
with:
|
||||||
|
python-version: '3.11'
|
||||||
|
|
||||||
|
- name: Install Poetry
|
||||||
|
run: |
|
||||||
|
pip install poetry==2.3.1
|
||||||
|
poetry --version
|
||||||
|
|
||||||
|
- name: Build package
|
||||||
|
run: |
|
||||||
|
poetry install --only-root
|
||||||
|
poetry build
|
||||||
|
|
||||||
|
- uses: actions/upload-artifact@v4
|
||||||
|
with:
|
||||||
|
name: dist
|
||||||
|
path: ./dist
|
||||||
|
|
||||||
|
pypi-publish:
|
||||||
|
needs: build
|
||||||
|
name: Upload release to PyPI
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
environment:
|
||||||
|
name: pypi
|
||||||
|
url: https://pypi.org/project/vban-cmd/
|
||||||
|
permissions:
|
||||||
|
id-token: write
|
||||||
|
steps:
|
||||||
|
- uses: actions/download-artifact@v4
|
||||||
|
with:
|
||||||
|
name: dist
|
||||||
|
path: ./dist
|
||||||
|
|
||||||
|
- name: Publish package distributions to PyPI
|
||||||
|
uses: pypa/gh-action-pypi-publish@release/v1
|
||||||
|
with:
|
||||||
|
packages-dir: ./dist
|
||||||
19
.github/workflows/ruff.yml
vendored
Normal file
19
.github/workflows/ruff.yml
vendored
Normal file
@@ -0,0 +1,19 @@
|
|||||||
|
name: Ruff
|
||||||
|
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
branches: [main]
|
||||||
|
|
||||||
|
pull_request:
|
||||||
|
branches: [main]
|
||||||
|
|
||||||
|
workflow_dispatch:
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
ruff:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v6
|
||||||
|
- uses: astral-sh/ruff-action@v3
|
||||||
|
with:
|
||||||
|
args: 'format --check --diff'
|
||||||
@@ -11,6 +11,13 @@ Before any major/minor/patch bump all unit tests will be run to verify they pass
|
|||||||
|
|
||||||
- [x]
|
- [x]
|
||||||
|
|
||||||
|
## [2.6.0] - 2026-02-26
|
||||||
|
|
||||||
|
### Added
|
||||||
|
|
||||||
|
- support for packet with [ident:1](https://github.com/onyx-and-iris/Voicemeeter-SDK/blob/3be2c1c36563afbd6df3da8436406c77d2cc1f10/VoicemeeterRemote.h#L982) in VBAN TEXT subprotocol.
|
||||||
|
- This includes Strip 3D, PEQ, comp, gate, denoiser and pitch parameters.
|
||||||
|
|
||||||
## [2.5.2] - 2025-01-25
|
## [2.5.2] - 2025-01-25
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
|||||||
51
README.md
51
README.md
@@ -171,9 +171,7 @@ example:
|
|||||||
print(vban.strip[4].comp.knob)
|
print(vban.strip[4].comp.knob)
|
||||||
```
|
```
|
||||||
|
|
||||||
Strip Comp properties are defined as write only.
|
Strip Comp `knob` is defined for all versions, all other parameters potato only.
|
||||||
|
|
||||||
`knob` defined for all versions, all other parameters potato only.
|
|
||||||
|
|
||||||
##### Strip.Gate
|
##### Strip.Gate
|
||||||
|
|
||||||
@@ -193,9 +191,7 @@ example:
|
|||||||
vban.strip[2].gate.attack = 300.8
|
vban.strip[2].gate.attack = 300.8
|
||||||
```
|
```
|
||||||
|
|
||||||
Strip Gate properties are defined as write only, potato version only.
|
Strip Gate `knob` is defined for all versions, all other parameters potato only.
|
||||||
|
|
||||||
`knob` defined for all versions, all other parameters potato only.
|
|
||||||
|
|
||||||
##### Strip.Denoiser
|
##### Strip.Denoiser
|
||||||
|
|
||||||
@@ -212,7 +208,32 @@ The following properties are available.
|
|||||||
- `on`: boolean
|
- `on`: boolean
|
||||||
- `ab`: boolean
|
- `ab`: boolean
|
||||||
|
|
||||||
Strip EQ properties are defined as write only, potato version only.
|
example:
|
||||||
|
|
||||||
|
```python
|
||||||
|
vban.strip[0].eq.ab = True
|
||||||
|
```
|
||||||
|
|
||||||
|
##### Strip.EQ.Channel.Cell
|
||||||
|
|
||||||
|
The following properties are available.
|
||||||
|
|
||||||
|
- `on`: boolean
|
||||||
|
- `type`: int, from 0 up to 6
|
||||||
|
- `f`: float, from 20.0 up to 20_000.0
|
||||||
|
- `gain`: float, from -36.0 up to 18.0
|
||||||
|
- `q`: float, from 0.3 up to 100
|
||||||
|
|
||||||
|
example:
|
||||||
|
|
||||||
|
```python
|
||||||
|
vban.strip[0].eq.channel[0].cell[2].on = True
|
||||||
|
vban.strip[1].eq.channel[0].cell[2].f = 5000
|
||||||
|
```
|
||||||
|
|
||||||
|
Strip EQ parameters are defined for PhysicalStrips, potato version only.
|
||||||
|
|
||||||
|
Only channel[0] properties are readable over VBAN.
|
||||||
|
|
||||||
##### Gainlayers
|
##### Gainlayers
|
||||||
|
|
||||||
@@ -400,8 +421,8 @@ You just need to define a key `extends` in the config TOML, that names the confi
|
|||||||
Three example 'extender' configs are included with the repo. You may load them with:
|
Three example 'extender' configs are included with the repo. You may load them with:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
import voicemeeterlib
|
import vban_cmd
|
||||||
with voicemeeterlib.api('banana') as vm:
|
with vban_cmd.api('banana') as vm:
|
||||||
vm.apply_config('extender')
|
vm.apply_config('extender')
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -528,13 +549,15 @@ with vban_cmd.api('banana', **opts) as vban:
|
|||||||
...
|
...
|
||||||
```
|
```
|
||||||
|
|
||||||
## Tests
|
### Run tests
|
||||||
|
|
||||||
First make sure you installed the [development dependencies](https://github.com/onyx-and-iris/vban-cmd-python#installation)
|
Install [poetry](https://python-poetry.org/docs/#installation) and then:
|
||||||
|
|
||||||
Then from tests directory:
|
```powershell
|
||||||
|
poetry poe test-basic
|
||||||
`pytest -v`
|
poetry poe test-banana
|
||||||
|
poetry poe test-potato
|
||||||
|
```
|
||||||
|
|
||||||
## Resources
|
## Resources
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import logging
|
import logging
|
||||||
|
import os
|
||||||
import tkinter as tk
|
import tkinter as tk
|
||||||
from tkinter import ttk
|
from tkinter import ttk
|
||||||
|
|
||||||
@@ -100,7 +101,14 @@ class App(tk.Tk):
|
|||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
with vban_cmd.api('banana', ldirty=True) as vban:
|
KIND_ID = 'banana'
|
||||||
|
conn = {
|
||||||
|
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
|
||||||
|
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
|
||||||
|
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
|
||||||
|
}
|
||||||
|
|
||||||
|
with vban_cmd.api(KIND_ID, ldirty=True, **conn) as vban:
|
||||||
app = App(vban)
|
app = App(vban)
|
||||||
app.mainloop()
|
app.mainloop()
|
||||||
|
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import os
|
||||||
import threading
|
import threading
|
||||||
from logging import config
|
from logging import config
|
||||||
|
|
||||||
@@ -92,8 +93,13 @@ class Observer:
|
|||||||
|
|
||||||
def main():
|
def main():
|
||||||
KIND_ID = 'potato'
|
KIND_ID = 'potato'
|
||||||
|
conn = {
|
||||||
|
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
|
||||||
|
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
|
||||||
|
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
|
||||||
|
}
|
||||||
|
|
||||||
with vban_cmd.api(KIND_ID) as vban:
|
with vban_cmd.api(KIND_ID, **conn) as vban:
|
||||||
stop_event = threading.Event()
|
stop_event = threading.Event()
|
||||||
|
|
||||||
with Observer(vban, stop_event):
|
with Observer(vban, stop_event):
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import logging
|
import logging
|
||||||
|
import os
|
||||||
|
|
||||||
import vban_cmd
|
import vban_cmd
|
||||||
|
|
||||||
@@ -23,8 +24,13 @@ class App:
|
|||||||
|
|
||||||
def main():
|
def main():
|
||||||
KIND_ID = 'banana'
|
KIND_ID = 'banana'
|
||||||
|
conn = {
|
||||||
|
'ip': os.environ.get('VBANCMD_IP', 'localhost'),
|
||||||
|
'port': int(os.environ.get('VBANCMD_PORT', 6980)),
|
||||||
|
'streamname': os.environ.get('VBANCMD_STREAMNAME', 'Command1'),
|
||||||
|
}
|
||||||
|
|
||||||
with vban_cmd.api(KIND_ID, pdirty=True, ldirty=True) as vban:
|
with vban_cmd.api(KIND_ID, pdirty=True, ldirty=True, **conn) as vban:
|
||||||
App(vban)
|
App(vban)
|
||||||
|
|
||||||
while _ := input('Press <Enter> to exit\n'):
|
while _ := input('Press <Enter> to exit\n'):
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "vban-cmd"
|
name = "vban-cmd"
|
||||||
version = "2.5.2"
|
version = "2.6.0"
|
||||||
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" }
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
|
import abc
|
||||||
import time
|
import time
|
||||||
from abc import abstractmethod
|
|
||||||
from typing import Union
|
from typing import Union
|
||||||
|
|
||||||
from .enums import NBS, BusModes
|
from .enums import NBS, BusModes
|
||||||
@@ -14,7 +14,7 @@ class Bus(IRemote):
|
|||||||
Defines concrete implementation for bus
|
Defines concrete implementation for bus
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@abstractmethod
|
@abc.abstractmethod
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,11 @@
|
|||||||
from enum import IntEnum
|
from enum import Enum, IntEnum, unique
|
||||||
|
|
||||||
|
|
||||||
|
@unique
|
||||||
|
class KindId(Enum):
|
||||||
|
BASIC = 1
|
||||||
|
BANANA = 2
|
||||||
|
POTATO = 3
|
||||||
|
|
||||||
|
|
||||||
class NBS(IntEnum):
|
class NBS(IntEnum):
|
||||||
@@ -11,5 +18,3 @@ BusModes = IntEnum(
|
|||||||
'normal amix bmix repeat composite tvmix upmix21 upmix41 upmix61 centeronly lfeonly rearonly',
|
'normal amix bmix repeat composite tvmix upmix21 upmix41 upmix61 centeronly lfeonly rearonly',
|
||||||
start=0,
|
start=0,
|
||||||
)
|
)
|
||||||
|
|
||||||
EQGains = IntEnum('EQGains', 'bass mid treble', start=0)
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
|
import abc
|
||||||
import logging
|
import logging
|
||||||
from abc import abstractmethod
|
|
||||||
from enum import IntEnum
|
from enum import IntEnum
|
||||||
from functools import cached_property
|
from functools import cached_property
|
||||||
from typing import Iterable
|
from typing import Iterable
|
||||||
@@ -115,7 +115,7 @@ class FactoryBase(VbanCmd):
|
|||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@abstractmethod
|
@abc.abstractmethod
|
||||||
def steps(self):
|
def steps(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
|
import abc
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from abc import ABCMeta, abstractmethod
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -78,7 +78,7 @@ class Modes:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class IRemote(metaclass=ABCMeta):
|
class IRemote(abc.ABC):
|
||||||
"""
|
"""
|
||||||
Common interface between base class and extended (higher) classes
|
Common interface between base class and extended (higher) classes
|
||||||
|
|
||||||
@@ -111,7 +111,7 @@ class IRemote(metaclass=ABCMeta):
|
|||||||
return ''.join(cmd)
|
return ''.join(cmd)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@abstractmethod
|
@abc.abstractmethod
|
||||||
def identifier(self):
|
def identifier(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -1,16 +1,9 @@
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from enum import Enum, unique
|
|
||||||
|
|
||||||
|
from .enums import KindId
|
||||||
from .error import VBANCMDError
|
from .error import VBANCMDError
|
||||||
|
|
||||||
|
|
||||||
@unique
|
|
||||||
class KindId(Enum):
|
|
||||||
BASIC = 1
|
|
||||||
BANANA = 2
|
|
||||||
POTATO = 3
|
|
||||||
|
|
||||||
|
|
||||||
class SingletonType(type):
|
class SingletonType(type):
|
||||||
"""ensure only a single instance of a kind map object"""
|
"""ensure only a single instance of a kind map object"""
|
||||||
|
|
||||||
@@ -22,12 +15,15 @@ class SingletonType(type):
|
|||||||
return cls._instances[cls]
|
return cls._instances[cls]
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass(frozen=True)
|
||||||
class KindMapClass(metaclass=SingletonType):
|
class KindMapClass(metaclass=SingletonType):
|
||||||
name: str
|
name: str
|
||||||
ins: tuple
|
ins: tuple
|
||||||
outs: tuple
|
outs: tuple
|
||||||
vban: tuple
|
vban: tuple
|
||||||
|
strip_channels: int
|
||||||
|
bus_channels: int
|
||||||
|
cells: int
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def phys_in(self):
|
def phys_in(self):
|
||||||
@@ -65,28 +61,37 @@ class KindMapClass(metaclass=SingletonType):
|
|||||||
return self.name.capitalize()
|
return self.name.capitalize()
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass(frozen=True)
|
||||||
class BasicMap(KindMapClass):
|
class BasicMap(KindMapClass):
|
||||||
name: str
|
name: str
|
||||||
ins: tuple = (2, 1)
|
ins: tuple = (2, 1)
|
||||||
outs: tuple = (1, 1)
|
outs: tuple = (1, 1)
|
||||||
vban: tuple = (4, 4, 1, 1)
|
vban: tuple = (4, 4, 1, 1)
|
||||||
|
strip_channels: int = 0
|
||||||
|
bus_channels: int = 0
|
||||||
|
cells: int = 0
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass(frozen=True)
|
||||||
class BananaMap(KindMapClass):
|
class BananaMap(KindMapClass):
|
||||||
name: str
|
name: str
|
||||||
ins: tuple = (3, 2)
|
ins: tuple = (3, 2)
|
||||||
outs: tuple = (3, 2)
|
outs: tuple = (3, 2)
|
||||||
vban: tuple = (8, 8, 1, 1)
|
vban: tuple = (8, 8, 1, 1)
|
||||||
|
strip_channels: int = 0
|
||||||
|
bus_channels: int = 8
|
||||||
|
cells: int = 6
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass(frozen=True)
|
||||||
class PotatoMap(KindMapClass):
|
class PotatoMap(KindMapClass):
|
||||||
name: str
|
name: str
|
||||||
ins: tuple = (5, 3)
|
ins: tuple = (5, 3)
|
||||||
outs: tuple = (5, 3)
|
outs: tuple = (5, 3)
|
||||||
vban: tuple = (8, 8, 1, 1)
|
vban: tuple = (8, 8, 1, 1)
|
||||||
|
strip_channels: int = 2
|
||||||
|
bus_channels: int = 8
|
||||||
|
cells: int = 6
|
||||||
|
|
||||||
|
|
||||||
def kind_factory(kind_id):
|
def kind_factory(kind_id):
|
||||||
|
|||||||
@@ -106,13 +106,23 @@ def xy_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}')
|
||||||
_type, axis = param.split('_')
|
|
||||||
if self.public_packets[NBS.one] is None:
|
if self.public_packets[NBS.one] is None:
|
||||||
return 0.0
|
return 0.0
|
||||||
x, y = getattr(
|
|
||||||
self.public_packets[NBS.one].strips[self.index], f'position_{_type.lower()}'
|
positions = self.public_packets[NBS.one].strips[self.index].positions
|
||||||
)
|
match param:
|
||||||
return x if axis == 'x' else y
|
case 'pan_x':
|
||||||
|
return positions.pan_x
|
||||||
|
case 'pan_y':
|
||||||
|
return positions.pan_y
|
||||||
|
case 'color_x':
|
||||||
|
return positions.color_x
|
||||||
|
case 'color_y':
|
||||||
|
return positions.color_y
|
||||||
|
case 'fx1':
|
||||||
|
return positions.fx1
|
||||||
|
case 'fx2':
|
||||||
|
return positions.fx2
|
||||||
|
|
||||||
def fset(self, val):
|
def fset(self, val):
|
||||||
self.setter(param, val)
|
self.setter(param, val)
|
||||||
@@ -129,12 +139,17 @@ def send_prop(param):
|
|||||||
self.logger.debug(f'getter: {cmd}')
|
self.logger.debug(f'getter: {cmd}')
|
||||||
if self.public_packets[NBS.one] is None:
|
if self.public_packets[NBS.one] is None:
|
||||||
return 0.0
|
return 0.0
|
||||||
val = getattr(self.public_packets[NBS.one].strips[self.index], f'send_{param}')
|
|
||||||
|
sends = self.public_packets[NBS.one].strips[self.index].sends
|
||||||
match param:
|
match param:
|
||||||
case 'reverb' | 'fx1':
|
case 'reverb':
|
||||||
return val[0]
|
return sends.reverb
|
||||||
case 'delay' | 'fx2':
|
case 'delay':
|
||||||
return val[1]
|
return sends.delay
|
||||||
|
case 'fx1':
|
||||||
|
return sends.fx1
|
||||||
|
case 'fx2':
|
||||||
|
return sends.fx2
|
||||||
|
|
||||||
def fset(self, val):
|
def fset(self, val):
|
||||||
self.setter(param, val)
|
self.setter(param, val)
|
||||||
|
|||||||
@@ -1,4 +1,6 @@
|
|||||||
|
import struct
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
from typing import NamedTuple
|
||||||
|
|
||||||
from .enums import NBS
|
from .enums import NBS
|
||||||
from .kinds import KindMapClass
|
from .kinds import KindMapClass
|
||||||
@@ -22,34 +24,34 @@ class VbanRtPacket:
|
|||||||
|
|
||||||
nbs: NBS
|
nbs: NBS
|
||||||
_kind: KindMapClass
|
_kind: KindMapClass
|
||||||
_voicemeeterType: bytes # data[28:29]
|
_voicemeeterType: bytes
|
||||||
_reserved: bytes # data[29:30]
|
_reserved: bytes
|
||||||
_buffersize: bytes # data[30:32]
|
_buffersize: bytes
|
||||||
_voicemeeterVersion: bytes # data[32:36]
|
_voicemeeterVersion: bytes
|
||||||
_optionBits: bytes # data[36:40]
|
_optionBits: bytes
|
||||||
_samplerate: bytes # data[40:44]
|
_samplerate: bytes
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class VbanRtPacketNBS0(VbanRtPacket):
|
class VbanRtPacketNBS0(VbanRtPacket):
|
||||||
"""Represents the body of a VBAN RT data packet with NBS 0"""
|
"""Represents the body of a VBAN RT data packet with NBS 0"""
|
||||||
|
|
||||||
_inputLeveldB100: bytes # data[44:112]
|
_inputLeveldB100: bytes
|
||||||
_outputLeveldB100: bytes # data[112:240]
|
_outputLeveldB100: bytes
|
||||||
_TransportBit: bytes # data[240:244]
|
_TransportBit: bytes
|
||||||
_stripState: bytes # data[244:276]
|
_stripState: bytes
|
||||||
_busState: bytes # data[276:308]
|
_busState: bytes
|
||||||
_stripGaindB100Layer1: bytes # data[308:324]
|
_stripGaindB100Layer1: bytes
|
||||||
_stripGaindB100Layer2: bytes # data[324:340]
|
_stripGaindB100Layer2: bytes
|
||||||
_stripGaindB100Layer3: bytes # data[340:356]
|
_stripGaindB100Layer3: bytes
|
||||||
_stripGaindB100Layer4: bytes # data[356:372]
|
_stripGaindB100Layer4: bytes
|
||||||
_stripGaindB100Layer5: bytes # data[372:388]
|
_stripGaindB100Layer5: bytes
|
||||||
_stripGaindB100Layer6: bytes # data[388:404]
|
_stripGaindB100Layer6: bytes
|
||||||
_stripGaindB100Layer7: bytes # data[404:420]
|
_stripGaindB100Layer7: bytes
|
||||||
_stripGaindB100Layer8: bytes # data[420:436]
|
_stripGaindB100Layer8: bytes
|
||||||
_busGaindB100: bytes # data[436:452]
|
_busGaindB100: bytes
|
||||||
_stripLabelUTF8c60: bytes # data[452:932]
|
_stripLabelUTF8c60: bytes
|
||||||
_busLabelUTF8c60: bytes # data[932:1412]
|
_busLabelUTF8c60: bytes
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes):
|
def from_bytes(cls, nbs: NBS, kind: KindMapClass, data: bytes):
|
||||||
@@ -216,64 +218,135 @@ class VbanRtPacketNBS0(VbanRtPacket):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class Audibility(NamedTuple):
|
||||||
|
knob: float
|
||||||
|
comp: float
|
||||||
|
gate: float
|
||||||
|
denoiser: float
|
||||||
|
|
||||||
|
|
||||||
|
class Positions(NamedTuple):
|
||||||
|
pan_x: float
|
||||||
|
pan_y: float
|
||||||
|
color_x: float
|
||||||
|
color_y: float
|
||||||
|
fx1: float
|
||||||
|
fx2: float
|
||||||
|
|
||||||
|
|
||||||
|
class EqGains(NamedTuple):
|
||||||
|
bass: float
|
||||||
|
mid: float
|
||||||
|
treble: float
|
||||||
|
|
||||||
|
|
||||||
|
class ParametricEQSettings(NamedTuple):
|
||||||
|
on: bool
|
||||||
|
type: int
|
||||||
|
gain: float
|
||||||
|
freq: float
|
||||||
|
q: float
|
||||||
|
|
||||||
|
|
||||||
|
class Sends(NamedTuple):
|
||||||
|
reverb: float
|
||||||
|
delay: float
|
||||||
|
fx1: float
|
||||||
|
fx2: float
|
||||||
|
|
||||||
|
|
||||||
|
class CompressorSettings(NamedTuple):
|
||||||
|
gain_in: float
|
||||||
|
attack_ms: float
|
||||||
|
release_ms: float
|
||||||
|
n_knee: float
|
||||||
|
ratio: float
|
||||||
|
threshold: float
|
||||||
|
c_enabled: bool
|
||||||
|
makeup: bool
|
||||||
|
gain_out: float
|
||||||
|
|
||||||
|
|
||||||
|
class GateSettings(NamedTuple):
|
||||||
|
threshold_in: float
|
||||||
|
damping_max: float
|
||||||
|
bp_sidechain: bool
|
||||||
|
attack_ms: float
|
||||||
|
hold_ms: float
|
||||||
|
release_ms: float
|
||||||
|
|
||||||
|
|
||||||
|
class DenoiserSettings(NamedTuple):
|
||||||
|
threshold: float
|
||||||
|
|
||||||
|
|
||||||
|
class PitchSettings(NamedTuple):
|
||||||
|
enabled: bool
|
||||||
|
dry_wet: float
|
||||||
|
value: float
|
||||||
|
formant_lo: float
|
||||||
|
formant_med: float
|
||||||
|
formant_high: float
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class VbanVMParamStrip:
|
class VbanVMParamStrip:
|
||||||
"""Represents the VBAN_VMPARAMSTRIP_PACKET structure"""
|
"""Represents the VBAN_VMPARAMSTRIP_PACKET structure"""
|
||||||
|
|
||||||
_mode: bytes # long = 4 bytes data[0:4]
|
_mode: bytes
|
||||||
_dblevel: bytes # float = 4 bytes data[4:8]
|
_dblevel: bytes
|
||||||
_audibility: bytes # short = 2 bytes data[8:10]
|
_audibility: bytes
|
||||||
_pos3D_x: bytes # short = 2 bytes data[10:12]
|
_pos3D_x: bytes
|
||||||
_pos3D_y: bytes # short = 2 bytes data[12:14]
|
_pos3D_y: bytes
|
||||||
_posColor_x: bytes # short = 2 bytes data[14:16]
|
_posColor_x: bytes
|
||||||
_posColor_y: bytes # short = 2 bytes data[16:18]
|
_posColor_y: bytes
|
||||||
_EQgain1: bytes # short = 2 bytes data[18:20]
|
_EQgain1: bytes
|
||||||
_EQgain2: bytes # short = 2 bytes data[20:22]
|
_EQgain2: bytes
|
||||||
_EQgain3: bytes # short = 2 bytes data[22:24]
|
_EQgain3: bytes
|
||||||
|
|
||||||
# First channel parametric EQ
|
# First channel parametric EQ
|
||||||
_PEQ_eqOn: bytes # 6 * char = 6 bytes data[24:30]
|
_PEQ_eqOn: bytes
|
||||||
_PEQ_eqtype: bytes # 6 * char = 6 bytes data[30:36]
|
_PEQ_eqtype: bytes
|
||||||
_PEQ_eqgain: bytes # 6 * float = 24 bytes data[36:60]
|
_PEQ_eqgain: bytes
|
||||||
_PEQ_eqfreq: bytes # 6 * float = 24 bytes data[60:84]
|
_PEQ_eqfreq: bytes
|
||||||
_PEQ_eqq: bytes # 6 * float = 24 bytes data[84:108]
|
_PEQ_eqq: bytes
|
||||||
|
|
||||||
_audibility_c: bytes # short = 2 bytes data[108:110]
|
_audibility_c: bytes
|
||||||
_audibility_g: bytes # short = 2 bytes data[110:112]
|
_audibility_g: bytes
|
||||||
_audibility_d: bytes # short = 2 bytes data[112:114]
|
_audibility_d: bytes
|
||||||
_posMod_x: bytes # short = 2 bytes data[114:116]
|
_posMod_x: bytes
|
||||||
_posMod_y: bytes # short = 2 bytes data[116:118]
|
_posMod_y: bytes
|
||||||
_send_reverb: bytes # short = 2 bytes data[118:120]
|
_send_reverb: bytes
|
||||||
_send_delay: bytes # short = 2 bytes data[120:122]
|
_send_delay: bytes
|
||||||
_send_fx1: bytes # short = 2 bytes data[122:124]
|
_send_fx1: bytes
|
||||||
_send_fx2: bytes # short = 2 bytes data[124:126]
|
_send_fx2: bytes
|
||||||
_dblimit: bytes # short = 2 bytes data[126:128]
|
_dblimit: bytes
|
||||||
_nKaraoke: bytes # short = 2 bytes data[128:130]
|
_nKaraoke: bytes
|
||||||
|
|
||||||
_COMP_gain_in: bytes # short = 2 bytes data[130:132]
|
_COMP_gain_in: bytes
|
||||||
_COMP_attack_ms: bytes # short = 2 bytes data[132:134]
|
_COMP_attack_ms: bytes
|
||||||
_COMP_release_ms: bytes # short = 2 bytes data[134:136]
|
_COMP_release_ms: bytes
|
||||||
_COMP_n_knee: bytes # short = 2 bytes data[136:138]
|
_COMP_n_knee: bytes
|
||||||
_COMP_comprate: bytes # short = 2 bytes data[138:140]
|
_COMP_comprate: bytes
|
||||||
_COMP_threshold: bytes # short = 2 bytes data[140:142]
|
_COMP_threshold: bytes
|
||||||
_COMP_c_enabled: bytes # short = 2 bytes data[142:144]
|
_COMP_c_enabled: bytes
|
||||||
_COMP_c_auto: bytes # short = 2 bytes data[144:146]
|
_COMP_c_auto: bytes
|
||||||
_COMP_gain_out: bytes # short = 2 bytes data[146:148]
|
_COMP_gain_out: bytes
|
||||||
|
|
||||||
_GATE_dBThreshold_in: bytes # short = 2 bytes data[148:150]
|
_GATE_dBThreshold_in: bytes
|
||||||
_GATE_dBDamping_max: bytes # short = 2 bytes data[150:152]
|
_GATE_dBDamping_max: bytes
|
||||||
_GATE_BP_Sidechain: bytes # short = 2 bytes data[152:154]
|
_GATE_BP_Sidechain: bytes
|
||||||
_GATE_attack_ms: bytes # short = 2 bytes data[154:156]
|
_GATE_attack_ms: bytes
|
||||||
_GATE_hold_ms: bytes # short = 2 bytes data[156:158]
|
_GATE_hold_ms: bytes
|
||||||
_GATE_release_ms: bytes # short = 2 bytes data[158:160]
|
_GATE_release_ms: bytes
|
||||||
|
|
||||||
_DenoiserThreshold: bytes # short = 2 bytes data[160:162]
|
_DenoiserThreshold: bytes
|
||||||
_PitchEnabled: bytes # short = 2 bytes data[162:164]
|
_PitchEnabled: bytes
|
||||||
_Pitch_DryWet: bytes # short = 2 bytes data[164:166]
|
_Pitch_DryWet: bytes
|
||||||
_Pitch_Value: bytes # short = 2 bytes data[166:168]
|
_Pitch_Value: bytes
|
||||||
_Pitch_formant_lo: bytes # short = 2 bytes data[168:170]
|
_Pitch_formant_lo: bytes
|
||||||
_Pitch_formant_med: bytes # short = 2 bytes data[170:172]
|
_Pitch_formant_med: bytes
|
||||||
_Pitch_formant_high: bytes # short = 2 bytes data[172:174]
|
_Pitch_formant_high: bytes
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_bytes(cls, data: bytes):
|
def from_bytes(cls, data: bytes):
|
||||||
@@ -333,59 +406,133 @@ class VbanVMParamStrip:
|
|||||||
return int.from_bytes(self._mode, 'little')
|
return int.from_bytes(self._mode, 'little')
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def position_pan(self) -> tuple[int, int]:
|
def audibility(self) -> Audibility:
|
||||||
return (
|
return Audibility(
|
||||||
|
round(int.from_bytes(self._audibility, 'little', signed=True) * 0.01, 2),
|
||||||
|
round(int.from_bytes(self._audibility_c, 'little', signed=True) * 0.01, 2),
|
||||||
|
round(int.from_bytes(self._audibility_g, 'little', signed=True) * 0.01, 2),
|
||||||
|
round(int.from_bytes(self._audibility_d, 'little', signed=True) * 0.01, 2),
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def positions(self) -> Positions:
|
||||||
|
return Positions(
|
||||||
round(int.from_bytes(self._pos3D_x, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._pos3D_x, 'little', signed=True) * 0.01, 2),
|
||||||
round(int.from_bytes(self._pos3D_y, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._pos3D_y, 'little', signed=True) * 0.01, 2),
|
||||||
)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def position_color(self) -> tuple[int, int]:
|
|
||||||
return (
|
|
||||||
round(int.from_bytes(self._posColor_x, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._posColor_x, 'little', signed=True) * 0.01, 2),
|
||||||
round(int.from_bytes(self._posColor_y, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._posColor_y, 'little', signed=True) * 0.01, 2),
|
||||||
)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def position_fx(self) -> tuple[int, int]:
|
|
||||||
return (
|
|
||||||
round(int.from_bytes(self._posMod_x, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._posMod_x, 'little', signed=True) * 0.01, 2),
|
||||||
round(int.from_bytes(self._posMod_y, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._posMod_y, 'little', signed=True) * 0.01, 2),
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def send_reverb(self) -> tuple[float, float]:
|
def eqgains(self) -> EqGains:
|
||||||
return (
|
return EqGains(
|
||||||
|
*[
|
||||||
|
round(
|
||||||
|
int.from_bytes(getattr(self, f'_EQgain{i}'), 'little', signed=True)
|
||||||
|
* 0.01,
|
||||||
|
2,
|
||||||
|
)
|
||||||
|
for i in range(1, 4)
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def parametric_eq(self) -> tuple[ParametricEQSettings, ...]:
|
||||||
|
return tuple(
|
||||||
|
ParametricEQSettings(
|
||||||
|
on=bool(int.from_bytes(self._PEQ_eqOn[i : i + 1], 'little')),
|
||||||
|
type=int.from_bytes(self._PEQ_eqtype[i : i + 1], 'little'),
|
||||||
|
freq=struct.unpack('<f', self._PEQ_eqfreq[i * 4 : (i + 1) * 4])[0],
|
||||||
|
gain=struct.unpack('<f', self._PEQ_eqgain[i * 4 : (i + 1) * 4])[0],
|
||||||
|
q=struct.unpack('<f', self._PEQ_eqq[i * 4 : (i + 1) * 4])[0],
|
||||||
|
)
|
||||||
|
for i in range(6)
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def sends(self) -> Sends:
|
||||||
|
return Sends(
|
||||||
round(int.from_bytes(self._send_reverb, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._send_reverb, 'little', signed=True) * 0.01, 2),
|
||||||
round(int.from_bytes(self._send_delay, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._send_delay, 'little', signed=True) * 0.01, 2),
|
||||||
)
|
|
||||||
|
|
||||||
send_delay = send_reverb
|
|
||||||
|
|
||||||
@property
|
|
||||||
def send_fx1(self) -> tuple[float, float]:
|
|
||||||
return (
|
|
||||||
round(int.from_bytes(self._send_fx1, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._send_fx1, 'little', signed=True) * 0.01, 2),
|
||||||
round(int.from_bytes(self._send_fx2, 'little', signed=True) * 0.01, 2),
|
round(int.from_bytes(self._send_fx2, 'little', signed=True) * 0.01, 2),
|
||||||
)
|
)
|
||||||
|
|
||||||
send_fx2 = send_fx1
|
|
||||||
|
|
||||||
@property
|
|
||||||
def eqgains(self) -> tuple[float, float, float]:
|
|
||||||
return tuple(
|
|
||||||
round(
|
|
||||||
int.from_bytes(getattr(self, f'_EQgain{i}'), 'little', signed=True)
|
|
||||||
* 0.01,
|
|
||||||
2,
|
|
||||||
)
|
|
||||||
for i in range(1, 4)
|
|
||||||
)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def karaoke(self) -> int:
|
def karaoke(self) -> int:
|
||||||
return int.from_bytes(self._nKaraoke, 'little')
|
return int.from_bytes(self._nKaraoke, 'little')
|
||||||
|
|
||||||
|
@property
|
||||||
|
def compressor(self) -> CompressorSettings:
|
||||||
|
return CompressorSettings(
|
||||||
|
gain_in=round(
|
||||||
|
int.from_bytes(self._COMP_gain_in, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
attack_ms=round(int.from_bytes(self._COMP_attack_ms, 'little') * 0.1, 2),
|
||||||
|
release_ms=round(int.from_bytes(self._COMP_release_ms, 'little') * 0.1, 2),
|
||||||
|
n_knee=round(int.from_bytes(self._COMP_n_knee, 'little') * 0.01, 2),
|
||||||
|
ratio=round(int.from_bytes(self._COMP_comprate, 'little') * 0.01, 2),
|
||||||
|
threshold=round(
|
||||||
|
int.from_bytes(self._COMP_threshold, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
c_enabled=bool(int.from_bytes(self._COMP_c_enabled, 'little')),
|
||||||
|
makeup=bool(int.from_bytes(self._COMP_c_auto, 'little')),
|
||||||
|
gain_out=round(
|
||||||
|
int.from_bytes(self._COMP_gain_out, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def gate(self) -> GateSettings:
|
||||||
|
return GateSettings(
|
||||||
|
threshold_in=round(
|
||||||
|
int.from_bytes(self._GATE_dBThreshold_in, 'little', signed=True) * 0.01,
|
||||||
|
2,
|
||||||
|
),
|
||||||
|
damping_max=round(
|
||||||
|
int.from_bytes(self._GATE_dBDamping_max, 'little', signed=True) * 0.01,
|
||||||
|
2,
|
||||||
|
),
|
||||||
|
bp_sidechain=round(
|
||||||
|
int.from_bytes(self._GATE_BP_Sidechain, 'little') * 0.1, 2
|
||||||
|
),
|
||||||
|
attack_ms=round(int.from_bytes(self._GATE_attack_ms, 'little') * 0.1, 2),
|
||||||
|
hold_ms=round(int.from_bytes(self._GATE_hold_ms, 'little') * 0.1, 2),
|
||||||
|
release_ms=round(int.from_bytes(self._GATE_release_ms, 'little') * 0.1, 2),
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def denoiser(self) -> DenoiserSettings:
|
||||||
|
return DenoiserSettings(
|
||||||
|
threshold=round(
|
||||||
|
int.from_bytes(self._DenoiserThreshold, 'little', signed=True) * 0.01, 2
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def pitch(self) -> PitchSettings:
|
||||||
|
return PitchSettings(
|
||||||
|
enabled=bool(int.from_bytes(self._PitchEnabled, 'little')),
|
||||||
|
dry_wet=round(
|
||||||
|
int.from_bytes(self._Pitch_DryWet, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
value=round(
|
||||||
|
int.from_bytes(self._Pitch_Value, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
formant_lo=round(
|
||||||
|
int.from_bytes(self._Pitch_formant_lo, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
formant_med=round(
|
||||||
|
int.from_bytes(self._Pitch_formant_med, 'little', signed=True) * 0.01, 2
|
||||||
|
),
|
||||||
|
formant_high=round(
|
||||||
|
int.from_bytes(self._Pitch_formant_high, 'little', signed=True) * 0.01,
|
||||||
|
2,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class VbanRtPacketNBS1(VbanRtPacket):
|
class VbanRtPacketNBS1(VbanRtPacket):
|
||||||
@@ -422,19 +569,38 @@ class VbanRtPacketNBS1(VbanRtPacket):
|
|||||||
class SubscribeHeader:
|
class SubscribeHeader:
|
||||||
"""Represents the header of an RT subscription packet"""
|
"""Represents the header of an RT subscription packet"""
|
||||||
|
|
||||||
ident: NBS = NBS.zero
|
nbs: NBS = NBS.zero
|
||||||
name = 'Register-RTP'
|
name: str = 'Register-RTP'
|
||||||
timeout = 15
|
timeout: int = 15
|
||||||
vban: bytes = 'VBAN'.encode()
|
|
||||||
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little')
|
@property
|
||||||
format_nbs: bytes = (ident.value & 0xFF).to_bytes(1, 'little')
|
def vban(self) -> bytes:
|
||||||
format_nbc: bytes = (VBAN_SERVICE_RTPACKETREGISTER).to_bytes(1, 'little')
|
return b'VBAN'
|
||||||
format_bit: bytes = (timeout & 0xFF).to_bytes(1, 'little') # timeout
|
|
||||||
streamname: bytes = name.encode('ascii') + bytes(16 - len(name))
|
@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
|
@classmethod
|
||||||
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
|
def to_bytes(cls, nbs: NBS, framecounter: int) -> bytes:
|
||||||
header = cls(ident=nbs)
|
header = cls(nbs=nbs)
|
||||||
|
|
||||||
data = bytearray()
|
data = bytearray()
|
||||||
data.extend(header.vban)
|
data.extend(header.vban)
|
||||||
data.extend(header.format_sr)
|
data.extend(header.format_sr)
|
||||||
@@ -451,30 +617,31 @@ class VbanRtPacketHeader:
|
|||||||
"""Represents the header of an RT response packet"""
|
"""Represents the header of an RT response packet"""
|
||||||
|
|
||||||
name: str = 'Voicemeeter-RTP'
|
name: str = 'Voicemeeter-RTP'
|
||||||
vban: bytes = 'VBAN'.encode()
|
format_sr: int = VBAN_PROTOCOL_SERVICE
|
||||||
format_sr: bytes = (VBAN_PROTOCOL_SERVICE).to_bytes(1, 'little')
|
format_nbs: int = 0
|
||||||
format_nbs: bytes = (0).to_bytes(1, 'little')
|
format_nbc: int = VBAN_SERVICE_RTPACKET
|
||||||
format_nbc: bytes = (VBAN_SERVICE_RTPACKET).to_bytes(1, 'little')
|
format_bit: int = 0
|
||||||
format_bit: bytes = (0).to_bytes(1, 'little')
|
|
||||||
streamname: bytes = name.encode('ascii') + bytes(16 - len(name))
|
@property
|
||||||
|
def vban(self) -> bytes:
|
||||||
|
return b'VBAN'
|
||||||
|
|
||||||
|
@property
|
||||||
|
def streamname(self) -> bytes:
|
||||||
|
return self.name.encode('ascii') + bytes(16 - len(self.name))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_bytes(cls, data: bytes):
|
def from_bytes(cls, data: bytes):
|
||||||
if len(data) < HEADER_SIZE:
|
if len(data) < HEADER_SIZE:
|
||||||
raise ValueError('Data is too short to be a valid VbanRTPPacketHeader')
|
raise ValueError('Data is too short to be a valid VbanRTPPacketHeader')
|
||||||
vban = data[0:4]
|
|
||||||
format_sr = data[4]
|
|
||||||
format_nbs = data[5]
|
|
||||||
format_nbc = data[6]
|
|
||||||
format_bit = data[7]
|
|
||||||
name = data[8:24].rstrip(b'\x00').decode('utf-8')
|
name = data[8:24].rstrip(b'\x00').decode('utf-8')
|
||||||
return cls(
|
return cls(
|
||||||
name=name,
|
name=name,
|
||||||
vban=vban,
|
format_sr=data[4] & VBAN_SERVICE_MASK,
|
||||||
format_sr=format_sr & VBAN_SERVICE_MASK,
|
format_nbs=data[5],
|
||||||
format_nbs=format_nbs,
|
format_nbc=data[6],
|
||||||
format_nbc=format_nbc,
|
format_bit=data[7],
|
||||||
format_bit=format_bit,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -485,21 +652,30 @@ class RequestHeader:
|
|||||||
name: str
|
name: str
|
||||||
bps_index: int
|
bps_index: int
|
||||||
channel: int
|
channel: int
|
||||||
vban: bytes = 'VBAN'.encode()
|
framecounter: int = 0
|
||||||
nbs: bytes = (0).to_bytes(1, 'little')
|
|
||||||
bit: bytes = (0x10).to_bytes(1, 'little')
|
|
||||||
framecounter: bytes = (0).to_bytes(4, 'little')
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def sr(self):
|
def vban(self) -> bytes:
|
||||||
|
return b'VBAN'
|
||||||
|
|
||||||
|
@property
|
||||||
|
def sr(self) -> bytes:
|
||||||
return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
|
return (VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def nbc(self):
|
def nbs(self) -> bytes:
|
||||||
|
return (0).to_bytes(1, 'little')
|
||||||
|
|
||||||
|
@property
|
||||||
|
def nbc(self) -> bytes:
|
||||||
return (self.channel).to_bytes(1, 'little')
|
return (self.channel).to_bytes(1, 'little')
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def streamname(self):
|
def bit(self) -> bytes:
|
||||||
|
return (0x10).to_bytes(1, 'little')
|
||||||
|
|
||||||
|
@property
|
||||||
|
def streamname(self) -> bytes:
|
||||||
return self.name.encode() + bytes(16 - len(self.name))
|
return self.name.encode() + bytes(16 - len(self.name))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@@ -509,6 +685,7 @@ class RequestHeader:
|
|||||||
header = cls(
|
header = cls(
|
||||||
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
|
name=name, bps_index=bps_index, channel=channel, framecounter=framecounter
|
||||||
)
|
)
|
||||||
|
|
||||||
data = bytearray()
|
data = bytearray()
|
||||||
data.extend(header.vban)
|
data.extend(header.vban)
|
||||||
data.extend(header.sr)
|
data.extend(header.sr)
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
|
import abc
|
||||||
import time
|
import time
|
||||||
from abc import abstractmethod
|
|
||||||
from typing import Union
|
from typing import Union
|
||||||
|
|
||||||
from . import kinds
|
from . import kinds
|
||||||
from .enums import NBS, EQGains
|
from .enums import NBS
|
||||||
from .iremote import IRemote
|
from .iremote import IRemote
|
||||||
from .meta import (
|
from .meta import (
|
||||||
channel_bool_prop,
|
channel_bool_prop,
|
||||||
@@ -21,7 +21,7 @@ class Strip(IRemote):
|
|||||||
Defines concrete implementation for strip
|
Defines concrete implementation for strip
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@abstractmethod
|
@abc.abstractmethod
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@@ -68,7 +68,7 @@ class PhysicalStrip(Strip):
|
|||||||
'comp': StripComp(remote, index),
|
'comp': StripComp(remote, index),
|
||||||
'gate': StripGate(remote, index),
|
'gate': StripGate(remote, index),
|
||||||
'denoiser': StripDenoiser(remote, index),
|
'denoiser': StripDenoiser(remote, index),
|
||||||
'eq': StripEQ(remote, index),
|
'eq': StripEQ.make(remote, index),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -76,12 +76,14 @@ class PhysicalStrip(Strip):
|
|||||||
return f'{type(self).__name__}{self.index}'
|
return f'{type(self).__name__}{self.index}'
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def device(self):
|
def audibility(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].audibility.knob
|
||||||
|
|
||||||
@property
|
@audibility.setter
|
||||||
def sr(self):
|
def audibility(self, val: float):
|
||||||
return
|
self.setter('audibility', val)
|
||||||
|
|
||||||
|
|
||||||
class StripComp(IRemote):
|
class StripComp(IRemote):
|
||||||
@@ -91,7 +93,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def knob(self) -> float:
|
def knob(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].audibility.comp
|
||||||
|
|
||||||
@knob.setter
|
@knob.setter
|
||||||
def knob(self, val: float):
|
def knob(self, val: float):
|
||||||
@@ -99,7 +103,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def gainin(self) -> float:
|
def gainin(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.gain_in
|
||||||
|
|
||||||
@gainin.setter
|
@gainin.setter
|
||||||
def gainin(self, val: float):
|
def gainin(self, val: float):
|
||||||
@@ -107,7 +113,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def ratio(self) -> float:
|
def ratio(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.ratio
|
||||||
|
|
||||||
@ratio.setter
|
@ratio.setter
|
||||||
def ratio(self, val: float):
|
def ratio(self, val: float):
|
||||||
@@ -115,7 +123,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def threshold(self) -> float:
|
def threshold(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.threshold
|
||||||
|
|
||||||
@threshold.setter
|
@threshold.setter
|
||||||
def threshold(self, val: float):
|
def threshold(self, val: float):
|
||||||
@@ -123,7 +133,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def attack(self) -> float:
|
def attack(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.attack_ms
|
||||||
|
|
||||||
@attack.setter
|
@attack.setter
|
||||||
def attack(self, val: float):
|
def attack(self, val: float):
|
||||||
@@ -131,7 +143,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def release(self) -> float:
|
def release(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.release_ms
|
||||||
|
|
||||||
@release.setter
|
@release.setter
|
||||||
def release(self, val: float):
|
def release(self, val: float):
|
||||||
@@ -139,7 +153,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def knee(self) -> float:
|
def knee(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.n_knee
|
||||||
|
|
||||||
@knee.setter
|
@knee.setter
|
||||||
def knee(self, val: float):
|
def knee(self, val: float):
|
||||||
@@ -147,7 +163,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def gainout(self) -> float:
|
def gainout(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].compressor.gain_out
|
||||||
|
|
||||||
@gainout.setter
|
@gainout.setter
|
||||||
def gainout(self, val: float):
|
def gainout(self, val: float):
|
||||||
@@ -155,7 +173,9 @@ class StripComp(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def makeup(self) -> bool:
|
def makeup(self) -> bool:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return False
|
||||||
|
return bool(self.public_packets[NBS.one].strips[self.index].compressor.makeup)
|
||||||
|
|
||||||
@makeup.setter
|
@makeup.setter
|
||||||
def makeup(self, val: bool):
|
def makeup(self, val: bool):
|
||||||
@@ -169,7 +189,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def knob(self) -> float:
|
def knob(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].audibility.gate
|
||||||
|
|
||||||
@knob.setter
|
@knob.setter
|
||||||
def knob(self, val: float):
|
def knob(self, val: float):
|
||||||
@@ -177,7 +199,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def threshold(self) -> float:
|
def threshold(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].gate.threshold_in
|
||||||
|
|
||||||
@threshold.setter
|
@threshold.setter
|
||||||
def threshold(self, val: float):
|
def threshold(self, val: float):
|
||||||
@@ -185,7 +209,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def damping(self) -> float:
|
def damping(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].gate.damping_max
|
||||||
|
|
||||||
@damping.setter
|
@damping.setter
|
||||||
def damping(self, val: float):
|
def damping(self, val: float):
|
||||||
@@ -193,7 +219,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def bpsidechain(self) -> int:
|
def bpsidechain(self) -> int:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].gate.bp_sidechain
|
||||||
|
|
||||||
@bpsidechain.setter
|
@bpsidechain.setter
|
||||||
def bpsidechain(self, val: int):
|
def bpsidechain(self, val: int):
|
||||||
@@ -201,7 +229,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def attack(self) -> float:
|
def attack(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].gate.attack_ms
|
||||||
|
|
||||||
@attack.setter
|
@attack.setter
|
||||||
def attack(self, val: float):
|
def attack(self, val: float):
|
||||||
@@ -209,7 +239,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def hold(self) -> float:
|
def hold(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].gate.hold_ms
|
||||||
|
|
||||||
@hold.setter
|
@hold.setter
|
||||||
def hold(self, val: float):
|
def hold(self, val: float):
|
||||||
@@ -217,7 +249,9 @@ class StripGate(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def release(self) -> float:
|
def release(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].gate.release_ms
|
||||||
|
|
||||||
@release.setter
|
@release.setter
|
||||||
def release(self, val: float):
|
def release(self, val: float):
|
||||||
@@ -231,7 +265,9 @@ class StripDenoiser(IRemote):
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def knob(self) -> float:
|
def knob(self) -> float:
|
||||||
return
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return self.public_packets[NBS.one].strips[self.index].audibility.denoiser
|
||||||
|
|
||||||
@knob.setter
|
@knob.setter
|
||||||
def knob(self, val: float):
|
def knob(self, val: float):
|
||||||
@@ -239,6 +275,25 @@ class StripDenoiser(IRemote):
|
|||||||
|
|
||||||
|
|
||||||
class StripEQ(IRemote):
|
class StripEQ(IRemote):
|
||||||
|
@classmethod
|
||||||
|
def make(cls, remote, i):
|
||||||
|
"""
|
||||||
|
Factory method for Strip EQ.
|
||||||
|
|
||||||
|
Returns a StripEQ class.
|
||||||
|
"""
|
||||||
|
STRIPEQ_cls = type(
|
||||||
|
'StripEQ',
|
||||||
|
(cls,),
|
||||||
|
{
|
||||||
|
'channel': tuple(
|
||||||
|
StripEQCh.make(remote, i, j)
|
||||||
|
for j in range(remote.kind.strip_channels)
|
||||||
|
)
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return STRIPEQ_cls(remote, i)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def identifier(self) -> str:
|
def identifier(self) -> str:
|
||||||
return f'strip[{self.index}].eq'
|
return f'strip[{self.index}].eq'
|
||||||
@@ -260,6 +315,140 @@ class StripEQ(IRemote):
|
|||||||
self.setter('ab', 1 if val else 0)
|
self.setter('ab', 1 if val else 0)
|
||||||
|
|
||||||
|
|
||||||
|
class StripEQCh(IRemote):
|
||||||
|
@classmethod
|
||||||
|
def make(cls, remote, i, j):
|
||||||
|
"""
|
||||||
|
Factory method for Strip EQ channel.
|
||||||
|
|
||||||
|
Returns a StripEQCh class.
|
||||||
|
"""
|
||||||
|
StripEQCh_cls = type(
|
||||||
|
'StripEQCh',
|
||||||
|
(cls,),
|
||||||
|
{
|
||||||
|
'cell': tuple(
|
||||||
|
StripEQChCell(remote, i, j, k) for k in range(remote.kind.cells)
|
||||||
|
)
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return StripEQCh_cls(remote, i, j)
|
||||||
|
|
||||||
|
def __init__(self, remote, i, j):
|
||||||
|
super().__init__(remote, i)
|
||||||
|
self.channel_index = j
|
||||||
|
|
||||||
|
@property
|
||||||
|
def identifier(self) -> str:
|
||||||
|
return f'Strip[{self.index}].eq.channel[{self.channel_index}]'
|
||||||
|
|
||||||
|
|
||||||
|
class StripEQChCell(IRemote):
|
||||||
|
def __init__(self, remote, i, j, k):
|
||||||
|
super().__init__(remote, i)
|
||||||
|
self.channel_index = j
|
||||||
|
self.cell_index = k
|
||||||
|
|
||||||
|
@property
|
||||||
|
def identifier(self) -> str:
|
||||||
|
return f'Strip[{self.index}].eq.channel[{self.channel_index}].cell[{self.cell_index}]'
|
||||||
|
|
||||||
|
@property
|
||||||
|
def on(self) -> bool:
|
||||||
|
if self.channel_index > 0:
|
||||||
|
self.logger.warning(
|
||||||
|
'Only channel 0 is supported over VBAN for Strip EQ cells'
|
||||||
|
)
|
||||||
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return False
|
||||||
|
return (
|
||||||
|
self.public_packets[NBS.one]
|
||||||
|
.strips[self.index]
|
||||||
|
.parametric_eq[self.cell_index]
|
||||||
|
.on
|
||||||
|
)
|
||||||
|
|
||||||
|
@on.setter
|
||||||
|
def on(self, val: bool):
|
||||||
|
self.setter('on', 1 if val else 0)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def type(self) -> int:
|
||||||
|
if self.channel_index > 0:
|
||||||
|
self.logger.warning(
|
||||||
|
'Only channel 0 is supported over VBAN for Strip EQ cells'
|
||||||
|
)
|
||||||
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0
|
||||||
|
return (
|
||||||
|
self.public_packets[NBS.one]
|
||||||
|
.strips[self.index]
|
||||||
|
.parametric_eq[self.cell_index]
|
||||||
|
.type
|
||||||
|
)
|
||||||
|
|
||||||
|
@type.setter
|
||||||
|
def type(self, val: int):
|
||||||
|
self.setter('type', val)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def f(self) -> float:
|
||||||
|
if self.channel_index > 0:
|
||||||
|
self.logger.warning(
|
||||||
|
'Only channel 0 is supported over VBAN for Strip EQ cells'
|
||||||
|
)
|
||||||
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return (
|
||||||
|
self.public_packets[NBS.one]
|
||||||
|
.strips[self.index]
|
||||||
|
.parametric_eq[self.cell_index]
|
||||||
|
.freq
|
||||||
|
)
|
||||||
|
|
||||||
|
@f.setter
|
||||||
|
def f(self, val: float):
|
||||||
|
self.setter('f', val)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def gain(self) -> float:
|
||||||
|
if self.channel_index > 0:
|
||||||
|
self.logger.warning(
|
||||||
|
'Only channel 0 is supported over VBAN for Strip EQ cells'
|
||||||
|
)
|
||||||
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return (
|
||||||
|
self.public_packets[NBS.one]
|
||||||
|
.strips[self.index]
|
||||||
|
.parametric_eq[self.cell_index]
|
||||||
|
.gain
|
||||||
|
)
|
||||||
|
|
||||||
|
@gain.setter
|
||||||
|
def gain(self, val: float):
|
||||||
|
self.setter('gain', val)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def q(self) -> float:
|
||||||
|
if self.channel_index > 0:
|
||||||
|
self.logger.warning(
|
||||||
|
'Only channel 0 is supported over VBAN for Strip EQ cells'
|
||||||
|
)
|
||||||
|
if self.public_packets[NBS.one] is None:
|
||||||
|
return 0.0
|
||||||
|
return (
|
||||||
|
self.public_packets[NBS.one]
|
||||||
|
.strips[self.index]
|
||||||
|
.parametric_eq[self.cell_index]
|
||||||
|
.q
|
||||||
|
)
|
||||||
|
|
||||||
|
@q.setter
|
||||||
|
def q(self, val: float):
|
||||||
|
self.setter('q', val)
|
||||||
|
|
||||||
|
|
||||||
class VirtualStrip(Strip):
|
class VirtualStrip(Strip):
|
||||||
@classmethod
|
@classmethod
|
||||||
def make(cls, remote, i, is_phys):
|
def make(cls, remote, i, is_phys):
|
||||||
@@ -296,7 +485,7 @@ class VirtualStrip(Strip):
|
|||||||
def bass(self) -> float:
|
def bass(self) -> float:
|
||||||
if self.public_packets[NBS.one] is None:
|
if self.public_packets[NBS.one] is None:
|
||||||
return 0.0
|
return 0.0
|
||||||
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.bass]
|
return self.public_packets[NBS.one].strips[self.index].eqgains.bass
|
||||||
|
|
||||||
@bass.setter
|
@bass.setter
|
||||||
def bass(self, val: float):
|
def bass(self, val: float):
|
||||||
@@ -306,7 +495,7 @@ class VirtualStrip(Strip):
|
|||||||
def mid(self) -> float:
|
def mid(self) -> float:
|
||||||
if self.public_packets[NBS.one] is None:
|
if self.public_packets[NBS.one] is None:
|
||||||
return 0.0
|
return 0.0
|
||||||
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.mid]
|
return self.public_packets[NBS.one].strips[self.index].eqgains.mid
|
||||||
|
|
||||||
@mid.setter
|
@mid.setter
|
||||||
def mid(self, val: float):
|
def mid(self, val: float):
|
||||||
@@ -318,7 +507,7 @@ class VirtualStrip(Strip):
|
|||||||
def treble(self) -> float:
|
def treble(self) -> float:
|
||||||
if self.public_packets[NBS.one] is None:
|
if self.public_packets[NBS.one] is None:
|
||||||
return 0.0
|
return 0.0
|
||||||
return self.public_packets[NBS.one].strips[self.index].eqgains[EQGains.treble]
|
return self.public_packets[NBS.one].strips[self.index].eqgains.treble
|
||||||
|
|
||||||
@treble.setter
|
@treble.setter
|
||||||
def treble(self, val: float):
|
def treble(self, val: float):
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
from abc import abstractmethod
|
import abc
|
||||||
|
|
||||||
from . import kinds
|
from . import kinds
|
||||||
from .iremote import IRemote
|
from .iremote import IRemote
|
||||||
@@ -11,7 +11,7 @@ class VbanStream(IRemote):
|
|||||||
Defines concrete implementation for vban stream
|
Defines concrete implementation for vban stream
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@abstractmethod
|
@abc.abstractmethod
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -1,8 +1,8 @@
|
|||||||
|
import abc
|
||||||
import logging
|
import logging
|
||||||
import socket
|
import socket
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
from abc import ABCMeta, abstractmethod
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from queue import Queue
|
from queue import Queue
|
||||||
from typing import Iterable, Union
|
from typing import Iterable, Union
|
||||||
@@ -18,8 +18,8 @@ from .worker import Producer, Subscriber, Updater
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class VbanCmd(metaclass=ABCMeta):
|
class VbanCmd(abc.ABC):
|
||||||
"""Base class responsible for communicating with the VBAN RT Packet Service"""
|
"""Abstract Base Class for Voicemeeter VBAN Command Interfaces"""
|
||||||
|
|
||||||
DELAY = 0.001
|
DELAY = 0.001
|
||||||
# fmt: off
|
# fmt: off
|
||||||
@@ -49,7 +49,7 @@ class VbanCmd(metaclass=ABCMeta):
|
|||||||
self.stop_event = None
|
self.stop_event = None
|
||||||
self.producer = None
|
self.producer = None
|
||||||
|
|
||||||
@abstractmethod
|
@abc.abstractmethod
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
"""Ensure subclasses override str magic method"""
|
"""Ensure subclasses override str magic method"""
|
||||||
pass
|
pass
|
||||||
@@ -58,7 +58,7 @@ class VbanCmd(metaclass=ABCMeta):
|
|||||||
try:
|
try:
|
||||||
import tomllib
|
import tomllib
|
||||||
except ModuleNotFoundError:
|
except ModuleNotFoundError:
|
||||||
import tomli as tomllib
|
import tomli as tomllib # type: ignore[import]
|
||||||
|
|
||||||
def get_filepath():
|
def get_filepath():
|
||||||
for pn in (
|
for pn in (
|
||||||
|
|||||||
@@ -39,9 +39,6 @@ class Subscriber(threading.Thread):
|
|||||||
sub_packet, (self._remote.ip, self._remote.port)
|
sub_packet, (self._remote.ip, self._remote.port)
|
||||||
)
|
)
|
||||||
self._framecounter = bump_framecounter(self._framecounter)
|
self._framecounter = bump_framecounter(self._framecounter)
|
||||||
self.logger.debug(
|
|
||||||
f'sent subscription for NBS {nbs.name} to {self._remote.ip}:{self._remote.port}'
|
|
||||||
)
|
|
||||||
|
|
||||||
self.wait_until_stopped(10)
|
self.wait_until_stopped(10)
|
||||||
except socket.gaierror as e:
|
except socket.gaierror as e:
|
||||||
@@ -102,27 +99,11 @@ class Producer(threading.Thread):
|
|||||||
|
|
||||||
match response_header.format_nbs:
|
match response_header.format_nbs:
|
||||||
case NBS.zero:
|
case NBS.zero:
|
||||||
"""
|
|
||||||
self.logger.debug(
|
|
||||||
'Received NB0 RTP Packet from %s, Size: %d bytes',
|
|
||||||
addr,
|
|
||||||
len(data),
|
|
||||||
)
|
|
||||||
"""
|
|
||||||
|
|
||||||
return VbanRtPacketNBS0.from_bytes(
|
return VbanRtPacketNBS0.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:
|
||||||
"""
|
|
||||||
self.logger.debug(
|
|
||||||
'Received NB1 RTP Packet from %s, Size: %d bytes',
|
|
||||||
addr,
|
|
||||||
len(data),
|
|
||||||
)
|
|
||||||
"""
|
|
||||||
|
|
||||||
return VbanRtPacketNBS1.from_bytes(
|
return VbanRtPacketNBS1.from_bytes(
|
||||||
nbs=NBS.one, kind=self._remote.kind, data=data
|
nbs=NBS.one, kind=self._remote.kind, data=data
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user