write to rich consoles

obsws_cli/app.py

@@ -4,6 +4,7 @@ from typing import Annotated
 
 import obsws_python as obsws
 import typer
+from rich.console import Console
 
 from . import (
     filter,
@@ -41,6 +42,9 @@ for module in (
 ):
     app.add_typer(module.app, name=module.__name__.split('.')[-1])
 
+out_console = Console()
+err_console = Console(stderr=True)
+
 
 @app.callback()
 def main(
@@ -71,6 +75,6 @@ def main(
 def version(ctx: typer.Context):
     """Get the OBS Client and WebSocket versions."""
     resp = ctx.obj.get_version()
-    typer.echo(
+    out_console.print(
         f'OBS Client version: {resp.obs_version} with WebSocket version: {resp.obs_web_socket_version}'
     )

obsws_cli/record.py

@@ -1,10 +1,13 @@
 """module for controlling OBS recording functionality."""
 
 import typer
+from rich.console import Console
 
 from .alias import AliasGroup
 
 app = typer.Typer(cls=AliasGroup)
+out_console = Console()
+err_console = Console(stderr=True)
 
 
 @app.callback()
@@ -27,11 +30,11 @@ def start(ctx: typer.Context):
         if paused:
             err_msg += ' Try resuming it.'
 
-        typer.echo(err_msg, err=True)
+        err_console.print(err_msg)
         raise typer.Exit(1)
 
     ctx.obj.start_record()
-    typer.echo('Recording started successfully.')
+    out_console.print('Recording started successfully.')
 
 
 @app.command('stop | st')
@@ -39,11 +42,11 @@ def stop(ctx: typer.Context):
     """Stop recording."""
     active, _ = _get_recording_status(ctx)
     if not active:
-        typer.echo('Recording is not in progress, cannot stop.', err=True)
+        err_console.print('Recording is not in progress, cannot stop.')
         raise typer.Exit(1)
 
     ctx.obj.stop_record()
-    typer.echo('Recording stopped successfully.')
+    out_console.print('Recording stopped successfully.')
 
 
 @app.command('toggle | tg')
@@ -51,9 +54,9 @@ def toggle(ctx: typer.Context):
     """Toggle recording."""
     resp = ctx.obj.toggle_record()
     if resp.output_active:
-        typer.echo('Recording started successfully.')
+        out_console.print('Recording started successfully.')
     else:
-        typer.echo('Recording stopped successfully.')
+        out_console.print('Recording stopped successfully.')
 
 
 @app.command('status | ss')
@@ -62,11 +65,11 @@ def status(ctx: typer.Context):
     active, paused = _get_recording_status(ctx)
     if active:
         if paused:
-            typer.echo('Recording is in progress and paused.')
+            out_console.print('Recording is in progress and paused.')
         else:
-            typer.echo('Recording is in progress.')
+            out_console.print('Recording is in progress.')
     else:
-        typer.echo('Recording is not in progress.')
+        out_console.print('Recording is not in progress.')
 
 
 @app.command('resume | r')
@@ -74,14 +77,14 @@ def resume(ctx: typer.Context):
     """Resume recording."""
     active, paused = _get_recording_status(ctx)
     if not active:
-        typer.echo('Recording is not in progress, cannot resume.', err=True)
+        err_console.print('Recording is not in progress, cannot resume.')
         raise typer.Exit(1)
     if not paused:
-        typer.echo('Recording is in progress but not paused, cannot resume.', err=True)
+        err_console.print('Recording is in progress but not paused, cannot resume.')
         raise typer.Exit(1)
 
     ctx.obj.resume_record()
-    typer.echo('Recording resumed successfully.')
+    out_console.print('Recording resumed successfully.')
 
 
 @app.command('pause | p')
@@ -89,13 +92,11 @@ def pause(ctx: typer.Context):
     """Pause recording."""
     active, paused = _get_recording_status(ctx)
     if not active:
-        typer.echo('Recording is not in progress, cannot pause.', err=True)
+        err_console.print('Recording is not in progress, cannot pause.')
         raise typer.Exit(1)
     if paused:
-        typer.echo(
+        err_console.print('Recording is in progress but already paused, cannot pause.')
-            'Recording is in progress but already paused, cannot pause.', err=True
-        )
         raise typer.Exit(1)
 
     ctx.obj.pause_record()
-    typer.echo('Recording paused successfully.')
+    out_console.print('Recording paused successfully.')

obsws_cli/replaybuffer.py

@@ -1,10 +1,13 @@
 """module containing commands for manipulating the replay buffer in OBS."""
 
 import typer
+from rich.console import Console
 
 from .alias import AliasGroup
 
 app = typer.Typer(cls=AliasGroup)
+out_console = Console()
+err_console = Console(stderr=True)
 
 
 @app.callback()
@@ -16,14 +19,14 @@ def main():
 def start(ctx: typer.Context):
     """Start the replay buffer."""
     ctx.obj.start_replay_buffer()
-    typer.echo('Replay buffer started.')
+    out_console.print('Replay buffer started.')
 
 
 @app.command('stop | st')
 def stop(ctx: typer.Context):
     """Stop the replay buffer."""
     ctx.obj.stop_replay_buffer()
-    typer.echo('Replay buffer stopped.')
+    out_console.print('Replay buffer stopped.')
 
 
 @app.command('toggle | tg')
@@ -31,9 +34,9 @@ def toggle(ctx: typer.Context):
     """Toggle the replay buffer."""
     resp = ctx.obj.toggle_replay_buffer()
     if resp.output_active:
-        typer.echo('Replay buffer is active.')
+        out_console.print('Replay buffer is active.')
     else:
-        typer.echo('Replay buffer is not active.')
+        out_console.print('Replay buffer is not active.')
 
 
 @app.command('status | ss')
@@ -41,13 +44,13 @@ def status(ctx: typer.Context):
     """Get the status of the replay buffer."""
     resp = ctx.obj.get_replay_buffer_status()
     if resp.output_active:
-        typer.echo('Replay buffer is active.')
+        out_console.print('Replay buffer is active.')
     else:
-        typer.echo('Replay buffer is not active.')
+        out_console.print('Replay buffer is not active.')
 
 
 @app.command('save | sv')
 def save(ctx: typer.Context):
     """Save the replay buffer."""
     ctx.obj.save_replay_buffer()
-    typer.echo('Replay buffer saved.')
+    out_console.print('Replay buffer saved.')

obsws_cli/stream.py

@@ -1,10 +1,13 @@
 """module for controlling OBS stream functionality."""
 
 import typer
+from rich.console import Console
 
 from .alias import AliasGroup
 
 app = typer.Typer(cls=AliasGroup)
+out_console = Console()
+err_console = Console(stderr=True)
 
 
 @app.callback()
@@ -23,11 +26,11 @@ def start(ctx: typer.Context):
     """Start streaming."""
     active, _ = _get_streaming_status(ctx)
     if active:
-        typer.echo('Streaming is already in progress, cannot start.', err=True)
+        err_console.print('Streaming is already in progress, cannot start.')
         raise typer.Exit(1)
 
     ctx.obj.start_stream()
-    typer.echo('Streaming started successfully.')
+    out_console.print('Streaming started successfully.')
 
 
 @app.command('stop | st')
@@ -35,11 +38,11 @@ def stop(ctx: typer.Context):
     """Stop streaming."""
     active, _ = _get_streaming_status(ctx)
     if not active:
-        typer.echo('Streaming is not in progress, cannot stop.', err=True)
+        err_console.print('Streaming is not in progress, cannot stop.')
         raise typer.Exit(1)
 
     ctx.obj.stop_stream()
-    typer.echo('Streaming stopped successfully.')
+    out_console.print('Streaming stopped successfully.')
 
 
 @app.command('toggle | tg')
@@ -47,9 +50,9 @@ def toggle(ctx: typer.Context):
     """Toggle streaming."""
     resp = ctx.obj.toggle_stream()
     if resp.output_active:
-        typer.echo('Streaming started successfully.')
+        out_console.print('Streaming started successfully.')
     else:
-        typer.echo('Streaming stopped successfully.')
+        out_console.print('Streaming stopped successfully.')
 
 
 @app.command('status | ss')
@@ -62,15 +65,19 @@ def status(ctx: typer.Context):
             minutes = int(seconds // 60)
             seconds = int(seconds % 60)
             if minutes > 0:
-                typer.echo(
+                out_console.print(
                     f'Streaming is in progress for {minutes} minutes and {seconds} seconds.'
                 )
             else:
                 if seconds > 0:
-                    typer.echo(f'Streaming is in progress for {seconds} seconds.')
+                    out_console.print(
+                        f'Streaming is in progress for {seconds} seconds.'
+                    )
                 else:
-                    typer.echo('Streaming is in progress for less than a second.')
+                    out_console.print(
+                        'Streaming is in progress for less than a second.'
+                    )
         else:
-            typer.echo('Streaming is in progress.')
+            out_console.print('Streaming is in progress.')
     else:
-        typer.echo('Streaming is not in progress.')
+        err_console.print('Streaming is not in progress.')

obsws_cli/studiomode.py

@@ -1,10 +1,13 @@
 """module containing commands for manipulating studio mode in OBS."""
 
 import typer
+from rich.console import Console
 
 from .alias import AliasGroup
 
 app = typer.Typer(cls=AliasGroup)
+out_console = Console()
+err_console = Console(stderr=True)
 
 
 @app.callback()
@@ -16,14 +19,14 @@ def main():
 def enable(ctx: typer.Context):
     """Enable studio mode."""
     ctx.obj.set_studio_mode_enabled(True)
-    typer.echo('Studio mode has been enabled.')
+    out_console.print('Studio mode has been enabled.')
 
 
 @app.command('disable | off')
 def disable(ctx: typer.Context):
     """Disable studio mode."""
     ctx.obj.set_studio_mode_enabled(False)
-    typer.echo('Studio mode has been disabled.')
+    out_console.print('Studio mode has been disabled.')
 
 
 @app.command('toggle | tg')
@@ -32,10 +35,10 @@ def toggle(ctx: typer.Context):
     resp = ctx.obj.get_studio_mode_enabled()
     if resp.studio_mode_enabled:
         ctx.obj.set_studio_mode_enabled(False)
-        typer.echo('Studio mode is now disabled.')
+        out_console.print('Studio mode is now disabled.')
     else:
         ctx.obj.set_studio_mode_enabled(True)
-        typer.echo('Studio mode is now enabled.')
+        out_console.print('Studio mode is now enabled.')
 
 
 @app.command('status | ss')
@@ -43,6 +46,6 @@ def status(ctx: typer.Context):
     """Get the status of studio mode."""
     resp = ctx.obj.get_studio_mode_enabled()
     if resp.studio_mode_enabled:
-        typer.echo('Studio mode is enabled.')
+        out_console.print('Studio mode is enabled.')
     else:
-        typer.echo('Studio mode is disabled.')
+        out_console.print('Studio mode is disabled.')

obsws_cli/virtualcam.py

@@ -1,10 +1,13 @@
 """module containing commands for manipulating virtual camera in OBS."""
 
 import typer
+from rich.console import Console
 
 from .alias import AliasGroup
 
 app = typer.Typer(cls=AliasGroup)
+out_console = Console()
+err_console = Console(stderr=True)
 
 
 @app.callback()
@@ -16,21 +19,21 @@ def main():
 def start(ctx: typer.Context):
     """Start the virtual camera."""
     ctx.obj.start_virtual_cam()
-    typer.echo('Virtual camera started.')
+    out_console.print('Virtual camera started.')
 
 
 @app.command('stop | p')
 def stop(ctx: typer.Context):
     """Stop the virtual camera."""
     ctx.obj.stop_virtual_cam()
-    typer.echo('Virtual camera stopped.')
+    out_console.print('Virtual camera stopped.')
 
 
 @app.command('toggle | tg')
 def toggle(ctx: typer.Context):
     """Toggle the virtual camera."""
     ctx.obj.toggle_virtual_cam()
-    typer.echo('Virtual camera toggled.')
+    out_console.print('Virtual camera toggled.')
 
 
 @app.command('status | ss')
@@ -38,6 +41,6 @@ def status(ctx: typer.Context):
     """Get the status of the virtual camera."""
     resp = ctx.obj.get_virtual_cam_status()
     if resp.output_active:
-        typer.echo('Virtual camera is enabled.')
+        out_console.print('Virtual camera is enabled.')
     else:
-        typer.echo('Virtual camera is disabled.')
+        out_console.print('Virtual camera is disabled.')