rename to vban-cli

src/vban_cli/__init__.py

@@ -0,0 +1,3 @@
+from importlib.metadata import version
+
+__version__ = version('vmr-cli')

src/vban_cli/app.py

@@ -0,0 +1,53 @@
+from dataclasses import dataclass
+from typing import Annotated
+
+import vban_cmd
+from cyclopts import App, Parameter, config
+
+from . import __version__ as version
+from . import bus, console, strip
+from .context import Context
+
+app = App(
+    config=config.Env(
+        'VMR_CLI_',
+    ),  # Environment variable prefix for configuration parameters
+    version=version,
+)
+app.command(strip.app.meta, name='strip')
+app.command(bus.app.meta, name='bus')
+
+
+@Parameter(name='*')
+@dataclass
+class VBANConfig:
+    kind: Annotated[str, Parameter(help='Kind of Voicemeeter')] = 'potato'
+    host: Annotated[str, Parameter(help='VBAN host')] = 'localhost'
+    port: Annotated[int, Parameter(help='VBAN port')] = 6980
+    streamname: Annotated[str, Parameter(help='VBAN stream name')] = 'Command1'
+
+
+@app.meta.default
+def launcher(
+    *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
+    vban_config: Annotated[VBANConfig, Parameter()] = VBANConfig(),
+):
+    with vban_cmd.api(
+        vban_config.kind,
+        ip=vban_config.host,
+        port=vban_config.port,
+        streamname=vban_config.streamname,
+    ) as client:
+        additional_kwargs = {}
+        command, bound, _ = app.parse_args(tokens)
+        additional_kwargs['ctx'] = Context(client=client)
+
+        return command(*bound.args, **bound.kwargs, **additional_kwargs)
+
+
+def run():
+    try:
+        app.meta()
+    except Exception as e:
+        console.err.print(f'Error: {e}')
+        return e.code

src/vban_cli/bus.py

