Compare commits

..

No commits in common. "bff5d396a4c4d65431967e21ec005619db9f32b1" and "a26ce74151035dd7544537d688e0fc9b57af8794" have entirely different histories.

21 changed files with 270 additions and 366 deletions

View File

@ -5,21 +5,6 @@ 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.17.0] - 2025-06-20
### Added
- input list, scene list and sceneitem list now accept --uuid flag.
- Active column added to scene list table.
### Changed
- scene list no longer prints the UUIDs by default, enable it with the --uuid flag.
### Fixed
- Issue with input list not printing all inputs if no filters were applied.
# [0.16.8] - 2025-06-07
### Added

View File

@ -81,10 +81,6 @@ obsws-cli obs-version
#### Scene
- list: List all scenes.
- flags:
*optional*
- --uuid: Show UUIDs of scenes
```console
obsws-cli scene list
@ -106,10 +102,6 @@ obsws-cli scene switch LIVE
#### Scene Item
- list: List all items in a scene.
- flags:
*optional*
- --uuid: Show UUIDs of scene items
*optional*
- args: <scene_name>
@ -275,7 +267,6 @@ obsws-cli group status START "test_group"
- --colour: Filter by colour source type.
- --ffmpeg: Filter by ffmpeg source type.
- --vlc: Filter by VLC source type.
- --uuid: Show UUIDs of inputs.
```console
obsws-cli input list

View File

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

View File

