keep the exit codes simple (0 or 1)

run() now handles exceptions/exit codes

.github/workflows/publish.yml

@@ -0,0 +1,39 @@
+name: Publish to PyPI
+
+on:
+  release:
+    types: [published]
+  push:
+    tags:
+      - 'v*.*.*'
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+
+    environment: pypi
+    permissions:
+      # This permission is needed for private repositories.
+      contents: read
+      # IMPORTANT: this permission is mandatory for trusted publishing
+      id-token: write
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.11'
+          cache: 'pip'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip 
+          pip install hatch
+
+      - name: Build package
+        run: hatch build
+
+      - name: Publish on PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1

.github/workflows/ruff.yml

@@ -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@v4
+      - uses: astral-sh/ruff-action@v3
+        with:
+          args: 'format --check --diff'

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+# [0.20.0] - 2025-07-14
+
+### Added
+
+-   text command group, see [Text](https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#text)
+
 # [0.19.0] - 2025-06-23
 
 ### Added

README.md

@@ -335,6 +335,22 @@ obsws-cli input unmute "Mic/Aux"
 obsws-cli input toggle "Mic/Aux"
 ```
 
+#### Text
+
+-   current: Get the current text for a text input.
+    -   args: <input_name>
+
+```console
+obsws-cli text current "My Text Input"
+```
+
+-   update: Update the text of a text input.
+    -   args: <input_name> <new_text>
+
+```console
+obsws-cli text update "My Text Input" "hi OBS!"
+```
+
 #### Record
 
 -   start: Start recording.

obsws_cli/__about__.py

@@ -1,4 +1,4 @@
 # SPDX-FileCopyrightText: 2025-present onyx-and-iris <code@onyxandiris.online>
 #
 # SPDX-License-Identifier: MIT
-__version__ = "0.19.1"
+__version__ = '0.20.2'

obsws_cli/__init__.py

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

obsws_cli/alias.py

@@ -44,6 +44,8 @@ class RootTyperAliasGroup(typer.core.TyperGroup):
                 cmd_name = 'stream'
             case 'sm':
                 cmd_name = 'studiomode'
+            case 't':
+                cmd_name = 'text'
             case 'vc':
                 cmd_name = 'virtualcam'
         return super().get_command(ctx, cmd_name)

obsws_cli/app.py

@@ -2,163 +2,116 @@
 
 import importlib
 import logging
-from typing import Annotated
+from dataclasses import dataclass
+from typing import Annotated, Any
 
 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, settings, styles
+from . import console, styles
-from .alias import RootTyperAliasGroup
+from .context import Context
+from .error import OBSWSCLIError
 
-app = typer.Typer(cls=RootTyperAliasGroup)
+app = App(
-for sub_typer in (
+    config=config.Env(
+        'OBS_'
+    ),  # Environment variable prefix for configuration parameters
+    version=version,
+)
+app.meta.group_parameters = Group('Session Parameters', sort_key=0)
+for sub_app in (
     'filter',
-    'group',
-    'hotkey',
-    'input',
-    'profile',
-    'projector',
-    'record',
-    'replaybuffer',
     'scene',
-    'scenecollection',
-    'sceneitem',
-    'screenshot',
-    'stream',
-    'studiomode',
-    'virtualcam',
 ):
-    module = importlib.import_module(f'.{sub_typer}', package=__package__)
+    module = importlib.import_module(f'.{sub_app}', package=__package__)
-    app.add_typer(module.app, name=sub_typer)
+    app.command(module.app)
 
 
-def version_callback(value: bool):
+@Parameter(name='*')
-    """Show the version of the CLI."""
+@dataclass
-    if value:
+class OBSConfig:
-        console.out.print(f'obsws-cli version: {version}')
+    """Dataclass to hold OBS connection parameters."""
-        raise typer.Exit()
+
+    host: str = 'localhost'
+    port: int = 4455
+    password: str = ''
 
 
-def setup_logging(debug: bool):
+@dataclass
+class StyleConfig:
+    """Dataclass to hold style parameters."""
+
+    name: str = 'disabled'
+    no_border: bool = False
+
+
+def setup_logging(type_, value: Any):
     """Set up logging for the application."""
-    log_level = logging.DEBUG if debug else logging.CRITICAL
+    log_level = logging.DEBUG if value else logging.CRITICAL
     logging.basicConfig(
         level=log_level,
         format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
     )
 
 
-def validate_style(value: str):
+@app.meta.default
-    """Validate and return the style."""
+def launcher(
-    if value not in styles.registry:
+    *tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
-        raise typer.BadParameter(
+    obs_config: OBSConfig = Annotated[
-            f'Invalid style: {value}. Available styles: {", ".join(styles.registry.keys())}'
+        OBSConfig,
-        )
+        Parameter(
-    return value
+            show=False, allow_leading_hyphen=True, help='OBS connection parameters'
-
-
-@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[
+    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(
+        Parameter(validator=setup_logging),
-            '--debug',
+    ] = False,
-            '-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."""
-    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 application.
+
+    Handles exceptions and prints error messages to the console.
+    """
+    try:
+        app.meta()
+    except OBSWSCLIError as e:
+        console.err.print(f'Error: {e}')
+        return e.code

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}]'

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

obsws_cli/enum.py

@@ -0,0 +1,10 @@
+"""module for exit codes used in the application."""
+
+from enum import IntEnum, auto
+
+
+class ExitCode(IntEnum):
+    """Exit codes for the application."""
+
+    SUCCESS = 0
+    ERROR = auto()

obsws_cli/error.py

@@ -0,0 +1,11 @@
+"""module containing error handling for OBS WebSocket CLI."""
+
+
+class OBSWSCLIError(Exception):
+    """Base class for OBS WebSocket CLI errors."""
+
+    def __init__(self, message: str, code: int = 1):
+        """Initialize the error with a message and an optional code."""
+        super().__init__(message)
+        self.message = message
+        self.code = code

obsws_cli/filter.py

@@ -3,44 +3,42 @@
 from typing import Annotated, Optional
 
 import obsws_python as obsws
-import typer
+from cyclopts import App, Argument, Parameter
 from rich.table import Table
 from rich.text import Text
 
 from . import console, util
-from .alias import SubTyperAliasGroup
+from .context import Context
+from .enum import ExitCode
+from .error import OBSWSCLIError
 
-app = typer.Typer(cls=SubTyperAliasGroup)
+app = App(name='filter')
 
 
-@app.callback()
+@app.command(name=['list', 'ls'])
-def main():
-    """Control filters in OBS scenes."""
-
-
-@app.command('list | ls')
 def list_(
-    ctx: typer.Context,
     source_name: Annotated[
         Optional[str],
-        typer.Argument(
+        Argument(
-            show_default='The current scene',
+            hint='The source to list filters for',
-            help='The source to list filters for',
         ),
     ] = None,
+    /,
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
 ):
     """List filters for a source."""
     if not source_name:
-        source_name = ctx.obj['obsws'].get_current_program_scene().scene_name
+        source_name = ctx.client.get_current_program_scene().scene_name
 
     try:
-        resp = ctx.obj['obsws'].get_source_filter_list(source_name)
+        resp = ctx.client.get_source_filter_list(source_name)
     except obsws.error.OBSSDKRequestError as e:
         if e.code == 600:
-            console.err.print(
+            raise OBSWSCLIError(
-                f'No source was found by the name of [yellow]{source_name}[/yellow].'
+                f'No source found by the name of [yellow]{source_name}[/yellow].',
+                code=ExitCode.ERROR,
             )
-            raise typer.Exit(1)
         else:
             raise
 
@@ -48,25 +46,25 @@ def list_(
         console.out.print(
             f'No filters found for source {console.highlight(ctx, source_name)}'
         )
-        raise typer.Exit()
+        return
 
     table = Table(
         title=f'Filters for Source: {source_name}',
         padding=(0, 2),
-        border_style=ctx.obj['style'].border,
+        border_style=ctx.style.border,
     )
 
     columns = [
-        (Text('Filter Name', justify='center'), 'left', ctx.obj['style'].column),
+        (Text('Filter Name', justify='center'), 'left', ctx.style.column),
-        (Text('Kind', justify='center'), 'left', ctx.obj['style'].column),
+        (Text('Kind', justify='center'), 'left', ctx.style.column),
         (Text('Enabled', justify='center'), 'center', None),
-        (Text('Settings', justify='center'), 'center', ctx.obj['style'].column),
+        (Text('Settings', justify='center'), 'center', ctx.style.column),
     ]
     for heading, justify, style in columns:
         table.add_column(heading, justify=justify, style=style)
 
     for filter in resp.filters:
-        resp = ctx.obj['obsws'].get_source_filter_default_settings(filter['filterKind'])
+        resp = ctx.client.get_source_filter_default_settings(filter['filterKind'])
         settings = resp.default_filter_settings | filter['filterSettings']
 
         table.add_row(
@@ -84,93 +82,85 @@ def list_(
     console.out.print(table)
 
 
-def _get_filter_enabled(ctx: typer.Context, source_name: str, filter_name: str):
+def _get_filter_enabled(ctx: Context, source_name: str, filter_name: str):
     """Get the status of a filter for a source."""
-    resp = ctx.obj['obsws'].get_source_filter(source_name, filter_name)
+    resp = ctx.client.get_source_filter(source_name, filter_name)
     return resp.filter_enabled
 
 
-@app.command('enable | on')
+@app.command(name=['enable', 'on'])
 def enable(
-    ctx: typer.Context,
     source_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The source to enable the filter for'),
-            ..., show_default=False, help='The source to enable the filter for'
-        ),
     ],
     filter_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The name of the filter to enable'),
-            ..., show_default=False, help='The name of the filter to enable'
-        ),
     ],
+    /,
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
 ):
     """Enable a filter for a source."""
     if _get_filter_enabled(ctx, source_name, filter_name):
-        console.err.print(
+        raise OBSWSCLIError(
-            f'Filter [yellow]{filter_name}[/yellow] is already enabled for source [yellow]{source_name}[/yellow]'
+            f'Filter [yellow]{filter_name}[/yellow] is already enabled for source [yellow]{source_name}[/yellow]',
+            code=ExitCode.ERROR,
         )
-        raise typer.Exit(1)
 
-    ctx.obj['obsws'].set_source_filter_enabled(source_name, filter_name, enabled=True)
+    ctx.client.set_source_filter_enabled(source_name, filter_name, enabled=True)
     console.out.print(
         f'Enabled filter {console.highlight(ctx, filter_name)} for source {console.highlight(ctx, source_name)}'
     )
 
 
-@app.command('disable | off')
+@app.command(name=['disable', 'off'])
 def disable(
-    ctx: typer.Context,
     source_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The source to disable the filter for'),
-            ..., show_default=False, help='The source to disable the filter for'
-        ),
     ],
     filter_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The name of the filter to disable'),
-            ..., show_default=False, help='The name of the filter to disable'
-        ),
     ],
+    /,
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
 ):
     """Disable a filter for a source."""
     if not _get_filter_enabled(ctx, source_name, filter_name):
-        console.err.print(
+        raise OBSWSCLIError(
-            f'Filter [yellow]{filter_name}[/yellow] is already disabled for source [yellow]{source_name}[/yellow]'
+            f'Filter [yellow]{filter_name}[/yellow] is already disabled for source [yellow]{source_name}[/yellow]',
+            code=ExitCode.ERROR,
         )
-        raise typer.Exit(1)
 
-    ctx.obj['obsws'].set_source_filter_enabled(source_name, filter_name, enabled=False)
+    ctx.client.set_source_filter_enabled(source_name, filter_name, enabled=False)
     console.out.print(
         f'Disabled filter {console.highlight(ctx, filter_name)} for source {console.highlight(ctx, source_name)}'
     )
 
 
-@app.command('toggle | tg')
+@app.command(name=['toggle', 'tg'])
 def toggle(
-    ctx: typer.Context,
     source_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The source to toggle the filter for'),
-            ..., show_default=False, help='The source to toggle the filter for'
-        ),
     ],
     filter_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The name of the filter to toggle'),
-            ..., show_default=False, help='The name of the filter to toggle'
-        ),
     ],
+    /,
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
 ):
     """Toggle a filter for a source."""
     is_enabled = _get_filter_enabled(ctx, source_name, filter_name)
     new_state = not is_enabled
 
-    ctx.obj['obsws'].set_source_filter_enabled(
+    ctx.client.set_source_filter_enabled(source_name, filter_name, enabled=new_state)
-        source_name, filter_name, enabled=new_state
-    )
     if new_state:
         console.out.print(
             f'Enabled filter {console.highlight(ctx, filter_name)} for source {console.highlight(ctx, source_name)}'
@@ -181,21 +171,19 @@ def toggle(
         )
 
 
-@app.command('status | ss')
+@app.command(name=['status', 'ss'])
 def status(
-    ctx: typer.Context,
     source_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The source to get the filter status for'),
-            ..., show_default=False, help='The source to get the filter status for'
-        ),
     ],
     filter_name: Annotated[
         str,
-        typer.Argument(
+        Argument(hint='The name of the filter to get the status for'),
-            ..., show_default=False, help='The name of the filter to get the status for'
-        ),
     ],
+    /,
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
 ):
     """Get the status of a filter for a source."""
     is_enabled = _get_filter_enabled(ctx, source_name, filter_name)

obsws_cli/scene.py

@@ -2,45 +2,43 @@
 
 from typing import Annotated
 
-import typer
+from cyclopts import App, Argument, Parameter
 from rich.table import Table
 from rich.text import Text
 
 from . import console, util, validate
-from .alias import SubTyperAliasGroup
+from .context import Context
+from .enum import ExitCode
+from .error import OBSWSCLIError
 
-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_(
-    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."""
-    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 +60,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 OBSWSCLIError(
-        raise typer.Exit(1)
+            'Studio mode is not enabled, cannot get preview scene.',
+            code=ExitCode.ERROR,
+        )
 
     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, 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[
         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 OBSWSCLIError(
-        raise typer.Exit(1)
+            'Studio mode is not enabled, cannot switch to preview scene.',
+            code=ExitCode.ERROR,
+        )
 
     if not validate.scene_in_scenes(ctx, scene_name):
-        console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
+        raise OBSWSCLIError(
-        raise typer.Exit(1)
+            f'Scene [yellow]{scene_name}[/yellow] not found.',
+            code=ExitCode.ERROR,
+        )
 
     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)}'
         )

obsws_cli/settings.py

@@ -16,6 +16,7 @@ class Settings(UserDict):
     The settings are expected to be in uppercase and should start with 'OBS_'.
 
     Example:
+    -------
         settings = Settings()
         host = settings['OBS_HOST']
         settings['OBS_PORT'] = 4455
@@ -64,12 +65,15 @@ def get(key: str) -> SettingsValue:
     """Get a setting value by key.
 
     Args:
+    ----
         key (str): The key of the setting to retrieve.
 
     Returns:
+    -------
         The value of the setting.
 
     Raises:
+    ------
         KeyError: If the key does not exist in the settings.
 
     """

obsws_cli/text.py

@@ -0,0 +1,78 @@
+"""module containing commands for manipulating text inputs."""
+
+from typing import Annotated, Optional
+
+import typer
+
+from . import console, validate
+from .alias import SubTyperAliasGroup
+
+app = typer.Typer(cls=SubTyperAliasGroup)
+
+
+@app.callback()
+def main():
+    """Control text inputs in OBS."""
+
+
+@app.command('current | get')
+def current(
+    ctx: typer.Context,
+    input_name: Annotated[str, typer.Argument(help='Name of the text input to get.')],
+):
+    """Get the current text for a text input."""
+    if not validate.input_in_inputs(ctx, input_name):
+        console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
+        raise typer.Exit(1)
+
+    resp = ctx.obj['obsws'].get_input_settings(name=input_name)
+    if not resp.input_kind.startswith('text_'):
+        console.err.print(
+            f'Input [yellow]{input_name}[/yellow] is not a text input.',
+        )
+        raise typer.Exit(1)
+
+    current_text = resp.input_settings.get('text', '')
+    if not current_text:
+        current_text = '(empty)'
+    console.out.print(
+        f'Current text for input {console.highlight(ctx, input_name)}: {current_text}',
+    )
+
+
+@app.command('update | set')
+def update(
+    ctx: typer.Context,
+    input_name: Annotated[
+        str, typer.Argument(help='Name of the text input to update.')
+    ],
+    new_text: Annotated[
+        Optional[str],
+        typer.Argument(
+            help='The new text to set for the input.',
+        ),
+    ] = None,
+):
+    """Update the text of a text input."""
+    if not validate.input_in_inputs(ctx, input_name):
+        console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
+        raise typer.Exit(1)
+
+    resp = ctx.obj['obsws'].get_input_settings(name=input_name)
+    if not resp.input_kind.startswith('text_'):
+        console.err.print(
+            f'Input [yellow]{input_name}[/yellow] is not a text input.',
+        )
+        raise typer.Exit(1)
+
+    ctx.obj['obsws'].set_input_settings(
+        name=input_name,
+        settings={'text': new_text},
+        overlay=True,
+    )
+
+    if not new_text:
+        new_text = '(empty)'
+    console.out.print(
+        f'Text for input {console.highlight(ctx, input_name)} updated to: {new_text}',
+    )

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(
+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."""
-    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)

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"

tests/conftest.py

@@ -71,6 +71,13 @@ def pytest_sessionstart(session):
         },
         sceneItemEnabled=True,
     )
+    session.obsws.create_input(
+        sceneName='pytest_scene',
+        inputName='pytest_text_input',
+        inputKind='text_gdiplus_v3',
+        inputSettings={'text': 'Hello, OBS!'},
+        sceneItemEnabled=True,
+    )
     resp = session.obsws.get_scene_item_list('pytest_scene')
     for item in resp.scene_items:
         if item['sourceName'] == 'pytest_input_2':

tests/test_text.py

@@ -0,0 +1,18 @@
+"""Unit tests for the text command in the OBS WebSocket CLI."""
+
+from typer.testing import CliRunner
+
+from obsws_cli.app import app
+
+runner = CliRunner(mix_stderr=False)
+
+
+def test_text_update():
+    """Test the text update command."""
+    result = runner.invoke(app, ['text', 'current', 'pytest_text_input'])
+    assert result.exit_code == 0
+    assert 'Current text for input pytest_text_input: Hello, OBS!' in result.stdout
+
+    result = runner.invoke(app, ['text', 'update', 'pytest_text_input', 'New Text'])
+    assert result.exit_code == 0
+    assert 'Text for input pytest_text_input updated to: New Text' in result.stdout