From 8349a196e83c4b79bb3f3c323c82baac10a487fb Mon Sep 17 00:00:00 2001 From: onyx-and-iris Date: Thu, 17 Jul 2025 03:18:04 +0100 Subject: [PATCH] root meta app + scene sub_app converted this could take a while... note, --version flag not implemented --- obsws_cli/__init__.py | 4 +- obsws_cli/app.py | 218 ++++++++++++++---------------------------- obsws_cli/console.py | 7 +- obsws_cli/context.py | 15 +++ obsws_cli/scene.py | 75 ++++++++------- obsws_cli/validate.py | 32 +++---- pyproject.toml | 9 +- 7 files changed, 154 insertions(+), 206 deletions(-) create mode 100644 obsws_cli/context.py diff --git a/obsws_cli/__init__.py b/obsws_cli/__init__.py index 367b928..f991c56 100644 --- a/obsws_cli/__init__.py +++ b/obsws_cli/__init__.py @@ -2,6 +2,6 @@ # # SPDX-License-Identifier: MIT -from .app import app +from .app import run -__all__ = ['app'] +__all__ = ['run'] diff --git a/obsws_cli/app.py b/obsws_cli/app.py index 0e68533..c66f884 100644 --- a/obsws_cli/app.py +++ b/obsws_cli/app.py @@ -1,165 +1,89 @@ """Command line interface for the OBS WebSocket API.""" import importlib -import logging +from dataclasses import dataclass from typing import Annotated import obsws_python as obsws -import typer +from cyclopts import App, Group, Parameter, config -from obsws_cli.__about__ import __version__ as version +from . import console, styles +from .context import Context -from . import console, settings, styles -from .alias import RootTyperAliasGroup +app = App( + config=config.Env( + 'OBS_' + ), # Environment variable prefix for configuration parameters +) +app.meta.group_parameters = Group('Session Parameters', sort_key=0) +for sub_app in ('scene',): + module = importlib.import_module(f'.{sub_app}', package=__package__) + app.command(module.app) -app = typer.Typer(cls=RootTyperAliasGroup) -for sub_typer in ( - 'filter', - 'group', - 'hotkey', - 'input', - 'profile', - 'projector', - 'record', - 'replaybuffer', - 'scene', - 'scenecollection', - 'sceneitem', - 'screenshot', - 'stream', - 'studiomode', - 'text', - 'virtualcam', + +@Parameter(name='*') +@dataclass +class OBSConfig: + """Dataclass to hold OBS connection parameters.""" + + host: str = 'localhost' + port: int = 4455 + password: str = '' + + +@dataclass +class StyleConfig: + """Dataclass to hold style parameters.""" + + name: str = 'disabled' + no_border: bool = False + + +@app.meta.default +def launcher( + *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)], + obs_config: OBSConfig = Annotated[ + OBSConfig, + Parameter( + show=False, allow_leading_hyphen=True, help='OBS connection parameters' + ), + ], + style_config: StyleConfig = Annotated[ + StyleConfig, + Parameter(show=False, allow_leading_hyphen=True, help='Style parameters'), + ], ): - module = importlib.import_module(f'.{sub_typer}', package=__package__) - app.add_typer(module.app, name=sub_typer) + """Initialize the OBS WebSocket client and return the context.""" + with obsws.ReqClient( + host=obs_config.host, + port=obs_config.port, + password=obs_config.password, + ) as client: + additional_kwargs = {} + command, bound, ignored = app.parse_args(tokens) + if 'ctx' in ignored: + # If 'ctx' is in ignored, it means it was not passed as an argument + # and we need to add it to the bound arguments. + additional_kwargs['ctx'] = ignored['ctx']( + client, + styles.request_style_obj(style_config.name, style_config.no_border), + ) + return command(*bound.args, **bound.kwargs, **additional_kwargs) -def version_callback(value: bool): - """Show the version of the CLI.""" - if value: - console.out.print(f'obsws-cli version: {version}') - raise typer.Exit() - - -def setup_logging(debug: bool): - """Set up logging for the application.""" - log_level = logging.DEBUG if debug else logging.CRITICAL - logging.basicConfig( - level=log_level, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - ) - - -def validate_style(value: str): - """Validate and return the style.""" - if value not in styles.registry: - raise typer.BadParameter( - f'Invalid style: {value}. Available styles: {", ".join(styles.registry.keys())}' - ) - return value - - -@app.callback() -def main( - ctx: typer.Context, - host: Annotated[ - str, - typer.Option( - '--host', - '-H', - envvar='OBS_HOST', - help='WebSocket host', - show_default='localhost', - ), - ] = settings.get('host'), - port: Annotated[ - int, - typer.Option( - '--port', - '-P', - envvar='OBS_PORT', - help='WebSocket port', - show_default=4455, - ), - ] = settings.get('port'), - password: Annotated[ - str, - typer.Option( - '--password', - '-p', - envvar='OBS_PASSWORD', - help='WebSocket password', - show_default=False, - ), - ] = settings.get('password'), - timeout: Annotated[ - int, - typer.Option( - '--timeout', - '-T', - envvar='OBS_TIMEOUT', - help='WebSocket timeout', - show_default=5, - ), - ] = settings.get('timeout'), - version: Annotated[ - bool, - typer.Option( - '--version', - '-v', - is_eager=True, - help='Show the CLI version and exit', - show_default=False, - callback=version_callback, - ), - ] = False, - style: Annotated[ - str, - typer.Option( - '--style', - '-s', - envvar='OBS_STYLE', - help='Set the style for the CLI output', - show_default='disabled', - callback=validate_style, - ), - ] = settings.get('style'), - no_border: Annotated[ - bool, - typer.Option( - '--no-border', - '-b', - envvar='OBS_STYLE_NO_BORDER', - help='Disable table border styling in the CLI output', - show_default=False, - ), - ] = settings.get('style_no_border'), - debug: Annotated[ - bool, - typer.Option( - '--debug', - '-d', - envvar='OBS_DEBUG', - is_eager=True, - help='Enable debug logging', - show_default=False, - callback=setup_logging, - hidden=True, - ), - ] = settings.get('debug'), +@app.command +def obs_version( + *, + ctx: Annotated[Context, Parameter(parse=False)], ): - """obsws_cli is a command line interface for the OBS WebSocket API.""" - ctx.ensure_object(dict) - ctx.obj['obsws'] = ctx.with_resource(obsws.ReqClient(**ctx.params)) - ctx.obj['style'] = styles.request_style_obj(style, no_border) - - -@app.command() -def obs_version(ctx: typer.Context): """Get the OBS Client and WebSocket versions.""" - resp = ctx.obj['obsws'].get_version() + resp = ctx.client.get_version() console.out.print( f'OBS Client version: {console.highlight(ctx, resp.obs_version)}' f' with WebSocket version: {console.highlight(ctx, resp.obs_web_socket_version)}' ) + + +def run(): + """Run the OBS WebSocket CLI.""" + app.meta() diff --git a/obsws_cli/console.py b/obsws_cli/console.py index abaef64..528fad4 100644 --- a/obsws_cli/console.py +++ b/obsws_cli/console.py @@ -1,12 +1,13 @@ """module for console output handling in obsws_cli.""" -import typer from rich.console import Console +from .context import Context + out = Console() err = Console(stderr=True, style='bold red') -def highlight(ctx: typer.Context, text: str) -> str: +def highlight(ctx: Context, text: str) -> str: """Highlight text using the current context's style.""" - return f'[{ctx.obj["style"].highlight}]{text}[/{ctx.obj["style"].highlight}]' + return f'[{ctx.style.highlight}]{text}[/{ctx.style.highlight}]' diff --git a/obsws_cli/context.py b/obsws_cli/context.py new file mode 100644 index 0000000..7421040 --- /dev/null +++ b/obsws_cli/context.py @@ -0,0 +1,15 @@ +"""module for managing the application context.""" + +from dataclasses import dataclass + +import obsws_python as obsws + +from . import styles + + +@dataclass +class Context: + """Context for the application, holding OBS and style configurations.""" + + client: obsws.ReqClient + style: styles.Style diff --git a/obsws_cli/scene.py b/obsws_cli/scene.py index f52ced6..8f3c784 100644 --- a/obsws_cli/scene.py +++ b/obsws_cli/scene.py @@ -2,45 +2,41 @@ from typing import Annotated -import typer +from cyclopts import App, Argument, CycloptsError, Parameter from rich.table import Table from rich.text import Text from . import console, util, validate -from .alias import SubTyperAliasGroup +from .context import Context -app = typer.Typer(cls=SubTyperAliasGroup) +app = App(name='scene') -@app.callback() -def main(): - """Control OBS scenes.""" - - -@app.command('list | ls') +@app.command(name=['list', 'ls']) def list_( - ctx: typer.Context, - uuid: Annotated[bool, typer.Option(help='Show UUIDs of scenes')] = False, + uuid: Annotated[bool, Parameter(help='Show UUIDs of scenes')] = False, + *, + ctx: Annotated[Context, Parameter(parse=False)], ): """List all scenes.""" - resp = ctx.obj['obsws'].get_scene_list() + resp = ctx.client.get_scene_list() scenes = ( (scene.get('sceneName'), scene.get('sceneUuid')) for scene in reversed(resp.scenes) ) - active_scene = ctx.obj['obsws'].get_current_program_scene().scene_name + active_scene = ctx.client.get_current_program_scene().scene_name - table = Table(title='Scenes', padding=(0, 2), border_style=ctx.obj['style'].border) + table = Table(title='Scenes', padding=(0, 2), border_style=ctx.style.border) if uuid: columns = [ - (Text('Scene Name', justify='center'), 'left', ctx.obj['style'].column), + (Text('Scene Name', justify='center'), 'left', ctx.style.column), (Text('Active', justify='center'), 'center', None), - (Text('UUID', justify='center'), 'left', ctx.obj['style'].column), + (Text('UUID', justify='center'), 'left', ctx.style.column), ] else: columns = [ - (Text('Scene Name', justify='center'), 'left', ctx.obj['style'].column), + (Text('Scene Name', justify='center'), 'left', ctx.style.column), (Text('Active', justify='center'), 'center', None), ] for heading, justify, style in columns: @@ -62,57 +58,64 @@ def list_( console.out.print(table) -@app.command('current | get') +@app.command(name=['current', 'get']) def current( - ctx: typer.Context, preview: Annotated[ - bool, typer.Option(help='Get the preview scene instead of the program scene') + bool, Parameter(help='Get the preview scene instead of the program scene') ] = False, + *, + ctx: Annotated[Context, Parameter(parse=False)], ): """Get the current program scene or preview scene.""" if preview and not validate.studio_mode_enabled(ctx): - console.err.print('Studio mode is not enabled, cannot get preview scene.') - raise typer.Exit(1) + raise CycloptsError( + 'Studio mode is not enabled, cannot get preview scene.', + console=console.err, + ) if preview: - resp = ctx.obj['obsws'].get_current_preview_scene() + resp = ctx.client.get_current_preview_scene() console.out.print( f'Current Preview Scene: {console.highlight(ctx, resp.current_preview_scene_name)}' ) else: - resp = ctx.obj['obsws'].get_current_program_scene() + resp = ctx.client.get_current_program_scene() console.out.print( f'Current Program Scene: {console.highlight(ctx, resp.current_program_scene_name)}' ) -@app.command('switch | set') +@app.command(name=['switch', 'set']) def switch( - ctx: typer.Context, - scene_name: Annotated[ - str, typer.Argument(..., help='Name of the scene to switch to') - ], + scene_name: Annotated[str, Argument(hint='Name of the scene to switch to')], + /, preview: Annotated[ bool, - typer.Option(help='Switch to the preview scene instead of the program scene'), + Parameter(help='Switch to the preview scene instead of the program scene'), ] = False, + *, + ctx: Annotated[Context, Parameter(parse=False)], ): """Switch to a scene.""" if preview and not validate.studio_mode_enabled(ctx): - console.err.print('Studio mode is not enabled, cannot set the preview scene.') - raise typer.Exit(1) + raise CycloptsError( + 'Studio mode is not enabled, cannot set the preview scene.', + console=console.err, + ) if not validate.scene_in_scenes(ctx, scene_name): - console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.') - raise typer.Exit(1) + raise CycloptsError( + f'Scene [yellow]{scene_name}[/yellow] not found.', + console=console.err, + ) if preview: - ctx.obj['obsws'].set_current_preview_scene(scene_name) + ctx.client.set_current_preview_scene(scene_name) console.out.print( f'Switched to preview scene: {console.highlight(ctx, scene_name)}' ) else: - ctx.obj['obsws'].set_current_program_scene(scene_name) + ctx.client.set_current_program_scene(scene_name) console.out.print( f'Switched to program scene: {console.highlight(ctx, scene_name)}' ) diff --git a/obsws_cli/validate.py b/obsws_cli/validate.py index ebfec1b..1cc2653 100644 --- a/obsws_cli/validate.py +++ b/obsws_cli/validate.py @@ -2,53 +2,53 @@ import typer +from .context import Context + # type alias for an option that is skipped when the command is run skipped_option = typer.Option(parser=lambda _: _, hidden=True, expose_value=False) -def input_in_inputs(ctx: typer.Context, input_name: str) -> bool: +def input_in_inputs(ctx: Context, input_name: str) -> bool: """Check if an input is in the input list.""" - inputs = ctx.obj['obsws'].get_input_list().inputs + inputs = ctx.client.get_input_list().inputs return any(input_.get('inputName') == input_name for input_ in inputs) -def scene_in_scenes(ctx: typer.Context, scene_name: str) -> bool: +def scene_in_scenes(ctx: Context, scene_name: str) -> bool: """Check if a scene exists in the list of scenes.""" - resp = ctx.obj['obsws'].get_scene_list() + resp = ctx.client.get_scene_list() return any(scene.get('sceneName') == scene_name for scene in resp.scenes) -def studio_mode_enabled(ctx: typer.Context) -> bool: +def studio_mode_enabled(ctx: Context) -> bool: """Check if studio mode is enabled.""" - resp = ctx.obj['obsws'].get_studio_mode_enabled() + resp = ctx.client.get_studio_mode_enabled() return resp.studio_mode_enabled def scene_collection_in_scene_collections( - ctx: typer.Context, scene_collection_name: str + ctx: Context, scene_collection_name: str ) -> bool: """Check if a scene collection exists.""" - resp = ctx.obj['obsws'].get_scene_collection_list() + resp = ctx.client.get_scene_collection_list() return any( collection == scene_collection_name for collection in resp.scene_collections ) -def item_in_scene_item_list( - ctx: typer.Context, scene_name: str, item_name: str -) -> bool: +def item_in_scene_item_list(ctx: Context, scene_name: str, item_name: str) -> bool: """Check if an item exists in a scene.""" - resp = ctx.obj['obsws'].get_scene_item_list(scene_name) + resp = ctx.client.get_scene_item_list(scene_name) return any(item.get('sourceName') == item_name for item in resp.scene_items) -def profile_exists(ctx: typer.Context, profile_name: str) -> bool: +def profile_exists(ctx: Context, profile_name: str) -> bool: """Check if a profile exists.""" - resp = ctx.obj['obsws'].get_profile_list() + resp = ctx.client.get_profile_list() return any(profile == profile_name for profile in resp.profiles) -def monitor_exists(ctx: typer.Context, monitor_index: int) -> bool: +def monitor_exists(ctx: Context, monitor_index: int) -> bool: """Check if a monitor exists.""" - resp = ctx.obj['obsws'].get_monitor_list() + resp = ctx.client.get_monitor_list() return any(monitor['monitorIndex'] == monitor_index for monitor in resp.monitors) diff --git a/pyproject.toml b/pyproject.toml index 845c26b..2262c87 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,12 @@ classifiers = [ "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy", ] -dependencies = ["typer>=0.16.0", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"] +dependencies = [ + "cyclopts>=3.22.2", + "typer>=0.16.0", + "obsws-python>=1.8.0", + "python-dotenv>=1.1.0", +] [project.urls] @@ -30,7 +35,7 @@ Issues = "https://github.com/onyx-and-iris/obsws-cli/issues" Source = "https://github.com/onyx-and-iris/obsws-cli" [project.scripts] -obsws-cli = "obsws_cli:app" +obsws-cli = "obsws_cli:run" [tool.hatch.version] path = "obsws_cli/__about__.py"