Compare commits

..

6 Commits

Author SHA1 Message Date
8465944f30 add CHANGELOG + link in readme 2025-04-26 16:50:46 +01:00
8dad6f8bd1 patch bump 2025-04-26 16:17:40 +01:00
e806c6345d upd shorthand aliases 2025-04-26 16:17:28 +01:00
467c14f570 remove errors module 2025-04-26 15:55:21 +01:00
7915547fca add stream, studiomode tests 2025-04-26 15:53:29 +01:00
5ca61d04c7 load stream subtyper
upd env var prefix
2025-04-26 15:53:05 +01:00
20 changed files with 154 additions and 67 deletions

1
.gitignore vendored
View File

@ -131,6 +131,7 @@ ENV/
env.bak/
venv.bak/
.hatch
.test.env
# Spyder project settings
.spyderproject

12
CHANGELOG.md Normal file
View File

@ -0,0 +1,12 @@
# Changelog
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.9.2] - 2025-04-26
### Added
- Initial release.

View File

@ -3,6 +3,11 @@
[![Hatch project](https://img.shields.io/badge/%F0%9F%A5%9A-Hatch-4051b5.svg)](https://github.com/pypa/hatch)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
A command line interface for OBS Websocket v5
For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
-----
## Table of Contents
@ -50,9 +55,9 @@ Store and load environment variables from:
- `user home directory / .config / obsws-cli / obsws.env`
```env
OBSWS_HOST=localhost
OBSWS_PORT=4455
OBSWS_PASSWORD=<websocket password>
OBS_HOST=localhost
OBS_PORT=4455
OBS_PASSWORD=<websocket password>
```
Flags can be used to override environment variables.

View File

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

View File

@ -17,6 +17,7 @@ from . import (
scene,
scenecollection,
sceneitem,
stream,
studiomode,
virtualcam,
)
@ -32,7 +33,7 @@ class Settings(BaseSettings):
Path.home() / '.config' / 'obsws-cli' / 'obsws.env',
),
env_file_encoding='utf-8',
env_prefix='OBSWS_',
env_prefix='OBS_',
)
HOST: str = 'localhost'
@ -51,6 +52,7 @@ for module in (
scene,
scenecollection,
sceneitem,
stream,
studiomode,
virtualcam,
):

View File

@ -1,18 +0,0 @@
"""Exceptions for obsws_cli."""
class ObswsCliError(Exception):
"""Base class for all exceptions raised by obsws_cli."""
def __init__(self, message: str):
"""Initialize the exception with a message."""
message = (
message.split('With message: ')[1]
if 'With message: ' in message
else message
)
super().__init__(message)
class ObswsCliBadParameter(ObswsCliError):
"""Exception raised when a bad parameter is passed to a command."""

View File

@ -44,7 +44,7 @@ def _get_group(group_name: str, resp: DataclassProtocol) -> dict | None:
return group
@app.command()
@app.command('show | sh')
def show(ctx: typer.Context, scene_name: str, group_name: str):
"""Show a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
@ -71,7 +71,7 @@ def show(ctx: typer.Context, scene_name: str, group_name: str):
typer.echo(f"Group '{group_name}' is now visible.")
@app.command()
@app.command('hide | h')
def hide(ctx: typer.Context, scene_name: str, group_name: str):
"""Hide a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):
@ -129,7 +129,7 @@ def toggle(ctx: typer.Context, scene_name: str, group_name: str):
typer.echo(f"Group '{group_name}' is now hidden.")
@app.command()
@app.command('status | ss')
def status(ctx: typer.Context, scene_name: str, group_name: str):
"""Get the status of a group in a scene."""
if not validate.scene_in_scenes(ctx, scene_name):

View File

