mirror of
https://github.com/onyx-and-iris/vban-cli.git
synced 2026-03-02 21:19:11 +00:00
Compare commits
8 Commits
a9c3168542
...
e84accd37a
| Author | SHA1 | Date | |
|---|---|---|---|
| e84accd37a | |||
| d9810ce270 | |||
| 4fff581c95 | |||
| 1f6811d5a0 | |||
| a1da5c7256 | |||
| 642337d987 | |||
| 51002edb39 | |||
| d30c9f828d |
@ -144,9 +144,9 @@ examples:
|
||||
```console
|
||||
vban-cli recorder play
|
||||
|
||||
vban-cli recorder rew
|
||||
vban-cli recorder pause
|
||||
|
||||
vban-cli recorder replay
|
||||
vban-cli recorder goto "00:01:30"
|
||||
```
|
||||
|
||||
see `vban-cli recorder --help` for more info.
|
||||
@ -180,6 +180,7 @@ see `vban-cli sendtext --help` for more info.
|
||||
1. The VBAN RT SERVICE subprotocol defines two packet structures [ident:0][ident-0] and [ident:1][ident-1]. Neither of them contain the data for Bus EQ parameters.
|
||||
2. Packet structure with [ident:1][ident-1] is emitted by the VBAN server only on pdirty events. This means we do not receive the current state of those parameters on initial subscription. Therefore any commands which are intended to fetch the value of parameters defined in packet [ident:1][ident-1] will not work in this CLI.
|
||||
3. Packet structure with [ident:1][ident-1] defines parameteric EQ data only for the [first channel][ident-1-peq].
|
||||
4. There doesn't appear to be any way to retrieve the current recorder status, ie, recording, playing, stopped etc. I don't see the data available in either packet structures [ident:0][ident-0] or [ident:1][ident-1].
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "vban-cli"
|
||||
version = "0.9.0"
|
||||
version = "0.9.3"
|
||||
description = "A command-line interface for Voicemeeter leveraging VBAN."
|
||||
readme = "README.md"
|
||||
license = { text = "LICENSE" }
|
||||
|
||||
@ -3,16 +3,21 @@ from typing import Annotated
|
||||
|
||||
import vban_cmd
|
||||
from cyclopts import App, Argument, Parameter, config
|
||||
from rich.traceback import install as install_rich_traceback
|
||||
|
||||
from . import __version__ as version
|
||||
from . import bus, command, console, recorder, strip
|
||||
from .context import Context
|
||||
from .error import VbanCLIConnectionError
|
||||
|
||||
app = App(
|
||||
config=config.Env(
|
||||
'VBAN_CLI_',
|
||||
), # Environment variable prefix for configuration parameters
|
||||
version=version,
|
||||
console=console.out,
|
||||
error_console=console.err,
|
||||
exit_on_error=True,
|
||||
)
|
||||
app.command(strip.app.meta, name='strip')
|
||||
app.command(bus.app.meta, name='bus')
|
||||
@ -20,6 +25,8 @@ app.command(command.app, name='command')
|
||||
app.command(recorder.app, name='recorder')
|
||||
app.register_install_completion_command()
|
||||
|
||||
install_rich_traceback(console=console.err)
|
||||
|
||||
|
||||
@Parameter(name='*')
|
||||
@dataclass
|
||||
@ -43,14 +50,17 @@ def launcher(
|
||||
if command.__name__ == 'sendtext':
|
||||
disable_rt_listeners = True
|
||||
|
||||
with vban_cmd.api(
|
||||
vban_config.kind,
|
||||
ip=vban_config.host,
|
||||
port=vban_config.port,
|
||||
streamname=vban_config.streamname,
|
||||
disable_rt_listeners=disable_rt_listeners,
|
||||
) as client:
|
||||
return command(*bound.args, **bound.kwargs, ctx=Context(client=client))
|
||||
try:
|
||||
with vban_cmd.api(
|
||||
vban_config.kind,
|
||||
ip=vban_config.host,
|
||||
port=vban_config.port,
|
||||
streamname=vban_config.streamname,
|
||||
disable_rt_listeners=disable_rt_listeners,
|
||||
) as client:
|
||||
return command(*bound.args, **bound.kwargs, ctx=Context(client=client))
|
||||
except vban_cmd.error.VBANCMDConnectionError as e:
|
||||
raise VbanCLIConnectionError(str(e)) from e
|
||||
|
||||
|
||||
@app.command(name='sendtext')
|
||||
@ -58,16 +68,12 @@ def sendtext(
|
||||
text: Annotated[str, Argument()],
|
||||
/,
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Send a text command to the current Voicemeeter/Matrix instance."""
|
||||
if resp := ctx.client.sendtext(text):
|
||||
console.out.print(resp)
|
||||
app.console.print(resp)
|
||||
|
||||
|
||||
def run():
|
||||
try:
|
||||
app.meta()
|
||||
except Exception as e:
|
||||
console.err.print(f'Error: {e}')
|
||||
return e.code
|
||||
app.meta()
|
||||
|
||||
@ -2,7 +2,6 @@ from typing import Annotated, Literal, Optional
|
||||
|
||||
from cyclopts import App, Argument, Parameter
|
||||
|
||||
from . import console
|
||||
from .context import Context
|
||||
from .help import BusHelpFormatter
|
||||
|
||||
@ -13,19 +12,18 @@ app = App(name='bus', help_formatter=BusHelpFormatter())
|
||||
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
index: Annotated[int, Argument()] = None,
|
||||
index: Annotated[int, Argument()],
|
||||
/,
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the bus parameters."""
|
||||
additional_kwargs = {}
|
||||
command, bound, _ = app.parse_args(tokens)
|
||||
if tokens[0] == 'eq':
|
||||
additional_kwargs['eq_kind'] = app.name[0]
|
||||
if index is not None:
|
||||
additional_kwargs['index'] = index
|
||||
if ctx is not None:
|
||||
additional_kwargs['ctx'] = ctx
|
||||
additional_kwargs['index'] = index
|
||||
additional_kwargs['ctx'] = ctx
|
||||
|
||||
return command(*bound.args, **bound.kwargs, **additional_kwargs)
|
||||
|
||||
@ -36,8 +34,8 @@ def mono(
|
||||
Optional[Literal['off', 'mono', 'stereoreverse']], Argument()
|
||||
] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the mono state of the specified bus.
|
||||
|
||||
@ -47,7 +45,7 @@ def mono(
|
||||
If provided, sets the mono state to this value. If not provided, the current mono state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(['off', 'mono', 'stereoreverse'][ctx.client.bus[index].mono])
|
||||
app.console.print(['off', 'mono', 'stereoreverse'][ctx.client.bus[index].mono])
|
||||
return
|
||||
ctx.client.bus[index].mono = ['off', 'mono', 'stereoreverse'].index(new_value)
|
||||
|
||||
@ -56,8 +54,8 @@ def mono(
|
||||
def mute(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the mute state of the specified bus.
|
||||
|
||||
@ -67,7 +65,7 @@ def mute(
|
||||
If provided, sets the mute state to this value. If not provided, the current mute state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.bus[index].mute)
|
||||
app.console.print(ctx.client.bus[index].mute)
|
||||
return
|
||||
ctx.client.bus[index].mute = new_value
|
||||
|
||||
@ -94,8 +92,8 @@ def mode(
|
||||
Argument(),
|
||||
] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the bus mode of the specified bus.
|
||||
|
||||
@ -105,6 +103,6 @@ def mode(
|
||||
If provided, sets the bus mode to this value. If not provided, the current bus mode is printed.
|
||||
"""
|
||||
if type_ is None:
|
||||
console.out.print(ctx.client.bus[index].mode.get())
|
||||
app.console.print(ctx.client.bus[index].mode.get())
|
||||
return
|
||||
setattr(ctx.client.bus[index].mode, type_, True)
|
||||
|
||||
@ -2,7 +2,6 @@ from typing import Annotated
|
||||
|
||||
from cyclopts import App, Parameter
|
||||
|
||||
from . import console
|
||||
from .context import Context
|
||||
from .help import BaseHelpFormatter
|
||||
|
||||
@ -12,38 +11,38 @@ app = App(name='command', help_formatter=BaseHelpFormatter())
|
||||
@app.command(name='show')
|
||||
def show(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)] = None,
|
||||
):
|
||||
"""Bring the Voicemeeter GUI to the foreground."""
|
||||
ctx.client.command.show()
|
||||
console.out.print('Voicemeeter GUI should now be in the foreground.')
|
||||
app.console.print('Voicemeeter GUI should now be in the foreground.')
|
||||
|
||||
|
||||
@app.command(name='hide')
|
||||
def hide(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Send the Voicemeeter GUI to the background."""
|
||||
ctx.client.command.hide()
|
||||
console.out.print('Voicemeeter GUI should now be in the background.')
|
||||
app.console.print('Voicemeeter GUI should now be in the background.')
|
||||
|
||||
|
||||
@app.command(name='shutdown')
|
||||
def shutdown(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Shut down Voicemeeter."""
|
||||
ctx.client.command.shutdown()
|
||||
console.out.print('Voicemeeter should now be shut down.')
|
||||
app.console.print('Voicemeeter should now be shut down.')
|
||||
|
||||
|
||||
@app.command(name='restart')
|
||||
def restart(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Restart the Voicemeeter engine."""
|
||||
ctx.client.command.restart()
|
||||
console.out.print('Voicemeeter engine should now be restarting.')
|
||||
app.console.print('Voicemeeter engine should now be restarting.')
|
||||
|
||||
@ -11,16 +11,14 @@ app = App(name='comp', help_formatter=StripHelpFormatter())
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
index: Annotated[int, Argument()] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the compressor parameters."""
|
||||
additional_kwargs = {}
|
||||
command, bound, _ = app.parse_args(tokens)
|
||||
if index is not None:
|
||||
additional_kwargs['index'] = index
|
||||
if ctx is not None:
|
||||
additional_kwargs['ctx'] = ctx
|
||||
additional_kwargs['index'] = index
|
||||
additional_kwargs['ctx'] = ctx
|
||||
|
||||
return command(*bound.args, **bound.kwargs, **additional_kwargs)
|
||||
|
||||
@ -29,8 +27,8 @@ def launcher(
|
||||
def knob(
|
||||
new_knob: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the knob of the specified compressor.
|
||||
|
||||
@ -41,7 +39,7 @@ def knob(
|
||||
"""
|
||||
if new_knob is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(ctx.client.strip[index].comp.knob)
|
||||
# app.console.print(ctx.client.strip[index].comp.knob)
|
||||
return
|
||||
ctx.client.strip[index].comp.knob = new_knob
|
||||
|
||||
@ -50,8 +48,8 @@ def knob(
|
||||
def input_gain(
|
||||
new_gain: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the input gain of the specified compressor.
|
||||
|
||||
@ -62,6 +60,6 @@ def input_gain(
|
||||
"""
|
||||
if new_gain is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(ctx.client.strip[index].comp.gainin)
|
||||
# app.console.print(ctx.client.strip[index].comp.gainin)
|
||||
return
|
||||
ctx.client.strip[index].comp.gainin = new_gain
|
||||
|
||||
@ -11,16 +11,14 @@ app = App(name='denoiser', help_formatter=StripHelpFormatter())
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
index: Annotated[int, Argument()] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the denoiser parameters."""
|
||||
additional_kwargs = {}
|
||||
command, bound, _ = app.parse_args(tokens)
|
||||
if index is not None:
|
||||
additional_kwargs['index'] = index
|
||||
if ctx is not None:
|
||||
additional_kwargs['ctx'] = ctx
|
||||
additional_kwargs['index'] = index
|
||||
additional_kwargs['ctx'] = ctx
|
||||
|
||||
return command(*bound.args, **bound.kwargs, **additional_kwargs)
|
||||
|
||||
@ -29,8 +27,8 @@ def launcher(
|
||||
def knob(
|
||||
new_knob: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the knob of the specified denoiser.
|
||||
|
||||
@ -41,6 +39,6 @@ def knob(
|
||||
"""
|
||||
if new_knob is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(ctx.client.strip[index].denoiser.knob)
|
||||
# app.console.print(ctx.client.strip[index].denoiser.knob)
|
||||
return
|
||||
ctx.client.strip[index].denoiser.knob = new_knob
|
||||
|
||||
@ -14,9 +14,9 @@ app.command(cell_app.meta, name='cell')
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
eq_kind: Annotated[str, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Argument()] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
eq_kind: Annotated[str, Parameter(parse=False)],
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the EQ parameters."""
|
||||
additional_kwargs = {}
|
||||
@ -37,7 +37,7 @@ def launcher(
|
||||
def on(
|
||||
new_state: Annotated[bool, Argument()] = None,
|
||||
*,
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the on state of the specified EQ band.
|
||||
|
||||
@ -48,16 +48,17 @@ def on(
|
||||
"""
|
||||
if new_state is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(target.on)
|
||||
# app.console.print(target.on)
|
||||
return
|
||||
target.on = new_state
|
||||
|
||||
|
||||
@cell_app.meta.default
|
||||
def cell_launcher(
|
||||
band: Annotated[int, Argument()] = None,
|
||||
band: Annotated[int, Argument()],
|
||||
/,
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the EQ Cell parameters.
|
||||
|
||||
@ -74,7 +75,7 @@ def cell_launcher(
|
||||
def cell_on(
|
||||
new_state: Annotated[bool, Argument()] = None,
|
||||
*,
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the on state of the specified EQ cell.
|
||||
|
||||
@ -85,7 +86,7 @@ def cell_on(
|
||||
"""
|
||||
if new_state is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(target.on)
|
||||
# app.console.print(target.on)
|
||||
return
|
||||
target.on = new_state
|
||||
|
||||
@ -94,7 +95,7 @@ def cell_on(
|
||||
def cell_freq(
|
||||
new_freq: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the frequency of the specified EQ cell.
|
||||
|
||||
@ -105,7 +106,7 @@ def cell_freq(
|
||||
"""
|
||||
if new_freq is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(target.f)
|
||||
# app.console.print(target.f)
|
||||
return
|
||||
target.f = new_freq
|
||||
|
||||
@ -114,7 +115,7 @@ def cell_freq(
|
||||
def cell_gain(
|
||||
new_gain: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the gain of the specified EQ cell.
|
||||
|
||||
@ -125,7 +126,7 @@ def cell_gain(
|
||||
"""
|
||||
if new_gain is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(target.gain)
|
||||
# app.console.print(target.gain)
|
||||
return
|
||||
target.gain = new_gain
|
||||
|
||||
@ -134,7 +135,7 @@ def cell_gain(
|
||||
def cell_q(
|
||||
new_q: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the Q of the specified EQ cell.
|
||||
|
||||
@ -145,7 +146,7 @@ def cell_q(
|
||||
"""
|
||||
if new_q is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(target.q)
|
||||
# app.console.print(target.q)
|
||||
return
|
||||
target.q = new_q
|
||||
|
||||
@ -154,7 +155,7 @@ def cell_q(
|
||||
def cell_type(
|
||||
new_type: Annotated[int, Argument()] = None,
|
||||
*,
|
||||
target: Annotated[object, Parameter(show=False)] = None,
|
||||
target: Annotated[object, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the type of the specified EQ cell.
|
||||
|
||||
@ -165,6 +166,6 @@ def cell_type(
|
||||
"""
|
||||
if new_type is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(target.type)
|
||||
# app.console.print(target.type)
|
||||
return
|
||||
target.type = new_type
|
||||
|
||||
14
src/vban_cli/error.py
Normal file
14
src/vban_cli/error.py
Normal file
@ -0,0 +1,14 @@
|
||||
class VbanCLIError(Exception):
|
||||
"""Base class for exceptions in this module."""
|
||||
|
||||
def __init__(self, message: str, code: int = 1):
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
|
||||
|
||||
class VbanCLIConnectionError(VbanCLIError):
|
||||
"""Exception raised for connection errors."""
|
||||
|
||||
|
||||
class VbanCLIValidationError(VbanCLIError):
|
||||
"""Exception raised for validation errors."""
|
||||
@ -2,7 +2,6 @@ from typing import Annotated
|
||||
|
||||
from cyclopts import App, Argument, Parameter
|
||||
|
||||
from . import console
|
||||
from .context import Context
|
||||
from .help import GainlayerHelpFormatter
|
||||
|
||||
@ -11,21 +10,18 @@ app = App(name='gainlayer', help_formatter=GainlayerHelpFormatter())
|
||||
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
gainlayer_index: Annotated[int, Argument()] = None,
|
||||
gainlayer_index: Annotated[int, Argument()],
|
||||
/,
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
index: Annotated[int, Argument()] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the gainlayers."""
|
||||
additional_kwargs = {}
|
||||
command, bound, _ = app.parse_args(tokens)
|
||||
if index is not None and gainlayer_index is not None:
|
||||
additional_kwargs['strip_index'] = index
|
||||
additional_kwargs['gainlayer_index'] = gainlayer_index
|
||||
else:
|
||||
raise ValueError('Both gainlayer_index and index must be provided.')
|
||||
if ctx is not None:
|
||||
additional_kwargs['ctx'] = ctx
|
||||
additional_kwargs['strip_index'] = index
|
||||
additional_kwargs['gainlayer_index'] = gainlayer_index
|
||||
additional_kwargs['ctx'] = ctx
|
||||
|
||||
return command(*bound.args, **bound.kwargs, **additional_kwargs)
|
||||
|
||||
@ -34,9 +30,9 @@ def launcher(
|
||||
def level(
|
||||
new_level: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
strip_index: Annotated[int, Parameter(show=False)] = None,
|
||||
gainlayer_index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
strip_index: Annotated[int, Parameter(parse=False)],
|
||||
gainlayer_index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the level of the specified gainlayer.
|
||||
|
||||
@ -46,6 +42,6 @@ def level(
|
||||
If provided, sets the level to this value. If not provided, the current level is printed.
|
||||
"""
|
||||
if new_level is None:
|
||||
console.out.print(ctx.client.strip[strip_index].gainlayer[gainlayer_index].gain)
|
||||
app.console.print(ctx.client.strip[strip_index].gainlayer[gainlayer_index].gain)
|
||||
return
|
||||
ctx.client.strip[strip_index].gainlayer[gainlayer_index].gain = new_level
|
||||
|
||||
@ -11,16 +11,14 @@ app = App(name='gate', help_formatter=StripHelpFormatter())
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
index: Annotated[int, Argument()] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the compressor parameters."""
|
||||
additional_kwargs = {}
|
||||
command, bound, _ = app.parse_args(tokens)
|
||||
if index is not None:
|
||||
additional_kwargs['index'] = index
|
||||
if ctx is not None:
|
||||
additional_kwargs['ctx'] = ctx
|
||||
additional_kwargs['index'] = index
|
||||
additional_kwargs['ctx'] = ctx
|
||||
|
||||
return command(*bound.args, **bound.kwargs, **additional_kwargs)
|
||||
|
||||
@ -29,8 +27,8 @@ def launcher(
|
||||
def knob(
|
||||
new_knob: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the knob of the specified gate.
|
||||
|
||||
@ -41,7 +39,7 @@ def knob(
|
||||
"""
|
||||
if new_knob is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(ctx.client.strip[index].gate.knob)
|
||||
# app.console.print(ctx.client.strip[index].gate.knob)
|
||||
return
|
||||
ctx.client.strip[index].gate.knob = new_knob
|
||||
|
||||
@ -50,8 +48,8 @@ def knob(
|
||||
def threshold(
|
||||
new_threshold: Annotated[float, Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the threshold of the specified gate.
|
||||
|
||||
@ -62,6 +60,6 @@ def threshold(
|
||||
"""
|
||||
if new_threshold is None:
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 2.
|
||||
# console.out.print(ctx.client.strip[index].gate.threshold)
|
||||
# app.console.print(ctx.client.strip[index].gate.threshold)
|
||||
return
|
||||
ctx.client.strip[index].gate.threshold = new_threshold
|
||||
|
||||
@ -3,7 +3,7 @@ from typing import Annotated
|
||||
|
||||
from cyclopts import App, Parameter, validators
|
||||
|
||||
from . import console, validation
|
||||
from . import validation
|
||||
from .context import Context
|
||||
from .help import BaseHelpFormatter
|
||||
|
||||
@ -13,71 +13,84 @@ app = App(name='recorder', help_formatter=BaseHelpFormatter())
|
||||
@app.command(name='play')
|
||||
def play(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Start the recorder playback."""
|
||||
ctx.client.recorder.play()
|
||||
console.out.print('Recorder playback started.')
|
||||
|
||||
|
||||
@app.command(name='stop')
|
||||
def stop(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
):
|
||||
"""Stop the recorder playback."""
|
||||
ctx.client.recorder.stop()
|
||||
console.out.print('Recorder playback stopped.')
|
||||
app.console.print('Recorder playback started.')
|
||||
|
||||
|
||||
@app.command(name='pause')
|
||||
def pause(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Pause the recorder playback."""
|
||||
ctx.client.recorder.pause()
|
||||
console.out.print('Recorder playback paused.')
|
||||
ctx.client.recorder.stop()
|
||||
app.console.print('Recorder playback paused.')
|
||||
|
||||
|
||||
@app.command(name='stop')
|
||||
def stop(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Stop the recorder playback/recording and reset to the beginning."""
|
||||
ctx.client.recorder.stop()
|
||||
ctx.client.recorder.goto('00:00:00')
|
||||
# We have no way of knowing if the recorder was playing or recording, so we print a generic message.
|
||||
# See https://github.com/onyx-and-iris/vban-cli?tab=readme-ov-file#implementation-notes - 4.
|
||||
app.console.print('Recorder stopped.')
|
||||
|
||||
|
||||
@app.command(name='replay')
|
||||
def replay(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Replay the recorder playback."""
|
||||
ctx.client.recorder.replay()
|
||||
console.out.print('Recorder playback replay started.')
|
||||
app.console.print('Recorder playback replay started.')
|
||||
|
||||
|
||||
@app.command(name='record')
|
||||
def record(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Start recording."""
|
||||
ctx.client.recorder.record()
|
||||
console.out.print('Recording started.')
|
||||
app.console.print('Recorder recording started.')
|
||||
|
||||
|
||||
@app.command(name='pause-recording')
|
||||
def pause_recording(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Pause the recorder recording."""
|
||||
ctx.client.recorder.pause()
|
||||
app.console.print('Recorder recording paused.')
|
||||
|
||||
|
||||
@app.command(name='ff')
|
||||
def ff(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Fast forward the recorder playback."""
|
||||
ctx.client.recorder.ff()
|
||||
console.out.print('Recorder playback fast forwarded.')
|
||||
app.console.print('Recorder playback fast forwarded.')
|
||||
|
||||
|
||||
@app.command(name='rew')
|
||||
def rew(
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Rewind the recorder playback."""
|
||||
ctx.client.recorder.rew()
|
||||
console.out.print('Recorder playback rewound.')
|
||||
app.console.print('Recorder playback rewound.')
|
||||
|
||||
|
||||
@app.command(name='load')
|
||||
@ -91,13 +104,13 @@ def load(
|
||||
],
|
||||
/,
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Load a file into the recorder.
|
||||
|
||||
note: This command may only work if vban-cli is running on localhost and may not work if vban-cli is running on a remote server."""
|
||||
ctx.client.recorder.load(file_path)
|
||||
console.out.print(f'Loaded file: {file_path}')
|
||||
app.console.print(f'Loaded file: {file_path}')
|
||||
|
||||
|
||||
@app.command(name='goto')
|
||||
@ -111,8 +124,8 @@ def goto(
|
||||
],
|
||||
/,
|
||||
*,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Go to a specific timestamp in the recorder playback."""
|
||||
ctx.client.recorder.goto(time_string)
|
||||
console.out.print(f'Went to timestamp {time_string} in recorder playback.')
|
||||
app.console.print(f'Went to timestamp {time_string} in recorder playback.')
|
||||
|
||||
@ -2,7 +2,7 @@ from typing import Annotated, Optional
|
||||
|
||||
from cyclopts import App, Argument, Parameter
|
||||
|
||||
from . import comp, console, denoiser, eq, gainlayer, gate
|
||||
from . import comp, denoiser, eq, gainlayer, gate
|
||||
from .context import Context
|
||||
from .help import StripHelpFormatter
|
||||
|
||||
@ -16,19 +16,18 @@ app.command(gainlayer.app.meta, name='gainlayer')
|
||||
|
||||
@app.meta.default
|
||||
def launcher(
|
||||
index: Annotated[int, Argument()] = None,
|
||||
index: Annotated[int, Argument()],
|
||||
/,
|
||||
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Control the strip parameters."""
|
||||
additional_kwargs = {}
|
||||
command, bound, _ = app.parse_args(tokens)
|
||||
if tokens[0] == 'eq':
|
||||
additional_kwargs['eq_kind'] = app.name[0]
|
||||
if index is not None:
|
||||
additional_kwargs['index'] = index
|
||||
if ctx is not None:
|
||||
additional_kwargs['ctx'] = ctx
|
||||
additional_kwargs['index'] = index
|
||||
additional_kwargs['ctx'] = ctx
|
||||
|
||||
return command(*bound.args, **bound.kwargs, **additional_kwargs)
|
||||
|
||||
@ -37,8 +36,8 @@ def launcher(
|
||||
def mono(
|
||||
new_state: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the mono state of the specified strip.
|
||||
|
||||
@ -48,7 +47,7 @@ def mono(
|
||||
If provided, sets the mono state to this value. If not provided, the current mono state is printed.
|
||||
"""
|
||||
if new_state is None:
|
||||
console.out.print(ctx.client.strip[index].mono)
|
||||
app.console.print(ctx.client.strip[index].mono)
|
||||
return
|
||||
ctx.client.strip[index].mono = new_state
|
||||
|
||||
@ -57,8 +56,8 @@ def mono(
|
||||
def solo(
|
||||
new_state: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the solo state of the specified strip.
|
||||
|
||||
@ -68,7 +67,7 @@ def solo(
|
||||
If provided, sets the solo state to this value. If not provided, the current solo state is printed.
|
||||
"""
|
||||
if new_state is None:
|
||||
console.out.print(ctx.client.strip[index].solo)
|
||||
app.console.print(ctx.client.strip[index].solo)
|
||||
return
|
||||
ctx.client.strip[index].solo = new_state
|
||||
|
||||
@ -77,8 +76,8 @@ def solo(
|
||||
def mute(
|
||||
new_state: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the mute state of the specified strip.
|
||||
|
||||
@ -88,7 +87,7 @@ def mute(
|
||||
If provided, sets the mute state to this value. If not provided, the current mute state is printed.
|
||||
"""
|
||||
if new_state is None:
|
||||
console.out.print(ctx.client.strip[index].mute)
|
||||
app.console.print(ctx.client.strip[index].mute)
|
||||
return
|
||||
ctx.client.strip[index].mute = new_state
|
||||
|
||||
@ -97,8 +96,8 @@ def mute(
|
||||
def gain(
|
||||
new_value: Annotated[Optional[float], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the gain of the specified strip.
|
||||
|
||||
@ -108,7 +107,7 @@ def gain(
|
||||
If provided, sets the gain to this value. If not provided, the current gain is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].gain)
|
||||
app.console.print(ctx.client.strip[index].gain)
|
||||
return
|
||||
ctx.client.strip[index].gain = new_value
|
||||
|
||||
@ -117,8 +116,8 @@ def gain(
|
||||
def a1(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the A1 state of the specified strip.
|
||||
|
||||
@ -128,7 +127,7 @@ def a1(
|
||||
If provided, sets the A1 state to this value. If not provided, the current A1 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].A1)
|
||||
app.console.print(ctx.client.strip[index].A1)
|
||||
return
|
||||
ctx.client.strip[index].A1 = new_value
|
||||
|
||||
@ -137,8 +136,8 @@ def a1(
|
||||
def a2(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the A2 state of the specified strip.
|
||||
|
||||
@ -148,7 +147,7 @@ def a2(
|
||||
If provided, sets the A2 state to this value. If not provided, the current A2 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].A2)
|
||||
app.console.print(ctx.client.strip[index].A2)
|
||||
return
|
||||
ctx.client.strip[index].A2 = new_value
|
||||
|
||||
@ -157,8 +156,8 @@ def a2(
|
||||
def a3(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the A3 state of the specified strip.
|
||||
|
||||
@ -168,7 +167,7 @@ def a3(
|
||||
If provided, sets the A3 state to this value. If not provided, the current A3 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].A3)
|
||||
app.console.print(ctx.client.strip[index].A3)
|
||||
return
|
||||
ctx.client.strip[index].A3 = new_value
|
||||
|
||||
@ -177,8 +176,8 @@ def a3(
|
||||
def a4(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the A4 state of the specified strip.
|
||||
|
||||
@ -188,7 +187,7 @@ def a4(
|
||||
If provided, sets the A4 state to this value. If not provided, the current A4 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].A4)
|
||||
app.console.print(ctx.client.strip[index].A4)
|
||||
return
|
||||
ctx.client.strip[index].A4 = new_value
|
||||
|
||||
@ -197,8 +196,8 @@ def a4(
|
||||
def a5(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the A5 state of the specified strip.
|
||||
|
||||
@ -208,7 +207,7 @@ def a5(
|
||||
If provided, sets the A5 state to this value. If not provided, the current A5 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].A5)
|
||||
app.console.print(ctx.client.strip[index].A5)
|
||||
return
|
||||
ctx.client.strip[index].A5 = new_value
|
||||
|
||||
@ -217,8 +216,8 @@ def a5(
|
||||
def b1(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the B1 state of the specified strip.
|
||||
|
||||
@ -228,7 +227,7 @@ def b1(
|
||||
If provided, sets the B1 state to this value. If not provided, the current B1 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].B1)
|
||||
app.console.print(ctx.client.strip[index].B1)
|
||||
return
|
||||
ctx.client.strip[index].B1 = new_value
|
||||
|
||||
@ -237,8 +236,8 @@ def b1(
|
||||
def b2(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the B2 state of the specified strip.
|
||||
|
||||
@ -248,7 +247,7 @@ def b2(
|
||||
If provided, sets the B2 state to this value. If not provided, the current B2 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].B2)
|
||||
app.console.print(ctx.client.strip[index].B2)
|
||||
return
|
||||
ctx.client.strip[index].B2 = new_value
|
||||
|
||||
@ -257,8 +256,8 @@ def b2(
|
||||
def b3(
|
||||
new_value: Annotated[Optional[bool], Argument()] = None,
|
||||
*,
|
||||
index: Annotated[int, Parameter(show=False)] = None,
|
||||
ctx: Annotated[Context, Parameter(show=False)] = None,
|
||||
index: Annotated[int, Parameter(parse=False)],
|
||||
ctx: Annotated[Context, Parameter(parse=False)],
|
||||
):
|
||||
"""Get or set the B3 state of the specified strip.
|
||||
|
||||
@ -268,6 +267,6 @@ def b3(
|
||||
If provided, sets the B3 state to this value. If not provided, the current B3 state is printed.
|
||||
"""
|
||||
if new_value is None:
|
||||
console.out.print(ctx.client.strip[index].B3)
|
||||
app.console.print(ctx.client.strip[index].B3)
|
||||
return
|
||||
ctx.client.strip[index].B3 = new_value
|
||||
|
||||
@ -1,9 +1,11 @@
|
||||
import re
|
||||
|
||||
from .error import VbanCLIValidationError
|
||||
|
||||
|
||||
def is_valid_time_string(type_, value: str) -> str:
|
||||
"""Validate if the given string is a valid time format (HH:MM:SS)."""
|
||||
pattern = r'^(?:[01]\d|2[0123]):(?:[012345]\d):(?:[012345]\d)$'
|
||||
if not re.match(pattern, value):
|
||||
raise ValueError('Invalid time format. Expected HH:MM:SS.')
|
||||
raise VbanCLIValidationError('Invalid time format. Expected HH:MM:SS.')
|
||||
return value
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user