Compare commits

..

No commits in common. "f223c51a7159e7ecc695343f685355f4260771ad" and "5189ee1d5b5053599793dd2d4574052b846e7836" have entirely different histories.

11 changed files with 74 additions and 144 deletions

View File

@ -96,10 +96,7 @@ obsws-cli scene switch LIVE
#### Scene Item
- list: List all items in a scene.
*optional*
- args: <scene_name>
- defaults to current scene
```console
obsws-cli sceneitem list LIVE

View File

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

View File

@ -4,7 +4,6 @@ from typing import Annotated
import obsws_python as obsws
import typer
from rich.console import Console
from . import (
filter,
@ -42,9 +41,6 @@ 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(
@ -75,6 +71,6 @@ def main(
def version(ctx: typer.Context):
"""Get the OBS Client and WebSocket versions."""
resp = ctx.obj.get_version()
out_console.print(
typer.echo(
f'OBS Client version: {resp.obs_version} with WebSocket version: {resp.obs_web_socket_version}'
)

View File

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

View File

@ -1,13 +1,10 @@
"""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()
@ -19,14 +16,14 @@ def main():
def start(ctx: typer.Context):
"""Start the replay buffer."""
ctx.obj.start_replay_buffer()
out_console.print('Replay buffer started.')
typer.echo('Replay buffer started.')
@app.command('stop | st')
def stop(ctx: typer.Context):
"""Stop the replay buffer."""
ctx.obj.stop_replay_buffer()
out_console.print('Replay buffer stopped.')
typer.echo('Replay buffer stopped.')
@app.command('toggle | tg')
@ -34,9 +31,9 @@ def toggle(ctx: typer.Context):
"""Toggle the replay buffer."""
resp = ctx.obj.toggle_replay_buffer()
if resp.output_active:
out_console.print('Replay buffer is active.')
typer.echo('Replay buffer is active.')
else:
out_console.print('Replay buffer is not active.')
typer.echo('Replay buffer is not active.')
@app.command('status | ss')
@ -44,13 +41,13 @@ def status(ctx: typer.Context):
"""Get the status of the replay buffer."""
resp = ctx.obj.get_replay_buffer_status()
if resp.output_active:
out_console.print('Replay buffer is active.')
typer.echo('Replay buffer is active.')
else:
out_console.print('Replay buffer is not active.')
typer.echo('Replay buffer is not active.')
@app.command('save | sv')
def save(ctx: typer.Context):
"""Save the replay buffer."""
ctx.obj.save_replay_buffer()
out_console.print('Replay buffer saved.')
typer.echo('Replay buffer saved.')

View File

@ -3,15 +3,11 @@
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True)
@app.callback()
@ -23,20 +19,8 @@ def main():
def list(ctx: typer.Context):
"""List all scenes."""
resp = ctx.obj.get_scene_list()
scenes = (
(scene.get('sceneIndex'), scene.get('sceneName'))
for scene in reversed(resp.scenes)
)
table = Table(title='Scenes')
table.add_column('Scene Name', justify='left', style='cyan')
for scene_index, scene_name in scenes:
table.add_row(
scene_name,
)
out_console.print(table)
scenes = (scene.get('sceneName') for scene in reversed(resp.scenes))
typer.echo('\n'.join(scenes))
@app.command('current | get')
@ -48,15 +32,15 @@ def current(
):
"""Get the current program scene or preview scene."""
if preview and not validate.studio_mode_enabled(ctx):
err_console.print('Studio mode is not enabled, cannot get preview scene.')
typer.echo('Studio mode is not enabled, cannot get preview scene.', err=True)
raise typer.Exit(1)
if preview:
resp = ctx.obj.get_current_preview_scene()
out_console.print(resp.current_preview_scene_name)
typer.echo(resp.current_preview_scene_name)
else:
resp = ctx.obj.get_current_program_scene()
out_console.print(resp.current_program_scene_name)
typer.echo(resp.current_program_scene_name)
@app.command('switch | set')
@ -70,16 +54,18 @@ def switch(
):
"""Switch to a scene."""
if preview and not validate.studio_mode_enabled(ctx):
err_console.print('Studio mode is not enabled, cannot set the preview scene.')
typer.echo(
'Studio mode is not enabled, cannot set the preview scene.', err=True
)
raise typer.Exit(1)
if not validate.scene_in_scenes(ctx, scene_name):
err_console.print(f"Scene '{scene_name}' not found.")
typer.echo(f"Scene '{scene_name}' not found.", err=True)
raise typer.Exit(1)
if preview:
ctx.obj.set_current_preview_scene(scene_name)
out_console.print(f'Switched to preview scene: {scene_name}')
typer.echo(f'Switched to preview scene: {scene_name}')
else:
ctx.obj.set_current_program_scene(scene_name)
out_console.print(f'Switched to program scene: {scene_name}')
typer.echo(f'Switched to program scene: {scene_name}')

View File

@ -1,15 +1,11 @@
"""module containing commands for manipulating scene collections."""
import typer
from rich.console import Console
from rich.table import Table
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True)
@app.callback()
@ -21,49 +17,44 @@ def main():
def list(ctx: typer.Context):
"""List all scene collections."""
resp = ctx.obj.get_scene_collection_list()
table = Table(title='Scene Collections')
table.add_column('Scene Collection Name', justify='left', style='cyan')
for scene_collection_name in resp.scene_collections:
table.add_row(scene_collection_name)
out_console.print(table)
typer.echo('\n'.join(resp.scene_collections))
@app.command('current | get')
def current(ctx: typer.Context):
"""Get the current scene collection."""
resp = ctx.obj.get_scene_collection_list()
out_console.print(resp.current_scene_collection_name)
typer.echo(resp.current_scene_collection_name)
@app.command('switch | set')
def switch(ctx: typer.Context, scene_collection_name: str):
"""Switch to a scene collection."""
if not validate.scene_collection_in_scene_collections(ctx, scene_collection_name):
err_console.print(f"Scene collection '{scene_collection_name}' not found.")
typer.echo(f"Scene collection '{scene_collection_name}' not found.", err=True)
raise typer.Exit(1)
current_scene_collection = (
ctx.obj.get_scene_collection_list().current_scene_collection_name
)
if scene_collection_name == current_scene_collection:
err_console.print(
f'Scene collection "{scene_collection_name}" is already active.'
typer.echo(
f'Scene collection "{scene_collection_name}" is already active.', err=True
)
raise typer.Exit(1)
ctx.obj.set_current_scene_collection(scene_collection_name)
out_console.print(f"Switched to scene collection '{scene_collection_name}'")
typer.echo(f"Switched to scene collection '{scene_collection_name}'")
@app.command('create | new')
def create(ctx: typer.Context, scene_collection_name: str):
"""Create a new scene collection."""
if validate.scene_collection_in_scene_collections(ctx, scene_collection_name):
err_console.print(f"Scene collection '{scene_collection_name}' already exists.")
typer.echo(
f"Scene collection '{scene_collection_name}' already exists.", err=True
)
raise typer.Exit(1)
ctx.obj.create_scene_collection(scene_collection_name)
out_console.print(f'Created scene collection {scene_collection_name}')
typer.echo(f'Created scene collection {scene_collection_name}')

View File

@ -4,15 +4,11 @@ from collections.abc import Callable
from typing import Annotated, Optional
import typer
from rich.console import Console
from rich.table import Table
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True)
@app.callback()
@ -21,34 +17,15 @@ def main():
@app.command('list | ls')
def list(
ctx: typer.Context,
scene_name: str = typer.Argument(
None, help='Scene name (optional, defaults to current scene)'
),
):
def list(ctx: typer.Context, scene_name: str):
"""List all items in a scene."""
if not scene_name:
scene_name = ctx.obj.get_current_program_scene().scene_name
if not validate.scene_in_scenes(ctx, scene_name):
err_console.print(f"Scene '{scene_name}' not found.")
raise typer.Exit(1)
typer.echo(f"Scene '{scene_name}' not found.", err=True)
typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
items = [item.get('sourceName') for item in resp.scene_items]
if not items:
err_console.print(f"No items found in scene '{scene_name}'.")
raise typer.Exit(1)
table = Table(title=f'Items in Scene: {scene_name}')
table.add_column('Item Name', justify='left', style='cyan')
for item in items:
table.add_row(item)
out_console.print(table)
items = (item.get('sourceName') for item in resp.scene_items)
typer.echo('\n'.join(items))
def _validate_scene_name_and_item_name(

View File

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

View File

@ -1,13 +1,10 @@
"""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()
@ -19,14 +16,14 @@ def main():
def enable(ctx: typer.Context):
"""Enable studio mode."""
ctx.obj.set_studio_mode_enabled(True)
out_console.print('Studio mode has been enabled.')
typer.echo('Studio mode has been enabled.')
@app.command('disable | off')
def disable(ctx: typer.Context):
"""Disable studio mode."""
ctx.obj.set_studio_mode_enabled(False)
out_console.print('Studio mode has been disabled.')
typer.echo('Studio mode has been disabled.')
@app.command('toggle | tg')
@ -35,10 +32,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)
out_console.print('Studio mode is now disabled.')
typer.echo('Studio mode is now disabled.')
else:
ctx.obj.set_studio_mode_enabled(True)
out_console.print('Studio mode is now enabled.')
typer.echo('Studio mode is now enabled.')
@app.command('status | ss')
@ -46,6 +43,6 @@ def status(ctx: typer.Context):
"""Get the status of studio mode."""
resp = ctx.obj.get_studio_mode_enabled()
if resp.studio_mode_enabled:
out_console.print('Studio mode is enabled.')
typer.echo('Studio mode is enabled.')
else:
out_console.print('Studio mode is disabled.')
typer.echo('Studio mode is disabled.')

View File

@ -1,13 +1,10 @@
"""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()
@ -19,21 +16,21 @@ def main():
def start(ctx: typer.Context):
"""Start the virtual camera."""
ctx.obj.start_virtual_cam()
out_console.print('Virtual camera started.')
typer.echo('Virtual camera started.')
@app.command('stop | p')
def stop(ctx: typer.Context):
"""Stop the virtual camera."""
ctx.obj.stop_virtual_cam()
out_console.print('Virtual camera stopped.')
typer.echo('Virtual camera stopped.')
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle the virtual camera."""
ctx.obj.toggle_virtual_cam()
out_console.print('Virtual camera toggled.')
typer.echo('Virtual camera toggled.')
@app.command('status | ss')
@ -41,6 +38,6 @@ def status(ctx: typer.Context):
"""Get the status of the virtual camera."""
resp = ctx.obj.get_virtual_cam_status()
if resp.output_active:
out_console.print('Virtual camera is enabled.')
typer.echo('Virtual camera is enabled.')
else:
out_console.print('Virtual camera is disabled.')
typer.echo('Virtual camera is disabled.')