@ -1,15 +1,15 @@
"""Command line interface for the OBS WebSocket API."""
import importlib
import logging
from typing import Annotated
import obsws_python as obsws
import typer
from rich.console import Console
from obsws_cli.__about__ import __version__ as obsws_cli_version
from . import console, settings
from . import settings
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
@ -33,23 +33,17 @@ for sub_typer in (
module = importlib.import_module(f'.{sub_typer}', package=__package__)
app.add_typer(module.app, name=sub_typer)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
def version_callback(value: bool):
"""Show the version of the CLI."""
if value:
console.out.print(f'obsws-cli version: {obsws_cli_version}')
out_console.print(f'obsws-cli version: {obsws_cli_version}')
raise typer.Exit()
def setup_logging(debug: bool):
"""Set up logging for the application."""
log_level = logging.DEBUG if debug else logging.CRITICAL
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
@app.callback()
def main(
ctx: typer.Context,
@ -66,11 +60,7 @@ def main(
port: Annotated[
int,
typer.Option(
'--port',
'-P',
envvar='OBS_PORT',
help='WebSocket port',
show_default=4455,
'--port', '-P', envvar='OBS_PORT', help='WebSocket port', show_default=4455
),
] = settings.get('port'),
password: Annotated[
@ -104,19 +94,6 @@ def main(
callback=version_callback,
),
] = False,
debug: Annotated[
bool,
typer.Option(
'--debug',
'-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."""
ctx.obj = ctx.with_resource(obsws.ReqClient(**ctx.params))
@ -126,6 +103,6 @@ def main(
def obs_version(ctx: typer.Context):
"""Get the OBS Client and WebSocket versions."""
resp = ctx.obj.get_version()
console.out.print(
out_console.print(
f'OBS Client version: {resp.obs_version} with WebSocket version: {resp.obs_web_socket_version}'
)

View File

@ -1,6 +0,0 @@
"""module for console output handling in obsws_cli."""
from rich.console import Console
out = Console()
err = Console(stderr=True, style='bold red')

View File

@ -4,12 +4,15 @@ from typing import Annotated, Optional
import obsws_python as obsws
import typer
from rich.console import Console
from rich.table import Table
from . import console, util
from . import util
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -36,7 +39,7 @@ def list_(
resp = ctx.obj.get_source_filter_list(source_name)
except obsws.error.OBSSDKRequestError as e:
if e.code == 600:
console.err.print(
err_console.print(
f'No source was found by the name of [yellow]{source_name}[/yellow].'
)
raise typer.Exit(1)
@ -44,7 +47,7 @@ def list_(
raise
if not resp.filters:
console.out.print(f'No filters found for source [yellow]{source_name}[/yellow]')
out_console.print(f'No filters found for source [yellow]{source_name}[/yellow]')
raise typer.Exit()
table = Table(title=f'Filters for Source: {source_name}', padding=(0, 2))
@ -74,7 +77,7 @@ def list_(
),
)
console.out.print(table)
out_console.print(table)
def _get_filter_enabled(ctx: typer.Context, source_name: str, filter_name: str):
@ -101,13 +104,13 @@ def enable(
):
"""Enable a filter for a source."""
if _get_filter_enabled(ctx, source_name, filter_name):
console.err.print(
err_console.print(
f'Filter [yellow]{filter_name}[/yellow] is already enabled for source [yellow]{source_name}[/yellow]'
)
raise typer.Exit(1)
ctx.obj.set_source_filter_enabled(source_name, filter_name, enabled=True)
console.out.print(
out_console.print(
f'Enabled filter [green]{filter_name}[/green] for source [green]{source_name}[/green]'
)
@ -130,13 +133,13 @@ def disable(
):
"""Disable a filter for a source."""
if not _get_filter_enabled(ctx, source_name, filter_name):
console.err.print(
err_console.print(
f'Filter [yellow]{filter_name}[/yellow] is already disabled for source [yellow]{source_name}[/yellow]'
)
raise typer.Exit(1)
ctx.obj.set_source_filter_enabled(source_name, filter_name, enabled=False)
console.out.print(
out_console.print(
f'Disabled filter [green]{filter_name}[/green] for source [green]{source_name}[/green]'
)
@ -163,11 +166,11 @@ def toggle(
ctx.obj.set_source_filter_enabled(source_name, filter_name, enabled=new_state)
if new_state:
console.out.print(
out_console.print(
f'Enabled filter [green]{filter_name}[/green] for source [green]{source_name}[/green]'
)
else:
console.out.print(
out_console.print(
f'Disabled filter [green]{filter_name}[/green] for source [green]{source_name}[/green]'
)
@ -191,10 +194,10 @@ def status(
"""Get the status of a filter for a source."""
is_enabled = _get_filter_enabled(ctx, source_name, filter_name)
if is_enabled:
console.out.print(
out_console.print(
f'Filter [green]{filter_name}[/green] is enabled for source [green]{source_name}[/green]'
)
else:
console.out.print(
out_console.print(
f'Filter [green]{filter_name}[/green] is disabled for source [green]{source_name}[/green]'
)

View File

@ -3,13 +3,16 @@
from typing import Annotated, Optional
import typer
from rich.console import Console
from rich.table import Table
from . import console, validate
from . import validate
from .alias import AliasGroup
from .protocols import DataclassProtocol
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -33,7 +36,7 @@ def list_(
scene_name = ctx.obj.get_current_program_scene().scene_name
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f"Scene '{scene_name}' not found.")
err_console.print(f"Scene '{scene_name}' not found.")
raise typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
@ -44,7 +47,7 @@ def list_(
]
if not groups:
console.out.print(f"No groups found in scene '{scene_name}'.")
out_console.print(f"No groups found in scene '{scene_name}'.")
raise typer.Exit()
table = Table(title=f'Groups in Scene: {scene_name}', padding=(0, 2))
@ -64,7 +67,7 @@ def list_(
':white_heavy_check_mark:' if is_enabled else ':x:',
)
console.out.print(table)
out_console.print(table)
def _get_group(group_name: str, resp: DataclassProtocol) -> dict | None:
@ -93,12 +96,12 @@ def show(
):
"""Show a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f"Scene '{scene_name}' not found.")
err_console.print(f"Scene '{scene_name}' not found.")
raise typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
console.err.print(
err_console.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
@ -109,7 +112,7 @@ def show(
enabled=True,
)
console.out.print(f'Group [green]{group_name}[/green] is now visible.')
out_console.print(f'Group [green]{group_name}[/green] is now visible.')
@app.command('hide | h')
@ -124,12 +127,12 @@ def hide(
):
"""Hide a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
err_console.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
console.err.print(
err_console.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
@ -140,7 +143,7 @@ def hide(
enabled=False,
)
console.out.print(f'Group [green]{group_name}[/green] is now hidden.')
out_console.print(f'Group [green]{group_name}[/green] is now hidden.')
@app.command('toggle | tg')
@ -155,12 +158,12 @@ def toggle(
):
"""Toggle a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
err_console.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
console.err.print(
err_console.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
@ -173,9 +176,9 @@ def toggle(
)
if new_state:
console.out.print(f'Group [green]{group_name}[/green] is now visible.')
out_console.print(f'Group [green]{group_name}[/green] is now visible.')
else:
console.out.print(f'Group [green]{group_name}[/green] is now hidden.')
out_console.print(f'Group [green]{group_name}[/green] is now hidden.')
@app.command('status | ss')
@ -190,12 +193,12 @@ def status(
):
"""Get the status of a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
err_console.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
console.err.print(
err_console.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
@ -206,6 +209,6 @@ def status(
)
if enabled.scene_item_enabled:
console.out.print(f'Group [green]{group_name}[/green] is now visible.')
out_console.print(f'Group [green]{group_name}[/green] is now visible.')
else:
console.out.print(f'Group [green]{group_name}[/green] is now hidden.')
out_console.print(f'Group [green]{group_name}[/green] is now hidden.')

View File

@ -3,12 +3,14 @@
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -29,7 +31,7 @@ def list_(
for hotkey in resp.hotkeys:
table.add_row(hotkey)
console.out.print(table)
out_console.print(table)
@app.command('trigger | tr')

View File

@ -2,14 +2,16 @@
from typing import Annotated
import obsws_python as obsws
import typer
from rich.console import Console
from rich.table import Table
from . import console, util, validate
from . import util, validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -25,7 +27,6 @@ def list_(
colour: Annotated[bool, typer.Option(help='Filter by colour source type.')] = False,
ffmpeg: Annotated[bool, typer.Option(help='Filter by ffmpeg source type.')] = False,
vlc: Annotated[bool, typer.Option(help='Filter by VLC source type.')] = False,
uuid: Annotated[bool, typer.Option(help='Show UUIDs of inputs.')] = False,
):
"""List all inputs."""
resp = ctx.obj.get_input_list()
@ -42,11 +43,11 @@ def list_(
if vlc:
kinds.append('vlc')
if not any([input, output, colour, ffmpeg, vlc]):
kinds = ctx.obj.get_input_kind_list(False).input_kinds
kinds = ['input', 'output', 'color', 'ffmpeg', 'vlc']
inputs = sorted(
(
(input_.get('inputName'), input_.get('inputKind'), input_.get('inputUuid'))
(input_.get('inputName'), input_.get('inputKind'))
for input_ in filter(
lambda input_: any(kind in input_.get('inputKind') for kind in kinds),
resp.inputs,
@ -56,19 +57,10 @@ def list_(
)
if not inputs:
console.out.print('No inputs found.')
out_console.print('No inputs found.')
raise typer.Exit()
table = Table(title='Inputs', padding=(0, 2))
if uuid:
columns = [
('Input Name', 'left', 'cyan'),
('Kind', 'center', 'cyan'),
('Muted', 'center', None),
('UUID', 'left', 'cyan'),
]
else:
table.title += ' (UUIDs hidden)'
columns = [
('Input Name', 'left', 'cyan'),
('Kind', 'center', 'cyan'),
@ -77,32 +69,22 @@ def list_(
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for input_name, input_kind, input_uuid in inputs:
for input_name, input_kind in inputs:
input_mark = ''
try:
if any(
kind in input_kind
for kind in ['input_capture', 'output_capture', 'ffmpeg', 'vlc']
):
input_muted = ctx.obj.get_input_mute(name=input_name).input_muted
input_mark = ':white_heavy_check_mark:' if input_muted else ':x:'
except obsws.error.OBSSDKRequestError as e:
if e.code == 604: # Input does not support audio
input_mark = 'N/A'
else:
raise
if uuid:
table.add_row(
input_name,
util.snakecase_to_titlecase(input_kind),
input_mark,
input_uuid,
)
else:
table.add_row(
input_name,
util.snakecase_to_titlecase(input_kind),
input_mark,
)
console.out.print(table)
out_console.print(table)
@app.command('mute | m')
@ -114,7 +96,7 @@ def mute(
):
"""Mute an input."""
if not validate.input_in_inputs(ctx, input_name):
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
err_console.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
ctx.obj.set_input_mute(
@ -122,7 +104,7 @@ def mute(
muted=True,
)
console.out.print(f'Input [green]{input_name}[/green] muted.')
out_console.print(f'Input [green]{input_name}[/green] muted.')
@app.command('unmute | um')
@ -135,7 +117,7 @@ def unmute(
):
"""Unmute an input."""
if not validate.input_in_inputs(ctx, input_name):
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
err_console.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
ctx.obj.set_input_mute(
@ -143,7 +125,7 @@ def unmute(
muted=False,
)
console.out.print(f'Input [green]{input_name}[/green] unmuted.')
out_console.print(f'Input [green]{input_name}[/green] unmuted.')
@app.command('toggle | tg')
@ -156,7 +138,7 @@ def toggle(
):
"""Toggle an input."""
if not validate.input_in_inputs(ctx, input_name):
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
err_console.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj.get_input_mute(name=input_name)
@ -168,10 +150,10 @@ def toggle(
)
if new_state:
console.out.print(
out_console.print(
f'Input [green]{input_name}[/green] muted.',
)
else:
console.out.print(
out_console.print(
f'Input [green]{input_name}[/green] unmuted.',
)

View File

@ -3,12 +3,15 @@
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from . import console, validate
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -35,14 +38,14 @@ def list_(ctx: typer.Context):
':white_heavy_check_mark:' if profile == resp.current_profile_name else '',
)
console.out.print(table)
out_console.print(table)
@app.command('current | get')
def current(ctx: typer.Context):
"""Get the current profile."""
resp = ctx.obj.get_profile_list()
console.out.print(resp.current_profile_name)
out_console.print(resp.current_profile_name)
@app.command('switch | set')
@ -57,18 +60,18 @@ def switch(
):
"""Switch to a profile."""
if not validate.profile_exists(ctx, profile_name):
console.err.print(f'Profile [yellow]{profile_name}[/yellow] not found.')
err_console.print(f'Profile [yellow]{profile_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj.get_profile_list()
if resp.current_profile_name == profile_name:
console.err.print(
err_console.print(
f'Profile [yellow]{profile_name}[/yellow] is already the current profile.'
)
raise typer.Exit(1)
ctx.obj.set_current_profile(profile_name)
console.out.print(f'Switched to profile [green]{profile_name}[/green].')
out_console.print(f'Switched to profile [green]{profile_name}[/green].')
@app.command('create | new')
@ -81,11 +84,11 @@ def create(
):
"""Create a new profile."""
if validate.profile_exists(ctx, profile_name):
console.err.print(f'Profile [yellow]{profile_name}[/yellow] already exists.')
err_console.print(f'Profile [yellow]{profile_name}[/yellow] already exists.')
raise typer.Exit(1)
ctx.obj.create_profile(profile_name)
console.out.print(f'Created profile [green]{profile_name}[/green].')
out_console.print(f'Created profile [green]{profile_name}[/green].')
@app.command('remove | rm')
@ -98,8 +101,8 @@ def remove(
):
"""Remove a profile."""
if not validate.profile_exists(ctx, profile_name):
console.err.print(f'Profile [yellow]{profile_name}[/yellow] not found.')
err_console.print(f'Profile [yellow]{profile_name}[/yellow] not found.')
raise typer.Exit(1)
ctx.obj.remove_profile(profile_name)
console.out.print(f'Removed profile [green]{profile_name}[/green].')
out_console.print(f'Removed profile [green]{profile_name}[/green].')

View File

@ -3,12 +3,14 @@
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -22,7 +24,7 @@ def list_monitors(ctx: typer.Context):
resp = ctx.obj.get_monitor_list()
if not resp.monitors:
console.out.print('No monitors found.')
out_console.print('No monitors found.')
return
monitors = sorted(
@ -37,7 +39,7 @@ def list_monitors(ctx: typer.Context):
for index, monitor in monitors:
table.add_row(str(index), monitor)
console.out.print(table)
out_console.print(table)
@app.command('open | o')
@ -67,13 +69,13 @@ def open(
monitor_index=monitor_index,
)
console.out.print(
out_console.print(
f'Opened projector for source [green]{source_name}[/] on monitor [green]{monitor["monitorName"]}[/].'
)
break
else:
console.err.print(
err_console.print(
f'Monitor with index [yellow]{monitor_index}[/yellow] not found.'
)
raise typer.Exit(code=1)

View File

@ -4,11 +4,13 @@ from pathlib import Path
from typing import Annotated, Optional
import typer
from rich.console import Console
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -31,11 +33,11 @@ def start(ctx: typer.Context):
if paused:
err_msg += ' Try resuming it.'
console.err.print(err_msg)
err_console.print(err_msg)
raise typer.Exit(1)
ctx.obj.start_record()
console.out.print('Recording started successfully.')
out_console.print('Recording started successfully.')
@app.command('stop | st')
@ -43,11 +45,11 @@ def stop(ctx: typer.Context):
"""Stop recording."""
active, _ = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot stop.')
err_console.print('Recording is not in progress, cannot stop.')
raise typer.Exit(1)
resp = ctx.obj.stop_record()
console.out.print(
out_console.print(
f'Recording stopped successfully. Saved to: [green]{resp.output_path}[/green]'
)
@ -57,9 +59,9 @@ def toggle(ctx: typer.Context):
"""Toggle recording."""
resp = ctx.obj.toggle_record()
if resp.output_active:
console.out.print('Recording started successfully.')
out_console.print('Recording started successfully.')
else:
console.out.print('Recording stopped successfully.')
out_console.print('Recording stopped successfully.')
@app.command('status | ss')
@ -68,11 +70,11 @@ def status(ctx: typer.Context):
active, paused = _get_recording_status(ctx)
if active:
if paused:
console.out.print('Recording is in progress and paused.')
out_console.print('Recording is in progress and paused.')
else:
console.out.print('Recording is in progress.')
out_console.print('Recording is in progress.')
else:
console.out.print('Recording is not in progress.')
out_console.print('Recording is not in progress.')
@app.command('resume | r')
@ -80,14 +82,14 @@ def resume(ctx: typer.Context):
"""Resume recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot resume.')
err_console.print('Recording is not in progress, cannot resume.')
raise typer.Exit(1)
if not paused:
console.err.print('Recording is in progress but not paused, cannot resume.')
err_console.print('Recording is in progress but not paused, cannot resume.')
raise typer.Exit(1)
ctx.obj.resume_record()
console.out.print('Recording resumed successfully.')
out_console.print('Recording resumed successfully.')
@app.command('pause | p')
@ -95,14 +97,14 @@ def pause(ctx: typer.Context):
"""Pause recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot pause.')
err_console.print('Recording is not in progress, cannot pause.')
raise typer.Exit(1)
if paused:
console.err.print('Recording is in progress but already paused, cannot pause.')
err_console.print('Recording is in progress but already paused, cannot pause.')
raise typer.Exit(1)
ctx.obj.pause_record()
console.out.print('Recording paused successfully.')
out_console.print('Recording paused successfully.')
@app.command('directory | d')
@ -122,8 +124,8 @@ def directory(
"""Get or set the recording directory."""
if record_directory is not None:
ctx.obj.set_record_directory(str(record_directory))
console.out.print(f'Recording directory updated to: {record_directory}')
out_console.print(f'Recording directory updated to: {record_directory}')
else:
console.out.print(
out_console.print(
f'Recording directory: [green]{ctx.obj.get_record_directory().record_directory}[/green]'
)

View File

@ -1,11 +1,13 @@
"""module containing commands for manipulating the replay buffer in OBS."""
import typer
from rich.console import Console
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -18,11 +20,11 @@ def start(ctx: typer.Context):
"""Start the replay buffer."""
resp = ctx.obj.get_replay_buffer_status()
if resp.output_active:
console.err.print('Replay buffer is already active.')
err_console.print('Replay buffer is already active.')
raise typer.Exit(1)
ctx.obj.start_replay_buffer()
console.out.print('Replay buffer started.')
out_console.print('Replay buffer started.')
@app.command('stop | st')
@ -30,11 +32,11 @@ def stop(ctx: typer.Context):
"""Stop the replay buffer."""
resp = ctx.obj.get_replay_buffer_status()
if not resp.output_active:
console.err.print('Replay buffer is not active.')
err_console.print('Replay buffer is not active.')
raise typer.Exit(1)
ctx.obj.stop_replay_buffer()
console.out.print('Replay buffer stopped.')
out_console.print('Replay buffer stopped.')
@app.command('toggle | tg')
@ -42,9 +44,9 @@ def toggle(ctx: typer.Context):
"""Toggle the replay buffer."""
resp = ctx.obj.toggle_replay_buffer()
if resp.output_active:
console.out.print('Replay buffer is active.')
out_console.print('Replay buffer is active.')
else:
console.out.print('Replay buffer is not active.')
out_console.print('Replay buffer is not active.')
@app.command('status | ss')
@ -52,13 +54,13 @@ def status(ctx: typer.Context):
"""Get the status of the replay buffer."""
resp = ctx.obj.get_replay_buffer_status()
if resp.output_active:
console.out.print('Replay buffer is active.')
out_console.print('Replay buffer is active.')
else:
console.out.print('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()
console.out.print('Replay buffer saved.')
out_console.print('Replay buffer saved.')

View File

@ -3,12 +3,15 @@
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from . import console, validate
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -17,10 +20,7 @@ def main():
@app.command('list | ls')
def list_(
ctx: typer.Context,
uuid: Annotated[bool, typer.Option(help='Show UUIDs of scenes')] = False,
):
def list_(ctx: typer.Context):
"""List all scenes."""
resp = ctx.obj.get_scene_list()
scenes = (
@ -28,43 +28,21 @@ def list_(
for scene in reversed(resp.scenes)
)
active_scene = ctx.obj.get_current_program_scene().scene_name
table = Table(title='Scenes', padding=(0, 2))
if uuid:
columns = [
('Scene Name', 'left', 'cyan'),
('Active', 'center', None),
('UUID', 'left', 'cyan'),
]
else:
table.title += ' (UUIDs hidden)'
columns = [
('Scene Name', 'left', 'cyan'),
('Active', 'center', None),
]
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for scene_name, scene_uuid in scenes:
if scene_name == active_scene:
scene_output = f'[bold green]{scene_name}[/bold green]'
else:
scene_output = f'[dim]{scene_name}[/dim]'
if uuid:
table.add_row(
scene_output,
':white_heavy_check_mark:' if scene_name == active_scene else '',
scene_name,
scene_uuid,
)
else:
table.add_row(
scene_output,
':white_heavy_check_mark:' if scene_name == active_scene else '',
)
console.out.print(table)
out_console.print(table)
@app.command('current | get')
@ -76,15 +54,15 @@ def current(
):
"""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.')
err_console.print('Studio mode is not enabled, cannot get preview scene.')
raise typer.Exit(1)
if preview:
resp = ctx.obj.get_current_preview_scene()
console.out.print(resp.current_preview_scene_name)
out_console.print(resp.current_preview_scene_name)
else:
resp = ctx.obj.get_current_program_scene()
console.out.print(resp.current_program_scene_name)
out_console.print(resp.current_program_scene_name)
@app.command('switch | set')
@ -100,16 +78,16 @@ def switch(
):
"""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.')
err_console.print('Studio mode is not enabled, cannot set the preview scene.')
raise typer.Exit(1)
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
err_console.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
if preview:
ctx.obj.set_current_preview_scene(scene_name)
console.out.print(f'Switched to preview scene: [green]{scene_name}[/green]')
out_console.print(f'Switched to preview scene: [green]{scene_name}[/green]')
else:
ctx.obj.set_current_program_scene(scene_name)
console.out.print(f'Switched to program scene: [green]{scene_name}[/green]')
out_console.print(f'Switched to program scene: [green]{scene_name}[/green]')

View File

@ -3,12 +3,15 @@
from typing import Annotated
import typer
from rich.console import Console
from rich.table import Table
from . import console, validate
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -27,14 +30,14 @@ def list_(ctx: typer.Context):
for scene_collection_name in resp.scene_collections:
table.add_row(scene_collection_name)
console.out.print(table)
out_console.print(table)
@app.command('current | get')
def current(ctx: typer.Context):
"""Get the current scene collection."""
resp = ctx.obj.get_scene_collection_list()
console.out.print(resp.current_scene_collection_name)
out_console.print(resp.current_scene_collection_name)
@app.command('switch | set')
@ -46,7 +49,7 @@ def switch(
):
"""Switch to a scene collection."""
if not validate.scene_collection_in_scene_collections(ctx, scene_collection_name):
console.err.print(
err_console.print(
f'Scene collection [yellow]{scene_collection_name}[/yellow] not found.'
)
raise typer.Exit(1)
@ -55,13 +58,13 @@ def switch(
ctx.obj.get_scene_collection_list().current_scene_collection_name
)
if scene_collection_name == current_scene_collection:
console.err.print(
err_console.print(
f'Scene collection [yellow]{scene_collection_name}[/yellow] is already active.'
)
raise typer.Exit(1)
ctx.obj.set_current_scene_collection(scene_collection_name)
console.out.print(
out_console.print(
f'Switched to scene collection [green]{scene_collection_name}[/green].'
)
@ -75,12 +78,12 @@ def create(
):
"""Create a new scene collection."""
if validate.scene_collection_in_scene_collections(ctx, scene_collection_name):
console.err.print(
err_console.print(
f'Scene collection [yellow]{scene_collection_name}[/yellow] already exists.'
)
raise typer.Exit(1)
ctx.obj.create_scene_collection(scene_collection_name)
console.out.print(
out_console.print(
f'Created scene collection [green]{scene_collection_name}[/green].'
)

View File

@ -2,13 +2,17 @@
from typing import Annotated, Optional
import obsws_python as obsws
import typer
from rich.console import Console
from rich.table import Table
from . import console, validate
from . import validate
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -26,14 +30,13 @@ def list_(
help='Scene name to list items for',
),
] = None,
uuid: Annotated[bool, typer.Option(help='Show UUIDs of scene items')] = False,
):
"""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):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
err_console.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj.get_scene_item_list(scene_name)
@ -44,7 +47,6 @@ def list_(
item.get('sourceName'),
item.get('isGroup'),
item.get('sceneItemEnabled'),
item.get('sourceUuid', 'N/A'), # Include source UUID
)
for item in resp.scene_items
),
@ -52,20 +54,10 @@ def list_(
)
if not items:
console.out.print(f'No items found in scene [green]{scene_name}[/green].')
out_console.print(f'No items found in scene [green]{scene_name}[/green].')
raise typer.Exit()
table = Table(title=f'Items in Scene: {scene_name}', padding=(0, 2))
if uuid:
columns = [
('Item ID', 'center', 'cyan'),
('Item Name', 'left', 'cyan'),
('In Group', 'left', 'cyan'),
('Enabled', 'center', None),
('UUID', 'left', 'cyan'),
]
else:
table.title += ' (UUIDs hidden)'
columns = [
('Item ID', 'center', 'cyan'),
('Item Name', 'left', 'cyan'),
@ -76,7 +68,7 @@ def list_(
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for item_id, item_name, is_group, is_enabled, source_uuid in items:
for item_id, item_name, is_group, is_enabled in items:
if is_group:
resp = ctx.obj.get_group_scene_item_list(item_name)
group_items = sorted(
@ -85,19 +77,12 @@ def list_(
gi.get('sceneItemId'),
gi.get('sourceName'),
gi.get('sceneItemEnabled'),
gi.get('sourceUuid', 'N/A'), # Include source UUID
)
for gi in resp.scene_items
),
key=lambda x: x[0], # Sort by sceneItemId
)
for (
group_item_id,
group_item_name,
group_item_enabled,
group_item_source_uuid,
) in group_items:
if uuid:
for group_item_id, group_item_name, group_item_enabled in group_items:
table.add_row(
str(group_item_id),
group_item_name,
@ -105,25 +90,6 @@ def list_(
':white_heavy_check_mark:'
if is_enabled and group_item_enabled
else ':x:',
group_item_source_uuid,
)
else:
table.add_row(
str(group_item_id),
group_item_name,
item_name,
':white_heavy_check_mark:'
if is_enabled and group_item_enabled
else ':x:',
)
else:
if uuid:
table.add_row(
str(item_id),
item_name,
'',
':white_heavy_check_mark:' if is_enabled else ':x:',
source_uuid,
)
else:
table.add_row(
@ -133,7 +99,7 @@ def list_(
':white_heavy_check_mark:' if is_enabled else ':x:',
)
console.out.print(table)
out_console.print(table)
def _validate_sources(
@ -144,21 +110,19 @@ def _validate_sources(
) -> bool:
"""Validate the scene name and item name."""
if not validate.scene_in_scenes(ctx, scene_name):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
err_console.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
return False
if group:
if not validate.item_in_scene_item_list(ctx, scene_name, group):
console.err.print(
err_console.print(
f'Group [yellow]{group}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
return False
else:
if not validate.item_in_scene_item_list(ctx, scene_name, item_name):
console.err.print(
f'Item [yellow]{item_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow]. Is the item in a group? '
f'If so use the [yellow]--group[/yellow] option to specify the parent group.\n'
'Use `obsws-cli sceneitem list` for a list of items in the scene.'
err_console.print(
f'Item [yellow]{item_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
return False
@ -177,12 +141,23 @@ def _get_scene_name_and_item_id(
scene_item_id = item.get('sceneItemId')
break
else:
console.err.print(
err_console.print(
f'Item [yellow]{item_name}[/yellow] not found in group [yellow]{group}[/yellow].'
)
raise typer.Exit(1)
else:
try:
resp = ctx.obj.get_scene_item_id(scene_name, item_name)
except obsws.error.OBSSDKRequestError as e:
if e.code == 600:
err_console.print(
f'Item [yellow]{item_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow]. Is the item in a group? '
'If so use the --group option to specify the parent group. '
'Use `obsws-cli sceneitem list` for a list of items in the scene.'
)
raise typer.Exit(1)
else:
raise
scene_item_id = resp.scene_item_id
return scene_name, scene_item_id
@ -216,7 +191,7 @@ def show(
)
if group:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in group [green]{group}[/green] in scene [green]{old_scene_name}[/green] has been shown.'
)
else:
@ -224,7 +199,7 @@ def show(
# This is to avoid confusion with the parent group name
# which is not the same as the scene name
# and is not needed in this case
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in scene [green]{scene_name}[/green] has been shown.'
)
@ -257,7 +232,7 @@ def hide(
)
if group:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in group [green]{group}[/green] in scene [green]{old_scene_name}[/green] has been hidden.'
)
else:
@ -265,7 +240,7 @@ def hide(
# This is to avoid confusion with the parent group name
# which is not the same as the scene name
# and is not needed in this case
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in scene [green]{scene_name}[/green] has been hidden.'
)
@ -307,11 +282,11 @@ def toggle(
if group:
if new_state:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in group [green]{group}[/green] in scene [green]{old_scene_name}[/green] has been shown.'
)
else:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in group [green]{group}[/green] in scene [green]{old_scene_name}[/green] has been hidden.'
)
else:
@ -320,11 +295,11 @@ def toggle(
# which is not the same as the scene name
# and is not needed in this case
if new_state:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in scene [green]{scene_name}[/green] has been shown.'
)
else:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in scene [green]{scene_name}[/green] has been hidden.'
)
@ -358,7 +333,7 @@ def visible(
)
if group:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in group [green]{group}[/green] in scene [green]{old_scene_name}[/green] is currently {"visible" if enabled.scene_item_enabled else "hidden"}.'
)
else:
@ -366,7 +341,7 @@ def visible(
# This is to avoid confusion with the parent group name
# which is not the same as the scene name
# and is not needed in this case
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in scene [green]{scene_name}[/green] is currently {"visible" if enabled.scene_item_enabled else "hidden"}.'
)
@ -472,7 +447,7 @@ def transform(
transform['scaleY'] = scale_y
if not transform:
console.err.print('No transform options provided.')
err_console.print('No transform options provided.')
raise typer.Exit(1)
transform = ctx.obj.set_scene_item_transform(
@ -482,7 +457,7 @@ def transform(
)
if group:
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in group [green]{group}[/green] in scene [green]{old_scene_name}[/green] has been transformed.'
)
else:
@ -490,6 +465,6 @@ def transform(
# This is to avoid confusion with the parent group name
# which is not the same as the scene name
# and is not needed in this case
console.out.print(
out_console.print(
f'Item [green]{item_name}[/green] in scene [green]{scene_name}[/green] has been transformed.'
)

View File

@ -5,11 +5,15 @@ from typing import Annotated
import obsws_python as obsws
import typer
from rich.console import Console
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(
stderr=True,
)
@app.callback()
@ -74,17 +78,17 @@ def save(
except obsws.error.OBSSDKRequestError as e:
match e.code:
case 403:
console.err.print(
err_console.print(
'The [yellow]image format[/yellow] (file extension) must be included in the file name, '
"for example: '/path/to/screenshot.png'.",
)
raise typer.Exit(1)
case 600:
console.err.print(
err_console.print(
f'No source was found by the name of [yellow]{source_name}[/yellow]'
)
raise typer.Exit(1)
case _:
raise
console.out.print(f'Screenshot saved to [green]{output_path}[/green].')
out_console.print(f'Screenshot saved to [green]{output_path}[/green].')

View File

@ -22,8 +22,6 @@ class Settings(UserDict):
"""
PREFIX = 'OBS_'
def __init__(self, *args, **kwargs):
"""Initialize the Settings object."""
kwargs.update(
@ -36,25 +34,19 @@ class Settings(UserDict):
def __getitem__(self, key: str) -> SettingsValue:
"""Get a setting value by key."""
key = key.upper()
if not key.startswith(Settings.PREFIX):
key = f'{Settings.PREFIX}{key}'
return self.data[key]
if not key.startswith('OBS_'):
key = f'OBS_{key}'
return self.data[key.upper()]
def __setitem__(self, key: str, value: SettingsValue):
"""Set a setting value by key."""
key = key.upper()
if not key.startswith(Settings.PREFIX):
key = f'{Settings.PREFIX}{key}'
self.data[key] = value
if not key.startswith('OBS_'):
key = f'OBS_{key}'
self.data[key.upper()] = value
_settings = Settings(
OBS_HOST='localhost',
OBS_PORT=4455,
OBS_PASSWORD='',
OBS_TIMEOUT=5,
OBS_DEBUG=False,
OBS_HOST='localhost', OBS_PORT=4455, OBS_PASSWORD='', OBS_TIMEOUT=5
)

View File

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

View File

@ -1,11 +1,13 @@
"""module containing commands for manipulating studio mode in OBS."""
import typer
from rich.console import Console
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -17,14 +19,14 @@ def main():
def enable(ctx: typer.Context):
"""Enable studio mode."""
ctx.obj.set_studio_mode_enabled(True)
console.out.print('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)
console.out.print('Studio mode has been disabled.')
out_console.print('Studio mode has been disabled.')
@app.command('toggle | tg')
@ -33,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)
console.out.print('Studio mode is now disabled.')
out_console.print('Studio mode is now disabled.')
else:
ctx.obj.set_studio_mode_enabled(True)
console.out.print('Studio mode is now enabled.')
out_console.print('Studio mode is now enabled.')
@app.command('status | ss')
@ -44,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:
console.out.print('Studio mode is enabled.')
out_console.print('Studio mode is enabled.')
else:
console.out.print('Studio mode is disabled.')
out_console.print('Studio mode is disabled.')

View File

@ -1,11 +1,13 @@
"""module containing commands for manipulating virtual camera in OBS."""
import typer
from rich.console import Console
from . import console
from .alias import AliasGroup
app = typer.Typer(cls=AliasGroup)
out_console = Console()
err_console = Console(stderr=True, style='bold red')
@app.callback()
@ -17,14 +19,14 @@ def main():
def start(ctx: typer.Context):
"""Start the virtual camera."""
ctx.obj.start_virtual_cam()
console.out.print('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()
console.out.print('Virtual camera stopped.')
out_console.print('Virtual camera stopped.')
@app.command('toggle | tg')
@ -32,9 +34,9 @@ def toggle(ctx: typer.Context):
"""Toggle the virtual camera."""
resp = ctx.obj.toggle_virtual_cam()
if resp.output_active:
console.out.print('Virtual camera is enabled.')
out_console.print('Virtual camera is enabled.')
else:
console.out.print('Virtual camera is disabled.')
out_console.print('Virtual camera is disabled.')
@app.command('status | ss')
@ -42,6 +44,6 @@ def status(ctx: typer.Context):
"""Get the status of the virtual camera."""
resp = ctx.obj.get_virtual_cam_status()
if resp.output_active:
console.out.print('Virtual camera is enabled.')
out_console.print('Virtual camera is enabled.')
else:
console.out.print('Virtual camera is disabled.')
out_console.print('Virtual camera is disabled.')