onyx_online/obsws-cli
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/obsws-cli.git synced 2025-05-16 22:40:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

check streaming status to clean up stack traces

Browse Source 
toggle command now invokes stop/start

added toggle alias tg

patch bump
This commit is contained in:
onyx-and-iris 2025-04-20 21:14:57 +01:00
parent edd3753d4d
commit 2175fb8788
2 changed files with 24 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
obsws_cli/__about__.py
View File

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

51
obsws_cli/stream.py
View File

@ -1,11 +1,10 @@
"""module for controlling OBS stream functionality.""" """module for controlling OBS stream functionality."""
import obsws_python as obsws
import typer import typer
from .errors import ObswsCliError from .alias import AliasGroup
app = typer.Typer() app = typer.Typer(cls=AliasGroup)
@app.callback() @app.callback()
@ -13,36 +12,34 @@ def main():
"""Control OBS stream functionality.""" """Control OBS stream functionality."""
def _get_streaming_status(ctx: typer.Context) -> tuple:
"""Get streaming status."""
resp = ctx.obj['obsws'].get_stream_status()
return resp.output_active, resp.output_duration
@app.command() @app.command()
def start(ctx: typer.Context): def start(ctx: typer.Context):
"""Start streaming.""" """Start streaming."""
try: active, _ = _get_streaming_status(ctx)
if active:
typer.echo('Streaming is already in progress, cannot start.')
raise typer.Exit(code=1)
ctx.obj['obsws'].start_stream() ctx.obj['obsws'].start_stream()
typer.echo('Streaming started successfully.') typer.echo('Streaming started successfully.')
except obsws.error.OBSSDKRequestError as e:
if e.code == 500:
raise ObswsCliError(
'Streaming is already in progress, cannot start.'
) from e
raise
@app.command() @app.command()
def stop(ctx: typer.Context): def stop(ctx: typer.Context):
"""Stop streaming.""" """Stop streaming."""
try: active, _ = _get_streaming_status(ctx)
if not active:
typer.echo('Streaming is not in progress, cannot stop.')
raise typer.Exit(code=1)
ctx.obj['obsws'].stop_stream() ctx.obj['obsws'].stop_stream()
typer.echo('Streaming stopped successfully.') typer.echo('Streaming stopped successfully.')
except obsws.error.OBSSDKRequestError as e:
if e.code == 501:
raise ObswsCliError('Streaming is not in progress, cannot stop.') from e
raise
def _get_streaming_status(ctx: typer.Context) -> tuple:
"""Get streaming status."""
resp = ctx.obj['obsws'].get_stream_status()
return resp.output_active, resp.output_duration
@app.command() @app.command()
@ -74,14 +71,6 @@ def toggle(ctx: typer.Context):
"""Toggle streaming.""" """Toggle streaming."""
active, _ = _get_streaming_status(ctx) active, _ = _get_streaming_status(ctx)
if active: if active:
try: ctx.invoke(stop, ctx=ctx)
ctx.obj['obsws'].stop_stream()
typer.echo('Streaming stopped successfully.')
except obsws.error.OBSSDKRequestError as e:
raise ObswsCliError(str(e)) from e
else: else:
try: ctx.invoke(start, ctx=ctx)
ctx.obj['obsws'].start_stream()
typer.echo('Streaming started successfully.')
except obsws.error.OBSSDKRequestError as e:
raise ObswsCliError(str(e)) from e