onyx_online/obsws-cli
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/obsws-cli.git synced 2026-04-07 01:23:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: onyx_online:switch-to-cyclopts
Branches Tags
onyx_online:main onyx_online:validation-callbacks onyx_online:add-media onyx_online:add-settings onyx_online:extend-input onyx_online:switch-to-cyclopts onyx_online:add-record-split-chapter onyx_online:add-styles onyx_online:add-filter-subtyper
onyx_online:v0.24.10 onyx_online:v0.24.9 onyx_online:v0.24.8 onyx_online:v0.24.7 onyx_online:v0.24.6 onyx_online:v0.24.5 onyx_online:v0.24.4 onyx_online:v0.24.3 onyx_online:v0.24.2 onyx_online:v0.24.1 onyx_online:v0.24.0 onyx_online:v0.23.0 onyx_online:v0.22.0 onyx_online:v0.21.0 onyx_online:v0.20.2 onyx_online:v0.20.1 onyx_online:v0.20.0
..
compare: onyx_online:v0.21.0
Branches Tags
onyx_online:main onyx_online:validation-callbacks onyx_online:add-media onyx_online:add-settings onyx_online:extend-input onyx_online:switch-to-cyclopts onyx_online:add-record-split-chapter onyx_online:add-styles onyx_online:add-filter-subtyper
onyx_online:v0.24.10 onyx_online:v0.24.9 onyx_online:v0.24.8 onyx_online:v0.24.7 onyx_online:v0.24.6 onyx_online:v0.24.5 onyx_online:v0.24.4 onyx_online:v0.24.3 onyx_online:v0.24.2 onyx_online:v0.24.1 onyx_online:v0.24.0 onyx_online:v0.23.0 onyx_online:v0.22.0 onyx_online:v0.21.0 onyx_online:v0.20.2 onyx_online:v0.20.1 onyx_online:v0.20.0

5 Commits
switch-to- ... v0.21.0

Author SHA1 Message Date
onyx-and-iris
48e0f6cecd bump typer dependency. 
release 0.17.0 fixes slow rich imports, see https://github.com/fastapi/typer/releases/tag/0.17.0

This is related to issue #2.

minor version bump
 2025-09-29 04:21:26 +01:00
onyx-and-iris
52e13922dc upd test delays to 500ms 2025-07-30 08:42:11 +01:00
onyx-and-iris
f335d8ffd2 move the version flag 2025-07-29 08:48:30 +01:00
onyx-and-iris
286cda8066 raise typer.Exit() on empty list queries 2025-07-29 08:17:52 +01:00
onyx-and-iris
e851219ced tests should now pass from fresh install 2025-07-29 08:03:24 +01:00
40 changed files with 1334 additions and 1658 deletions
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-License-Identifier: MIT
__version__ = '0.20.2'
__version__ = '0.21.0'

4
obsws_cli/__init__.py
View File

@@ -2,6 +2,6 @@
#
# SPDX-License-Identifier: MIT
from .app import run
from .app import app
__all__ = ['run']
__all__ = ['app']

73
obsws_cli/alias.py Normal file
View File

@@ -0,0 +1,73 @@
"""module defining a custom group class for handling command name aliases."""
import re
import typer
class RootTyperAliasGroup(typer.core.TyperGroup):
"""A custom group class to handle command name aliases for the root typer."""
def __init__(self, *args, **kwargs):
"""Initialize the AliasGroup."""
super().__init__(*args, **kwargs)
self.no_args_is_help = True
def get_command(self, ctx, cmd_name):
"""Get a command by name."""
match cmd_name:
case 'f':
cmd_name = 'filter'
case 'g':
cmd_name = 'group'
case 'hk':
cmd_name = 'hotkey'
case 'i':
cmd_name = 'input'
case 'prf':
cmd_name = 'profile'
case 'prj':
cmd_name = 'projector'
case 'rc':
cmd_name = 'record'
case 'rb':
cmd_name = 'replaybuffer'
case 'sc':
cmd_name = 'scene'
case 'scc':
cmd_name = 'scenecollection'
case 'si':
cmd_name = 'sceneitem'
case 'ss':
cmd_name = 'screenshot'
case 'st':
cmd_name = 'stream'
case 'sm':
cmd_name = 'studiomode'
case 't':
cmd_name = 'text'
case 'vc':
cmd_name = 'virtualcam'
return super().get_command(ctx, cmd_name)
class SubTyperAliasGroup(typer.core.TyperGroup):
"""A custom group class to handle command name aliases for sub typers."""
_CMD_SPLIT_P = re.compile(r' ?[,|] ?')
def __init__(self, *args, **kwargs):
"""Initialize the AliasGroup."""
super().__init__(*args, **kwargs)
self.no_args_is_help = True
def get_command(self, ctx, cmd_name):
"""Get a command by name."""
cmd_name = self._group_cmd_name(cmd_name)
return super().get_command(ctx, cmd_name)
def _group_cmd_name(self, default_name):
for cmd in self.commands.values():
if cmd.name and default_name in self._CMD_SPLIT_P.split(cmd.name):
return cmd.name
return default_name

208
obsws_cli/app.py
View File

