34 Commits

Author SHA1 Message Date
3b184c6531 upd dependencies 2025-07-24 04:21:58 +01:00
8c37ce1fc0 remove typer import 2025-07-24 04:21:48 +01:00
436e4d5345 remove alias, settings 2025-07-24 04:21:40 +01:00
2ef89be184 convert virtualcam commands 2025-07-24 04:09:49 +01:00
506aff833c convert text commands 2025-07-24 04:08:07 +01:00
eb939b735c convert studiomode commands 2025-07-24 04:05:01 +01:00
bb7a468dd5 convert stream commands 2025-07-24 03:59:37 +01:00
e77627b845 convert screenshot commands 2025-07-24 03:52:50 +01:00
93b066090b fix aliases 2025-07-24 03:49:31 +01:00
1ce832dfde convert sceneitem commands 2025-07-24 03:46:11 +01:00
e8664f0117 convert scenecollection commands 2025-07-24 02:34:15 +01:00
a3dff0f739 convert replaybuffer commands 2025-07-24 02:29:58 +01:00
6da9df5ceb convert record commands 2025-07-24 02:26:18 +01:00
75fc18273e convert projector commands 2025-07-24 02:20:17 +01:00
e658819719 convert profile commands 2025-07-24 02:14:38 +01:00
4451fbf22c conver input commands 2025-07-24 02:06:05 +01:00
132b283347 convert hotkey commands 2025-07-24 01:48:55 +01:00
ae8ff20cf4 convert group commands 2025-07-24 01:39:02 +01:00
de1c604c46 update the --help message
add descriptions for filter + scene command groups

Usage now after main CLI description
2025-07-24 01:38:42 +01:00
105aaf29b7 keep the exit codes simple (0 or 1) 2025-07-17 04:34:44 +01:00
eb34a1833f convert filter sub app 2025-07-17 04:29:21 +01:00
abbab5c746 raise OBSWSCLIError 2025-07-17 04:28:59 +01:00
f0eb518609 implement --version + debug validator
run() now handles exceptions/exit codes
2025-07-17 04:28:40 +01:00
032b957670 add custom error class + exit codes 2025-07-17 04:28:10 +01:00
8349a196e8 root meta app + scene sub_app converted
this could take a while...

note, --version flag not implemented
2025-07-17 03:18:04 +01:00
f852a733c3 upd publish action 2025-07-14 03:27:52 +01:00
44dadcee23 upd publish action 2025-07-14 03:25:52 +01:00
ed4531c305 revert publish action 2025-07-14 03:23:25 +01:00
ec42a4cdd9 patch bump 2025-07-14 03:21:29 +01:00
6123c92d00 upd publish action 2025-07-14 03:21:06 +01:00
1ceb95ab16 fix environment name 2025-07-14 03:12:35 +01:00
f06e2d3982 upd publish action 2025-07-14 03:10:04 +01:00
39dff3cc28 patch bump 2025-07-14 03:02:53 +01:00
967c4ab699 upd publish action 2025-07-14 02:58:25 +01:00
28 changed files with 927 additions and 1022 deletions

View File

@@ -11,7 +11,7 @@ jobs:
deploy:
runs-on: ubuntu-latest
environment: release
environment: pypi
permissions:
# This permission is needed for private repositories.
contents: read
@@ -36,5 +36,4 @@ jobs:
run: hatch build
- name: Publish on PyPI
if: github.ref == 'refs/heads/main'
run: hatch publish
uses: pypa/gh-action-pypi-publish@release/v1

View File

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

View File

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

View File

