root meta app + scene sub_app converted

this could take a while...

note, --version flag not implemented
This commit is contained in:
onyx-and-iris 2025-07-17 03:18:04 +01:00
parent f852a733c3
commit 8349a196e8
7 changed files with 154 additions and 206 deletions

View File

@ -2,6 +2,6 @@
# #
# SPDX-License-Identifier: MIT # SPDX-License-Identifier: MIT
from .app import app from .app import run
__all__ = ['app'] __all__ = ['run']

View File

@ -1,165 +1,89 @@
"""Command line interface for the OBS WebSocket API.""" """Command line interface for the OBS WebSocket API."""
import importlib import importlib
import logging from dataclasses import dataclass
from typing import Annotated from typing import Annotated
import obsws_python as obsws 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 app = App(
from .alias import RootTyperAliasGroup config=config.Env(
'OBS_'
app = typer.Typer(cls=RootTyperAliasGroup) ), # Environment variable prefix for configuration parameters
for sub_typer in (
'filter',
'group',
'hotkey',
'input',
'profile',
'projector',
'record',
'replaybuffer',
'scene',
'scenecollection',
'sceneitem',
'screenshot',
'stream',
'studiomode',
'text',
'virtualcam',
):
module = importlib.import_module(f'.{sub_typer}', package=__package__)
app.add_typer(module.app, name=sub_typer)
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',
) )
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)
def validate_style(value: str): @Parameter(name='*')
"""Validate and return the style.""" @dataclass
if value not in styles.registry: class OBSConfig:
raise typer.BadParameter( """Dataclass to hold OBS connection parameters."""
f'Invalid style: {value}. Available styles: {", ".join(styles.registry.keys())}'
) host: str = 'localhost'
return value port: int = 4455
password: str = ''
@app.callback() @dataclass
def main( class StyleConfig:
ctx: typer.Context, """Dataclass to hold style parameters."""
host: Annotated[
str, name: str = 'disabled'
typer.Option( no_border: bool = False
'--host',
'-H',
envvar='OBS_HOST', @app.meta.default
help='WebSocket host', def launcher(
show_default='localhost', *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'
), ),
] = settings.get('host'), ],
port: Annotated[ style_config: StyleConfig = Annotated[
int, StyleConfig,
typer.Option( Parameter(show=False, allow_leading_hyphen=True, help='Style parameters'),
'--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'),
): ):
"""obsws_cli is a command line interface for the OBS WebSocket API.""" """Initialize the OBS WebSocket client and return the context."""
ctx.ensure_object(dict) with obsws.ReqClient(
ctx.obj['obsws'] = ctx.with_resource(obsws.ReqClient(**ctx.params)) host=obs_config.host,
ctx.obj['style'] = styles.request_style_obj(style, no_border) 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)
@app.command() @app.command
def obs_version(ctx: typer.Context): def obs_version(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the OBS Client and WebSocket versions.""" """Get the OBS Client and WebSocket versions."""
resp = ctx.obj['obsws'].get_version() resp = ctx.client.get_version()
console.out.print( console.out.print(
f'OBS Client version: {console.highlight(ctx, resp.obs_version)}' f'OBS Client version: {console.highlight(ctx, resp.obs_version)}'
f' with WebSocket version: {console.highlight(ctx, resp.obs_web_socket_version)}' f' with WebSocket version: {console.highlight(ctx, resp.obs_web_socket_version)}'
) )
def run():
"""Run the OBS WebSocket CLI."""
app.meta()

View File

@ -1,12 +1,13 @@
"""module for console output handling in obsws_cli.""" """module for console output handling in obsws_cli."""
import typer
from rich.console import Console from rich.console import Console
from .context import Context
out = Console() out = Console()
err = Console(stderr=True, style='bold red') 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.""" """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}]'

15
obsws_cli/context.py Normal file
View File

@ -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

View File

@ -2,45 +2,41 @@
from typing import Annotated from typing import Annotated
import typer from cyclopts import App, Argument, CycloptsError, Parameter
from rich.table import Table from rich.table import Table
from rich.text import Text from rich.text import Text
from . import console, util, validate from . import console, util, validate
from .alias import SubTyperAliasGroup from .context import Context
app = typer.Typer(cls=SubTyperAliasGroup) app = App(name='scene')
@app.callback() @app.command(name=['list', 'ls'])
def main():
"""Control OBS scenes."""
@app.command('list | ls')
def list_( def list_(
ctx: typer.Context, uuid: Annotated[bool, Parameter(help='Show UUIDs of scenes')] = False,
uuid: Annotated[bool, typer.Option(help='Show UUIDs of scenes')] = False, *,
ctx: Annotated[Context, Parameter(parse=False)],
): ):
"""List all scenes.""" """List all scenes."""
resp = ctx.obj['obsws'].get_scene_list() resp = ctx.client.get_scene_list()
scenes = ( scenes = (
(scene.get('sceneName'), scene.get('sceneUuid')) (scene.get('sceneName'), scene.get('sceneUuid'))
for scene in reversed(resp.scenes) 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: if uuid:
columns = [ 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('Active', justify='center'), 'center', None),
(Text('UUID', justify='center'), 'left', ctx.obj['style'].column), (Text('UUID', justify='center'), 'left', ctx.style.column),
] ]
else: else:
columns = [ 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('Active', justify='center'), 'center', None),
] ]
for heading, justify, style in columns: for heading, justify, style in columns:
@ -62,57 +58,64 @@ def list_(
console.out.print(table) console.out.print(table)
@app.command('current | get') @app.command(name=['current', 'get'])
def current( def current(
ctx: typer.Context,
preview: Annotated[ 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, ] = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
): ):
"""Get the current program scene or preview scene.""" """Get the current program scene or preview scene."""
if preview and not validate.studio_mode_enabled(ctx): if preview and not validate.studio_mode_enabled(ctx):
console.err.print('Studio mode is not enabled, cannot get preview scene.') raise CycloptsError(
raise typer.Exit(1) 'Studio mode is not enabled, cannot get preview scene.',
console=console.err,
)
if preview: if preview:
resp = ctx.obj['obsws'].get_current_preview_scene() resp = ctx.client.get_current_preview_scene()
console.out.print( console.out.print(
f'Current Preview Scene: {console.highlight(ctx, resp.current_preview_scene_name)}' f'Current Preview Scene: {console.highlight(ctx, resp.current_preview_scene_name)}'
) )
else: else:
resp = ctx.obj['obsws'].get_current_program_scene() resp = ctx.client.get_current_program_scene()
console.out.print( console.out.print(
f'Current Program Scene: {console.highlight(ctx, resp.current_program_scene_name)}' f'Current Program Scene: {console.highlight(ctx, resp.current_program_scene_name)}'
) )
@app.command('switch | set') @app.command(name=['switch', 'set'])
def switch( def switch(
ctx: typer.Context, scene_name: Annotated[str, Argument(hint='Name of the scene to switch to')],
scene_name: Annotated[ /,
str, typer.Argument(..., help='Name of the scene to switch to')
],
preview: Annotated[ preview: Annotated[
bool, 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, ] = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
): ):
"""Switch to a scene.""" """Switch to a scene."""
if preview and not validate.studio_mode_enabled(ctx): if preview and not validate.studio_mode_enabled(ctx):
console.err.print('Studio mode is not enabled, cannot set the preview scene.') raise CycloptsError(
raise typer.Exit(1) 'Studio mode is not enabled, cannot set the preview scene.',
console=console.err,
)
if not validate.scene_in_scenes(ctx, scene_name): if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.') raise CycloptsError(
raise typer.Exit(1) f'Scene [yellow]{scene_name}[/yellow] not found.',
console=console.err,
)
if preview: if preview:
ctx.obj['obsws'].set_current_preview_scene(scene_name) ctx.client.set_current_preview_scene(scene_name)
console.out.print( console.out.print(
f'Switched to preview scene: {console.highlight(ctx, scene_name)}' f'Switched to preview scene: {console.highlight(ctx, scene_name)}'
) )
else: else:
ctx.obj['obsws'].set_current_program_scene(scene_name) ctx.client.set_current_program_scene(scene_name)
console.out.print( console.out.print(
f'Switched to program scene: {console.highlight(ctx, scene_name)}' f'Switched to program scene: {console.highlight(ctx, scene_name)}'
) )

View File

@ -2,53 +2,53 @@
import typer import typer
from .context import Context
# type alias for an option that is skipped when the command is run # type alias for an option that is skipped when the command is run
skipped_option = typer.Option(parser=lambda _: _, hidden=True, expose_value=False) 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.""" """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) 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.""" """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) 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.""" """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 return resp.studio_mode_enabled
def scene_collection_in_scene_collections( def scene_collection_in_scene_collections(
ctx: typer.Context, scene_collection_name: str ctx: Context, scene_collection_name: str
) -> bool: ) -> bool:
"""Check if a scene collection exists.""" """Check if a scene collection exists."""
resp = ctx.obj['obsws'].get_scene_collection_list() resp = ctx.client.get_scene_collection_list()
return any( return any(
collection == scene_collection_name for collection in resp.scene_collections collection == scene_collection_name for collection in resp.scene_collections
) )
def item_in_scene_item_list( def item_in_scene_item_list(ctx: Context, scene_name: str, item_name: str) -> bool:
ctx: typer.Context, scene_name: str, item_name: str
) -> bool:
"""Check if an item exists in a scene.""" """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) 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.""" """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) 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.""" """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) return any(monitor['monitorIndex'] == monitor_index for monitor in resp.monitors)

View File

@ -21,7 +21,12 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy", "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] [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" Source = "https://github.com/onyx-and-iris/obsws-cli"
[project.scripts] [project.scripts]
obsws-cli = "obsws_cli:app" obsws-cli = "obsws_cli:run"
[tool.hatch.version] [tool.hatch.version]
path = "obsws_cli/__about__.py" path = "obsws_cli/__about__.py"