mirror of
https://github.com/onyx-and-iris/obsws-cli.git
synced 2026-03-03 16:59:17 +00:00
Compare commits
9 Commits
7a73ec35f6
...
3f3b331363
| Author | SHA1 | Date | |
|---|---|---|---|
| 3f3b331363 | |||
| 2535fe85c5 | |||
| 7d4485ec05 | |||
| 2c2501e017 | |||
| 356684e5d4 | |||
| f7e51f8488 | |||
| 8da29ce90e | |||
| 72c6bcee49 | |||
| dceafba065 |
@ -5,6 +5,14 @@ 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.23.0] - 2026-01-09
|
||||
|
||||
### Added
|
||||
|
||||
- new subcommands added to input, see [Input](https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#input)
|
||||
- settings command group, see [Settings](https://github.com/onyx-and-iris/obsws-cli?tab=readme-ov-file#settings)
|
||||
|
||||
|
||||
# [0.20.0] - 2025-07-14
|
||||
|
||||
### Added
|
||||
|
||||
101
README.md
101
README.md
@ -298,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:
|
||||
|
||||
@ -315,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>
|
||||
|
||||
@ -335,6 +355,32 @@ 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.
|
||||
@ -676,6 +722,61 @@ obsws-cli projector open --monitor-index=1 "test_group"
|
||||
obsws-cli screenshot save --width=2560 --height=1440 "Scene" "C:\Users\me\Videos\screenshot.png"
|
||||
```
|
||||
|
||||
#### Settings
|
||||
|
||||
- show: Show current OBS settings.
|
||||
- flags:
|
||||
|
||||
*optional*
|
||||
- --video: Show video settings.
|
||||
- --record: Show recording settings.
|
||||
- --profile: Show profile settings.
|
||||
|
||||
```console
|
||||
obsws-cli settings show --video --record
|
||||
```
|
||||
|
||||
- profile: Get/set OBS profile settings.
|
||||
- args: <category> <name> <value>
|
||||
|
||||
```console
|
||||
obsws-cli settings profile SimpleOutput VBitrate
|
||||
|
||||
obsws-cli settings profile SimpleOutput VBitrate 6000
|
||||
```
|
||||
|
||||
- stream-service: Get/set OBS stream service settings.
|
||||
- flags:
|
||||
- --key: Stream key.
|
||||
- --server: Stream server URL.
|
||||
|
||||
*optional*
|
||||
- args: <type>
|
||||
|
||||
```console
|
||||
obsws-cli settings stream-service
|
||||
|
||||
obsws-cli settings stream-service --key='live_xyzxyzxyzxyz' rtmp_common
|
||||
```
|
||||
|
||||
- video: Get/set OBS video settings.
|
||||
- flags:
|
||||
|
||||
*optional*
|
||||
- --base-width: Base (canvas) width.
|
||||
- --base-height: Base (canvas) height.
|
||||
- --output-width: Output (scaled) width.
|
||||
- --output-height: Output (scaled) height.
|
||||
- --fps-num: Frames per second numerator.
|
||||
- --fps-den: Frames per second denominator.
|
||||
|
||||
```console
|
||||
obsws-cli settings video
|
||||
|
||||
obsws-cli settings video --base-width=1920 --base-height=1080
|
||||
```
|
||||
|
||||
|
||||
## License
|
||||
|
||||
`obsws-cli` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
# SPDX-FileCopyrightText: 2025-present onyx-and-iris <code@onyxandiris.online>
|
||||
#
|
||||
# SPDX-License-Identifier: MIT
|
||||
__version__ = '0.21.0'
|
||||
__version__ = '0.23.0'
|
||||
|
||||
@ -9,7 +9,7 @@ import typer
|
||||
|
||||
from obsws_cli.__about__ import __version__ as version
|
||||
|
||||
from . import console, settings, styles
|
||||
from . import config, console, styles
|
||||
from .alias import RootTyperAliasGroup
|
||||
|
||||
app = typer.Typer(cls=RootTyperAliasGroup)
|
||||
@ -30,6 +30,7 @@ for sub_typer in (
|
||||
'studiomode',
|
||||
'text',
|
||||
'virtualcam',
|
||||
'settings',
|
||||
):
|
||||
module = importlib.import_module(f'.{sub_typer}', package=__package__)
|
||||
app.add_typer(module.app, name=sub_typer)
|
||||
@ -72,7 +73,7 @@ def main(
|
||||
help='WebSocket host',
|
||||
show_default='localhost',
|
||||
),
|
||||
] = settings.get('host'),
|
||||
] = config.get('host'),
|
||||
port: Annotated[
|
||||
int,
|
||||
typer.Option(
|
||||
@ -82,7 +83,7 @@ def main(
|
||||
help='WebSocket port',
|
||||
show_default=4455,
|
||||
),
|
||||
] = settings.get('port'),
|
||||
] = config.get('port'),
|
||||
password: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
@ -92,7 +93,7 @@ def main(
|
||||
help='WebSocket password',
|
||||
show_default=False,
|
||||
),
|
||||
] = settings.get('password'),
|
||||
] = config.get('password'),
|
||||
timeout: Annotated[
|
||||
int,
|
||||
typer.Option(
|
||||
@ -102,7 +103,7 @@ def main(
|
||||
help='WebSocket timeout',
|
||||
show_default=5,
|
||||
),
|
||||
] = settings.get('timeout'),
|
||||
] = config.get('timeout'),
|
||||
style: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
@ -113,7 +114,7 @@ def main(
|
||||
show_default='disabled',
|
||||
callback=validate_style,
|
||||
),
|
||||
] = settings.get('style'),
|
||||
] = config.get('style'),
|
||||
no_border: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
@ -123,7 +124,7 @@ def main(
|
||||
help='Disable table border styling in the CLI output',
|
||||
show_default=False,
|
||||
),
|
||||
] = settings.get('style_no_border'),
|
||||
] = config.get('style_no_border'),
|
||||
version: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
@ -147,7 +148,7 @@ def main(
|
||||
callback=setup_logging,
|
||||
hidden=True,
|
||||
),
|
||||
] = settings.get('debug'),
|
||||
] = config.get('debug'),
|
||||
):
|
||||
"""obsws_cli is a command line interface for the OBS WebSocket API."""
|
||||
ctx.ensure_object(dict)
|
||||
|
||||
80
obsws_cli/config.py
Normal file
80
obsws_cli/config.py
Normal file
@ -0,0 +1,80 @@
|
||||
"""module for settings management for obsws-cli."""
|
||||
|
||||
from collections import UserDict
|
||||
from pathlib import Path
|
||||
|
||||
from dotenv import dotenv_values
|
||||
|
||||
ConfigValue = str | int
|
||||
|
||||
|
||||
class Config(UserDict):
|
||||
"""A class to manage config for obsws-cli.
|
||||
|
||||
This class extends UserDict to provide a dictionary-like interface for config.
|
||||
It loads config from environment variables and .env files.
|
||||
The config values are expected to be in uppercase and should start with 'OBS_'.
|
||||
|
||||
Example:
|
||||
-------
|
||||
config = Config()
|
||||
host = config['OBS_HOST']
|
||||
config['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) -> ConfigValue:
|
||||
"""Get a setting value by key."""
|
||||
key = key.upper()
|
||||
if not key.startswith(Config.PREFIX):
|
||||
key = f'{Config.PREFIX}{key}'
|
||||
return self.data[key]
|
||||
|
||||
def __setitem__(self, key: str, value: ConfigValue):
|
||||
"""Set a setting value by key."""
|
||||
key = key.upper()
|
||||
if not key.startswith(Config.PREFIX):
|
||||
key = f'{Config.PREFIX}{key}'
|
||||
self.data[key] = value
|
||||
|
||||
|
||||
_config = Config(
|
||||
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) -> ConfigValue:
|
||||
"""Get a setting value by key.
|
||||
|
||||
Args:
|
||||
----
|
||||
key (str): The key of the config to retrieve.
|
||||
|
||||
Returns:
|
||||
-------
|
||||
The value of the config.
|
||||
|
||||
Raises:
|
||||
------
|
||||
KeyError: If the key does not exist in the config.
|
||||
|
||||
"""
|
||||
return _config[key]
|
||||
@ -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,
|
||||
@ -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)}.',
|
||||
)
|
||||
|
||||
@ -1,80 +1,337 @@
|
||||
"""module for settings management for obsws-cli."""
|
||||
"""module for settings management."""
|
||||
|
||||
from collections import UserDict
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Optional
|
||||
|
||||
from dotenv import dotenv_values
|
||||
import typer
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
SettingsValue = str | int
|
||||
from . import console, util
|
||||
from .alias import SubTyperAliasGroup
|
||||
|
||||
app = typer.Typer(cls=SubTyperAliasGroup)
|
||||
|
||||
|
||||
class Settings(UserDict):
|
||||
"""A class to manage settings for obsws-cli.
|
||||
@app.callback()
|
||||
def main():
|
||||
"""Manage OBS settings."""
|
||||
|
||||
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
|
||||
@app.command('show | sh')
|
||||
def show(
|
||||
ctx: typer.Context,
|
||||
video: Annotated[
|
||||
bool, typer.Option('--video', '-v', help='Show video settings.')
|
||||
] = False,
|
||||
record: Annotated[
|
||||
bool, typer.Option('--record', '-r', help='Show recording settings.')
|
||||
] = False,
|
||||
profile: Annotated[
|
||||
bool, typer.Option('--profile', '-p', help='Show profile settings.')
|
||||
] = False,
|
||||
):
|
||||
"""Show current OBS settings."""
|
||||
if not any([video, record, profile]):
|
||||
video = True
|
||||
record = True
|
||||
profile = True
|
||||
|
||||
"""
|
||||
|
||||
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'),
|
||||
}
|
||||
resp = ctx.obj['obsws'].get_video_settings()
|
||||
video_table = Table(
|
||||
title='Video Settings',
|
||||
padding=(0, 2),
|
||||
border_style=ctx.obj['style'].border,
|
||||
)
|
||||
video_columns = (
|
||||
('Setting', 'left', ctx.obj['style'].column),
|
||||
('Value', 'left', ctx.obj['style'].column),
|
||||
)
|
||||
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]
|
||||
for header_text, justify, style in video_columns:
|
||||
video_table.add_column(
|
||||
Text(header_text, justify='center'),
|
||||
justify=justify,
|
||||
style=style,
|
||||
)
|
||||
|
||||
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
|
||||
for setting in resp.attrs():
|
||||
video_table.add_row(
|
||||
util.snakecase_to_titlecase(setting),
|
||||
str(getattr(resp, setting)),
|
||||
style='' if video_table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
|
||||
if video:
|
||||
console.out.print(video_table)
|
||||
|
||||
resp = ctx.obj['obsws'].get_record_directory()
|
||||
record_table = Table(
|
||||
title='Recording Settings',
|
||||
padding=(0, 2),
|
||||
border_style=ctx.obj['style'].border,
|
||||
)
|
||||
record_columns = (
|
||||
('Setting', 'left', ctx.obj['style'].column),
|
||||
('Value', 'left', ctx.obj['style'].column),
|
||||
)
|
||||
for header_text, justify, style in record_columns:
|
||||
record_table.add_column(
|
||||
Text(header_text, justify='center'),
|
||||
justify=justify,
|
||||
style=style,
|
||||
)
|
||||
|
||||
record_table.add_row(
|
||||
'Directory',
|
||||
resp.record_directory,
|
||||
style='' if record_table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
|
||||
if record:
|
||||
console.out.print(record_table)
|
||||
|
||||
profile_table = Table(
|
||||
title='Profile Settings',
|
||||
padding=(0, 2),
|
||||
border_style=ctx.obj['style'].border,
|
||||
)
|
||||
profile_columns = (
|
||||
('Setting', 'left', ctx.obj['style'].column),
|
||||
('Value', 'left', ctx.obj['style'].column),
|
||||
)
|
||||
for header_text, justify, style in profile_columns:
|
||||
profile_table.add_column(
|
||||
Text(header_text, justify='center'),
|
||||
justify=justify,
|
||||
style=style,
|
||||
)
|
||||
|
||||
params = [
|
||||
('Output', 'Mode', 'Output Mode'),
|
||||
('SimpleOutput', 'StreamEncoder', 'Simple Streaming Encoder'),
|
||||
('SimpleOutput', 'RecEncoder', 'Simple Recording Encoder'),
|
||||
('SimpleOutput', 'RecFormat2', 'Simple Recording Video Format'),
|
||||
('SimpleOutput', 'RecAudioEncoder', 'Simple Recording Audio Format'),
|
||||
('SimpleOutput', 'RecQuality', 'Simple Recording Quality'),
|
||||
('AdvOut', 'Encoder', 'Advanced Streaming Encoder'),
|
||||
('AdvOut', 'RecEncoder', 'Advanced Recording Encoder'),
|
||||
('AdvOut', 'RecType', 'Advanced Recording Type'),
|
||||
('AdvOut', 'RecFormat2', 'Advanced Recording Video Format'),
|
||||
('AdvOut', 'RecAudioEncoder', 'Advanced Recording Audio Format'),
|
||||
]
|
||||
|
||||
for category, name, display_name in params:
|
||||
resp = ctx.obj['obsws'].get_profile_parameter(
|
||||
category=category,
|
||||
name=name,
|
||||
)
|
||||
if resp.parameter_value is not None:
|
||||
profile_table.add_row(
|
||||
display_name,
|
||||
str(resp.parameter_value),
|
||||
style='' if profile_table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
|
||||
if profile:
|
||||
console.out.print(profile_table)
|
||||
|
||||
|
||||
_settings = Settings(
|
||||
OBS_HOST='localhost',
|
||||
OBS_PORT=4455,
|
||||
OBS_PASSWORD='',
|
||||
OBS_TIMEOUT=5,
|
||||
OBS_DEBUG=False,
|
||||
OBS_STYLE='disabled',
|
||||
OBS_STYLE_NO_BORDER=False,
|
||||
@app.command('profile | pr')
|
||||
def profile(
|
||||
ctx: typer.Context,
|
||||
category: Annotated[
|
||||
str,
|
||||
typer.Argument(
|
||||
...,
|
||||
help='Profile parameter category (e.g., SimpleOutput, AdvOut).',
|
||||
),
|
||||
],
|
||||
name: Annotated[
|
||||
str,
|
||||
typer.Argument(
|
||||
...,
|
||||
help='Profile parameter name (e.g., StreamEncoder, RecFormat2).',
|
||||
),
|
||||
],
|
||||
value: Annotated[
|
||||
Optional[str],
|
||||
typer.Argument(
|
||||
...,
|
||||
help='Value to set for the profile parameter. If omitted, the current value is retrieved.',
|
||||
),
|
||||
] = None,
|
||||
):
|
||||
"""Get/set OBS profile settings."""
|
||||
if value is None:
|
||||
resp = ctx.obj['obsws'].get_profile_parameter(
|
||||
category=category,
|
||||
name=name,
|
||||
)
|
||||
console.out.print(
|
||||
f'Parameter Value for [bold]{name}[/bold]: '
|
||||
f'[green]{resp.parameter_value}[/green]'
|
||||
)
|
||||
else:
|
||||
ctx.obj['obsws'].set_profile_parameter(
|
||||
category=category,
|
||||
name=name,
|
||||
value=value,
|
||||
)
|
||||
console.out.print(
|
||||
f'Set Parameter [bold]{name}[/bold] to [green]{value}[/green]'
|
||||
)
|
||||
|
||||
|
||||
def get(key: str) -> SettingsValue:
|
||||
"""Get a setting value by key.
|
||||
@app.command('stream-service | ss')
|
||||
def stream_service(
|
||||
ctx: typer.Context,
|
||||
type_: Annotated[
|
||||
Optional[str],
|
||||
typer.Argument(
|
||||
...,
|
||||
help='Stream service type (e.g., Twitch, YouTube). If omitted, current settings are retrieved.',
|
||||
),
|
||||
] = None,
|
||||
key: Annotated[
|
||||
Optional[str],
|
||||
typer.Option('--key', '-k', help='Stream key to set. Optional.'),
|
||||
] = None,
|
||||
server: Annotated[
|
||||
Optional[str],
|
||||
typer.Option('--server', '-s', help='Stream server to set. Optional.'),
|
||||
] = None,
|
||||
):
|
||||
"""Get/set OBS stream service settings."""
|
||||
if type_ is None:
|
||||
resp = ctx.obj['obsws'].get_stream_service_settings()
|
||||
table = Table(
|
||||
title='Stream Service Settings',
|
||||
padding=(0, 2),
|
||||
border_style=ctx.obj['style'].border,
|
||||
)
|
||||
columns = (
|
||||
('Setting', 'left', ctx.obj['style'].column),
|
||||
('Value', 'left', ctx.obj['style'].column),
|
||||
)
|
||||
for header_text, justify, style in columns:
|
||||
table.add_column(
|
||||
Text(header_text, justify='center'),
|
||||
justify=justify,
|
||||
style=style,
|
||||
)
|
||||
table.add_row(
|
||||
'Type',
|
||||
resp.stream_service_type,
|
||||
style='' if table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
table.add_row(
|
||||
'Server',
|
||||
resp.stream_service_settings.get('server', ''),
|
||||
style='' if table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
table.add_row(
|
||||
'Key',
|
||||
resp.stream_service_settings.get('key', ''),
|
||||
style='' if table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
console.out.print(table)
|
||||
else:
|
||||
current_settings = ctx.obj['obsws'].get_stream_service_settings()
|
||||
if key is None:
|
||||
key = current_settings.stream_service_settings.get('key', '')
|
||||
if server is None:
|
||||
server = current_settings.stream_service_settings.get('server', '')
|
||||
|
||||
Args:
|
||||
----
|
||||
key (str): The key of the setting to retrieve.
|
||||
ctx.obj['obsws'].set_stream_service_settings(
|
||||
ss_type=type_,
|
||||
ss_settings={'key': key, 'server': server},
|
||||
)
|
||||
console.out.print('Stream service settings updated.')
|
||||
|
||||
Returns:
|
||||
-------
|
||||
The value of the setting.
|
||||
|
||||
Raises:
|
||||
------
|
||||
KeyError: If the key does not exist in the settings.
|
||||
@app.command('video | vi')
|
||||
def video(
|
||||
ctx: typer.Context,
|
||||
base_width: Annotated[
|
||||
Optional[int],
|
||||
typer.Option('--base-width', '-bw', help='Set base (canvas) width.'),
|
||||
] = None,
|
||||
base_height: Annotated[
|
||||
Optional[int],
|
||||
typer.Option('--base-height', '-bh', help='Set base (canvas) height.'),
|
||||
] = None,
|
||||
output_width: Annotated[
|
||||
Optional[int],
|
||||
typer.Option('--output-width', '-ow', help='Set output (scaled) width.'),
|
||||
] = None,
|
||||
output_height: Annotated[
|
||||
Optional[int],
|
||||
typer.Option('--output-height', '-oh', help='Set output (scaled) height.'),
|
||||
] = None,
|
||||
fps_num: Annotated[
|
||||
Optional[int],
|
||||
typer.Option('--fps-num', '-fn', help='Set FPS numerator.'),
|
||||
] = None,
|
||||
fps_den: Annotated[
|
||||
Optional[int],
|
||||
typer.Option('--fps-den', '-fd', help='Set FPS denominator.'),
|
||||
] = None,
|
||||
):
|
||||
"""Get/set OBS video settings."""
|
||||
if not any(
|
||||
[
|
||||
base_width,
|
||||
base_height,
|
||||
output_width,
|
||||
output_height,
|
||||
fps_num,
|
||||
fps_den,
|
||||
]
|
||||
):
|
||||
resp = ctx.obj['obsws'].get_video_settings()
|
||||
table = Table(
|
||||
title='Video Settings',
|
||||
padding=(0, 2),
|
||||
border_style=ctx.obj['style'].border,
|
||||
)
|
||||
columns = (
|
||||
('Setting', 'left', ctx.obj['style'].column),
|
||||
('Value', 'left', ctx.obj['style'].column),
|
||||
)
|
||||
for header_text, justify, style in columns:
|
||||
table.add_column(
|
||||
Text(header_text, justify='center'),
|
||||
justify=justify,
|
||||
style=style,
|
||||
)
|
||||
for setting in resp.attrs():
|
||||
table.add_row(
|
||||
util.snakecase_to_titlecase(setting),
|
||||
str(getattr(resp, setting)),
|
||||
style='' if table.row_count % 2 == 0 else 'dim',
|
||||
)
|
||||
console.out.print(table)
|
||||
else:
|
||||
current_settings = ctx.obj['obsws'].get_video_settings()
|
||||
if base_width is None:
|
||||
base_width = current_settings.base_width
|
||||
if base_height is None:
|
||||
base_height = current_settings.base_height
|
||||
if output_width is None:
|
||||
output_width = current_settings.output_width
|
||||
if output_height is None:
|
||||
output_height = current_settings.output_height
|
||||
if fps_num is None:
|
||||
fps_num = current_settings.fps_num
|
||||
if fps_den is None:
|
||||
fps_den = current_settings.fps_den
|
||||
|
||||
"""
|
||||
return _settings[key]
|
||||
ctx.obj['obsws'].set_video_settings(
|
||||
base_width=base_width,
|
||||
base_height=base_height,
|
||||
out_width=output_width,
|
||||
out_height=output_height,
|
||||
numerator=fps_num,
|
||||
denominator=fps_den,
|
||||
)
|
||||
console.out.print('Video settings updated.')
|
||||
|
||||
@ -18,13 +18,14 @@ def main():
|
||||
@app.command('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,
|
||||
typer.Argument(
|
||||
help='Name of the text input to get.', callback=validate.input_in_inputs
|
||||
),
|
||||
],
|
||||
):
|
||||
"""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(
|
||||
@ -44,7 +45,10 @@ def current(
|
||||
def update(
|
||||
ctx: typer.Context,
|
||||
input_name: Annotated[
|
||||
str, typer.Argument(help='Name of the text input to update.')
|
||||
str,
|
||||
typer.Argument(
|
||||
help='Name of the text input to update.', callback=validate.input_in_inputs
|
||||
),
|
||||
],
|
||||
new_text: Annotated[
|
||||
Optional[str],
|
||||
@ -54,10 +58,6 @@ def update(
|
||||
] = None,
|
||||
):
|
||||
"""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(
|
||||
|
||||
@ -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
|
||||
|
||||
@ -21,7 +21,7 @@ classifiers = [
|
||||
"Programming Language :: Python :: Implementation :: CPython",
|
||||
"Programming Language :: Python :: Implementation :: PyPy",
|
||||
]
|
||||
dependencies = ["typer>=0.19.2", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
|
||||
dependencies = ["typer>=0.21.1", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
|
||||
|
||||
|
||||
[project.urls]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user