@ -42,7 +42,7 @@ def list(
typer.echo('\n'.join(input_.get('inputName') for input_ in inputs))
@app.command()
@app.command('mute | m')
def mute(ctx: typer.Context, input_name: str):
"""Mute an input."""
if not validate.input_in_inputs(ctx, input_name):
@ -60,7 +60,7 @@ def mute(ctx: typer.Context, input_name: str):
typer.echo(f"Input '{input_name}' muted.")
@app.command()
@app.command('unmute | um')
def unmute(ctx: typer.Context, input_name: str):
"""Unmute an input."""
if not validate.input_in_inputs(ctx, input_name):

View File

@ -21,14 +21,14 @@ def list(ctx: typer.Context):
typer.echo(profile)
@app.command()
@app.command('current | get')
def current(ctx: typer.Context):
"""Get the current profile."""
resp = ctx.obj.get_profile_list()
typer.echo(resp.current_profile_name)
@app.command()
@app.command('switch | set')
def switch(ctx: typer.Context, profile_name: str):
"""Switch to a profile."""
if not validate.profile_exists(ctx, profile_name):
@ -50,7 +50,7 @@ def switch(ctx: typer.Context, profile_name: str):
typer.echo(f"Switched to profile '{profile_name}'.")
@app.command()
@app.command('create | new')
def create(ctx: typer.Context, profile_name: str):
"""Create a new profile."""
if validate.profile_exists(ctx, profile_name):
@ -64,7 +64,7 @@ def create(ctx: typer.Context, profile_name: str):
typer.echo(f"Created profile '{profile_name}'.")
@app.command()
@app.command('remove | rm')
def remove(ctx: typer.Context, profile_name: str):
"""Remove a profile."""
if not validate.profile_exists(ctx, profile_name):

View File

@ -18,7 +18,7 @@ def _get_recording_status(ctx: typer.Context) -> tuple:
return resp.output_active, resp.output_paused
@app.command()
@app.command('start | s')
def start(ctx: typer.Context):
"""Start recording."""
active, paused = _get_recording_status(ctx)
@ -34,7 +34,7 @@ def start(ctx: typer.Context):
typer.echo('Recording started successfully.')
@app.command()
@app.command('stop | st')
def stop(ctx: typer.Context):
"""Stop recording."""
active, _ = _get_recording_status(ctx)
@ -46,7 +46,7 @@ def stop(ctx: typer.Context):
typer.echo('Recording stopped successfully.')
@app.command()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get recording status."""
active, paused = _get_recording_status(ctx)
@ -69,7 +69,7 @@ def toggle(ctx: typer.Context):
ctx.invoke(start, ctx=ctx)
@app.command()
@app.command('resume | r')
def resume(ctx: typer.Context):
"""Resume recording."""
active, paused = _get_recording_status(ctx)
@ -84,7 +84,7 @@ def resume(ctx: typer.Context):
typer.echo('Recording resumed successfully.')
@app.command()
@app.command('pause | p')
def pause(ctx: typer.Context):
"""Pause recording."""
active, paused = _get_recording_status(ctx)

View File

@ -12,21 +12,21 @@ def main():
"""Control profiles in OBS."""
@app.command()
@app.command('start | s')
def start(ctx: typer.Context):
"""Start the replay buffer."""
ctx.obj.start_replay_buffer()
typer.echo('Replay buffer started.')
@app.command()
@app.command('stop | st')
def stop(ctx: typer.Context):
"""Stop the replay buffer."""
ctx.obj.stop_replay_buffer()
typer.echo('Replay buffer stopped.')
@app.command()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get the status of the replay buffer."""
resp = ctx.obj.get_replay_buffer_status()
@ -36,7 +36,7 @@ def status(ctx: typer.Context):
typer.echo('Replay buffer is not active.')
@app.command()
@app.command('save | sv')
def save(ctx: typer.Context):
"""Save the replay buffer."""
ctx.obj.save_replay_buffer()

View File

@ -77,7 +77,7 @@ def _get_scene_name_and_item_id(
@_validate_scene_name_and_item_name
@app.command()
@app.command('show | sh')
def show(
ctx: typer.Context,
scene_name: str,
@ -99,7 +99,7 @@ def show(
@_validate_scene_name_and_item_name
@app.command()
@app.command('hide | h')
def hide(
ctx: typer.Context,
scene_name: str,
@ -164,7 +164,7 @@ def toggle(
@_validate_scene_name_and_item_name
@app.command()
@app.command('visible | v')
def visible(
ctx: typer.Context,
scene_name: str,
@ -206,7 +206,7 @@ def visible(
@_validate_scene_name_and_item_name
@app.command()
@app.command('transform | t')
def transform(
ctx: typer.Context,
scene_name: str,

View File

@ -18,7 +18,7 @@ def _get_streaming_status(ctx: typer.Context) -> tuple:
return resp.output_active, resp.output_duration
@app.command()
@app.command('start | s')
def start(ctx: typer.Context):
"""Start streaming."""
active, _ = _get_streaming_status(ctx)
@ -30,7 +30,7 @@ def start(ctx: typer.Context):
typer.echo('Streaming started successfully.')
@app.command()
@app.command('stop | st')
def stop(ctx: typer.Context):
"""Stop streaming."""
active, _ = _get_streaming_status(ctx)
@ -42,7 +42,7 @@ def stop(ctx: typer.Context):
typer.echo('Streaming stopped successfully.')
@app.command()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get streaming status."""
active, duration = _get_streaming_status(ctx)

View File

@ -12,19 +12,19 @@ def main():
"""Control studio mode in OBS."""
@app.command()
@app.command('enable | on')
def enable(ctx: typer.Context):
"""Enable studio mode."""
ctx.obj.set_studio_mode_enabled(True)
@app.command()
@app.command('disable | off')
def disable(ctx: typer.Context):
"""Disable studio mode."""
ctx.obj.set_studio_mode_enabled(False)
@app.command()
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle studio mode."""
resp = ctx.obj.get_studio_mode_enabled()
@ -36,7 +36,7 @@ def toggle(ctx: typer.Context):
typer.echo('Studio mode is now enabled.')
@app.command()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get the status of studio mode."""
resp = ctx.obj.get_studio_mode_enabled()

View File

@ -12,28 +12,28 @@ def main():
"""Control virtual camera in OBS."""
@app.command()
@app.command('start | s')
def start(ctx: typer.Context):
"""Start the virtual camera."""
ctx.obj.start_virtual_cam()
typer.echo('Virtual camera started.')
@app.command()
@app.command('stop | p')
def stop(ctx: typer.Context):
"""Stop the virtual camera."""
ctx.obj.stop_virtual_cam()
typer.echo('Virtual camera stopped.')
@app.command()
@app.command('toggle | tg')
def toggle(ctx: typer.Context):
"""Toggle the virtual camera."""
ctx.obj.toggle_virtual_cam()
typer.echo('Virtual camera toggled.')
@app.command()
@app.command('status | ss')
def status(ctx: typer.Context):
"""Get the status of the virtual camera."""
resp = ctx.obj.get_virtual_cam_status()

View File

@ -40,10 +40,9 @@ path = "obsws_cli/__about__.py"
[tool.hatch.envs.default.scripts]
cli = "obsws-cli {args:}"
test = "pytest {args:obsws_cli tests}"
[tool.hatch.envs.hatch-test]
dependencies = ["pytest>=8.3.5", "pytest-dotenv"]
dependencies = ["pytest>=8.3.5"]
[tool.hatch.envs.types]
extra-dependencies = ["mypy>=1.0.0"]

View File

@ -3,6 +3,7 @@
import os
import obsws_python as obsws
from dotenv import find_dotenv, load_dotenv
def pytest_configure(config):
@ -19,9 +20,9 @@ def pytest_sessionstart(session):
"""
# Initialize the OBS WebSocket client
session.obsws = obsws.ReqClient(
host=os.environ['OBSWS_HOST'],
port=os.environ['OBSWS_PORT'],
password=os.environ['OBSWS_PASSWORD'],
host=os.environ['OBS_HOST'],
port=os.environ['OBS_PORT'],
password=os.environ['OBS_PASSWORD'],
timeout=5,
)
resp = session.obsws.get_version()
@ -32,6 +33,17 @@ def pytest_sessionstart(session):
)
print(' '.join(out))
load_dotenv(find_dotenv('.test.env'))
session.obsws.set_stream_service_settings(
'rtmp_common',
{
'service': 'Twitch',
'server': 'auto',
'key': os.environ['OBS_STREAM_KEY'],
},
)
session.obsws.set_current_scene_collection('test-collection')
session.obsws.create_scene('pytest')
@ -68,6 +80,10 @@ def pytest_sessionfinish(session, exitstatus):
"""
session.obsws.remove_scene('pytest')
resp = session.obsws.get_stream_status()
if resp.output_active:
session.obsws.stop_stream()
# Close the OBS WebSocket client connection
session.obsws.disconnect()

View File

@ -9,35 +9,35 @@ runner = CliRunner()
def test_group_list():
"""Test the group list command."""
result = runner.invoke(app, ['group', 'list', 'pytest00'])
result = runner.invoke(app, ['group', 'list', 'Scene'])
assert result.exit_code == 0
assert 'test_group' in result.stdout
def test_group_show():
"""Test the group show command."""
result = runner.invoke(app, ['group', 'show', 'pytest00', 'test_group'])
result = runner.invoke(app, ['group', 'show', 'Scene', 'test_group'])
assert result.exit_code == 0
assert "Group 'test_group' is now visible." in result.stdout
def test_group_toggle():
"""Test the group toggle command."""
result = runner.invoke(app, ['group', 'hide', 'pytest00', 'test_group'])
result = runner.invoke(app, ['group', 'hide', 'Scene', 'test_group'])
assert result.exit_code == 0
assert "Group 'test_group' is now hidden." in result.stdout
result = runner.invoke(app, ['group', 'toggle', 'pytest00', 'test_group'])
result = runner.invoke(app, ['group', 'toggle', 'Scene', 'test_group'])
assert result.exit_code == 0
assert "Group 'test_group' is now visible." in result.stdout
def test_group_status():
"""Test the group status command."""
result = runner.invoke(app, ['group', 'show', 'pytest00', 'test_group'])
result = runner.invoke(app, ['group', 'show', 'Scene', 'test_group'])
assert result.exit_code == 0
assert "Group 'test_group' is now visible." in result.stdout
result = runner.invoke(app, ['group', 'status', 'pytest00', 'test_group'])
result = runner.invoke(app, ['group', 'status', 'Scene', 'test_group'])
assert result.exit_code == 0
assert "Group 'test_group' is now visible." in result.stdout

43
tests/test_stream.py Normal file
View File

@ -0,0 +1,43 @@
"""Unit tests for the stream commands in the OBS WebSocket CLI."""
import time
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
def test_stream_start():
"""Test the stream start command."""
result = runner.invoke(app, ['stream', 'status'])
assert result.exit_code == 0
active = 'Streaming is in progress' in result.stdout
result = runner.invoke(app, ['stream', 'start'])
assert result.exit_code == 0
time.sleep(1) # Wait for the stream to start
if active:
assert 'Streaming is already in progress, cannot start.' in result.stdout
else:
assert 'Streaming started successfully.' in result.stdout
def test_stream_stop():
"""Test the stream stop command."""
result = runner.invoke(app, ['stream', 'status'])
assert result.exit_code == 0
active = 'Streaming is in progress' in result.stdout
result = runner.invoke(app, ['stream', 'stop'])
assert result.exit_code == 0
time.sleep(1) # Wait for the stream to stop
if active:
assert 'Streaming stopped successfully.' in result.stdout
else:
assert 'Streaming is not in progress, cannot stop.' in result.stdout

27
tests/test_studiomode.py Normal file
View File

@ -0,0 +1,27 @@
"""Unit tests for the studio mode command in the OBS WebSocket CLI."""
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
def test_studio_enable():
"""Test the studio enable command."""
result = runner.invoke(app, ['studiomode', 'enable'])
assert result.exit_code == 0
result = runner.invoke(app, ['studiomode', 'status'])
assert result.exit_code == 0
assert 'Studio mode is enabled.' in result.stdout
def test_studio_disable():
"""Test the studio disable command."""
result = runner.invoke(app, ['studiomode', 'disable'])
assert result.exit_code == 0
result = runner.invoke(app, ['studiomode', 'status'])
assert result.exit_code == 0
assert 'Studio mode is disabled.' in result.stdout