@@ -0,0 +1,107 @@
+from typing import Annotated, Literal, Optional
+
+from cyclopts import App, Argument, Parameter
+
+from . import console
+from .context import Context
+from .help import CustomHelpFormatter
+
+app = App(name='bus', help_formatter=CustomHelpFormatter())
+# The VBAN protocl does not yet define a packet structure for these bus parameters.
+# Hopefully that will come in the form of a 'VBAN_VMPARAMBUS_PACKET'.
+# app.command(eq.app.meta, name='eq')
+
+
+@app.meta.default
+def launcher(
+    index: Annotated[int, Argument()] = None,
+    *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Control the bus 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
+
+    return command(*bound.args, **bound.kwargs, **additional_kwargs)
+
+
+@app.command(name='mono')
+def mono(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the mono state of the specified bus.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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(ctx.client.bus[index].mono)
+        return
+    ctx.client.bus[index].mono = new_value
+
+
+@app.command(name='mute')
+def mute(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the mute state of the specified bus.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.bus[index].mute = new_value
+
+
+@app.command(name='mode')
+def mode(
+    type_: Annotated[
+        Optional[
+            Literal[
+                'normal',
+                'amix',
+                'bmix',
+                'repeat',
+                'composite',
+                'tvmix',
+                'upmix21',
+                'upmix41',
+                'upmix61',
+                'centeronly',
+                'lfeonly',
+                'rearonly',
+            ]
+        ],
+        Argument(),
+    ] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the bus mode of the specified bus.
+
+    Parameters
+    ----------
+    type_ : str, optional
+        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())
+        return
+    setattr(ctx.client.bus[index].mode, type_, True)

src/vban_cli/console.py

@@ -0,0 +1,4 @@
+from rich.console import Console
+
+out = Console()
+err = Console(stderr=True)

src/vban_cli/context.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+from vban_cmd.vbancmd import VbanCmd
+
+
+@dataclass
+class Context:
+    client: VbanCmd

src/vban_cli/eq.py

@@ -0,0 +1,53 @@
+from typing import Annotated
+
+from cyclopts import App, Argument, Parameter
+
+from .context import Context
+from .help import CustomHelpFormatter
+
+app = App(name='eq', help_formatter=CustomHelpFormatter())
+
+
+@app.meta.default
+def launcher(
+    band: Annotated[int, Argument()] = None,
+    *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
+    index: Annotated[int, Argument()] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Control the EQ parameters.
+
+    Only channel 0 is supported, as the VBAN protocol only exposes parameters for this channel.
+    """
+    additional_kwargs = {}
+    command, bound, _ = app.parse_args(tokens)
+    if index is not None:
+        additional_kwargs['index'] = index
+    if band is not None:
+        additional_kwargs['band'] = band
+    if ctx is not None:
+        additional_kwargs['ctx'] = ctx
+
+    return command(*bound.args, **bound.kwargs, **additional_kwargs)
+
+
+@app.command(name='on')
+def on(
+    new_state: Annotated[bool, Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    band: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the on state of the specified EQ band.
+
+    Parameters
+    ----------
+    new_state : bool
+        If provided, sets the on state to this value. If not provided, the current on state is printed.
+    """
+    if new_state is None:
+        # This doesn't work because the VBAN protocol doesn't send an initial NBS1 packet.
+        # console.out.print(ctx.client.strip[index].eq.channel[0].cell[band].on)
+        return
+    ctx.client.strip[index].eq.channel[0].cell[band].on = new_state

src/vban_cli/help.py

@@ -0,0 +1,43 @@
+import re
+
+from cyclopts.help import DefaultFormatter, HelpPanel
+from rich.console import Console, ConsoleOptions
+
+
+class CustomHelpFormatter(DefaultFormatter):
+    """Custom help formatter that injects an index argument into the usage line and filters it out from the parameters list.
+
+    This formatter modifies the usage line to include an <index> argument after the 'strip' command,
+    and filters out any parameters related to 'index' from the Parameters section of the help output.
+    """
+
+    def render_usage(self, console: Console, options: ConsoleOptions, usage) -> None:
+        """Render the usage line with index argument injected."""
+        if usage:
+            modified_usage = re.sub(
+                r'(\S+\s+[a-z]+)\s+(COMMAND)', r'\1 <index> \2', str(usage)
+            )
+            console.print(f'[bold]Usage:[/bold] {modified_usage}')
+
+    def __call__(
+        self, console: Console, options: ConsoleOptions, panel: HelpPanel
+    ) -> None:
+        """Render a help panel, filtering out the index parameter from Parameters sections."""
+        if panel.title == 'Parameters':
+            filtered_entries = [
+                entry
+                for entry in panel.entries
+                if not (
+                    entry.names and any('index' in name.lower() for name in entry.names)
+                )
+            ]
+
+            filtered_panel = HelpPanel(
+                title=panel.title,
+                entries=filtered_entries,
+                description=panel.description,
+                format=panel.format,
+            )
+            super().__call__(console, options, filtered_panel)
+        else:
+            super().__call__(console, options, panel)

src/vban_cli/strip.py

@@ -0,0 +1,267 @@
+from typing import Annotated, Optional
+
+from cyclopts import App, Argument, Parameter
+
+from . import console, eq
+from .context import Context
+from .help import CustomHelpFormatter
+
+app = App(name='strip', help_formatter=CustomHelpFormatter())
+app.command(eq.app.meta, name='eq')
+
+
+@app.meta.default
+def launcher(
+    index: Annotated[int, Argument()] = None,
+    *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Control the strip 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
+
+    return command(*bound.args, **bound.kwargs, **additional_kwargs)
+
+
+@app.command(name='mono')
+def mono(
+    new_state: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the mono state of the specified strip.
+
+    Parameters
+    ----------
+    new_state : bool, optional
+        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)
+        return
+    ctx.client.strip[index].mono = new_state
+
+
+@app.command(name='solo')
+def solo(
+    new_state: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the solo state of the specified strip.
+
+    Parameters
+    ----------
+    new_state : bool, optional
+        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)
+        return
+    ctx.client.strip[index].solo = new_state
+
+
+@app.command(name='mute')
+def mute(
+    new_state: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the mute state of the specified strip.
+
+    Parameters
+    ----------
+    new_state : bool, optional
+        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)
+        return
+    ctx.client.strip[index].mute = new_state
+
+
+@app.command(name='gain')
+def gain(
+    new_value: Annotated[Optional[float], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the gain of the specified strip.
+
+    Parameters
+    ----------
+    new_value : float, optional
+        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)
+        return
+    ctx.client.strip[index].gain = new_value
+
+
+@app.command(name='A1')
+def a1(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the A1 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].A1 = new_value
+
+
+@app.command(name='A2')
+def a2(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the A2 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].A2 = new_value
+
+
+@app.command(name='A3')
+def a3(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the A3 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].A3 = new_value
+
+
+@app.command(name='A4')
+def a4(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the A4 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].A4 = new_value
+
+
+@app.command(name='A5')
+def a5(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the A5 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].A5 = new_value
+
+
+@app.command(name='B1')
+def b1(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the B1 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].B1 = new_value
+
+
+@app.command(name='B2')
+def b2(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the B2 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].B2 = new_value
+
+
+@app.command(name='B3')
+def b3(
+    new_value: Annotated[Optional[bool], Argument()] = None,
+    *,
+    index: Annotated[int, Parameter(show=False)] = None,
+    ctx: Annotated[Context, Parameter(show=False)] = None,
+):
+    """Get or set the B3 state of the specified strip.
+
+    Parameters
+    ----------
+    new_value : bool, optional
+        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)
+        return
+    ctx.client.strip[index].B3 = new_value