@@ -1,73 +0,0 @@
"""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

View File

@@ -2,18 +2,27 @@
import importlib
import logging
from typing import Annotated
from dataclasses import dataclass
from typing import Annotated, Any
import obsws_python as obsws
import typer
from cyclopts import App, Group, Parameter, config
from obsws_cli.__about__ import __version__ as version
from . import console, settings, styles
from .alias import RootTyperAliasGroup
from . import console, styles
from .context import Context
from .error import OBSWSCLIError
app = typer.Typer(cls=RootTyperAliasGroup)
for sub_typer in (
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 (
'filter',
'group',
'hotkey',
@@ -31,135 +40,93 @@ for sub_typer in (
'text',
'virtualcam',
):
module = importlib.import_module(f'.{sub_typer}', package=__package__)
app.add_typer(module.app, name=sub_typer)
module = importlib.import_module(f'.{sub_app}', package=__package__)
app.command(module.app)
def version_callback(value: bool):
"""Show the version of the CLI."""
if value:
console.out.print(f'obsws-cli version: {version}')
raise typer.Exit()
@Parameter(name='*')
@dataclass
class OBSConfig:
"""Dataclass to hold OBS connection parameters."""
host: str = 'localhost'
port: int = 4455
password: str = ''
def setup_logging(debug: bool):
@dataclass
class StyleConfig:
"""Dataclass to hold style parameters."""
name: str = 'disabled'
no_border: bool = False
def setup_logging(type_, value: Any):
"""Set up logging for the application."""
log_level = logging.DEBUG if debug else logging.CRITICAL
log_level = logging.DEBUG if value else logging.CRITICAL
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
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',
@app.meta.default
def launcher(
*tokens: Annotated[str, Parameter(show=False, allow_leading_hyphen=True)],
obs_config: OBSConfig = Annotated[
OBSConfig,
Parameter(
show=False, allow_leading_hyphen=True, help='OBS connection parameters'
),
] = 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'),
version: Annotated[
bool,
typer.Option(
'--version',
'-v',
is_eager=True,
help='Show the CLI version and exit',
show_default=False,
callback=version_callback,
),
] = False,
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'),
],
style_config: StyleConfig = Annotated[
StyleConfig,
Parameter(show=False, allow_leading_hyphen=True, help='Style parameters'),
],
debug: Annotated[
bool,
typer.Option(
'--debug',
'-d',
envvar='OBS_DEBUG',
is_eager=True,
help='Enable debug logging',
show_default=False,
callback=setup_logging,
hidden=True,
),
] = settings.get('debug'),
Parameter(validator=setup_logging),
] = False,
):
"""obsws_cli is a command line interface for the OBS WebSocket API."""
ctx.ensure_object(dict)
ctx.obj['obsws'] = ctx.with_resource(obsws.ReqClient(**ctx.params))
ctx.obj['style'] = styles.request_style_obj(style, no_border)
"""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)
@app.command()
def obs_version(ctx: typer.Context):
@app.command
def obs_version(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the OBS Client and WebSocket versions."""
resp = ctx.obj['obsws'].get_version()
resp = ctx.client.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

View File

@@ -1,12 +1,13 @@
"""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: typer.Context, text: str) -> str:
def highlight(ctx: Context, text: str) -> str:
"""Highlight text using the current context's style."""
return f'[{ctx.obj["style"].highlight}]{text}[/{ctx.obj["style"].highlight}]'
return f'[{ctx.style.highlight}]{text}[/{ctx.style.highlight}]'

15
obsws_cli/context.py Normal file
View File

@@ -0,0 +1,15 @@
"""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 Normal file
View File

@@ -0,0 +1,10 @@
"""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 Normal file
View File

