convert stream commands

obsws_cli/app.py

@@ -35,6 +35,7 @@ for sub_app in (
     'scenecollection',
     'sceneitem',
     'screenshot',
+    'stream',
 ):
     module = importlib.import_module(f'.{sub_app}', package=__package__)
     app.command(module.app)

obsws_cli/stream.py

@@ -1,60 +1,75 @@
 """module for controlling OBS stream functionality."""
 
-import typer
+from typing import Annotated
+
+from cyclopts import App, Parameter
 
 from . import console
-from .alias import SubTyperAliasGroup
+from .context import Context
+from .enum import ExitCode
+from .error import OBSWSCLIError
 
-app = typer.Typer(cls=SubTyperAliasGroup)
+app = App(name='stream', help='Commands for controlling OBS stream functionality.')
 
 
-@app.callback()
+def _get_streaming_status(ctx: Context) -> tuple:
-def main():
-    """Control OBS stream functionality."""
-
-
-def _get_streaming_status(ctx: typer.Context) -> tuple:
     """Get streaming status."""
-    resp = ctx.obj['obsws'].get_stream_status()
+    resp = ctx.client.get_stream_status()
     return resp.output_active, resp.output_duration
 
 
-@app.command('start | s')
+@app.command(name=['start', 's'])
-def start(ctx: typer.Context):
+def start(
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
+):
     """Start streaming."""
     active, _ = _get_streaming_status(ctx)
     if active:
-        console.err.print('Streaming is already in progress, cannot start.')
+        raise OBSWSCLIError(
-        raise typer.Exit(1)
+            'Streaming is already in progress, cannot start.',
+            code=ExitCode.ERROR,
+        )
 
-    ctx.obj['obsws'].start_stream()
+    ctx.client.start_stream()
     console.out.print('Streaming started successfully.')
 
 
-@app.command('stop | st')
+@app.command(name=['stop', 'st'])
-def stop(ctx: typer.Context):
+def stop(
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
+):
     """Stop streaming."""
     active, _ = _get_streaming_status(ctx)
     if not active:
-        console.err.print('Streaming is not in progress, cannot stop.')
+        raise OBSWSCLIError(
-        raise typer.Exit(1)
+            'Streaming is not in progress, cannot stop.',
+            code=ExitCode.ERROR,
+        )
 
-    ctx.obj['obsws'].stop_stream()
+    ctx.client.stop_stream()
     console.out.print('Streaming stopped successfully.')
 
 
-@app.command('toggle | tg')
+@app.command(name=['toggle', 'tg'])
-def toggle(ctx: typer.Context):
+def toggle(
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
+):
     """Toggle streaming."""
-    resp = ctx.obj['obsws'].toggle_stream()
+    resp = ctx.client.toggle_stream()
     if resp.output_active:
         console.out.print('Streaming started successfully.')
     else:
         console.out.print('Streaming stopped successfully.')
 
 
-@app.command('status | ss')
+@app.command(name=['status', 'ss'])
-def status(ctx: typer.Context):
+def status(
+    *,
+    ctx: Annotated[Context, Parameter(parse=False)],
+):
     """Get streaming status."""
     active, duration = _get_streaming_status(ctx)
     if active: