43 Commits

Author SHA1 Message Date
8da29ce90e upd typer dep version
obsws-cli minor version bump
2026-01-09 09:21:43 +00:00
72c6bcee49 upd README with new input subcommands 2026-01-09 09:21:06 +00:00
dceafba065 extend input command group 2026-01-09 09:20:45 +00:00
7a73ec35f6 remove lazyimports env 2025-09-29 20:51:17 +01:00
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
52e13922dc upd test delays to 500ms 2025-07-30 08:42:11 +01:00
f335d8ffd2 move the version flag 2025-07-29 08:48:30 +01:00
286cda8066 raise typer.Exit() on empty list queries 2025-07-29 08:17:52 +01:00
e851219ced tests should now pass from fresh install 2025-07-29 08:03:24 +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
dc128720c7 hatch fmt 2025-07-14 02:48:21 +01:00
2e3f4267cd add workflows 2025-07-14 02:45:13 +01:00
000431ab82 add 0.20.0 to CHANGELOG 2025-07-14 02:32:59 +01:00
ec3e31cc4f add Text section to README 2025-07-14 02:32:21 +01:00
cda0bbedb9 minor bump 2025-07-14 02:32:09 +01:00
d0c96b853d add text unit tests 2025-07-14 02:31:47 +01:00
040a41daa7 add text command group 2025-07-14 02:31:35 +01:00
0c72a10fb7 bump obsws-python version 2025-07-01 09:30:04 +01:00
f882302d16 fixes missing argument 2025-07-01 09:29:56 +01:00
98e0d98cc7 typo 2025-06-27 13:45:24 +01:00
c6b22c7cf2 use console.highlight() 2025-06-27 13:29:39 +01:00
c3e55200db move style section
add link to style section in ToC.

add imgs.
2025-06-27 13:14:54 +01:00
4d37714aaf patch bump 2025-06-27 12:57:49 +01:00
157e1a167c fixes bug when setting --style=disabled (we were stil getting coloured check/cross marks) 2025-06-27 12:57:34 +01:00
d628c5d3a4 rename heading variables 2025-06-27 12:53:10 +01:00
4bf8edb692 add 0.19.0 to CHANGELOG 2025-06-23 09:11:26 +01:00
d68326f37a add record split/chapter to README 2025-06-23 09:11:15 +01:00
a001455dad add record split/chapter commands 2025-06-23 09:10:53 +01:00
4632260961 add --style validation
add Disabled class to style registry

patch bump
2025-06-22 12:35:21 +01:00
55a7da67db reword 2025-06-22 10:19:46 +01:00
7bec573ef9 by setting values in the default style to 'none' we avoid the rich markup errors in console.highlight
add comment to util.check_mark and test only NO_COLOR

patch bump
2025-06-22 10:14:46 +01:00
55e60ff977 in case NO_COLOR is set manually
patch bump
2025-06-22 02:49:32 +01:00
922efddf7a check if we're in colourless mode before passing back highlighted text.
pass context to check_mark so we can do the same there.

Fixes  rich.errors.MarkupError
2025-06-22 01:57:58 +01:00
4a0147aa8a import as version 2025-06-22 00:38:19 +01:00
cec76df1d1 add 0.18.0 to CHANGELOG 2025-06-21 23:47:19 +01:00
40 changed files with 862 additions and 153 deletions

39
.github/workflows/publish.yml vendored Normal file
View File

@@ -0,0 +1,39 @@
name: Publish to PyPI
on:
release:
types: [published]
push:
tags:
- 'v*.*.*'
jobs:
deploy:
runs-on: ubuntu-latest
environment: pypi
permissions:
# This permission is needed for private repositories.
contents: read
# IMPORTANT: this permission is mandatory for trusted publishing
id-token: write
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
cache: 'pip'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install hatch
- name: Build package
run: hatch build
- name: Publish on PyPI
uses: pypa/gh-action-pypi-publish@release/v1

19
.github/workflows/ruff.yml vendored Normal file
View File

@@ -0,0 +1,19 @@
name: Ruff
on:
push:
branches: [main]
pull_request:
branches: [main]
workflow_dispatch:
jobs:
ruff:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: astral-sh/ruff-action@v3
with:
args: 'format --check --diff'

View File

@@ -5,6 +5,28 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
# [0.20.0] - 2025-07-14
### Added
- text command group, see [Text](https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#text)
# [0.19.0] - 2025-06-23
### Added
- record split and record chapter commands, see [Record](https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#record)
- As of OBS 30.2.0, the only file format supporting *record chapter* is Hybrid MP4.
# [0.18.0] - 2025-06-21
### Added
- Various colouring styles, see [Style](https://github.com/onyx-and-iris/obsws-cli/tree/main?tab=readme-ov-file#style)
- colouring is applied to list tables as well as highlighted information in stdout/stderr output.
- table border styling may be optionally disabled with the --no-border flag.
# [0.17.3] - 2025-06-20
### Added

137
README.md
View File

@@ -14,6 +14,7 @@ For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
- [Installation](#installation)
- [Configuration](#configuration)
- [Style](#style)
- [Commands](#root-typer)
- [License](#license)
@@ -68,6 +69,37 @@ OBS_PASSWORD=<websocket password>
Flags can be used to override environment variables.
## Style
Styling is opt-in, by default you will get a colourless output:
![colourless](./img/colourless.png)
You may enable styling with the --style/-s flag:
```console
obsws-cli --style="cyan" sceneitem list
```
Available styles: _red, magenta, purple, blue, cyan, green, yellow, orange, white, grey, navy, black_
![coloured](./img/coloured-border.png)
Optionally you may disable border colouring with the --no-border flag:
![coloured-no-border](./img/coloured-no-border.png)
```console
obsws-cli --style="cyan" --no-border sceneitem list
```
Or with environment variables:
```env
OBS_STYLE=cyan
OBS_STYLE_NO_BORDER=true
```
## Root Typer
- obs-version: Get the OBS Client and WebSocket versions.
@@ -266,6 +298,20 @@ obsws-cli group status START "test_group"
#### Input
- create: Create a new input.
- args: <input_name> <input_kind>
```console
obsws-cli input create 'stream mix' 'wasapi_input_capture'
```
- remove: Remove an input.
- args: <input_name>
```console
obsws-cli input remove 'stream mix'
```
- list: List all inputs.
- flags:
@@ -283,6 +329,12 @@ obsws-cli input list
obsws-cli input list --input --colour
```
- list-kinds: List all input kinds.
```console
obsws-cli input list-kinds
```
- mute: Mute an input.
- args: <input_name>
@@ -303,6 +355,48 @@ obsws-cli input unmute "Mic/Aux"
obsws-cli input toggle "Mic/Aux"
```
- volume: Set the volume of an input.
- args: <input_name> <volume>
```console
obsws-cli input volume -- 'Desktop Audio' -38.9
```
- show: Show information for an input in the current scene.
- args: <input_name>
- flags:
*optional*
- --verbose: List all available input devices.
```console
obsws-cli input show 'Mic/Aux' --verbose
```
- update: Name of the input to update.
- args: <input_name> <device_name>
```console
obsws-cli input update 'Mic/Aux' 'Voicemeeter Out B1 (VB-Audio Voicemeeter VAIO)'
```
#### Text
- current: Get the current text for a text input.
- args: <input_name>
```console
obsws-cli text current "My Text Input"
```
- update: Update the text of a text input.
- args: <input_name> <new_text>
```console
obsws-cli text update "My Text Input" "hi OBS!"
```
#### Record
- start: Start recording.
@@ -354,6 +448,21 @@ obsws-cli record directory "/home/me/obs-vids/"
obsws-cli record directory "C:/Users/me/Videos"
```
- split: Split the current recording.
```console
obsws-cli record split
```
- chapter: Create a chapter in the current recording.
*optional*
- args: <chapter_name>
```console
obsws-cli record chapter "Chapter Name"
```
#### Stream
- start: Start streaming.
@@ -613,34 +722,6 @@ obsws-cli projector open --monitor-index=1 "test_group"
obsws-cli screenshot save --width=2560 --height=1440 "Scene" "C:\Users\me\Videos\screenshot.png"
```
## Style
By default styling is disabled but you may enable it with:
- --style/-s: Style used in output.
- OBS_STYLE
- --no-border/-b: Disable table border styling in output.
- OBS_STYLE_NO_BORDER
Available styles:
- red
- magenta
- purple
- blue
- cyan
- green
- yellow
- orange
- white
- grey
- navy
- black
```console
obsws-cli --style=cyan --no-border scene list
```
## License
`obsws-cli` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.

BIN
img/coloured-border.png Executable file

Binary file not shown.

After

Width:  |  Height:  |  Size: 6.2 KiB

BIN
img/coloured-no-border.png Executable file

Binary file not shown.

After

Width:  |  Height:  |  Size: 6.2 KiB

BIN
img/colourless.png Executable file

Binary file not shown.

After

Width:  |  Height:  |  Size: 5.4 KiB

View File

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

View File

@@ -4,4 +4,4 @@
from .app import app
__all__ = ["app"]
__all__ = ['app']

View File

@@ -44,6 +44,8 @@ class RootTyperAliasGroup(typer.core.TyperGroup):
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)

View File

@@ -7,7 +7,7 @@ from typing import Annotated
import obsws_python as obsws
import typer
from obsws_cli.__about__ import __version__ as obsws_cli_version
from obsws_cli.__about__ import __version__ as version
from . import console, settings, styles
from .alias import RootTyperAliasGroup
@@ -28,6 +28,7 @@ for sub_typer in (
'screenshot',
'stream',
'studiomode',
'text',
'virtualcam',
):
module = importlib.import_module(f'.{sub_typer}', package=__package__)
@@ -37,7 +38,7 @@ for sub_typer in (
def version_callback(value: bool):
"""Show the version of the CLI."""
if value:
console.out.print(f'obsws-cli version: {obsws_cli_version}')
console.out.print(f'obsws-cli version: {version}')
raise typer.Exit()
@@ -50,6 +51,15 @@ def setup_logging(debug: bool):
)
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,
@@ -93,17 +103,6 @@ def main(
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(
@@ -112,6 +111,7 @@ def main(
envvar='OBS_STYLE',
help='Set the style for the CLI output',
show_default='disabled',
callback=validate_style,
),
] = settings.get('style'),
no_border: Annotated[
@@ -124,6 +124,17 @@ def main(
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,
typer.Option(
@@ -140,7 +151,9 @@ def main(
):
"""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['obsws'] = ctx.with_resource(
obsws.ReqClient(host=host, port=port, password=password, timeout=timeout)
)
ctx.obj['style'] = styles.request_style_obj(style, no_border)

View File

@@ -62,8 +62,8 @@ def list_(
(Text('Enabled', justify='center'), 'center', None),
(Text('Settings', justify='center'), 'center', ctx.obj['style'].column),
]
for name, justify, style in columns:
table.add_column(name, justify=justify, style=style)
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'])

View File

@@ -61,8 +61,8 @@ def list_(
(Text('Group Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Enabled', justify='center'), 'center', None),
]
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for item_id, group_name, is_enabled in groups:
table.add_row(

View File

@@ -24,6 +24,10 @@ def list_(
"""List all hotkeys."""
resp = ctx.obj['obsws'].get_hotkey_list()
if not resp.hotkeys:
console.out.print('No hotkeys found.')
raise typer.Exit()
table = Table(
title='Hotkeys',
padding=(0, 2),

View File

@@ -18,6 +18,69 @@ def main():
"""Control inputs in OBS."""
@app.command('create | add')
def create(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the input to create.',
callback=validate.input_not_in_inputs,
),
],
input_kind: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Kind of the input to create.',
callback=validate.kind_in_input_kinds,
),
],
):
"""Create a new input."""
current_scene = (
ctx.obj['obsws'].get_current_program_scene().current_program_scene_name
)
try:
ctx.obj['obsws'].create_input(
inputName=input_name,
inputKind=input_kind,
sceneItemEnabled=True,
sceneName=current_scene,
inputSettings={},
)
except obsws.error.OBSSDKRequestError as e:
console.err.print(f'Failed to create input: [yellow]{e}[/yellow]')
raise typer.Exit(1)
console.out.print(
f'Input {console.highlight(ctx, input_name)} of kind '
f'{console.highlight(ctx, input_kind)} created.',
)
@app.command('remove | rm')
def remove(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the input to remove.',
callback=validate.input_in_inputs,
),
],
):
"""Remove an input."""
ctx.obj['obsws'].remove_input(name=input_name)
console.out.print(f'Input {console.highlight(ctx, input_name)} removed.')
@app.command('list | ls')
def list_(
ctx: typer.Context,
@@ -74,8 +137,8 @@ def list_(
(Text('Kind', justify='center'), 'center', ctx.obj['style'].column),
(Text('Muted', justify='center'), 'center', None),
]
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for input_name, input_kind, input_uuid in inputs:
input_mark = ''
@@ -105,18 +168,47 @@ def list_(
console.out.print(table)
@app.command('list-kinds | ls-k')
def list_kinds(
ctx: typer.Context,
):
"""List all input kinds."""
resp = ctx.obj['obsws'].get_input_kind_list(False)
kinds = sorted(resp.input_kinds)
if not kinds:
console.out.print('No input kinds found.')
raise typer.Exit()
table = Table(
title='Input Kinds', padding=(0, 2), border_style=ctx.obj['style'].border
)
table.add_column(
Text('Input Kind', justify='center'),
justify='left',
style=ctx.obj['style'].column,
)
for kind in kinds:
table.add_row(util.snakecase_to_titlecase(kind))
console.out.print(table)
@app.command('mute | m')
def mute(
ctx: typer.Context,
input_name: Annotated[
str, typer.Argument(..., show_default=False, help='Name of the input to mute.')
str,
typer.Argument(
...,
show_default=False,
help='Name of the input to mute.',
callback=validate.input_in_inputs,
),
],
):
"""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)
ctx.obj['obsws'].set_input_mute(
name=input_name,
muted=True,
@@ -130,14 +222,15 @@ def unmute(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Name of the input to unmute.'),
typer.Argument(
...,
show_default=False,
help='Name of the input to unmute.',
callback=validate.input_in_inputs,
),
],
):
"""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)
ctx.obj['obsws'].set_input_mute(
name=input_name,
muted=False,
@@ -151,14 +244,15 @@ def toggle(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(..., show_default=False, help='Name of the input to toggle.'),
typer.Argument(
...,
show_default=False,
help='Name of the input to toggle.',
callback=validate.input_in_inputs,
),
],
):
"""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)
resp = ctx.obj['obsws'].get_input_mute(name=input_name)
new_state = not resp.input_muted
@@ -175,3 +269,188 @@ def toggle(
console.out.print(
f'Input {console.highlight(ctx, input_name)} unmuted.',
)
@app.command('volume | vol')
def volume(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the input to set volume for.',
callback=validate.input_in_inputs,
),
],
volume: Annotated[
float,
typer.Argument(
...,
show_default=False,
help='Volume level to set (-90 to 0).',
min=-90,
max=0,
),
],
):
"""Set the volume of an input."""
ctx.obj['obsws'].set_input_volume(
name=input_name,
vol_db=volume,
)
console.out.print(
f'Input {console.highlight(ctx, input_name)} volume set to {console.highlight(ctx, volume)}.',
)
@app.command('show | s')
def show(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the input to show.',
callback=validate.input_in_inputs,
),
],
verbose: Annotated[
bool, typer.Option(help='List all available input devices.')
] = False,
):
"""Show information for an input in the current scene."""
input_list = ctx.obj['obsws'].get_input_list()
for input_ in input_list.inputs:
if input_['inputName'] == input_name:
input_kind = input_['inputKind']
break
for prop in ['device', 'device_id']:
try:
device_id = (
ctx.obj['obsws']
.get_input_settings(
name=input_name,
)
.input_settings.get(prop)
)
if device_id:
break
except obsws.error.OBSSDKRequestError:
continue
else:
device_id = '(N/A)'
for device in (
ctx.obj['obsws']
.get_input_properties_list_property_items(
input_name=input_name,
prop_name=prop,
)
.property_items
):
if device.get('itemValue') == device_id:
device_id = device.get('itemName')
break
table = Table(
title='Input Information', padding=(0, 2), border_style=ctx.obj['style'].border
)
columns = [
(Text('Input Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Kind', justify='center'), 'left', ctx.obj['style'].column),
(Text('Device', justify='center'), 'left', ctx.obj['style'].column),
]
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
table.add_row(
input_name,
util.snakecase_to_titlecase(input_kind),
device_id,
)
console.out.print(table)
if verbose:
resp = ctx.obj['obsws'].get_input_properties_list_property_items(
input_name=input_name,
prop_name=prop,
)
table = Table(
title='Devices',
padding=(0, 2),
border_style=ctx.obj['style'].border,
)
columns = [
(Text('Name', justify='center'), 'left', ctx.obj['style'].column),
]
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for i, item in enumerate(resp.property_items):
table.add_row(
item.get('itemName'),
style='' if i % 2 == 0 else 'dim',
)
console.out.print(table)
@app.command('update | upd')
def update(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the input to update.',
callback=validate.input_in_inputs,
),
],
device_name: Annotated[
str,
typer.Argument(
...,
show_default=False,
help='Name of the device to set for the input.',
),
],
):
"""Update a setting for an input."""
device_id = None
for prop in ['device', 'device_id']:
try:
for device in (
ctx.obj['obsws']
.get_input_properties_list_property_items(
input_name=input_name,
prop_name=prop,
)
.property_items
):
if device.get('itemName') == device_name:
device_id = device.get('itemValue')
break
except obsws.error.OBSSDKRequestError:
continue
if device_id:
break
if not device_id:
console.err.print(
f'Failed to find device ID for device '
f'{console.highlight(ctx, device_name)}.',
)
raise typer.Exit(1)
ctx.obj['obsws'].set_input_settings(
name=input_name, settings={prop: device_id}, overlay=True
)
console.out.print(
f'Input {console.highlight(ctx, input_name)} updated to use device '
f'{console.highlight(ctx, device_name)}.',
)

View File

@@ -22,6 +22,10 @@ def list_(ctx: typer.Context):
"""List profiles."""
resp = ctx.obj['obsws'].get_profile_list()
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
)
@@ -29,13 +33,15 @@ def list_(ctx: typer.Context):
(Text('Profile Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Current', justify='center'), 'center', None),
]
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for profile in resp.profiles:
table.add_row(
profile,
util.check_mark(profile == resp.current_profile_name, empty_if_false=True),
util.check_mark(
ctx, profile == resp.current_profile_name, empty_if_false=True
),
)
console.out.print(table)

View File

@@ -21,16 +21,15 @@ def main():
def list_monitors(ctx: typer.Context):
"""List available monitors."""
resp = ctx.obj['obsws'].get_monitor_list()
if not resp.monitors:
console.out.print('No monitors found.')
return
monitors = sorted(
((m['monitorIndex'], m['monitorName']) for m in resp.monitors),
key=lambda m: m[0],
)
if not monitors:
console.out.print('No monitors found.')
raise typer.Exit()
table = Table(
title='Available Monitors',
padding=(0, 2),

View File

@@ -130,3 +130,43 @@ def directory(
console.out.print(
f'Recording directory: {console.highlight(ctx, resp.record_directory)}'
)
@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 typer.Exit(1)
if paused:
console.err.print('Recording is paused, cannot split.')
raise typer.Exit(1)
ctx.obj['obsws'].split_record_file()
console.out.print('Recording split successfully.')
@app.command('chapter | ch')
def chapter(
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."""
active, paused = _get_recording_status(ctx)
if not active:
console.err.print('Recording is not in progress, cannot create chapter.')
raise typer.Exit(1)
if paused:
console.err.print('Recording is paused, cannot create chapter.')
raise typer.Exit(1)
ctx.obj['obsws'].create_record_chapter(chapter_name)
console.out.print(
f'Chapter {console.highlight(ctx, chapter_name or "unnamed")} created successfully.'
)

View File

@@ -29,6 +29,10 @@ def list_(
for scene in reversed(resp.scenes)
)
if not scenes:
console.out.print('No scenes found.')
raise typer.Exit()
active_scene = ctx.obj['obsws'].get_current_program_scene().scene_name
table = Table(title='Scenes', padding=(0, 2), border_style=ctx.obj['style'].border)
@@ -43,8 +47,8 @@ def list_(
(Text('Scene Name', justify='center'), 'left', ctx.obj['style'].column),
(Text('Active', justify='center'), 'center', None),
]
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for scene_name, scene_uuid in scenes:
if uuid:
@@ -108,7 +112,11 @@ def switch(
if preview:
ctx.obj['obsws'].set_current_preview_scene(scene_name)
console.out.print(f'Switched to preview scene: [green]{scene_name}[/green]')
console.out.print(
f'Switched to preview scene: {console.highlight(ctx, scene_name)}'
)
else:
ctx.obj['obsws'].set_current_program_scene(scene_name)
console.out.print(f'Switched to program scene: [green]{scene_name}[/green]')
console.out.print(
f'Switched to program scene: {console.highlight(ctx, scene_name)}'
)

View File

@@ -21,6 +21,10 @@ 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),

View File

@@ -78,8 +78,8 @@ def list_(
('Enabled', 'center', None),
]
# Add columns to the table
for column, justify, style in columns:
table.add_column(column, justify=justify, style=style)
for heading, justify, style in columns:
table.add_column(heading, justify=justify, style=style)
for item_id, item_name, is_group, is_enabled, source_uuid in items:
if is_group:

View File

@@ -16,6 +16,7 @@ class Settings(UserDict):
The settings are expected to be in uppercase and should start with 'OBS_'.
Example:
-------
settings = Settings()
host = settings['OBS_HOST']
settings['OBS_PORT'] = 4455
@@ -64,12 +65,15 @@ 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.
"""

View File

@@ -3,15 +3,15 @@
import os
from dataclasses import dataclass
_registry = {}
registry = {}
def register_style(cls):
"""Register a style class."""
key = cls.__name__.lower()
if key in _registry:
if key in registry:
raise ValueError(f'Style {key} is already registered.')
_registry[key] = cls
registry[key] = cls
return cls
@@ -19,11 +19,10 @@ def register_style(cls):
class Style:
"""Base class for styles."""
name: str = 'no_colour'
description: str = 'Style disabled'
border: str | None = None
column: str | None = None
highlight: str | None = None
name: str
border: str
column: str
highlight: str
no_border: bool = False
def __post_init__(self):
@@ -32,9 +31,16 @@ class Style:
if self.no_border:
self.border = None
def __str__(self):
"""Return a string representation of the style."""
return f'{self.name} - {self.description}'
@register_style
@dataclass
class Disabled(Style):
"""Disabled style."""
name: str = 'disabled'
border: str = 'none'
column: str = 'none'
highlight: str = 'none'
@register_style
@@ -43,10 +49,9 @@ class Red(Style):
"""Red style."""
name: str = 'red'
description: str = 'Red text color'
border: str = 'red3'
highlight: str = 'red1'
column: str = 'red1'
highlight: str = 'red1'
@register_style
@@ -55,10 +60,9 @@ class Magenta(Style):
"""Magenta style."""
name: str = 'magenta'
description: str = 'Magenta text color'
border: str = 'magenta3'
highlight: str = 'orchid1'
column: str = 'orchid1'
highlight: str = 'orchid1'
@register_style
@@ -67,10 +71,9 @@ class Purple(Style):
"""Purple style."""
name: str = 'purple'
description: str = 'Purple text color'
border: str = 'medium_purple4'
highlight: str = 'medium_purple'
column: str = 'medium_purple'
highlight: str = 'medium_purple'
@register_style
@@ -79,10 +82,9 @@ class Blue(Style):
"""Blue style."""
name: str = 'blue'
description: str = 'Blue text color'
border: str = 'cornflower_blue'
highlight: str = 'sky_blue2'
column: str = 'sky_blue2'
highlight: str = 'sky_blue2'
@register_style
@@ -91,10 +93,9 @@ class Cyan(Style):
"""Cyan style."""
name: str = 'cyan'
description: str = 'Cyan text color'
border: str = 'dark_cyan'
highlight: str = 'cyan'
column: str = 'cyan'
highlight: str = 'cyan'
@register_style
@@ -103,10 +104,9 @@ class Green(Style):
"""Green style."""
name: str = 'green'
description: str = 'Green text color'
border: str = 'green4'
highlight: str = 'spring_green3'
column: str = 'spring_green3'
highlight: str = 'spring_green3'
@register_style
@@ -115,10 +115,9 @@ class Yellow(Style):
"""Yellow style."""
name: str = 'yellow'
description: str = 'Yellow text color'
border: str = 'yellow3'
highlight: str = 'wheat1'
column: str = 'wheat1'
highlight: str = 'wheat1'
@register_style
@@ -127,10 +126,9 @@ class Orange(Style):
"""Orange style."""
name: str = 'orange'
description: str = 'Orange text color'
border: str = 'dark_orange'
highlight: str = 'orange1'
column: str = 'orange1'
highlight: str = 'orange1'
@register_style
@@ -139,10 +137,9 @@ class White(Style):
"""White style."""
name: str = 'white'
description: str = 'White text color'
border: str = 'grey82'
highlight: str = 'grey100'
column: str = 'grey100'
highlight: str = 'grey100'
@register_style
@@ -151,10 +148,9 @@ class Grey(Style):
"""Grey style."""
name: str = 'grey'
description: str = 'Grey text color'
border: str = 'grey50'
highlight: str = 'grey70'
column: str = 'grey70'
highlight: str = 'grey70'
@register_style
@@ -163,10 +159,9 @@ class Navy(Style):
"""Navy Blue style."""
name: str = 'navyblue'
description: str = 'Navy Blue text color'
border: str = 'deep_sky_blue4'
highlight: str = 'light_sky_blue3'
column: str = 'light_sky_blue3'
highlight: str = 'light_sky_blue3'
@register_style
@@ -175,17 +170,14 @@ class Black(Style):
"""Black style."""
name: str = 'black'
description: str = 'Black text color'
border: str = 'grey19'
column: str = 'grey11'
highlight: str = 'grey11'
def request_style_obj(style_name: str, no_border: bool) -> Style:
"""Entry point for style objects. Returns a Style object based on the style name."""
style_name = str(style_name).lower() # coerce the type to string and lowercase it
if style_name == 'disabled':
os.environ['NO_COLOR'] = '1'
if style_name not in _registry:
os.environ['NO_COLOR'] = '1' # Disable colour output
return Style()
return _registry[style_name](no_border=no_border)
return registry[style_name.lower()](no_border=no_border)

78
obsws_cli/text.py Normal file
View File

@@ -0,0 +1,78 @@
"""module containing commands for manipulating text inputs."""
from typing import Annotated, Optional
import typer
from . import console, validate
from .alias import SubTyperAliasGroup
app = typer.Typer(cls=SubTyperAliasGroup)
@app.callback()
def main():
"""Control text inputs in OBS."""
@app.command('current | get')
def current(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
help='Name of the text input to get.', callback=validate.input_in_inputs
),
],
):
"""Get the current text for a text input."""
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 typer.Exit(1)
current_text = resp.input_settings.get('text', '')
if not current_text:
current_text = '(empty)'
console.out.print(
f'Current text for input {console.highlight(ctx, input_name)}: {current_text}',
)
@app.command('update | set')
def update(
ctx: typer.Context,
input_name: Annotated[
str,
typer.Argument(
help='Name of the text input to update.', callback=validate.input_in_inputs
),
],
new_text: Annotated[
Optional[str],
typer.Argument(
help='The new text to set for the input.',
),
] = None,
):
"""Update the text of a text input."""
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 typer.Exit(1)
ctx.obj['obsws'].set_input_settings(
name=input_name,
settings={'text': new_text},
overlay=True,
)
if not new_text:
new_text = '(empty)'
console.out.print(
f'Text for input {console.highlight(ctx, input_name)} updated to: {new_text}',
)

View File

@@ -13,6 +13,10 @@ def check_mark(value: bool, empty_if_false: bool = False) -> str:
if empty_if_false and not value:
return ''
# rich gracefully handles the absence of colour throughout the rest of the application,
# but here we must handle it manually.
# If NO_COLOR is set, we return plain text symbols.
# Otherwise, we return coloured symbols.
if os.getenv('NO_COLOR', '') != '':
return '' if value else ''
return '' if value else ''

View File

@@ -2,14 +2,28 @@
import typer
from . import console
# 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: typer.Context, input_name: str) -> bool:
"""Check if an input is in the input list."""
inputs = ctx.obj['obsws'].get_input_list().inputs
return any(input_.get('inputName') == input_name for input_ in inputs)
"""Ensure the given input exists in the list of inputs."""
resp = ctx.obj['obsws'].get_input_list()
if not any(input.get('inputName') == input_name for input in resp.inputs):
console.err.print(f'Input [yellow]{input_name}[/yellow] does not exist.')
raise typer.Exit(1)
return input_name
def input_not_in_inputs(ctx: typer.Context, input_name: str) -> bool:
"""Ensure an input does not already exist in the list of inputs."""
resp = ctx.obj['obsws'].get_input_list()
if any(input.get('inputName') == input_name for input in resp.inputs):
console.err.print(f'Input [yellow]{input_name}[/yellow] already exists.')
raise typer.Exit(1)
return input_name
def scene_in_scenes(ctx: typer.Context, scene_name: str) -> bool:
@@ -52,3 +66,12 @@ def monitor_exists(ctx: typer.Context, monitor_index: int) -> bool:
"""Check if a monitor exists."""
resp = ctx.obj['obsws'].get_monitor_list()
return any(monitor['monitorIndex'] == monitor_index for monitor in resp.monitors)
def kind_in_input_kinds(ctx: typer.Context, input_kind: str) -> str:
"""Check if an input kind is valid."""
resp = ctx.obj['obsws'].get_input_kind_list(False)
if not any(kind == input_kind for kind in resp.input_kinds):
console.err.print(f'Input kind [yellow]{input_kind}[/yellow] not found.')
raise typer.Exit(1)
return input_kind

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.7.2", "python-dotenv>=1.1.0"]
dependencies = ["typer>=0.21.1", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
[project.urls]
@@ -42,9 +42,6 @@ dependencies = ["click-man>=0.5.1"]
cli = "obsws-cli {args:}"
man = "python man/generate.py --output=./man"
[tool.hatch.envs.lazyimports.scripts]
cli = "obsws-cli {args:}"
[tool.hatch.envs.hatch-test]
randomize = true

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',
@@ -71,6 +117,13 @@ def pytest_sessionstart(session):
},
sceneItemEnabled=True,
)
session.obsws.create_input(
sceneName='pytest_scene',
inputName='pytest_text_input',
inputKind='text_gdiplus_v3',
inputSettings={'text': 'Hello, OBS!'},
sceneItemEnabled=True,
)
resp = session.obsws.get_scene_item_list('pytest_scene')
for item in resp.scene_items:
if item['sourceName'] == 'pytest_input_2':
@@ -124,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:
@@ -142,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()

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():

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():

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():

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

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:

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

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():

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():

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

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():

18
tests/test_text.py Normal file
View File

@@ -0,0 +1,18 @@
"""Unit tests for the text command in the OBS WebSocket CLI."""
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
def test_text_update():
"""Test the text update command."""
result = runner.invoke(app, ['text', 'current', 'pytest_text_input'])
assert result.exit_code == 0
assert 'Current text for input pytest_text_input: Hello, OBS!' in result.stdout
result = runner.invoke(app, ['text', 'update', 'pytest_text_input', 'New Text'])
assert result.exit_code == 0
assert 'Text for input pytest_text_input updated to: New Text' in result.stdout

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():