@@ -0,0 +1,11 @@
"""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

View File

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

View File

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

View File

@@ -2,37 +2,33 @@
from typing import Annotated
import typer
from cyclopts import App, Argument, Parameter
from rich.table import Table
from rich.text import Text
from . import console
from .alias import SubTyperAliasGroup
from .context import Context
app = typer.Typer(cls=SubTyperAliasGroup)
app = App(name='hotkey', help='Commands for managing hotkeys in OBS')
@app.callback()
def main():
"""Control hotkeys in OBS."""
@app.command('list | ls')
@app.command(name=['list', 'ls'])
def list_(
ctx: typer.Context,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""List all hotkeys."""
resp = ctx.obj['obsws'].get_hotkey_list()
resp = ctx.client.get_hotkey_list()
table = Table(
title='Hotkeys',
padding=(0, 2),
border_style=ctx.obj['style'].border,
border_style=ctx.style.border,
)
table.add_column(
Text('Hotkey Name', justify='center'),
justify='left',
style=ctx.obj['style'].column,
style=ctx.style.column,
)
for i, hotkey in enumerate(resp.hotkeys):
@@ -41,40 +37,40 @@ def list_(
console.out.print(table)
@app.command('trigger | tr')
@app.command(name=['trigger', 'tr'])
def trigger(
ctx: typer.Context,
hotkey: Annotated[
str, typer.Argument(..., show_default=False, help='The hotkey to trigger')
],
hotkey: Annotated[str, Argument(hint='The hotkey to trigger')],
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Trigger a hotkey by name."""
ctx.obj['obsws'].trigger_hotkey_by_name(hotkey)
ctx.client.trigger_hotkey_by_name(hotkey)
@app.command('trigger-sequence | trs')
@app.command(name=['trigger-sequence', 'trs'])
def trigger_sequence(
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',
Argument(
hint='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')
bool, Parameter(help='Press shift when triggering the hotkey')
] = False,
ctrl: Annotated[
bool, typer.Option(..., help='Press control when triggering the hotkey')
bool, Parameter(help='Press control when triggering the hotkey')
] = False,
alt: Annotated[
bool, typer.Option(..., help='Press alt when triggering the hotkey')
bool, Parameter(help='Press alt when triggering the hotkey')
] = False,
cmd: Annotated[
bool, typer.Option(..., help='Press cmd when triggering the hotkey')
bool, Parameter(help='Press cmd when triggering the hotkey')
] = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Trigger a hotkey by sequence."""
ctx.obj['obsws'].trigger_hotkey_by_key_sequence(key_id, shift, ctrl, alt, cmd)
ctx.client.trigger_hotkey_by_key_sequence(key_id, shift, ctrl, alt, cmd)

View File

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

View File

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

View File

@@ -2,25 +2,25 @@
from typing import Annotated
import typer
from cyclopts import App, Argument, Parameter
from rich.table import Table
from rich.text import Text
from . import console
from .alias import SubTyperAliasGroup
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
app = typer.Typer(cls=SubTyperAliasGroup)
app = App(name='projector', help='Commands for managing projectors in OBS')
@app.callback()
def main():
"""Control projectors in OBS."""
@app.command('list-monitors | ls-m')
def list_monitors(ctx: typer.Context):
@app.command(name=['list-monitors', 'ls-m'])
def list_monitors(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""List available monitors."""
resp = ctx.obj['obsws'].get_monitor_list()
resp = ctx.client.get_monitor_list()
if not resp.monitors:
console.out.print('No monitors found.')
@@ -34,13 +34,13 @@ def list_monitors(ctx: typer.Context):
table = Table(
title='Available Monitors',
padding=(0, 2),
border_style=ctx.obj['style'].border,
border_style=ctx.style.border,
)
table.add_column(
Text('Index', justify='center'), justify='center', style=ctx.obj['style'].column
Text('Index', justify='center'), justify='center', style=ctx.style.column
)
table.add_column(
Text('Name', justify='center'), justify='left', style=ctx.obj['style'].column
Text('Name', justify='center'), justify='left', style=ctx.style.column
)
for index, monitor in monitors:
@@ -49,29 +49,30 @@ def list_monitors(ctx: typer.Context):
console.out.print(table)
@app.command('open | o')
@app.command(name=['open', 'o'])
def open(
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.',
Argument(
hint='Name of the source to project.',
),
] = '',
/,
monitor_index: Annotated[
int,
Parameter(help='Index of the monitor to open the projector on.'),
] = 0,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Open a fullscreen projector for a source on a specific monitor."""
if not source_name:
source_name = ctx.obj['obsws'].get_current_program_scene().scene_name
source_name = ctx.client.get_current_program_scene().scene_name
monitors = ctx.obj['obsws'].get_monitor_list().monitors
monitors = ctx.client.get_monitor_list().monitors
for monitor in monitors:
if monitor['monitorIndex'] == monitor_index:
ctx.obj['obsws'].open_source_projector(
ctx.client.open_source_projector(
source_name=source_name,
monitor_index=monitor_index,
)
@@ -82,8 +83,8 @@ def open(
break
else:
console.err.print(
raise OBSWSCLIError(
f'Monitor with index [yellow]{monitor_index}[/yellow] not found. '
f'Use [yellow]obsws-cli projector ls-m[/yellow] to see available monitors.'
f'Use [yellow]obsws-cli projector ls-m[/yellow] to see available monitors.',
ExitCode.ERROR,
)
raise typer.Exit(code=1)

View File

@@ -3,67 +3,75 @@
from pathlib import Path
from typing import Annotated, Optional
import typer
from cyclopts import App, Argument, Parameter
from . import console
from .alias import SubTyperAliasGroup
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
app = typer.Typer(cls=SubTyperAliasGroup)
app = App(name='record', help='Commands for controlling OBS recording functionality.')
@app.callback()
def main():
"""Control OBS recording functionality."""
def _get_recording_status(ctx: typer.Context) -> tuple:
def _get_recording_status(ctx: Context) -> tuple:
"""Get recording status."""
resp = ctx.obj['obsws'].get_record_status()
resp = ctx.client.get_record_status()
return resp.output_active, resp.output_paused
@app.command('start | s')
def start(ctx: typer.Context):
@app.command(name=['start', 's'])
def start(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""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)
console.err.print(err_msg)
raise typer.Exit(1)
ctx.obj['obsws'].start_record()
ctx.client.start_record()
console.out.print('Recording started successfully.')
@app.command('stop | st')
def stop(ctx: typer.Context):
@app.command(name=['stop', 'st'])
def stop(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Stop recording."""
active, _ = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot stop.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is not in progress, cannot stop.', ExitCode.ERROR
)
resp = ctx.obj['obsws'].stop_record()
resp = ctx.client.stop_record()
console.out.print(
f'Recording stopped successfully. Saved to: {console.highlight(ctx, resp.output_path)}'
)
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
@app.command(name=['toggle', 'tg'])
def toggle(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Toggle recording."""
resp = ctx.obj['obsws'].toggle_record()
resp = ctx.client.toggle_record()
if resp.output_active:
console.out.print('Recording started successfully.')
else:
console.out.print('Recording stopped successfully.')
@app.command('status | ss')
def status(ctx: typer.Context):
@app.command(name=['status', 'ss'])
def status(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get recording status."""
active, paused = _get_recording_status(ctx)
if active:
@@ -75,98 +83,114 @@ def status(ctx: typer.Context):
console.out.print('Recording is not in progress.')
@app.command('resume | r')
def resume(ctx: typer.Context):
@app.command(name=['resume', 'r'])
def resume(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Resume recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot resume.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is not in progress, cannot resume.', ExitCode.ERROR
)
if not paused:
console.err.print('Recording is in progress but not paused, cannot resume.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is in progress but not paused, cannot resume.', ExitCode.ERROR
)
ctx.obj['obsws'].resume_record()
ctx.client.resume_record()
console.out.print('Recording resumed successfully.')
@app.command('pause | p')
def pause(ctx: typer.Context):
@app.command(name=['pause', 'p'])
def pause(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Pause recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot pause.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is not in progress, cannot pause.', ExitCode.ERROR
)
if paused:
console.err.print('Recording is in progress but already paused, cannot pause.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is in progress but already paused, cannot pause.', ExitCode.ERROR
)
ctx.obj['obsws'].pause_record()
ctx.client.pause_record()
console.out.print('Recording paused successfully.')
@app.command('directory | d')
@app.command(name=['directory', 'd'])
def directory(
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.',
Argument(
hint='Directory to set for recording.',
),
] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get or set the recording directory."""
if record_directory is not None:
ctx.obj['obsws'].set_record_directory(str(record_directory))
ctx.client.set_record_directory(str(record_directory))
console.out.print(
f'Recording directory updated to: {console.highlight(ctx, record_directory)}'
)
else:
resp = ctx.obj['obsws'].get_record_directory()
resp = ctx.client.get_record_directory()
console.out.print(
f'Recording directory: {console.highlight(ctx, resp.record_directory)}'
)
@app.command('split | sp')
def split(ctx: typer.Context):
@app.command(name=['split', 'sp'])
def split(
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Split the current recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot split.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is not in progress, cannot split.', ExitCode.ERROR
)
if paused:
console.err.print('Recording is paused, cannot split.')
raise typer.Exit(1)
raise OBSWSCLIError('Recording is paused, cannot split.', ExitCode.ERROR)
ctx.obj['obsws'].split_record_file()
ctx.client.split_record_file()
console.out.print('Recording split successfully.')
@app.command('chapter | ch')
@app.command(name=['chapter', 'ch'])
def chapter(
ctx: typer.Context,
chapter_name: Annotated[
Optional[str],
typer.Argument(
help='Name of the chapter to create.',
Argument(
hint='Name of the chapter to create.',
),
] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Create a chapter in the current recording."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot create chapter.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is not in progress, cannot create chapter.', ExitCode.ERROR
)
if paused:
console.err.print('Recording is paused, cannot create chapter.')
raise typer.Exit(1)
raise OBSWSCLIError(
'Recording is paused, cannot create chapter.', ExitCode.ERROR
)
ctx.obj['obsws'].create_record_chapter(chapter_name)
ctx.client.create_record_chapter(chapter_name)
console.out.print(
f'Chapter {console.highlight(ctx, chapter_name or "unnamed")} created successfully.'
)

View File

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

View File

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

View File

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

View File

@@ -2,41 +2,42 @@
from typing import Annotated, Optional
import typer
from cyclopts import App, Argument, Parameter
from rich.table import Table
from . import console, util, validate
from .alias import SubTyperAliasGroup
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
app = typer.Typer(cls=SubTyperAliasGroup)
app = App(name='sceneitem', help='Commands for controlling scene items in OBS.')
@app.callback()
def main():
"""Control items in OBS scenes."""
@app.command('list | ls')
@app.command(name=['list', 'ls'])
def list_(
ctx: typer.Context,
scene_name: Annotated[
Optional[str],
typer.Argument(
show_default='The current scene',
help='Scene name to list items for',
Argument(
hint='Scene name to list items for',
),
] = None,
uuid: Annotated[bool, typer.Option(help='Show UUIDs of scene items')] = False,
/,
uuid: Annotated[bool, Parameter(help='Show UUIDs of scene items')] = False,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""List all items in a scene."""
if not scene_name:
scene_name = ctx.obj['obsws'].get_current_program_scene().scene_name
scene_name = ctx.client.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 typer.Exit(1)
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
exit_code=ExitCode.ERROR,
)
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
resp = ctx.client.get_scene_item_list(scene_name)
items = sorted(
(
(
@@ -52,10 +53,10 @@ def list_(
)
if not items:
console.out.print(
f'No items found in scene {console.highlight(ctx, scene_name)}.'
raise OBSWSCLIError(
f'No items found in scene [yellow]{scene_name}[/yellow].',
exit_code=ExitCode.SUCCESS,
)
raise typer.Exit()
table = Table(
title=f'Items in Scene: {scene_name}',
@@ -138,36 +139,39 @@ def list_(
def _validate_sources(
ctx: typer.Context,
ctx: 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):
console.err.print(f'Scene [yellow]{scene_name}[/yellow] not found.')
return False
raise OBSWSCLIError(
f'Scene [yellow]{scene_name}[/yellow] not found.',
exit_code=ExitCode.ERROR,
)
if group:
if not validate.item_in_scene_item_list(ctx, scene_name, group):
console.err.print(
f'Group [yellow]{group}[/yellow] not found in scene [yellow]{scene_name}[/yellow].'
raise OBSWSCLIError(
f'Group [yellow]{group}[/yellow] not found in scene [yellow]{scene_name}[/yellow].',
exit_code=ExitCode.ERROR,
)
return False
else:
if not validate.item_in_scene_item_list(ctx, scene_name, item_name):
console.err.print(
raise OBSWSCLIError(
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.'
'Use [yellow]obsws-cli sceneitem ls[/yellow] for a list of items in the scene.',
exit_code=ExitCode.ERROR,
)
return False
return True
def _get_scene_name_and_item_id(
ctx: typer.Context, scene_name: str, item_name: str, group: Optional[str] = None
ctx: 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:
@@ -178,10 +182,10 @@ def _get_scene_name_and_item_id(
scene_item_id = item.get('sceneItemId')
break
else:
console.err.print(
f'Item [yellow]{item_name}[/yellow] not found in group [yellow]{group}[/yellow].'
raise OBSWSCLIError(
f'Item [yellow]{item_name}[/yellow] not found in group [yellow]{group}[/yellow].',
exit_code=ExitCode.ERROR,
)
raise typer.Exit(1)
else:
resp = ctx.obj['obsws'].get_scene_item_id(scene_name, item_name)
scene_item_id = resp.scene_item_id
@@ -189,21 +193,20 @@ def _get_scene_name_and_item_id(
return scene_name, scene_item_id
@app.command('show | sh')
@app.command(name=['show', 'sh'])
def show(
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
scene_name: Annotated[str, Argument(hint='Scene name the item is in')],
item_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Item name to show in the scene'),
Argument(hint='Item name to show in the scene'),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
/,
group: Annotated[Optional[str], Parameter(help='Parent group name')] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Show an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
_validate_sources(ctx, scene_name, item_name, group)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
@@ -231,21 +234,20 @@ def show(
)
@app.command('hide | h')
@app.command(name=['hide', 'h'])
def hide(
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
scene_name: Annotated[str, Argument(hint='Scene name the item is in')],
item_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Item name to hide in the scene'),
Argument(hint='Item name to hide in the scene'),
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
/,
group: Annotated[Optional[str], Parameter(help='Parent group name')] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Hide an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
_validate_sources(ctx, scene_name, item_name, group)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
@@ -272,36 +274,30 @@ def hide(
)
@app.command('toggle | tg')
@app.command(name=['toggle', 'tg'])
def toggle(
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,
scene_name: Annotated[str, Argument(hint='Scene name the item is in')],
item_name: Annotated[str, Argument(hint='Item name to toggle in the scene')],
/,
group: Annotated[Optional[str], Parameter(help='Parent group name')] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Toggle an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
_validate_sources(ctx, scene_name, item_name, group)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
ctx, scene_name, item_name, group
)
enabled = ctx.obj['obsws'].get_scene_item_enabled(
enabled = ctx.client.get_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
)
new_state = not enabled.scene_item_enabled
ctx.obj['obsws'].set_scene_item_enabled(
ctx.client.set_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
enabled=new_state,
@@ -333,30 +329,26 @@ def toggle(
)
@app.command('visible | v')
@app.command(name=['visible', 'v'])
def visible(
ctx: typer.Context,
scene_name: Annotated[
str, typer.Argument(..., show_default=False, help='Scene name the item is in')
],
scene_name: Annotated[str, Argument(hint='Scene name the item is in')],
item_name: Annotated[
str,
typer.Argument(
..., show_default=False, help='Item name to check visibility in the scene'
),
str, Argument(hint='Item name to check visibility in the scene')
],
group: Annotated[Optional[str], typer.Option(help='Parent group name')] = None,
/,
group: Annotated[Optional[str], Parameter(help='Parent group name')] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Check if an item in a scene is visible."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
_validate_sources(ctx, scene_name, item_name, group)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
ctx, scene_name, item_name, group
)
enabled = ctx.obj['obsws'].get_scene_item_enabled(
enabled = ctx.client.get_scene_item_enabled(
scene_name=scene_name,
item_id=int(scene_item_id),
)
@@ -377,68 +369,62 @@ def visible(
)
@app.command('transform | t')
@app.command(name=['transform', 't'])
def transform(
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,
scene_name: Annotated[str, Argument(hint='Scene name the item is in')],
item_name: Annotated[str, Argument(hint='Item name to transform in the scene')],
/,
group: Annotated[Optional[str], Parameter(help='Parent group name')] = None,
alignment: Annotated[
Optional[int], typer.Option(help='Alignment of the item in the scene')
Optional[int], Parameter(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')
Optional[int], Parameter(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')
Optional[float], Parameter(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')
Optional[str], Parameter(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')
Optional[float], Parameter(help='Width of the item in the scene')
] = None,
crop_to_bounds: Annotated[
Optional[bool], typer.Option(help='Crop the item to the bounds')
Optional[bool], Parameter(help='Crop the item to the bounds')
] = None,
crop_bottom: Annotated[
Optional[float], typer.Option(help='Bottom crop of the item in the scene')
Optional[float], Parameter(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')
Optional[float], Parameter(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')
Optional[float], Parameter(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')
Optional[float], Parameter(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')
Optional[float], Parameter(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')
Optional[float], Parameter(help='Y position of the item in the scene')
] = None,
rotation: Annotated[
Optional[float], typer.Option(help='Rotation of the item in the scene')
Optional[float], Parameter(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')
Optional[float], Parameter(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')
Optional[float], Parameter(help='Y scale of the item in the scene')
] = None,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Set the transform of an item in a scene."""
if not _validate_sources(ctx, scene_name, item_name, group):
raise typer.Exit(1)
_validate_sources(ctx, scene_name, item_name, group)
old_scene_name = scene_name
scene_name, scene_item_id = _get_scene_name_and_item_id(
@@ -478,10 +464,12 @@ def transform(
transform['scaleY'] = scale_y
if not transform:
console.err.print('No transform options provided.')
raise typer.Exit(1)
raise OBSWSCLIError(
'No transform options provided. Use at least one of the transform options.',
exit_code=ExitCode.ERROR,
)
transform = ctx.obj['obsws'].set_scene_item_transform(
transform = ctx.client.set_scene_item_transform(
scene_name=scene_name,
item_id=int(scene_item_id),
transform=transform,

View File

@@ -4,66 +4,57 @@ from pathlib import Path
from typing import Annotated
import obsws_python as obsws
import typer
from cyclopts import App, Argument, Parameter
from . import console
from .alias import SubTyperAliasGroup
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
app = typer.Typer(cls=SubTyperAliasGroup)
app = App(name='screenshot', help='Commands for taking screenshots using OBS.')
@app.callback()
def main():
"""Take screenshots using OBS."""
@app.command('save | sv')
@app.command(name=['save', 'sv'])
def save(
ctx: typer.Context,
source_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the source to take a screenshot of.',
Argument(
hint='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).',
Argument(
hint='Path to save the screenshot (must include file name and extension).',
),
],
/,
width: Annotated[
float,
typer.Option(
Parameter(
help='Width of the screenshot.',
),
] = 1920,
height: Annotated[
float,
typer.Option(
Parameter(
help='Height of the screenshot.',
),
] = 1080,
quality: Annotated[
float,
typer.Option(
min=-1,
max=100,
Parameter(
help='Quality of the screenshot.',
),
] = -1,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Take a screenshot and save it to a file."""
try:
ctx.obj['obsws'].save_source_screenshot(
ctx.client.save_source_screenshot(
name=source_name,
img_format=output_path.suffix.lstrip('.').lower(),
file_path=str(output_path),
@@ -74,16 +65,16 @@ def save(
except obsws.error.OBSSDKRequestError as e:
match e.code:
case 403:
console.err.print(
raise OBSWSCLIError(
'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:
console.err.print(
f'No source was found by the name of [yellow]{source_name}[/yellow]'
raise OBSWSCLIError(
'No source was found by the name of [yellow]{source_name}[/yellow]',
code=ExitCode.ERROR,
)
raise typer.Exit(1)
case _:
raise

View File

@@ -1,80 +0,0 @@
"""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]

View File

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

View File

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

View File

@@ -2,35 +2,34 @@
from typing import Annotated, Optional
import typer
from cyclopts import App, Argument, Parameter
from . import console, validate
from .alias import SubTyperAliasGroup
from .context import Context
from .enum import ExitCode
from .error import OBSWSCLIError
app = typer.Typer(cls=SubTyperAliasGroup)
app = App(name='text', help='Commands for controlling text inputs in OBS.')
@app.callback()
def main():
"""Control text inputs in OBS."""
@app.command('current | get')
@app.command(name=['current', 'get'])
def current(
ctx: typer.Context,
input_name: Annotated[str, typer.Argument(help='Name of the text input to get.')],
input_name: Annotated[str, Argument(hint='Name of the text input to get.')],
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Get the current text for a text input."""
if not validate.input_in_inputs(ctx, input_name):
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj['obsws'].get_input_settings(name=input_name)
if not resp.input_kind.startswith('text_'):
console.err.print(
f'Input [yellow]{input_name}[/yellow] is not a text input.',
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.', code=ExitCode.ERROR
)
resp = ctx.client.get_input_settings(name=input_name)
if not resp.input_kind.startswith('text_'):
raise OBSWSCLIError(
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:
@@ -40,32 +39,31 @@ def current(
)
@app.command('update | set')
@app.command(name=['update', 'set'])
def update(
ctx: typer.Context,
input_name: Annotated[
str, typer.Argument(help='Name of the text input to update.')
],
input_name: Annotated[str, Argument(hint='Name of the text input to update.')],
new_text: Annotated[
Optional[str],
typer.Argument(
help='The new text to set for the input.',
),
Argument(hint='The new text to set for the input.'),
] = None,
/,
*,
ctx: Annotated[Context, Parameter(parse=False)],
):
"""Update the text of a text input."""
if not validate.input_in_inputs(ctx, input_name):
console.err.print(f'Input [yellow]{input_name}[/yellow] not found.')
raise typer.Exit(1)
resp = ctx.obj['obsws'].get_input_settings(name=input_name)
if not resp.input_kind.startswith('text_'):
console.err.print(
f'Input [yellow]{input_name}[/yellow] is not a text input.',
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] not found.', code=ExitCode.ERROR
)
raise typer.Exit(1)
ctx.obj['obsws'].set_input_settings(
resp = ctx.client.get_input_settings(name=input_name)
if not resp.input_kind.startswith('text_'):
raise OBSWSCLIError(
f'Input [yellow]{input_name}[/yellow] is not a text input.',
code=ExitCode.ERROR,
)
ctx.client.set_input_settings(
name=input_name,
settings={'text': new_text},
overlay=True,

View File

@@ -1,54 +1,49 @@
"""module containing validation functions."""
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)
from .context import Context
def input_in_inputs(ctx: typer.Context, input_name: str) -> bool:
def input_in_inputs(ctx: Context, input_name: str) -> bool:
"""Check if an input is in the input list."""
inputs = ctx.obj['obsws'].get_input_list().inputs
inputs = ctx.client.get_input_list().inputs
return any(input_.get('inputName') == input_name for input_ in inputs)
def scene_in_scenes(ctx: typer.Context, scene_name: str) -> bool:
def scene_in_scenes(ctx: Context, scene_name: str) -> bool:
"""Check if a scene exists in the list of scenes."""
resp = ctx.obj['obsws'].get_scene_list()
resp = ctx.client.get_scene_list()
return any(scene.get('sceneName') == scene_name for scene in resp.scenes)
def studio_mode_enabled(ctx: typer.Context) -> bool:
def studio_mode_enabled(ctx: Context) -> bool:
"""Check if studio mode is enabled."""
resp = ctx.obj['obsws'].get_studio_mode_enabled()
resp = ctx.client.get_studio_mode_enabled()
return resp.studio_mode_enabled
def scene_collection_in_scene_collections(
ctx: typer.Context, scene_collection_name: str
ctx: Context, scene_collection_name: str
) -> bool:
"""Check if a scene collection exists."""
resp = ctx.obj['obsws'].get_scene_collection_list()
resp = ctx.client.get_scene_collection_list()
return any(
collection == scene_collection_name for collection in resp.scene_collections
)
def item_in_scene_item_list(
ctx: typer.Context, scene_name: str, item_name: str
) -> bool:
def item_in_scene_item_list(ctx: Context, scene_name: str, item_name: str) -> bool:
"""Check if an item exists in a scene."""
resp = ctx.obj['obsws'].get_scene_item_list(scene_name)
resp = ctx.client.get_scene_item_list(scene_name)
return any(item.get('sourceName') == item_name for item in resp.scene_items)
def profile_exists(ctx: typer.Context, profile_name: str) -> bool:
def profile_exists(ctx: Context, profile_name: str) -> bool:
"""Check if a profile exists."""
resp = ctx.obj['obsws'].get_profile_list()
resp = ctx.client.get_profile_list()
return any(profile == profile_name for profile in resp.profiles)
def monitor_exists(ctx: typer.Context, monitor_index: int) -> bool:
def monitor_exists(ctx: Context, monitor_index: int) -> bool:
"""Check if a monitor exists."""
resp = ctx.obj['obsws'].get_monitor_list()
resp = ctx.client.get_monitor_list()
return any(monitor['monitorIndex'] == monitor_index for monitor in resp.monitors)

View File

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

View File

@@ -21,7 +21,7 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
]
dependencies = ["typer>=0.16.0", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
dependencies = ["cyclopts>=3.22.2", "obsws-python>=1.8.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:app"
obsws-cli = "obsws_cli:run"
[tool.hatch.version]
path = "obsws_cli/__about__.py"