@@ -2,27 +2,18 @@
import importlib
import logging
from dataclasses import dataclass
from typing import Annotated, Any
from typing import Annotated
import obsws_python as obsws
from cyclopts import App, Group, Parameter, config
import typer
from obsws_cli.__about__ import __version__ as version
from . import console, styles
from .context import Context
from .error import OBSWSCLIError
from . import console, settings, styles
from .alias import RootTyperAliasGroup
app = App(
config=config.Env(
'OBS_'
), # Environment variable prefix for configuration parameters
version=version,
usage='[bold][yellow]Usage:[/yellow] [white]obsws-cli [OPTIONS] COMMAND [ARGS]...[/white][/bold]',
)
app.meta.group_parameters = Group('Options', sort_key=0)
for sub_app in (
app = typer.Typer(cls=RootTyperAliasGroup)
for sub_typer in (
'filter',
'group',
'hotkey',
@@ -40,98 +31,137 @@ for sub_app in (
'text',
'virtualcam',
):
module = importlib.import_module(f'.{sub_app}', package=__package__)
app.command(module.app)
module = importlib.import_module(f'.{sub_typer}', package=__package__)
app.add_typer(module.app, name=sub_typer)
@Parameter(name='*')
@dataclass
class OBSConfig:
"""Dataclass to hold OBS connection parameters.
Attributes:
host (str): The hostname or IP address of the OBS WebSocket server.
port (int): The port number of the OBS WebSocket server.
password (str): The password for the OBS WebSocket server, if required.
"""
host: str = 'localhost'
port: int = 4455
password: str = ''
def version_callback(value: bool):
"""Show the version of the CLI."""
if value:
console.out.print(f'obsws-cli version: {version}')
raise typer.Exit()
@dataclass
class StyleConfig:
"""Dataclass to hold style parameters.
Attributes:
name (str): The name of the style to use for console output.
no_border (bool): Whether to style the borders in the console output.
"""
name: str = 'disabled'
no_border: bool = False
def setup_logging(type_, value: Any):
def setup_logging(debug: bool):
"""Set up logging for the application."""
log_level = logging.DEBUG if value else logging.CRITICAL
log_level = logging.DEBUG if debug else logging.CRITICAL
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
@app.meta.default
def launcher(
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
obs_config: OBSConfig,
style_config: StyleConfig,
def validate_style(value: str):
"""Validate and return the style."""
if value not in styles.registry:
raise typer.BadParameter(
f'Invalid style: {value}. Available styles: {", ".join(styles.registry.keys())}'
)
return value
@app.callback()
def main(
ctx: typer.Context,
host: Annotated[
str,
typer.Option(
'--host',
'-H',
envvar='OBS_HOST',
help='WebSocket host',
show_default='localhost',
),
] = settings.get('host'),
port: Annotated[
int,
typer.Option(
'--port',
'-P',
envvar='OBS_PORT',
help='WebSocket port',
show_default=4455,
),
] = settings.get('port'),
password: Annotated[
str,
typer.Option(
'--password',
'-p',
envvar='OBS_PASSWORD',
help='WebSocket password',
show_default=False,
),
] = settings.get('password'),
timeout: Annotated[
int,
typer.Option(
'--timeout',
'-T',
envvar='OBS_TIMEOUT',
help='WebSocket timeout',
show_default=5,
),
] = settings.get('timeout'),
style: Annotated[
str,
typer.Option(
'--style',
'-s',
envvar='OBS_STYLE',
help='Set the style for the CLI output',
show_default='disabled',
callback=validate_style,
),
] = settings.get('style'),
no_border: Annotated[
bool,
typer.Option(
'--no-border',
'-b',
envvar='OBS_STYLE_NO_BORDER',
help='Disable table border styling in the CLI output',
show_default=False,
),
] = settings.get('style_no_border'),
version: Annotated[
bool,
typer.Option(
'--version',
'-v',
is_eager=True,
help='Show the CLI version and exit',
show_default=False,
callback=version_callback,
),
] = False,
debug: Annotated[
bool,
Parameter(validator=setup_logging, help='Enable debug logging'),
] = False,
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'),
):
"""Command line interface for the OBS WebSocket API."""
with obsws.ReqClient(
host=obs_config.host,
port=obs_config.port,
password=obs_config.password,
) as client:
additional_kwargs = {}
command, bound, ignored = app.parse_args(tokens)
if 'ctx' in ignored:
# If 'ctx' is in ignored, it means it was not passed as an argument
# and we need to add it to the bound arguments.
additional_kwargs['ctx'] = ignored['ctx'](
client,
styles.request_style_obj(style_config.name, style_config.no_border),
)
return command(*bound.args, **bound.kwargs, **additional_kwargs)
"""obsws_cli is a command line interface for the OBS WebSocket API."""
ctx.ensure_object(dict)
ctx.obj['obsws'] = ctx.with_resource(
obsws.ReqClient(host=host, port=port, password=password, timeout=timeout)
)
ctx.obj['style'] = styles.request_style_obj(style, no_border)
@app.command
def obs_version(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
@app.command()
def obs_version(ctx: typer.Context):
"""Get the OBS Client and WebSocket versions."""
resp = ctx.client.get_version()
resp = ctx.obj['obsws'].get_version()
console.out.print(
f'OBS Client version: {console.highlight(ctx, resp.obs_version)}'
f' with WebSocket version: {console.highlight(ctx, resp.obs_web_socket_version)}'
)
def run():
"""Run the OBS WebSocket CLI application.
Handles exceptions and prints error messages to the console.
"""
try:
app.meta()
except OBSWSCLIError as e:
console.err.print(f'Error: {e}')
return e.code

7
obsws_cli/console.py
View File

@@ -1,13 +1,12 @@
"""module for console output handling in obsws_cli."""
import typer
from rich.console import Console
from .context import Context
out = Console()
err = Console(stderr=True, style='bold red')
def highlight(ctx: Context, text: str) -> str:
def highlight(ctx: typer.Context, text: str) -> str:
"""Highlight text using the current context's style."""
return f'[{ctx.style.highlight}]{text}[/{ctx.style.highlight}]'
return f'[{ctx.obj["style"].highlight}]{text}[/{ctx.obj["style"].highlight}]'

15
obsws_cli/context.py
View File

@@ -1,15 +0,0 @@
"""module for managing the application context."""
from dataclasses import dataclass
import obsws_python as obsws
from . import styles
@dataclass
class Context:
"""Context for the application, holding OBS and style configurations."""
client: obsws.ReqClient
style: styles.Style

10
obsws_cli/enum.py
View File

@@ -1,10 +0,0 @@
"""module for exit codes used in the application."""
from enum import IntEnum, auto
class ExitCode(IntEnum):
"""Exit codes for the application."""
SUCCESS = 0
ERROR = auto()

11
obsws_cli/error.py
View File

@@ -1,11 +0,0 @@
"""module containing error handling for OBS WebSocket CLI."""
class OBSWSCLIError(Exception):
"""Base class for OBS WebSocket CLI errors."""
def __init__(self, message: str, code: int = 1):
"""Initialize the error with a message and an optional code."""
super().__init__(message)
self.message = message
self.code = code

216
obsws_cli/filter.py
View File

@@ -3,46 +3,44 @@
from typing import Annotated, Optional
import obsws_python as obsws
from cyclopts import App, Parameter
import typer
from rich.table import Table
from rich.text import Text
from . import console, util
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='filter', help='Commands for managing filters in OBS sources')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
@app.callback()
def main():
"""Control filters in OBS scenes."""
@app.command('list | ls')
def list_(
source_name: Optional[str] = None,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
source_name: Annotated[
Optional[str],
typer.Argument(
show_default='The current scene',
help='The source to list filters for',
),
] = None,
):
"""List filters for a source.
Parameters
----------
source_name : str, optional
The name of the source to list filters for. If not provided, the current program scene's source will be used.
ctx : Context
The context containing the OBS client and other settings.
"""
"""List filters for a source."""
if not source_name:
source_name = ctx.client.get_current_program_scene().scene_name
source_name = ctx.obj['obsws'].get_current_program_scene().scene_name
try:
resp = ctx.client.get_source_filter_list(source_name)
resp = ctx.obj['obsws'].get_source_filter_list(source_name)
except obsws.error.OBSSDKRequestError as e:
if e.code == 600:
raise OBSWSCLIError(
f'No source found by the name of [yellow]{source_name}[/yellow].',
code=ExitCode.ERROR,
console.err.print(
f'No source was found by the name of [yellow]{source_name}[/yellow].'
)
raise typer.Exit(1)
else:
raise
@@ -50,25 +48,25 @@ def list_(
console.out.print(
f'No filters found for source {console.highlight(ctx, source_name)}'
)
return
raise typer.Exit()
table = Table(
title=f'Filters for Source: {source_name}',
padding=(0, 2),
border_style=ctx.style.border,
border_style=ctx.obj['style'].border,
)
columns = [
(Text('Filter Name', justify='center'), 'left', ctx.style.column),
(Text('Kind', justify='center'), 'left', ctx.style.column),
(Text('Filter Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Kind', justify='center'), 'left', ctx.obj['style'].column),
(Text('Enabled', justify='center'), 'center', None),
(Text('Settings', justify='center'), 'center', ctx.style.column),
(Text('Settings', justify='center'), 'center', ctx.obj['style'].column),
]
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for filter in resp.filters:
resp = ctx.client.get_source_filter_default_settings(filter['filterKind'])
resp = ctx.obj['obsws'].get_source_filter_default_settings(filter['filterKind'])
settings = resp.default_filter_settings | filter['filterSettings']
table.add_row(
@@ -86,100 +84,93 @@ def list_(
console.out.print(table)
def _get_filter_enabled(ctx: Context, source_name: str, filter_name: str):
def _get_filter_enabled(ctx: typer.Context, source_name: str, filter_name: str):
"""Get the status of a filter for a source."""
resp = ctx.client.get_source_filter(source_name, filter_name)
resp = ctx.obj['obsws'].get_source_filter(source_name, filter_name)
return resp.filter_enabled
@app.command(name=['enable', 'on'])
@app.command('enable | on')
def enable(
source_name: str,
filter_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
source_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The source to enable the filter for'
),
],
filter_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The name of the filter to enable'
),
],
):
"""Enable a filter for a source.
Parameters
----------
source_name : str
The name of the source to enable the filter for.
filter_name : str
The name of the filter to enable.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Enable a filter for a source."""
if _get_filter_enabled(ctx, source_name, filter_name):
raise OBSWSCLIError(
f'Filter [yellow]{filter_name}[/yellow] is already enabled for source [yellow]{source_name}[/yellow]',
code=ExitCode.ERROR,
console.err.print(
f'Filter [yellow]{filter_name}[/yellow] is already enabled for source [yellow]{source_name}[/yellow]'
)
raise typer.Exit(1)
ctx.client.set_source_filter_enabled(source_name, filter_name, enabled=True)
ctx.obj['obsws'].set_source_filter_enabled(source_name, filter_name, enabled=True)
console.out.print(
f'Enabled filter {console.highlight(ctx, filter_name)} for source {console.highlight(ctx, source_name)}'
)
@app.command(name=['disable', 'off'])
@app.command('disable | off')
def disable(
source_name: str,
filter_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
source_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The source to disable the filter for'
),
],
filter_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The name of the filter to disable'
),
],
):
"""Disable a filter for a source.
Parameters
----------
source_name : str
The name of the source to disable the filter for.
filter_name : str
The name of the filter to disable.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Disable a filter for a source."""
if not _get_filter_enabled(ctx, source_name, filter_name):
raise OBSWSCLIError(
f'Filter [yellow]{filter_name}[/yellow] is already disabled for source [yellow]{source_name}[/yellow]',
code=ExitCode.ERROR,
console.err.print(
f'Filter [yellow]{filter_name}[/yellow] is already disabled for source [yellow]{source_name}[/yellow]'
)
raise typer.Exit(1)
ctx.client.set_source_filter_enabled(source_name, filter_name, enabled=False)
ctx.obj['obsws'].set_source_filter_enabled(source_name, filter_name, enabled=False)
console.out.print(
f'Disabled filter {console.highlight(ctx, filter_name)} for source {console.highlight(ctx, source_name)}'
)
@app.command(name=['toggle', 'tg'])
@app.command('toggle | tg')
def toggle(
source_name: str,
filter_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
source_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The source to toggle the filter for'
),
],
filter_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The name of the filter to toggle'
),
],
):
"""Toggle a filter for a source.
Parameters
----------
source_name : str
The name of the source to toggle the filter for.
filter_name : str
The name of the filter to toggle.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Toggle a filter for a source."""
is_enabled = _get_filter_enabled(ctx, source_name, filter_name)
new_state = not is_enabled
ctx.client.set_source_filter_enabled(source_name, filter_name, enabled=new_state)
ctx.obj['obsws'].set_source_filter_enabled(
source_name, filter_name, enabled=new_state
)
if new_state:
console.out.print(
f'Enabled filter {console.highlight(ctx, filter_name)} for source {console.highlight(ctx, source_name)}'
@@ -190,26 +181,23 @@ def toggle(
)
@app.command(name=['status', 'ss'])
@app.command('status | ss')
def status(
source_name: str,
filter_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
source_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The source to get the filter status for'
),
],
filter_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='The name of the filter to get the status for'
),
],
):
"""Get the status of a filter for a source.
Parameters
----------
source_name : str
The name of the source to check the filter status for.
filter_name : str
The name of the filter to check the status for.
ctx : Context
The context containing the OBS client and other settings.
"""
"""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(

230
obsws_cli/group.py
View File

@@ -2,47 +2,42 @@
from typing import Annotated, Optional
from cyclopts import App, Parameter
import typer
from rich.table import Table
from rich.text import Text
from . import console, util, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
from .protocols import DataclassProtocol
app = App(name='group', help='Commands for managing groups in OBS scenes')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
@app.callback()
def main():
"""Control groups in OBS scenes."""
@app.command('list | ls')
def list_(
scene_name: Optional[str] = None,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
Optional[str],
typer.Argument(
show_default='The current scene',
help='Scene name to list groups for',
),
] = None,
):
"""List groups in a scene.
Parameters
----------
scene_name : str, optional
The name of the scene to list groups for. If not provided, the current program scene
will be used.
ctx : Context
The context containing the OBS client and other settings.
"""
"""List groups in a scene."""
if not scene_name:
scene_name = ctx.client.get_current_program_scene().scene_name
scene_name = ctx.obj['obsws'].get_current_program_scene().scene_name
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f"Scene '{scene_name}' not found.")
raise typer.Exit(1)
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
groups = [
(item.get('sceneItemId'), item.get('sourceName'), item.get('sceneItemEnabled'))
for item in resp.scene_items
@@ -53,17 +48,17 @@ def list_(
console.out.print(
f'No groups found in scene {console.highlight(ctx, scene_name)}.'
)
return
raise typer.Exit()
table = Table(
title=f'Groups in Scene: {scene_name}',
padding=(0, 2),
border_style=ctx.style.border,
border_style=ctx.obj['style'].border,
)
columns = [
(Text('ID', justify='center'), 'center', ctx.style.column),
(Text('Group Name', justify='center'), 'left', ctx.style.column),
(Text('ID', justify='center'), 'center', ctx.obj['style'].column),
(Text('Group Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Enabled', justify='center'), 'center', None),
]
for heading, justify, style in columns:
@@ -92,40 +87,30 @@ def _get_group(group_name: str, resp: DataclassProtocol) -> dict | None:
return group
@app.command(name=['show', 'sh'])
@app.command('show | sh')
def show(
scene_name: str,
group_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Scene name the group is in'),
],
group_name: Annotated[
str, typer.Argument(..., show_default=False, help='Group name to show')
],
):
"""Show a group in a scene.
Parameters
----------
scene_name : str
The name of the scene where the group is located.
group_name : str
The name of the group to show.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Show a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f"Scene '{scene_name}' not found.")
raise typer.Exit(1)
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
raise OBSWSCLIError(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].',
code=ExitCode.ERROR,
console.err.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
ctx.client.set_scene_item_enabled(
ctx.obj['obsws'].set_scene_item_enabled(
scene_name=scene_name,
item_id=int(group.get('sceneItemId')),
enabled=True,
@@ -134,40 +119,29 @@ def show(
console.out.print(f'Group {console.highlight(ctx, group_name)} is now visible.')
@app.command(name=['hide', 'h'])
@app.command('hide | h')
def hide(
scene_name: str,
group_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the group is in')
],
group_name: Annotated[
str, typer.Argument(..., show_default=False, help='Group name to hide')
],
):
"""Hide a group in a scene.
Parameters
----------
scene_name : str
The name of the scene where the group is located.
group_name : str
The name of the group to hide.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Hide a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
raise OBSWSCLIError(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].',
code=ExitCode.ERROR,
console.err.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
ctx.client.set_scene_item_enabled(
ctx.obj['obsws'].set_scene_item_enabled(
scene_name=scene_name,
item_id=int(group.get('sceneItemId')),
enabled=False,
@@ -176,41 +150,30 @@ def hide(
console.out.print(f'Group {console.highlight(ctx, group_name)} is now hidden.')
@app.command(name=['toggle', 'tg'])
@app.command('toggle | tg')
def toggle(
scene_name: str,
group_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the group is in')
],
group_name: Annotated[
str, typer.Argument(..., show_default=False, help='Group name to toggle')
],
):
"""Toggle a group in a scene.
Parameters
----------
scene_name : str
The name of the scene where the group is located.
group_name : str
The name of the group to toggle.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Toggle a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
raise OBSWSCLIError(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].',
code=ExitCode.ERROR,
console.err.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
new_state = not group.get('sceneItemEnabled')
ctx.client.set_scene_item_enabled(
ctx.obj['obsws'].set_scene_item_enabled(
scene_name=scene_name,
item_id=int(group.get('sceneItemId')),
enabled=new_state,
@@ -222,40 +185,29 @@ def toggle(
console.out.print(f'Group {console.highlight(ctx, group_name)} is now hidden.')
@app.command(name=['status', 'ss'])
@app.command('status | ss')
def status(
scene_name: str,
group_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the group is in')
],
group_name: Annotated[
str, typer.Argument(..., show_default=False, help='Group name to check status')
],
):
"""Get the status of a group in a scene.
Parameters
----------
scene_name : str
The name of the scene where the group is located.
group_name : str
The name of the group to check.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Get the status of a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
if (group := _get_group(group_name, resp)) is None:
raise OBSWSCLIError(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].',
code=ExitCode.ERROR,
console.err.print(
f'Group [yellow]{group_name}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
)
raise typer.Exit(1)
enabled = ctx.client.get_scene_item_enabled(
enabled = ctx.obj['obsws'].get_scene_item_enabled(
scene_name=scene_name,
item_id=int(group.get('sceneItemId')),
)

108
obsws_cli/hotkey.py
View File

@@ -2,40 +2,41 @@
from typing import Annotated
from cyclopts import App, Parameter
import typer
from rich.table import Table
from rich.text import Text
from . import console
from .context import Context
from .alias import SubTyperAliasGroup
app = App(name='hotkey', help='Commands for managing hotkeys in OBS')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
@app.callback()
def main():
"""Control hotkeys in OBS."""
@app.command('list | ls')
def list_(
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
):
"""List all hotkeys.
"""List all hotkeys."""
resp = ctx.obj['obsws'].get_hotkey_list()
Parameters
----------
ctx : Context
The context containing the OBS client to interact with.
"""
resp = ctx.client.get_hotkey_list()
if not resp.hotkeys:
console.out.print('No hotkeys found.')
raise typer.Exit()
table = Table(
title='Hotkeys',
padding=(0, 2),
border_style=ctx.style.border,
border_style=ctx.obj['style'].border,
)
table.add_column(
Text('Hotkey Name', justify='center'),
justify='left',
style=ctx.style.column,
style=ctx.obj['style'].column,
)
for i, hotkey in enumerate(resp.hotkeys):
@@ -44,53 +45,40 @@ def list_(
console.out.print(table)
@app.command(name=['trigger', 'tr'])
@app.command('trigger | tr')
def trigger(
hotkey: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
hotkey: Annotated[
str, typer.Argument(..., show_default=False, help='The hotkey to trigger')
],
):
"""Trigger a hotkey by name.
Parameters
----------
hotkey : str
The name of the hotkey to trigger.
ctx : Context
The context containing the OBS client to interact with.
"""
ctx.client.trigger_hotkey_by_name(hotkey)
"""Trigger a hotkey by name."""
ctx.obj['obsws'].trigger_hotkey_by_name(hotkey)
@app.command(name=['trigger-sequence', 'trs'])
@app.command('trigger-sequence | trs')
def trigger_sequence(
key_id: str,
/,
shift: bool = False,
ctrl: bool = False,
alt: bool = False,
cmd: bool = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
key_id: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='The OBS key ID to trigger, see https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#hotkey for more info',
),
],
shift: Annotated[
bool, typer.Option(..., help='Press shift when triggering the hotkey')
] = False,
ctrl: Annotated[
bool, typer.Option(..., help='Press control when triggering the hotkey')
] = False,
alt: Annotated[
bool, typer.Option(..., help='Press alt when triggering the hotkey')
] = False,
cmd: Annotated[
bool, typer.Option(..., help='Press cmd when triggering the hotkey')
] = False,
):
"""Trigger a hotkey by sequence.
Parameters
----------
key_id : str
The OBS key ID to trigger, see https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#hotkey for more info
shift : bool, optional
Press shift when triggering the hotkey (default is False)
ctrl : bool, optional
Press control when triggering the hotkey (default is False)
alt : bool, optional
Press alt when triggering the hotkey (default is False)
cmd : bool, optional
Press cmd when triggering the hotkey (default is False)
ctx : Context
The context containing the OBS client to interact with.
"""
ctx.client.trigger_hotkey_by_key_sequence(key_id, shift, ctrl, alt, cmd)
"""Trigger a hotkey by sequence."""
ctx.obj['obsws'].trigger_hotkey_by_key_sequence(key_id, shift, ctrl, alt, cmd)

164
obsws_cli/input.py
View File

@@ -3,50 +3,33 @@
from typing import Annotated
import obsws_python as obsws
from cyclopts import App, Parameter
import typer
from rich.table import Table
from rich.text import Text
from . import console, util, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='input', help='Commands for managing inputs in OBS')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
@app.callback()
def main():
"""Control inputs in OBS."""
@app.command('list | ls')
def list_(
input: bool = False,
output: bool = False,
colour: bool = False,
ffmpeg: bool = False,
vlc: bool = False,
uuid: bool = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
input: Annotated[bool, typer.Option(help='Filter by input type.')] = False,
output: Annotated[bool, typer.Option(help='Filter by output type.')] = False,
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.
Parameters
----------
input:
Filter by input type.
output:
Filter by output type.
colour:
Filter by colour source type.
ffmpeg:
Filter by ffmpeg source type.
vlc:
Filter by VLC source type.
uuid:
Show UUIDs of inputs.
ctx:
The context containing the client and style.
"""
resp = ctx.client.get_input_list()
"""List all inputs."""
resp = ctx.obj['obsws'].get_input_list()
kinds = []
if input:
@@ -60,7 +43,7 @@ def list_(
if vlc:
kinds.append('vlc')
if not any([input, output, colour, ffmpeg, vlc]):
kinds = ctx.client.get_input_kind_list(False).input_kinds
kinds = ctx.obj['obsws'].get_input_kind_list(False).input_kinds
inputs = sorted(
(
@@ -74,21 +57,21 @@ def list_(
)
if not inputs:
console.out.print('No inputs found matching the specified filters.')
return
console.out.print('No inputs found.')
raise typer.Exit()
table = Table(title='Inputs', padding=(0, 2), border_style=ctx.style.border)
table = Table(title='Inputs', padding=(0, 2), border_style=ctx.obj['style'].border)
if uuid:
columns = [
(Text('Input Name', justify='center'), 'left', ctx.style.column),
(Text('Kind', justify='center'), 'center', ctx.style.column),
(Text('Input Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Kind', justify='center'), 'center', ctx.obj['style'].column),
(Text('Muted', justify='center'), 'center', None),
(Text('UUID', justify='center'), 'left', ctx.style.column),
(Text('UUID', justify='center'), 'left', ctx.obj['style'].column),
]
else:
columns = [
(Text('Input Name', justify='center'), 'left', ctx.style.column),
(Text('Kind', justify='center'), 'center', ctx.style.column),
(Text('Input Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Kind', justify='center'), 'center', ctx.obj['style'].column),
(Text('Muted', justify='center'), 'center', None),
]
for heading, justify, style in columns:
@@ -97,7 +80,7 @@ def list_(
for input_name, input_kind, input_uuid in inputs:
input_mark = ''
try:
input_muted = ctx.client.get_input_mute(name=input_name).input_muted
input_muted = ctx.obj['obsws'].get_input_mute(name=input_name).input_muted
input_mark = util.check_mark(input_muted)
except obsws.error.OBSSDKRequestError as e:
if e.code == 604: # Input does not support audio
@@ -122,30 +105,19 @@ def list_(
console.out.print(table)
@app.command(name=['mute', 'm'])
@app.command('mute | m')
def mute(
input_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
input_name: Annotated[
str, typer.Argument(..., show_default=False, help='Name of the input to mute.')
],
):
"""Mute an input.
Parameters
----------
input_name: str
Name of the input to mute.
ctx: Context
The context containing the client and style.
"""
"""Mute an input."""
if not validate.input_in_inputs(ctx, input_name):
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
ctx.client.set_input_mute(
ctx.obj['obsws'].set_input_mute(
name=input_name,
muted=True,
)
@@ -153,30 +125,20 @@ def mute(
console.out.print(f'Input {console.highlight(ctx, input_name)} muted.')
@app.command(name=['unmute', 'um'])
@app.command('unmute | um')
def unmute(
input_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Name of the input to unmute.'),
],
):
"""Unmute an input.
Parameters
----------
input_name: str
Name of the input to unmute.
ctx: Context
The context containing the client and style.
"""
"""Unmute an input."""
if not validate.input_in_inputs(ctx, input_name):
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
ctx.client.set_input_mute(
ctx.obj['obsws'].set_input_mute(
name=input_name,
muted=False,
)
@@ -184,33 +146,23 @@ def unmute(
console.out.print(f'Input {console.highlight(ctx, input_name)} unmuted.')
@app.command(name=['toggle', 'tg'])
@app.command('toggle | tg')
def toggle(
input_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Name of the input to toggle.'),
],
):
"""Toggle an input.
Parameters
----------
input_name: str
Name of the input to toggle.
ctx: Context
The context containing the client and style.
"""
"""Toggle an input."""
if not validate.input_in_inputs(ctx, input_name):
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.client.get_input_mute(name=input_name)
resp = ctx.obj['obsws'].get_input_mute(name=input_name)
new_state = not resp.input_muted
ctx.client.set_input_mute(
ctx.obj['obsws'].set_input_mute(
name=input_name,
muted=new_state,
)

157
obsws_cli/profile.py
View File

@@ -2,45 +2,40 @@
from typing import Annotated
from cyclopts import App, Parameter
import typer
from rich.table import Table
from rich.text import Text
from . import console, util, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='profile', help='Commands for managing profiles in OBS')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
def list_(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""List profiles.
@app.callback()
def main():
"""Control profiles in OBS."""
Parameters
----------
ctx: Context
The context containing the client and style.
"""
resp = ctx.client.get_profile_list()
@app.command('list | ls')
def list_(ctx: typer.Context):
"""List profiles."""
resp = ctx.obj['obsws'].get_profile_list()
table = Table(title='Profiles', padding=(0, 2), border_style=ctx.style.border)
if not resp.profiles:
console.out.print('No profiles found.')
raise typer.Exit()
table = Table(
title='Profiles', padding=(0, 2), border_style=ctx.obj['style'].border
)
columns = [
(Text('Profile Name', justify='center'), 'left', ctx.style.column),
(Text('Profile Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Current', justify='center'), 'center', None),
]
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
if not resp.profiles:
console.out.print('No profiles found.')
return
for profile in resp.profiles:
table.add_row(
profile,
@@ -52,110 +47,70 @@ def list_(
console.out.print(table)
@app.command(name=['current', 'get'])
def current(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the current profile.
Parameters
----------
ctx: Context
The context containing the client and style.
"""
resp = ctx.client.get_profile_list()
@app.command('current | get')
def current(ctx: typer.Context):
"""Get the current profile."""
resp = ctx.obj['obsws'].get_profile_list()
console.out.print(
f'Current profile: {console.highlight(ctx, resp.current_profile_name)}'
)
@app.command(name=['switch', 'set'])
@app.command('switch | set')
def switch(
profile_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
profile_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='Name of the profile to switch to'
),
],
):
"""Switch to a profile.
Parameters
----------
profile_name: str
Name of the profile to switch to.
ctx: Context
The context containing the client and style.
"""
"""Switch to a profile."""
if not validate.profile_exists(ctx, profile_name):
console.err.print(f'Profile [yellow]{profile_name}[/yellow] not found.')
raise OBSWSCLIError(
f'Profile [yellow]{profile_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
raise typer.Exit(1)
resp = ctx.client.get_profile_list()
resp = ctx.obj['obsws'].get_profile_list()
if resp.current_profile_name == profile_name:
raise OBSWSCLIError(
f'Profile [yellow]{profile_name}[/yellow] is already the current profile.',
code=ExitCode.ERROR,
console.err.print(
f'Profile [yellow]{profile_name}[/yellow] is already the current profile.'
)
raise typer.Exit(1)
ctx.client.set_current_profile(profile_name)
ctx.obj['obsws'].set_current_profile(profile_name)
console.out.print(f'Switched to profile {console.highlight(ctx, profile_name)}.')
@app.command(name=['create', 'new'])
@app.command('create | new')
def create(
profile_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
profile_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Name of the profile to create.'),
],
):
"""Create a new profile.
Parameters
----------
profile_name: str
Name of the profile to create.
ctx: Context
The context containing the client and style.
"""
"""Create a new profile."""
if validate.profile_exists(ctx, profile_name):
raise OBSWSCLIError(
f'Profile [yellow]{profile_name}[/yellow] already exists.',
code=ExitCode.ERROR,
)
console.err.print(f'Profile [yellow]{profile_name}[/yellow] already exists.')
raise typer.Exit(1)
ctx.client.create_profile(profile_name)
ctx.obj['obsws'].create_profile(profile_name)
console.out.print(f'Created profile {console.highlight(ctx, profile_name)}.')
@app.command(name=['remove', 'rm'])
@app.command('remove | rm')
def remove(
profile_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
profile_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Name of the profile to remove.'),
],
):
"""Remove a profile.
Parameters
----------
profile_name: str
Name of the profile to remove.
ctx: Context
The context containing the client and style.
"""
"""Remove a profile."""
if not validate.profile_exists(ctx, profile_name):
console.err.print(f'Profile [yellow]{profile_name}[/yellow] not found.')
raise OBSWSCLIError(
f'Profile [yellow]{profile_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
raise typer.Exit(1)
ctx.client.remove_profile(profile_name)
ctx.obj['obsws'].remove_profile(profile_name)
console.out.print(f'Removed profile {console.highlight(ctx, profile_name)}.')

88
obsws_cli/projector.py
View File

@@ -1,57 +1,45 @@
"""module containing commands for manipulating projectors in OBS."""
from typing import Annotated, Optional
from typing import Annotated
from cyclopts import App, Parameter, validators
import typer
from rich.table import Table
from rich.text import Text
from . import console
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='projector', help='Commands for managing projectors in OBS')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list-monitors', 'ls-m'])
def list_monitors(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""List available monitors.
@app.callback()
def main():
"""Control projectors in OBS."""
Parameters
----------
ctx : Context
The context containing the OBS client and configuration.
"""
resp = ctx.client.get_monitor_list()
if not resp.monitors:
console.out.print('No monitors found.')
return
@app.command('list-monitors | ls-m')
def list_monitors(ctx: typer.Context):
"""List available monitors."""
resp = ctx.obj['obsws'].get_monitor_list()
monitors = sorted(
((m['monitorIndex'], m['monitorName']) for m in resp.monitors),
key=lambda m: m[0],
)
if not monitors:
console.out.print('No monitors available.')
return
console.out.print('No monitors found.')
raise typer.Exit()
table = Table(
title='Available Monitors',
padding=(0, 2),
border_style=ctx.style.border,
border_style=ctx.obj['style'].border,
)
table.add_column(
Text('Index', justify='center'), justify='center', style=ctx.style.column
Text('Index', justify='center'), justify='center', style=ctx.obj['style'].column
)
table.add_column(
Text('Name', justify='center'), justify='left', style=ctx.style.column
Text('Name', justify='center'), justify='left', style=ctx.obj['style'].column
)
for index, monitor in monitors:
@@ -60,33 +48,29 @@ def list_monitors(
console.out.print(table)
@app.command(name=['open', 'o'])
@app.command('open | o')
def open(
source_name: Optional[str] = None,
/,
monitor_index: Annotated[int, Parameter(validator=validators.Number(gte=0))] = 0,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
monitor_index: Annotated[
int,
typer.Option(help='Index of the monitor to open the projector on.'),
] = 0,
source_name: Annotated[
str,
typer.Argument(
show_default='The current scene',
help='Name of the source to project.',
),
] = '',
):
"""Open a fullscreen projector for a source on a specific monitor.
Parameters
----------
source_name : str, optional
The name of the source to project. If not provided, the current program scene will be used.
monitor_index : int, optional
The index of the monitor to open the projector on. Defaults to 0 (the primary monitor).
ctx : Context
The context containing the OBS client and configuration.
"""
"""Open a fullscreen projector for a source on a specific monitor."""
if not source_name:
source_name = ctx.client.get_current_program_scene().scene_name
source_name = ctx.obj['obsws'].get_current_program_scene().scene_name
monitors = ctx.client.get_monitor_list().monitors
monitors = ctx.obj['obsws'].get_monitor_list().monitors
for monitor in monitors:
if monitor['monitorIndex'] == monitor_index:
ctx.client.open_source_projector(
ctx.obj['obsws'].open_source_projector(
source_name=source_name,
monitor_index=monitor_index,
)
@@ -97,8 +81,8 @@ def open(
break
else:
raise OBSWSCLIError(
console.err.print(
f'Monitor with index [yellow]{monitor_index}[/yellow] not found. '
f'Use [yellow]obsws-cli projector ls-m[/yellow] to see available monitors.',
ExitCode.ERROR,
f'Use [yellow]obsws-cli projector ls-m[/yellow] to see available monitors.'
)
raise typer.Exit(code=1)

245
obsws_cli/record.py
View File

@@ -3,104 +3,68 @@
from pathlib import Path
from typing import Annotated, Optional
from cyclopts import App, Parameter
import typer
from . import console
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='record', help='Commands for controlling OBS recording functionality.')
app = typer.Typer(cls=SubTyperAliasGroup)
def _get_recording_status(ctx: Context) -> tuple:
@app.callback()
def main():
"""Control OBS recording functionality."""
def _get_recording_status(ctx: typer.Context) -> tuple:
"""Get recording status."""
resp = ctx.client.get_record_status()
resp = ctx.obj['obsws'].get_record_status()
return resp.output_active, resp.output_paused
@app.command(name=['start', 's'])
def start(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Start recording.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
@app.command('start | s')
def start(ctx: typer.Context):
"""Start recording."""
active, paused = _get_recording_status(ctx)
if active:
err_msg = 'Recording is already in progress, cannot start.'
if paused:
err_msg += ' Try resuming it.'
raise OBSWSCLIError(err_msg, ExitCode.ERROR)
ctx.client.start_record()
console.err.print(err_msg)
raise typer.Exit(1)
ctx.obj['obsws'].start_record()
console.out.print('Recording started successfully.')
@app.command(name=['stop', 'st'])
def stop(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Stop recording.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
@app.command('stop | st')
def stop(ctx: typer.Context):
"""Stop recording."""
active, _ = _get_recording_status(ctx)
if not active:
raise OBSWSCLIError(
'Recording is not in progress, cannot stop.', ExitCode.ERROR
)
console.err.print('Recording is not in progress, cannot stop.')
raise typer.Exit(1)
resp = ctx.client.stop_record()
resp = ctx.obj['obsws'].stop_record()
console.out.print(
f'Recording stopped successfully. Saved to: {console.highlight(ctx, resp.output_path)}'
)
@app.command(name=['toggle', 'tg'])
def toggle(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Toggle recording.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.toggle_record()
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle recording."""
resp = ctx.obj['obsws'].toggle_record()
if resp.output_active:
console.out.print('Recording started successfully.')
else:
console.out.print('Recording stopped successfully.')
@app.command(name=['status', 'ss'])
def status(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get recording status.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get recording status."""
active, paused = _get_recording_status(ctx)
if active:
if paused:
@@ -111,143 +75,98 @@ def status(
console.out.print('Recording is not in progress.')
@app.command(name=['resume', 'r'])
def resume(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Resume recording.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
@app.command('resume | r')
def resume(ctx: typer.Context):
"""Resume recording."""
active, paused = _get_recording_status(ctx)
if not active:
raise OBSWSCLIError(
'Recording is not in progress, cannot resume.', ExitCode.ERROR
)
console.err.print('Recording is not in progress, cannot resume.')
raise typer.Exit(1)
if not paused:
raise OBSWSCLIError(
'Recording is in progress but not paused, cannot resume.', ExitCode.ERROR
)
console.err.print('Recording is in progress but not paused, cannot resume.')
raise typer.Exit(1)
ctx.client.resume_record()
ctx.obj['obsws'].resume_record()
console.out.print('Recording resumed successfully.')
@app.command(name=['pause', 'p'])
def pause(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Pause recording.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
@app.command('pause | p')
def pause(ctx: typer.Context):
"""Pause recording."""
active, paused = _get_recording_status(ctx)
if not active:
raise OBSWSCLIError(
'Recording is not in progress, cannot pause.', ExitCode.ERROR
)
console.err.print('Recording is not in progress, cannot pause.')
raise typer.Exit(1)
if paused:
raise OBSWSCLIError(
'Recording is in progress but already paused, cannot pause.', ExitCode.ERROR
)
console.err.print('Recording is in progress but already paused, cannot pause.')
raise typer.Exit(1)
ctx.client.pause_record()
ctx.obj['obsws'].pause_record()
console.out.print('Recording paused successfully.')
@app.command(name=['directory', 'd'])
@app.command('directory | d')
def directory(
# Since the CLI and OBS may be running on different platforms,
# we won't validate the path here.
record_directory: Optional[Path] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
record_directory: Annotated[
Optional[Path],
# Since the CLI and OBS may be running on different platforms,
# we won't validate the path here.
typer.Argument(
file_okay=False,
dir_okay=True,
help='Directory to set for recording.',
),
] = None,
):
"""Get or set the recording directory.
Parameters
----------
record_directory: Optional[Path]
The directory to set for recording. If not provided, the current recording directory is displayed.
ctx: Context
The context containing the OBS client and other settings.
"""
"""Get or set the recording directory."""
if record_directory is not None:
ctx.client.set_record_directory(str(record_directory))
ctx.obj['obsws'].set_record_directory(str(record_directory))
console.out.print(
f'Recording directory updated to: {console.highlight(ctx, record_directory)}'
)
else:
resp = ctx.client.get_record_directory()
resp = ctx.obj['obsws'].get_record_directory()
console.out.print(
f'Recording directory: {console.highlight(ctx, resp.record_directory)}'
)
@app.command(name=['split', 'sp'])
def split(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Split the current recording.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
@app.command('split | sp')
def split(ctx: typer.Context):
"""Split the current recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot split.')
raise OBSWSCLIError(
'Recording is not in progress, cannot split.', ExitCode.ERROR
)
raise typer.Exit(1)
if paused:
raise OBSWSCLIError('Recording is paused, cannot split.', ExitCode.ERROR)
console.err.print('Recording is paused, cannot split.')
raise typer.Exit(1)
ctx.client.split_record_file()
ctx.obj['obsws'].split_record_file()
console.out.print('Recording split successfully.')
@app.command(name=['chapter', 'ch'])
@app.command('chapter | ch')
def chapter(
chapter_name: Optional[str] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
chapter_name: Annotated[
Optional[str],
typer.Argument(
help='Name of the chapter to create.',
),
] = None,
):
"""Create a chapter in the current recording.
Parameters
----------
chapter_name: Optional[str]
The name of the chapter to create. If not provided, an unnamed chapter is created.
ctx: Context
The context containing the OBS client and other settings.
"""
"""Create a chapter in the current recording."""
active, paused = _get_recording_status(ctx)
if not active:
raise OBSWSCLIError(
'Recording is not in progress, cannot create chapter.', ExitCode.ERROR
)
console.err.print('Recording is not in progress, cannot create chapter.')
raise typer.Exit(1)
if paused:
raise OBSWSCLIError(
'Recording is paused, cannot create chapter.', ExitCode.ERROR
)
console.err.print('Recording is paused, cannot create chapter.')
raise typer.Exit(1)
ctx.client.create_record_chapter(chapter_name)
ctx.obj['obsws'].create_record_chapter(chapter_name)
console.out.print(
f'Chapter {console.highlight(ctx, chapter_name or "unnamed")} created successfully.'
)

113
obsws_cli/replaybuffer.py
View File

@@ -1,113 +1,64 @@
"""module containing commands for manipulating the replay buffer in OBS."""
from typing import Annotated
from cyclopts import App, Parameter
import typer
from . import console
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(
name='replaybuffer', help='Commands for controlling the replay buffer in OBS.'
)
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['start', 's'])
def start(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Start the replay buffer.
@app.callback()
def main():
"""Control profiles in OBS."""
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.get_replay_buffer_status()
@app.command('start | s')
def start(ctx: typer.Context):
"""Start the replay buffer."""
resp = ctx.obj['obsws'].get_replay_buffer_status()
if resp.output_active:
raise OBSWSCLIError('Replay buffer is already active.', ExitCode.ERROR)
console.err.print('Replay buffer is already active.')
raise typer.Exit(1)
ctx.client.start_replay_buffer()
ctx.obj['obsws'].start_replay_buffer()
console.out.print('Replay buffer started.')
@app.command(name=['stop', 'st'])
def stop(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Stop the replay buffer.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.get_replay_buffer_status()
@app.command('stop | st')
def stop(ctx: typer.Context):
"""Stop the replay buffer."""
resp = ctx.obj['obsws'].get_replay_buffer_status()
if not resp.output_active:
raise OBSWSCLIError('Replay buffer is not active.', ExitCode.ERROR)
console.err.print('Replay buffer is not active.')
raise typer.Exit(1)
ctx.client.stop_replay_buffer()
ctx.obj['obsws'].stop_replay_buffer()
console.out.print('Replay buffer stopped.')
@app.command(name=['toggle', 'tg'])
def toggle(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Toggle the replay buffer.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.toggle_replay_buffer()
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle the replay buffer."""
resp = ctx.obj['obsws'].toggle_replay_buffer()
if resp.output_active:
console.out.print('Replay buffer is active.')
else:
console.out.print('Replay buffer is not active.')
@app.command(name=['status', 'ss'])
def status(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the status of the replay buffer.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.get_replay_buffer_status()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get the status of the replay buffer."""
resp = ctx.obj['obsws'].get_replay_buffer_status()
if resp.output_active:
console.out.print('Replay buffer is active.')
else:
console.out.print('Replay buffer is not active.')
@app.command(name=['save', 'sv'])
def save(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Save the replay buffer.
Parameters
----------
ctx: Context
The context containing the OBS client and other settings.
"""
ctx.client.save_replay_buffer()
@app.command('save | sv')
def save(ctx: typer.Context):
"""Save the replay buffer."""
ctx.obj['obsws'].save_replay_buffer()
console.out.print('Replay buffer saved.')

119
obsws_cli/scene.py
View File

@@ -2,35 +2,28 @@
from typing import Annotated
from cyclopts import App, Parameter
import typer
from rich.table import Table
from rich.text import Text
from . import console, util, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='scene', help='Commands for managing OBS scenes')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
@app.callback()
def main():
"""Control OBS scenes."""
@app.command('list | ls')
def list_(
uuid: bool = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
uuid: Annotated[bool, typer.Option(help='Show UUIDs of scenes')] = False,
):
"""List all scenes.
Parameters
----------
uuid : bool
Show UUIDs of scenes.
ctx : Context
The context containing the OBS client and configuration.
"""
resp = ctx.client.get_scene_list()
"""List all scenes."""
resp = ctx.obj['obsws'].get_scene_list()
scenes = (
(scene.get('sceneName'), scene.get('sceneUuid'))
for scene in reversed(resp.scenes)
@@ -38,20 +31,20 @@ def list_(
if not scenes:
console.out.print('No scenes found.')
return
raise typer.Exit()
active_scene = ctx.client.get_current_program_scene().scene_name
active_scene = ctx.obj['obsws'].get_current_program_scene().scene_name
table = Table(title='Scenes', padding=(0, 2), border_style=ctx.style.border)
table = Table(title='Scenes', padding=(0, 2), border_style=ctx.obj['style'].border)
if uuid:
columns = [
(Text('Scene Name', justify='center'), 'left', ctx.style.column),
(Text('Scene Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Active', justify='center'), 'center', None),
(Text('UUID', justify='center'), 'left', ctx.style.column),
(Text('UUID', justify='center'), 'left', ctx.obj['style'].column),
]
else:
columns = [
(Text('Scene Name', justify='center'), 'left', ctx.style.column),
(Text('Scene Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Active', justify='center'), 'center', None),
]
for heading, justify, style in columns:
@@ -73,79 +66,57 @@ def list_(
console.out.print(table)
@app.command(name=['current', 'get'])
@app.command('current | get')
def current(
preview: bool = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
preview: Annotated[
bool, typer.Option(help='Get the preview scene instead of the program scene')
] = False,
):
"""Get the current program scene or preview scene.
Parameters
----------
preview : bool
If True, get the preview scene instead of the program scene.
ctx : Context
The context containing the OBS client and configuration.
"""
"""Get the current program scene or preview scene."""
if preview and not validate.studio_mode_enabled(ctx):
raise OBSWSCLIError(
'Studio mode is not enabled, cannot get preview scene.',
code=ExitCode.ERROR,
)
console.err.print('Studio mode is not enabled, cannot get preview scene.')
raise typer.Exit(1)
if preview:
resp = ctx.client.get_current_preview_scene()
resp = ctx.obj['obsws'].get_current_preview_scene()
console.out.print(
f'Current Preview Scene: {console.highlight(ctx, resp.current_preview_scene_name)}'
)
else:
resp = ctx.client.get_current_program_scene()
resp = ctx.obj['obsws'].get_current_program_scene()
console.out.print(
f'Current Program Scene: {console.highlight(ctx, resp.current_program_scene_name)}'
)
@app.command(name=['switch', 'set'])
@app.command('switch | set')
def switch(
scene_name: str,
/,
preview: bool = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., help='Name of the scene to switch to')
],
preview: Annotated[
bool,
typer.Option(help='Switch to the preview scene instead of the program scene'),
] = False,
):
"""Switch to a scene.
Parameters
----------
scene_name : str
The name of the scene to switch to.
preview : bool
If True, switch to the preview scene instead of the program scene.
ctx : Context
The context containing the OBS client and configuration.
"""
"""Switch to a scene."""
if preview and not validate.studio_mode_enabled(ctx):
raise OBSWSCLIError(
'Studio mode is not enabled, cannot switch to preview scene.',
code=ExitCode.ERROR,
)
console.err.print('Studio mode is not enabled, cannot set the preview scene.')
raise typer.Exit(1)
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
code=ExitCode.ERROR,
)
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
raise typer.Exit(1)
if preview:
ctx.client.set_current_preview_scene(scene_name)
ctx.obj['obsws'].set_current_preview_scene(scene_name)
console.out.print(
f'Switched to preview scene: {console.highlight(ctx, scene_name)}'
)
else:
ctx.client.set_current_program_scene(scene_name)
ctx.obj['obsws'].set_current_program_scene(scene_name)
console.out.print(
f'Switched to program scene: {console.highlight(ctx, scene_name)}'
)

127
obsws_cli/scenecollection.py
View File

@@ -2,44 +2,37 @@
from typing import Annotated
from cyclopts import App, Parameter
import typer
from rich.table import Table
from . import console, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(
name='scenecollection', help='Commands for controlling scene collections in OBS.'
)
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
def list_(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""List all scene collections.
@app.callback()
def main():
"""Control scene collections in OBS."""
Parameters
----------
ctx : Context
The context containing the OBS client and configuration.
"""
resp = ctx.client.get_scene_collection_list()
@app.command('list | ls')
def list_(ctx: typer.Context):
"""List all scene collections."""
resp = ctx.obj['obsws'].get_scene_collection_list()
if not resp.scene_collections:
console.out.print('No scene collections found.')
raise typer.Exit()
table = Table(
title='Scene Collections',
padding=(0, 2),
border_style=ctx.style.border,
border_style=ctx.obj['style'].border,
)
table.add_column(
'Scene Collection Name', justify='left', style=ctx.obj['style'].column
)
table.add_column('Scene Collection Name', justify='left', style=ctx.style.column)
if not resp.scene_collections:
console.out.print('No scene collections found.')
return
for scene_collection_name in resp.scene_collections:
table.add_row(scene_collection_name)
@@ -47,87 +40,59 @@ def list_(
console.out.print(table)
@app.command(name=['current', 'get'])
def current(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the current scene collection.
Parameters
----------
ctx : Context
The context containing the OBS client and configuration.
"""
resp = ctx.client.get_scene_collection_list()
@app.command('current | get')
def current(ctx: typer.Context):
"""Get the current scene collection."""
resp = ctx.obj['obsws'].get_scene_collection_list()
console.out.print(
f'Current scene collection: {console.highlight(ctx, resp.current_scene_collection_name)}'
)
@app.command(name=['switch', 'set'])
@app.command('switch | set')
def switch(
scene_collection_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_collection_name: Annotated[
str, typer.Argument(..., help='Name of the scene collection to switch to')
],
):
"""Switch to a scene collection.
Parameters
----------
scene_collection_name : str
The name of the scene collection to switch to.
ctx : Context
The context containing the OBS client and configuration.
"""
"""Switch to a scene collection."""
if not validate.scene_collection_in_scene_collections(ctx, scene_collection_name):
raise OBSWSCLIError(
f'Scene collection [yellow]{scene_collection_name}[/yellow] not found.',
exit_code=ExitCode.ERROR,
console.err.print(
f'Scene collection [yellow]{scene_collection_name}[/yellow] not found.'
)
raise typer.Exit(1)
current_scene_collection = (
ctx.client.get_scene_collection_list().current_scene_collection_name
ctx.obj['obsws'].get_scene_collection_list().current_scene_collection_name
)
if scene_collection_name == current_scene_collection:
raise OBSWSCLIError(
f'Scene collection [yellow]{scene_collection_name}[/yellow] is already active.',
exit_code=ExitCode.ERROR,
console.err.print(
f'Scene collection [yellow]{scene_collection_name}[/yellow] is already active.'
)
raise typer.Exit(1)
ctx.client.set_current_scene_collection(scene_collection_name)
ctx.obj['obsws'].set_current_scene_collection(scene_collection_name)
console.out.print(
f'Switched to scene collection {console.highlight(ctx, scene_collection_name)}.'
)
@app.command(name=['create', 'new'])
@app.command('create | new')
def create(
scene_collection_name: str,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_collection_name: Annotated[
str, typer.Argument(..., help='Name of the scene collection to create')
],
):
"""Create a new scene collection.
Parameters
----------
scene_collection_name : str
The name of the scene collection to create.
ctx : Context
The context containing the OBS client and configuration.
"""
"""Create a new scene collection."""
if validate.scene_collection_in_scene_collections(ctx, scene_collection_name):
raise OBSWSCLIError(
f'Scene collection [yellow]{scene_collection_name}[/yellow] already exists.',
exit_code=ExitCode.ERROR,
console.err.print(
f'Scene collection [yellow]{scene_collection_name}[/yellow] already exists.'
)
raise typer.Exit(1)
ctx.client.create_scene_collection(scene_collection_name)
ctx.obj['obsws'].create_scene_collection(scene_collection_name)
console.out.print(
f'Created scene collection {console.highlight(ctx, scene_collection_name)}.'
)

400
obsws_cli/sceneitem.py
View File

@@ -2,49 +2,41 @@
from typing import Annotated, Optional
from cyclopts import App, Parameter
import typer
from rich.table import Table
from . import console, util, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='sceneitem', help='Commands for controlling scene items in OBS.')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['list', 'ls'])
@app.callback()
def main():
"""Control items in OBS scenes."""
@app.command('list | ls')
def list_(
scene_name: Optional[str] = None,
/,
uuid: bool = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
Optional[str],
typer.Argument(
show_default='The current scene',
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.
Parameters
----------
scene_name : str, optional
The name of the scene to list items for. If not provided, the current program scene
will be used.
uuid : bool
Show UUIDs of scene items.
ctx : Context
The context containing the OBS client and configuration.
"""
"""List all items in a scene."""
if not scene_name:
scene_name = ctx.client.get_current_program_scene().scene_name
scene_name = ctx.obj['obsws'].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.')
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
exit_code=ExitCode.ERROR,
)
raise typer.Exit(1)
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
items = sorted(
(
(
@@ -63,26 +55,26 @@ def list_(
console.out.print(
f'No items found in scene {console.highlight(ctx, scene_name)}.'
)
return
raise typer.Exit()
table = Table(
title=f'Items in Scene: {scene_name}',
padding=(0, 2),
border_style=ctx.style.border,
border_style=ctx.obj['style'].border,
)
if uuid:
columns = [
('Item ID', 'center', ctx.style.column),
('Item Name', 'left', ctx.style.column),
('In Group', 'left', ctx.style.column),
('Item ID', 'center', ctx.obj['style'].column),
('Item Name', 'left', ctx.obj['style'].column),
('In Group', 'left', ctx.obj['style'].column),
('Enabled', 'center', None),
('UUID', 'left', ctx.style.column),
('UUID', 'left', ctx.obj['style'].column),
]
else:
columns = [
('Item ID', 'center', ctx.style.column),
('Item Name', 'left', ctx.style.column),
('In Group', 'left', ctx.style.column),
('Item ID', 'center', ctx.obj['style'].column),
('Item Name', 'left', ctx.obj['style'].column),
('In Group', 'left', ctx.obj['style'].column),
('Enabled', 'center', None),
]
# Add columns to the table
@@ -91,7 +83,7 @@ def list_(
for item_id, item_name, is_group, is_enabled, source_uuid in items:
if is_group:
resp = ctx.client.get_group_scene_item_list(item_name)
resp = ctx.obj['obsws'].get_group_scene_item_list(item_name)
group_items = sorted(
(
(
@@ -146,91 +138,79 @@ def list_(
def _validate_sources(
ctx: Context,
ctx: typer.Context,
scene_name: str,
item_name: str,
group: Optional[str] = None,
):
) -> bool:
"""Validate the scene name and item name."""
if not validate.scene_in_scenes(ctx, scene_name):
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
exit_code=ExitCode.ERROR,
)
console.err.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):
raise OBSWSCLIError(
f'Group [yellow]{group}[/yellow] not found in scene [yellow]{scene_name}[/yellow].',
exit_code=ExitCode.ERROR,
console.err.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):
raise OBSWSCLIError(
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 [yellow]obsws-cli sceneitem ls[/yellow] for a list of items in the scene.',
exit_code=ExitCode.ERROR,
'Use [yellow]obsws-cli sceneitem ls[/yellow] for a list of items in the scene.'
)
return False
return True
def _get_scene_name_and_item_id(
ctx: Context,
scene_name: str,
item_name: str,
group: Optional[str] = None,
ctx: typer.Context, scene_name: str, item_name: str, group: Optional[str] = None
):
"""Get the scene name and item ID for the given scene and item name."""
if group:
resp = ctx.client.get_group_scene_item_list(group)
resp = ctx.obj['obsws'].get_group_scene_item_list(group)
for item in resp.scene_items:
if item.get('sourceName') == item_name:
scene_name = group
scene_item_id = item.get('sceneItemId')
break
else:
raise OBSWSCLIError(
f'Item [yellow]{item_name}[/yellow] not found in group [yellow]{group}[/yellow].',
exit_code=ExitCode.ERROR,
console.err.print(
f'Item [yellow]{item_name}[/yellow] not found in group [yellow]{group}[/yellow].'
)
raise typer.Exit(1)
else:
resp = ctx.client.get_scene_item_id(scene_name, item_name)
resp = ctx.obj['obsws'].get_scene_item_id(scene_name, item_name)
scene_item_id = resp.scene_item_id
return scene_name, scene_item_id
@app.command(name=['show', 'sh'])
@app.command('show | sh')
def show(
scene_name: str,
item_name: str,
/,
group: Optional[str] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
item_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Item name to show in the scene'),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
):
"""Show an item in a scene.
Parameters
----------
scene_name : str
The name of the scene the item is in.
item_name : str
The name of the item to show in the scene.
group : str, optional
The name of the parent group the item is in, if applicable.
ctx : Context
The context containing the OBS client and configuration.
"""
_validate_sources(ctx, scene_name, item_name, group)
"""Show an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
ctx, scene_name, item_name, group
)
ctx.client.set_scene_item_enabled(
ctx.obj['obsws'].set_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
enabled=True,
@@ -251,37 +231,28 @@ def show(
)
@app.command(name=['hide', 'h'])
@app.command('hide | h')
def hide(
scene_name: str,
item_name: str,
/,
group: Optional[str] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
item_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Item name to hide in the scene'),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
):
"""Hide an item in a scene.
Parameters
----------
scene_name : str
The name of the scene the item is in.
item_name : str
The name of the item to hide in the scene.
group : str, optional
The name of the parent group the item is in, if applicable.
ctx : Context
The context containing the OBS client and configuration.
"""
_validate_sources(ctx, scene_name, item_name, group)
"""Hide an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
ctx, scene_name, item_name, group
)
ctx.client.set_scene_item_enabled(
ctx.obj['obsws'].set_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
enabled=False,
@@ -301,43 +272,36 @@ def hide(
)
@app.command(name=['toggle', 'tg'])
@app.command('toggle | tg')
def toggle(
scene_name: str,
item_name: str,
/,
group: Optional[str] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
item_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='Item name to toggle in the scene'
),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
):
"""Toggle an item in a scene.
Parameters
----------
scene_name : str
The name of the scene the item is in.
item_name : str
The name of the item to toggle in the scene.
group : str, optional
The name of the parent group the item is in, if applicable.
ctx : Context
The context containing the OBS client and configuration.
"""
_validate_sources(ctx, scene_name, item_name, group)
"""Toggle an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
ctx, scene_name, item_name, group
)
enabled = ctx.client.get_scene_item_enabled(
enabled = ctx.obj['obsws'].get_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
)
new_state = not enabled.scene_item_enabled
ctx.client.set_scene_item_enabled(
ctx.obj['obsws'].set_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
enabled=new_state,
@@ -369,37 +333,30 @@ def toggle(
)
@app.command(name=['visible', 'v'])
@app.command('visible | v')
def visible(
scene_name: str,
item_name: str,
/,
group: Optional[str] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
item_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='Item name to check visibility in the scene'
),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
):
"""Check if an item in a scene is visible.
Parameters
----------
scene_name : str
The name of the scene the item is in.
item_name : str
The name of the item to check visibility in the scene.
group : str, optional
The name of the parent group the item is in, if applicable.
ctx : Context
The context containing the OBS client and configuration.
"""
_validate_sources(ctx, scene_name, item_name, group)
"""Check if an item in a scene is visible."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
ctx, scene_name, item_name, group
)
enabled = ctx.client.get_scene_item_enabled(
enabled = ctx.obj['obsws'].get_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
)
@@ -420,75 +377,68 @@ def visible(
)
@app.command(name=['transform', 't'])
@app.command('transform | t')
def transform(
scene_name: str,
item_name: str,
/,
group: Optional[str] = None,
alignment: Optional[int] = None,
bounds_alignment: Optional[int] = None,
bounds_height: Optional[float] = None,
bounds_type: Optional[str] = None,
bounds_width: Optional[float] = None,
crop_to_bounds: Optional[bool] = None,
crop_bottom: Optional[float] = None,
crop_left: Optional[float] = None,
crop_right: Optional[float] = None,
crop_top: Optional[float] = None,
position_x: Optional[float] = None,
position_y: Optional[float] = None,
rotation: Optional[float] = None,
scale_x: Optional[float] = None,
scale_y: Optional[float] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
item_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='Item name to transform in the scene'
),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
alignment: Annotated[
Optional[int], typer.Option(help='Alignment of the item in the scene')
] = None,
bounds_alignment: Annotated[
Optional[int], typer.Option(help='Bounds alignment of the item in the scene')
] = None,
bounds_height: Annotated[
Optional[float], typer.Option(help='Height of the item in the scene')
] = None,
bounds_type: Annotated[
Optional[str], typer.Option(help='Type of bounds for the item in the scene')
] = None,
bounds_width: Annotated[
Optional[float], typer.Option(help='Width of the item in the scene')
] = None,
crop_to_bounds: Annotated[
Optional[bool], typer.Option(help='Crop the item to the bounds')
] = None,
crop_bottom: Annotated[
Optional[float], typer.Option(help='Bottom crop of the item in the scene')
] = None,
crop_left: Annotated[
Optional[float], typer.Option(help='Left crop of the item in the scene')
] = None,
crop_right: Annotated[
Optional[float], typer.Option(help='Right crop of the item in the scene')
] = None,
crop_top: Annotated[
Optional[float], typer.Option(help='Top crop of the item in the scene')
] = None,
position_x: Annotated[
Optional[float], typer.Option(help='X position of the item in the scene')
] = None,
position_y: Annotated[
Optional[float], typer.Option(help='Y position of the item in the scene')
] = None,
rotation: Annotated[
Optional[float], typer.Option(help='Rotation of the item in the scene')
] = None,
scale_x: Annotated[
Optional[float], typer.Option(help='X scale of the item in the scene')
] = None,
scale_y: Annotated[
Optional[float], typer.Option(help='Y scale of the item in the scene')
] = None,
):
"""Set the transform of an item in a scene.
Parameters
----------
scene_name : str
The name of the scene the item is in.
item_name : str
The name of the item to transform in the scene.
group : str, optional
The name of the parent group the item is in, if applicable.
alignment : int, optional
Alignment of the item in the scene.
bounds_alignment : int, optional
Bounds alignment of the item in the scene.
bounds_height : float, optional
Height of the item in the scene.
bounds_type : str, optional
Type of bounds for the item in the scene.
bounds_width : float, optional
Width of the item in the scene.
crop_to_bounds : bool, optional
Crop the item to the bounds.
crop_bottom : float, optional
Bottom crop of the item in the scene.
crop_left : float, optional
Left crop of the item in the scene.
crop_right : float, optional
Right crop of the item in the scene.
crop_top : float, optional
Top crop of the item in the scene.
position_x : float, optional
X position of the item in the scene.
position_y : float, optional
Y position of the item in the scene.
rotation : float, optional
Rotation of the item in the scene.
scale_x : float, optional
X scale of the item in the scene.
scale_y : float, optional
Y scale of the item in the scene.
ctx : Context
The context containing the OBS client and configuration.
"""
_validate_sources(ctx, scene_name, item_name, group)
"""Set the transform of an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
@@ -528,12 +478,10 @@ def transform(
transform['scaleY'] = scale_y
if not transform:
raise OBSWSCLIError(
'No transform options provided. Use at least one of the transform options.',
exit_code=ExitCode.ERROR,
)
console.err.print('No transform options provided.')
raise typer.Exit(1)
transform = ctx.client.set_scene_item_transform(
transform = ctx.obj['obsws'].set_scene_item_transform(
scene_name=scene_name,
item_id=int(scene_item_id),
transform=transform,

97
obsws_cli/screenshot.py
View File

@@ -4,51 +4,66 @@ from pathlib import Path
from typing import Annotated
import obsws_python as obsws
from cyclopts import App, Parameter, validators
import typer
from . import console
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='screenshot', help='Commands for taking screenshots using OBS.')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['save', 'sv'])
@app.callback()
def main():
"""Take screenshots using OBS."""
@app.command('save | sv')
def save(
source_name: str,
# Since the CLI and OBS may be running on different platforms,
# we won't validate the path here.
output_path: Path,
/,
width: float = 1920,
height: float = 1080,
ctx: typer.Context,
source_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the source to take a screenshot of.',
),
],
output_path: Annotated[
Path,
# Since the CLI and OBS may be running on different platforms,
# we won't validate the path here.
typer.Argument(
...,
show_default=False,
file_okay=True,
dir_okay=False,
help='Path to save the screenshot (must include file name and extension).',
),
],
width: Annotated[
float,
typer.Option(
help='Width of the screenshot.',
),
] = 1920,
height: Annotated[
float,
typer.Option(
help='Height of the screenshot.',
),
] = 1080,
quality: Annotated[
float, Parameter(validator=validators.Number(gte=-1, lte=100))
] = -1.0,
*,
ctx: Annotated[Context, Parameter(parse=False)],
float,
typer.Option(
min=-1,
max=100,
help='Quality of the screenshot.',
),
] = -1,
):
"""Take a screenshot and save it to a file.
Parameters
----------
source_name : str
Name of the source to take a screenshot of.
output_path : Path
Path to save the screenshot (must include file name and extension).
width : float
Width of the screenshot.
height : float
Height of the screenshot.
quality : float
Quality of the screenshot. A value of -1 uses the default quality.
ctx : Context
Context containing the OBS WebSocket client instance.
"""
"""Take a screenshot and save it to a file."""
try:
ctx.client.save_source_screenshot(
ctx.obj['obsws'].save_source_screenshot(
name=source_name,
img_format=output_path.suffix.lstrip('.').lower(),
file_path=str(output_path),
@@ -59,16 +74,16 @@ def save(
except obsws.error.OBSSDKRequestError as e:
match e.code:
case 403:
raise OBSWSCLIError(
console.err.print(
'The [yellow]image format[/yellow] (file extension) must be included in the file name, '
"for example: '/path/to/screenshot.png'.",
code=ExitCode.ERROR,
)
raise typer.Exit(1)
case 600:
raise OBSWSCLIError(
'No source was found by the name of [yellow]{source_name}[/yellow]',
code=ExitCode.ERROR,
console.err.print(
f'No source was found by the name of [yellow]{source_name}[/yellow]'
)
raise typer.Exit(1)
case _:
raise

80
obsws_cli/settings.py Normal file
View File

@@ -0,0 +1,80 @@
"""module for settings management for obsws-cli."""
from collections import UserDict
from pathlib import Path
from dotenv import dotenv_values
SettingsValue = str | int
class Settings(UserDict):
"""A class to manage settings for obsws-cli.
This class extends UserDict to provide a dictionary-like interface for settings.
It loads settings from environment variables and .env files.
The settings are expected to be in uppercase and should start with 'OBS_'.
Example:
-------
settings = Settings()
host = settings['OBS_HOST']
settings['OBS_PORT'] = 4455
"""
PREFIX = 'OBS_'
def __init__(self, *args, **kwargs):
"""Initialize the Settings object."""
kwargs.update(
{
**dotenv_values('.env'),
**dotenv_values(Path.home() / '.config' / 'obsws-cli' / 'obsws.env'),
}
)
super().__init__(*args, **kwargs)
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]
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
_settings = Settings(
OBS_HOST='localhost',
OBS_PORT=4455,
OBS_PASSWORD='',
OBS_TIMEOUT=5,
OBS_DEBUG=False,
OBS_STYLE='disabled',
OBS_STYLE_NO_BORDER=False,
)
def get(key: str) -> SettingsValue:
"""Get a setting value by key.
Args:
----
key (str): The key of the setting to retrieve.
Returns:
-------
The value of the setting.
Raises:
------
KeyError: If the key does not exist in the settings.
"""
return _settings[key]

101
obsws_cli/stream.py
View File

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

85
obsws_cli/studiomode.py
View File

@@ -1,85 +1,48 @@
"""module containing commands for manipulating studio mode in OBS."""
from typing import Annotated
from cyclopts import App, Parameter
import typer
from . import console
from .context import Context
from .alias import SubTyperAliasGroup
app = App(name='studiomode', help='Commands for controlling studio mode in OBS.')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['enable', 'on'])
def enable(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Enable studio mode.
@app.callback()
def main():
"""Control studio mode in OBS."""
Parameters
----------
ctx : Context
Context containing the OBS WebSocket client instance.
"""
ctx.client.set_studio_mode_enabled(True)
@app.command('enable | on')
def enable(ctx: typer.Context):
"""Enable studio mode."""
ctx.obj['obsws'].set_studio_mode_enabled(True)
console.out.print('Studio mode has been enabled.')
@app.command(name=['disable', 'off'])
def disable(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Disable studio mode.
Parameters
----------
ctx : Context
Context containing the OBS WebSocket client instance.
"""
ctx.client.set_studio_mode_enabled(False)
@app.command('disable | off')
def disable(ctx: typer.Context):
"""Disable studio mode."""
ctx.obj['obsws'].set_studio_mode_enabled(False)
console.out.print('Studio mode has been disabled.')
@app.command(name=['toggle', 'tg'])
def toggle(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Toggle studio mode.
Parameters
----------
ctx : Context
Context containing the OBS WebSocket client instance.
"""
resp = ctx.client.get_studio_mode_enabled()
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle studio mode."""
resp = ctx.obj['obsws'].get_studio_mode_enabled()
if resp.studio_mode_enabled:
ctx.client.set_studio_mode_enabled(False)
ctx.obj['obsws'].set_studio_mode_enabled(False)
console.out.print('Studio mode is now disabled.')
else:
ctx.client.set_studio_mode_enabled(True)
ctx.obj['obsws'].set_studio_mode_enabled(True)
console.out.print('Studio mode is now enabled.')
@app.command(name=['status', 'ss'])
def status(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the status of studio mode.
Parameters
----------
ctx : Context
Context containing the OBS WebSocket client instance.
"""
resp = ctx.client.get_studio_mode_enabled()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get the status of studio mode."""
resp = ctx.obj['obsws'].get_studio_mode_enabled()
if resp.studio_mode_enabled:
console.out.print('Studio mode is enabled.')
else:

86
obsws_cli/text.py
View File

@@ -2,43 +2,35 @@
from typing import Annotated, Optional
from cyclopts import App, Parameter
import typer
from . import console, validate
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
from .alias import SubTyperAliasGroup
app = App(name='text', help='Commands for controlling text inputs in OBS.')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['current', 'get'])
@app.callback()
def main():
"""Control text inputs in OBS."""
@app.command('current | get')
def current(
input_name: str,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
input_name: Annotated[str, typer.Argument(help='Name of the text input to get.')],
):
"""Get the current text for a text input.
Parameters
----------
input_name : str
The name of the text input to retrieve the current text from.
ctx : Context
The context containing the OBS client and other settings.
"""
"""Get the current text for a text input."""
if not validate.input_in_inputs(ctx, input_name):
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.', code=ExitCode.ERROR
)
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.client.get_input_settings(name=input_name)
resp = ctx.obj['obsws'].get_input_settings(name=input_name)
if not resp.input_kind.startswith('text_'):
raise OBSWSCLIError(
console.err.print(
f'Input [yellow]{input_name}[/yellow] is not a text input.',
code=ExitCode.ERROR,
)
raise typer.Exit(1)
current_text = resp.input_settings.get('text', '')
if not current_text:
@@ -48,40 +40,32 @@ def current(
)
@app.command(name=['update', 'set'])
@app.command('update | set')
def update(
input_name: str,
new_text: Optional[str] = None,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
ctx: typer.Context,
input_name: Annotated[
str, typer.Argument(help='Name of the text input to update.')
],
new_text: Annotated[
Optional[str],
typer.Argument(
help='The new text to set for the input.',
),
] = None,
):
"""Update the text of a text input.
Parameters
----------
input_name : str
The name of the text input to update.
new_text : Optional[str]
The new text to set for the input. If not provided, the text will be cleared
(set to an empty string).
ctx : Context
The context containing the OBS client and other settings.
"""
"""Update the text of a text input."""
if not validate.input_in_inputs(ctx, input_name):
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.', code=ExitCode.ERROR
)
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.client.get_input_settings(name=input_name)
resp = ctx.obj['obsws'].get_input_settings(name=input_name)
if not resp.input_kind.startswith('text_'):
raise OBSWSCLIError(
console.err.print(
f'Input [yellow]{input_name}[/yellow] is not a text input.',
code=ExitCode.ERROR,
)
raise typer.Exit(1)
ctx.client.set_input_settings(
ctx.obj['obsws'].set_input_settings(
name=input_name,
settings={'text': new_text},
overlay=True,

35
obsws_cli/validate.py
View File

@@ -1,49 +1,54 @@
"""module containing validation functions."""
from .context import Context
import typer
# type alias for an option that is skipped when the command is run
skipped_option = typer.Option(parser=lambda _: _, hidden=True, expose_value=False)
def input_in_inputs(ctx: Context, input_name: str) -> bool:
def input_in_inputs(ctx: typer.Context, input_name: str) -> bool:
"""Check if an input is in the input list."""
inputs = ctx.client.get_input_list().inputs
inputs = ctx.obj['obsws'].get_input_list().inputs
return any(input_.get('inputName') == input_name for input_ in inputs)
def scene_in_scenes(ctx: Context, scene_name: str) -> bool:
def scene_in_scenes(ctx: typer.Context, scene_name: str) -> bool:
"""Check if a scene exists in the list of scenes."""
resp = ctx.client.get_scene_list()
resp = ctx.obj['obsws'].get_scene_list()
return any(scene.get('sceneName') == scene_name for scene in resp.scenes)
def studio_mode_enabled(ctx: Context) -> bool:
def studio_mode_enabled(ctx: typer.Context) -> bool:
"""Check if studio mode is enabled."""
resp = ctx.client.get_studio_mode_enabled()
resp = ctx.obj['obsws'].get_studio_mode_enabled()
return resp.studio_mode_enabled
def scene_collection_in_scene_collections(
ctx: Context, scene_collection_name: str
ctx: typer.Context, scene_collection_name: str
) -> bool:
"""Check if a scene collection exists."""
resp = ctx.client.get_scene_collection_list()
resp = ctx.obj['obsws'].get_scene_collection_list()
return any(
collection == scene_collection_name for collection in resp.scene_collections
)
def item_in_scene_item_list(ctx: Context, scene_name: str, item_name: str) -> bool:
def item_in_scene_item_list(
ctx: typer.Context, scene_name: str, item_name: str
) -> bool:
"""Check if an item exists in a scene."""
resp = ctx.client.get_scene_item_list(scene_name)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
return any(item.get('sourceName') == item_name for item in resp.scene_items)
def profile_exists(ctx: Context, profile_name: str) -> bool:
def profile_exists(ctx: typer.Context, profile_name: str) -> bool:
"""Check if a profile exists."""
resp = ctx.client.get_profile_list()
resp = ctx.obj['obsws'].get_profile_list()
return any(profile == profile_name for profile in resp.profiles)
def monitor_exists(ctx: Context, monitor_index: int) -> bool:
def monitor_exists(ctx: typer.Context, monitor_index: int) -> bool:
"""Check if a monitor exists."""
resp = ctx.client.get_monitor_list()
resp = ctx.obj['obsws'].get_monitor_list()
return any(monitor['monitorIndex'] == monitor_index for monitor in resp.monitors)

81
obsws_cli/virtualcam.py
View File

@@ -1,83 +1,46 @@
"""module containing commands for manipulating virtual camera in OBS."""
from typing import Annotated
from cyclopts import App, Parameter
import typer
from . import console
from .context import Context
from .alias import SubTyperAliasGroup
app = App(name='virtualcam', help='Commands for controlling the virtual camera in OBS.')
app = typer.Typer(cls=SubTyperAliasGroup)
@app.command(name=['start', 's'])
def start(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Start the virtual camera.
@app.callback()
def main():
"""Control virtual camera in OBS."""
Parameters
----------
ctx : Context
The context containing the OBS client and other settings.
"""
ctx.client.start_virtual_cam()
@app.command('start | s')
def start(ctx: typer.Context):
"""Start the virtual camera."""
ctx.obj['obsws'].start_virtual_cam()
console.out.print('Virtual camera started.')
@app.command(name=['stop', 'p'])
def stop(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Stop the virtual camera.
Parameters
----------
ctx : Context
The context containing the OBS client and other settings.
"""
ctx.client.stop_virtual_cam()
@app.command('stop | p')
def stop(ctx: typer.Context):
"""Stop the virtual camera."""
ctx.obj['obsws'].stop_virtual_cam()
console.out.print('Virtual camera stopped.')
@app.command(name=['toggle', 'tg'])
def toggle(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Toggle the virtual camera.
Parameters
----------
ctx : Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.toggle_virtual_cam()
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle the virtual camera."""
resp = ctx.obj['obsws'].toggle_virtual_cam()
if resp.output_active:
console.out.print('Virtual camera is enabled.')
else:
console.out.print('Virtual camera is disabled.')
@app.command(name=['status', 'ss'])
def status(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the status of the virtual camera.
Parameters
----------
ctx : Context
The context containing the OBS client and other settings.
"""
resp = ctx.client.get_virtual_cam_status()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get the status of the virtual camera."""
resp = ctx.obj['obsws'].get_virtual_cam_status()
if resp.output_active:
console.out.print('Virtual camera is enabled.')
else:

4
pyproject.toml
View File

@@ -21,7 +21,7 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
]
dependencies = ["cyclopts>=3.22.2", "obsws-python>=1.8.0"]
dependencies = ["typer>=0.19.2", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
[project.urls]
@@ -30,7 +30,7 @@ Issues = "https://github.com/onyx-and-iris/obsws-cli/issues"
Source = "https://github.com/onyx-and-iris/obsws-cli"
[project.scripts]
obsws-cli = "obsws_cli:run"
obsws-cli = "obsws_cli:app"
[tool.hatch.version]
path = "obsws_cli/__about__.py"

54
tests/conftest.py
View File

@@ -1,6 +1,7 @@
"""pytest configuration file."""
import os
import time
import obsws_python as obsws
from dotenv import find_dotenv, load_dotenv
@@ -44,9 +45,54 @@ def pytest_sessionstart(session):
},
)
session.obsws.set_current_scene_collection('test-collection')
session.obsws.create_profile('pytest_profile')
time.sleep(0.1) # Wait for the profile to be created
session.obsws.set_profile_parameter(
'SimpleOutput',
'RecRB',
'true',
)
# hack to ensure the replay buffer is enabled
session.obsws.set_current_profile('Untitled')
session.obsws.set_current_profile('pytest_profile')
session.obsws.create_scene('pytest_scene')
# Ensure Desktop Audio is created.
desktop_audio_kinds = {
'windows': 'wasapi_output_capture',
'linux': 'pulse_output_capture',
'darwin': 'coreaudio_output_capture',
}
platform = os.environ.get('OBS_TESTS_PLATFORM', os.uname().sysname.lower())
try:
session.obsws.create_input(
sceneName='pytest_scene',
inputName='Desktop Audio',
inputKind=desktop_audio_kinds[platform],
inputSettings={'device_id': 'default'},
sceneItemEnabled=True,
)
except obsws.error.OBSSDKRequestError as e:
if e.code == 601:
"""input already exists, continue."""
# Ensure Mic/Aux is created.
mic_kinds = {
'windows': 'wasapi_input_capture',
'linux': 'pulse_input_capture',
'darwin': 'coreaudio_input_capture',
}
try:
session.obsws.create_input(
sceneName='pytest_scene',
inputName='Mic/Aux',
inputKind=mic_kinds[platform],
inputSettings={'device_id': 'default'},
sceneItemEnabled=True,
)
except obsws.error.OBSSDKRequestError as e:
if e.code == 601:
"""input already exists, continue."""
session.obsws.create_input(
sceneName='pytest_scene',
inputName='pytest_input',
@@ -131,7 +177,7 @@ def pytest_sessionfinish(session, exitstatus):
session.obsws.remove_scene('pytest_scene')
session.obsws.set_current_scene_collection('default')
session.obsws.set_current_scene_collection('Untitled')
resp = session.obsws.get_stream_status()
if resp.output_active:
@@ -149,6 +195,8 @@ def pytest_sessionfinish(session, exitstatus):
if resp.studio_mode_enabled:
session.obsws.set_studio_mode_enabled(False)
session.obsws.remove_profile('pytest_profile')
# Close the OBS WebSocket client connection
session.obsws.disconnect()

2
tests/test_filter.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_filter_list():

10
tests/test_group.py
View File

@@ -1,10 +1,18 @@
"""Unit tests for the group command in the OBS WebSocket CLI."""
import os
import pytest
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
if os.environ.get('OBS_TESTS_SKIP_GROUP_TESTS'):
pytest.skip(
'Skipping group tests as per environment variable', allow_module_level=True
)
def test_group_list():

2
tests/test_hotkey.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_hotkey_list():

12
tests/test_input.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_input_list():
@@ -13,10 +13,7 @@ def test_input_list():
assert result.exit_code == 0
assert 'Desktop Audio' in result.stdout
assert 'Mic/Aux' in result.stdout
assert all(
item in result.stdout
for item in ('Colour Source', 'Colour Source 2', 'Colour Source 3')
)
assert all(item in result.stdout for item in ('pytest_input', 'pytest_input_2'))
def test_input_list_filter_input():
@@ -39,9 +36,6 @@ def test_input_list_filter_colour():
"""Test the input list command with colour filter."""
result = runner.invoke(app, ['input', 'list', '--colour'])
assert result.exit_code == 0
assert all(
item in result.stdout
for item in ('Colour Source', 'Colour Source 2', 'Colour Source 3')
)
assert all(item in result.stdout for item in ('pytest_input', 'pytest_input_2'))
assert 'Desktop Audio' not in result.stdout
assert 'Mic/Aux' not in result.stdout

4
tests/test_record.py
View File

@@ -6,7 +6,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_record_start():
@@ -49,7 +49,9 @@ def test_record_toggle():
result = runner.invoke(app, ['record', 'toggle'])
assert result.exit_code == 0
time.sleep(0.5) # Wait for the recording to toggle
if active:
assert 'Recording stopped successfully.' in result.stdout
else:

24
tests/test_replaybuffer.py
View File

@@ -1,10 +1,20 @@
"""Unit tests for the replaybuffer command in the OBS WebSocket CLI."""
import os
import time
import pytest
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
if os.environ.get('OBS_TESTS_SKIP_REPLAYBUFFER_TESTS'):
pytest.skip(
'Skipping replaybuffer tests as per environment variable',
allow_module_level=True,
)
def test_replaybuffer_start():
@@ -14,6 +24,9 @@ def test_replaybuffer_start():
active = 'Replay buffer is active.' in resp.stdout
resp = runner.invoke(app, ['replaybuffer', 'start'])
time.sleep(0.5) # Wait for the replay buffer to start
if active:
assert resp.exit_code != 0
assert 'Replay buffer is already active.' in resp.stderr
@@ -29,6 +42,9 @@ def test_replaybuffer_stop():
active = 'Replay buffer is active.' in resp.stdout
resp = runner.invoke(app, ['replaybuffer', 'stop'])
time.sleep(0.5) # Wait for the replay buffer to stop
if not active:
assert resp.exit_code != 0
assert 'Replay buffer is not active.' in resp.stderr
@@ -44,9 +60,11 @@ def test_replaybuffer_toggle():
active = 'Replay buffer is active.' in resp.stdout
resp = runner.invoke(app, ['replaybuffer', 'toggle'])
assert resp.exit_code == 0
time.sleep(0.5) # Wait for the replay buffer to toggle
if active:
assert resp.exit_code == 0
assert 'Replay buffer is not active.' in resp.stdout
else:
assert resp.exit_code == 0
assert 'Replay buffer is active.' in resp.stdout

2
tests/test_scene.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_scene_list():

2
tests/test_sceneitem.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_sceneitem_list():

8
tests/test_stream.py
View File

@@ -6,7 +6,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_stream_start():
@@ -23,7 +23,7 @@ def test_stream_start():
else:
assert result.exit_code == 0
assert 'Streaming started successfully.' in result.stdout
time.sleep(1) # Wait for the streaming to start
time.sleep(0.5) # Wait for the streaming to start
def test_stream_stop():
@@ -37,7 +37,7 @@ def test_stream_stop():
if active:
assert result.exit_code == 0
assert 'Streaming stopped successfully.' in result.stdout
time.sleep(1) # Wait for the streaming to stop
time.sleep(0.5) # Wait for the streaming to stop
else:
assert result.exit_code != 0
assert 'Streaming is not in progress, cannot stop.' in result.stderr
@@ -52,7 +52,7 @@ def test_stream_toggle():
result = runner.invoke(app, ['stream', 'toggle'])
assert result.exit_code == 0
time.sleep(1) # Wait for the stream to toggle
time.sleep(0.5) # Wait for the stream to toggle
if active:
assert 'Streaming stopped successfully.' in result.stdout

2
tests/test_studiomode.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_studio_enable():

2
tests/test_text.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_text_update():

2
tests/test_version.py
View File

@@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner(mix_stderr=False)
runner = CliRunner()
def test_version():