tessl/pypi-obsws-python

A Python SDK for OBS Studio WebSocket v5.0

—

Pending

Overview

Eval results

Files

Event Client

Name: tessl/pypi-obsws-python
Author: tessl

The EventClient provides an asynchronous interface for receiving and handling real-time events from OBS Studio. It uses a callback-based system where you register handler functions that are triggered when specific events occur.

Core Event System

Event Client Initialization

class EventClient:
    def __init__(self, host='localhost', port=4455, password='', subs=Subs.LOW_VOLUME, timeout=None):
        """
        Initialize event client.
        
        Parameters:
        - host (str): OBS WebSocket host address
        - port (int): OBS WebSocket port number
        - password (str): OBS WebSocket password  
        - subs (int): Event subscription flags (Subs enum values)
        - timeout (float, optional): Connection timeout in seconds
        """
    
    def __enter__(self):
        """Context manager entry."""
        
    def __exit__(self, exc_type, exc_value, exc_traceback):
        """Context manager exit with automatic disconnect."""
        
    def disconnect(self):
        """Stop listening for events and close connection."""
        
    unsubscribe = disconnect  # Alias for disconnect

Callback Management

The EventClient provides a callback property that manages event handler registration:

class Callback:
    def register(self, fns):
        """
        Register callback function(s) for events.
        
        Parameters:
        - fns (callable or list): Single function or list of functions to register
        """
    
    def deregister(self, fns):
        """
        Deregister callback function(s).
        
        Parameters:
        - fns (callable or list): Single function or list of functions to deregister
        """
    
    def get(self):
        """
        Get list of registered event names.
        
        Returns:
        List of event names (CamelCase) that have registered handlers
        """
    
    def clear(self):
        """Clear all registered callback functions."""

Event Handler Conventions

Event handler functions must follow specific naming conventions:

Use snake_case naming with "on_" prefix
Function name must match the OBS event name converted to snake_case
Handler receives a single data parameter containing event information

Event Name Mapping

OBS Event Name → Handler Function Name:

SceneCreated → on_scene_created
InputMuteStateChanged → on_input_mute_state_changed
CurrentProgramSceneChanged → on_current_program_scene_changed
StreamStateChanged → on_stream_state_changed

Event Data Access

Event data is provided as dataclass objects with snake_case attributes:

def on_scene_created(data):
    # Access event attributes
    print(f"Scene created: {data.scene_name}")
    print(f"Is group: {data.is_group}")
    
    # Inspect all available attributes
    print(f"Available attributes: {data.attrs()}")

Common Event Categories

Scene Events

def on_scene_created(data):
    """
    Triggered when a scene is created.
    
    Event data attributes:
    - scene_name (str): Name of created scene
    - is_group (bool): Whether scene is a group
    """

def on_scene_removed(data):
    """
    Triggered when a scene is removed.
    
    Event data attributes:
    - scene_name (str): Name of removed scene
    - is_group (bool): Whether scene was a group
    """

def on_scene_name_changed(data):
    """
    Triggered when a scene is renamed.
    
    Event data attributes:
    - old_scene_name (str): Previous scene name
    - scene_name (str): New scene name
    """

def on_current_program_scene_changed(data):
    """
    Triggered when program scene changes.
    
    Event data attributes:
    - scene_name (str): New program scene name
    """

def on_current_preview_scene_changed(data):
    """
    Triggered when preview scene changes.
    
    Event data attributes:
    - scene_name (str): New preview scene name
    """

def on_scene_list_changed(data):
    """
    Triggered when scene list changes.
    
    Event data attributes:
    - scenes (list): Updated scene list
    """

Input Events

def on_input_created(data):
    """
    Triggered when an input is created.
    
    Event data attributes:
    - input_name (str): Name of created input
    - input_kind (str): Type/kind of input
    - unversioned_input_kind (str): Unversioned input kind
    - input_settings (dict): Input settings
    - default_input_settings (dict): Default settings
    """

def on_input_removed(data):
    """
    Triggered when an input is removed.
    
    Event data attributes:
    - input_name (str): Name of removed input
    """

def on_input_name_changed(data):
    """
    Triggered when input is renamed.
    
    Event data attributes:
    - old_input_name (str): Previous input name
    - input_name (str): New input name
    """

def on_input_mute_state_changed(data):
    """
    Triggered when input mute state changes.
    
    Event data attributes:
    - input_name (str): Input name
    - input_muted (bool): New mute state
    """

def on_input_volume_changed(data):
    """
    Triggered when input volume changes.
    
    Event data attributes:
    - input_name (str): Input name
    - input_volume_mul (float): Volume multiplier
    - input_volume_db (float): Volume in dB
    """

def on_input_settings_changed(data):
    """
    Triggered when input settings change.
    
    Event data attributes:
    - input_name (str): Input name
    - input_settings (dict): New input settings
    """

Stream and Record Events

def on_stream_state_changed(data):
    """
    Triggered when streaming state changes.
    
    Event data attributes:
    - output_active (bool): Stream active state
    - output_state (str): Stream state name
    """

def on_record_state_changed(data):
    """
    Triggered when recording state changes.
    
    Event data attributes:
    - output_active (bool): Recording active state
    - output_state (str): Recording state name
    - output_path (str, optional): Recording file path
    """

def on_record_file_changed(data):
    """
    Triggered when recording file changes.
    
    Event data attributes:
    - new_output_path (str): New recording file path
    """

def on_replay_buffer_state_changed(data):
    """
    Triggered when replay buffer state changes.
    
    Event data attributes:
    - output_active (bool): Replay buffer active state
    - output_state (str): Replay buffer state name
    """

def on_replay_buffer_saved(data):
    """
    Triggered when replay buffer is saved.
    
    Event data attributes:
    - saved_replay_path (str): Path to saved replay file
    """

Scene Item Events

def on_scene_item_created(data):
    """
    Triggered when scene item is created.
    
    Event data attributes:
    - scene_name (str): Scene name
    - source_name (str): Source name
    - scene_item_id (int): Scene item ID
    - scene_item_index (int): Scene item index
    """

def on_scene_item_removed(data):
    """
    Triggered when scene item is removed.
    
    Event data attributes:
    - scene_name (str): Scene name
    - source_name (str): Source name
    - scene_item_id (int): Scene item ID
    """

def on_scene_item_enable_state_changed(data):
    """
    Triggered when scene item enable state changes.
    
    Event data attributes:
    - scene_name (str): Scene name
    - scene_item_id (int): Scene item ID
    - scene_item_enabled (bool): New enable state
    """

def on_scene_item_lock_state_changed(data):
    """
    Triggered when scene item lock state changes.
    
    Event data attributes:
    - scene_name (str): Scene name
    - scene_item_id (int): Scene item ID
    - scene_item_locked (bool): New lock state
    """

def on_scene_item_transform_changed(data):
    """
    Triggered when scene item transform changes (HIGH_VOLUME event).
    
    Event data attributes:
    - scene_name (str): Scene name
    - scene_item_id (int): Scene item ID
    - scene_item_transform (dict): New transform data
    """

Filter Events

def on_source_filter_created(data):
    """
    Triggered when source filter is created.
    
    Event data attributes:
    - source_name (str): Source name
    - filter_name (str): Filter name
    - filter_kind (str): Filter type
    - filter_index (int): Filter index
    - filter_settings (dict): Filter settings
    - default_filter_settings (dict): Default settings
    """

def on_source_filter_removed(data):
    """
    Triggered when source filter is removed.
    
    Event data attributes:
    - source_name (str): Source name
    - filter_name (str): Filter name
    """

def on_source_filter_enable_state_changed(data):
    """
    Triggered when filter enable state changes.
    
    Event data attributes:
    - source_name (str): Source name
    - filter_name (str): Filter name
    - filter_enabled (bool): New enable state
    """

Usage Examples

Basic Event Listening

import obsws_python as obs

# Create event client with low-volume events
client = obs.EventClient(subs=obs.Subs.LOW_VOLUME)

# Define event handlers
def on_scene_created(data):
    print(f"New scene created: {data.scene_name}")

def on_input_mute_state_changed(data):
    print(f"Input '{data.input_name}' mute changed to: {data.input_muted}")

def on_stream_state_changed(data):
    if data.output_active:
        print("Stream started!")
    else:
        print("Stream stopped!")

# Register callbacks
client.callback.register([
    on_scene_created,
    on_input_mute_state_changed,
    on_stream_state_changed
])

# Keep client running
try:
    input("Press Enter to stop listening...\n")
finally:
    client.disconnect()

Event Logging System

import obsws_python as obs
import logging
from datetime import datetime

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def log_event(event_name):
    """Decorator to log event occurrences."""
    def decorator(func):
        def wrapper(data):
            timestamp = datetime.now().strftime("%H:%M:%S")
            logger.info(f"[{timestamp}] {event_name}: {data.attrs()}")
            return func(data)
        return wrapper
    return decorator

# Event handlers with logging
@log_event("Scene Changed")
def on_current_program_scene_changed(data):
    print(f"Switched to scene: {data.scene_name}")

@log_event("Input Created")
def on_input_created(data):
    print(f"New input: {data.input_name} ({data.input_kind})")

@log_event("Recording State")
def on_record_state_changed(data):
    state = "started" if data.output_active else "stopped"
    print(f"Recording {state}")
    if hasattr(data, 'output_path') and data.output_path:
        print(f"Recording to: {data.output_path}")

# Initialize client and register handlers
with obs.EventClient(subs=obs.Subs.LOW_VOLUME) as client:
    client.callback.register([
        on_current_program_scene_changed,
        on_input_created,
        on_record_state_changed
    ])
    
    print("Event logging started. Registered events:")
    for event in client.callback.get():
        print(f"  - {event}")
    
    input("Press Enter to stop...\n")

High-Volume Event Monitoring

import obsws_python as obs
import time

# Monitor high-frequency events
client = obs.EventClient(subs=obs.Subs.HIGH_VOLUME)

volume_updates = 0
transform_updates = 0

def on_input_volume_meters(data):
    """Handle volume meter updates (very frequent)."""
    global volume_updates
    volume_updates += 1
    
    # Log every 100 updates to avoid spam
    if volume_updates % 100 == 0:
        print(f"Volume meter updates: {volume_updates}")

def on_scene_item_transform_changed(data):
    """Handle scene item transform changes (frequent during dragging)."""
    global transform_updates
    transform_updates += 1
    
    if transform_updates % 10 == 0:
        print(f"Transform updates: {transform_updates} (Item {data.scene_item_id})")

def on_input_active_state_changed(data):
    """Handle input active state changes."""
    state = "active" if data.input_active else "inactive"
    print(f"Input '{data.input_name}' is now {state}")

client.callback.register([
    on_input_volume_meters,
    on_scene_item_transform_changed,
    on_input_active_state_changed
])

print("Monitoring high-volume events...")
print("Warning: This will generate many events!")

try:
    # Run for 30 seconds
    time.sleep(30)
finally:
    client.disconnect()
    print(f"Final counts - Volume: {volume_updates}, Transform: {transform_updates}")

Dynamic Event Registration

import obsws_python as obs

class OBSEventManager:
    def __init__(self):
        self.client = obs.EventClient(subs=obs.Subs.LOW_VOLUME)
        self.handlers = {}
    
    def add_scene_monitoring(self):
        """Add scene-related event handlers."""
        def on_scene_created(data):
            print(f"📄 Scene created: {data.scene_name}")
            
        def on_current_program_scene_changed(data):
            print(f"🎬 Now showing: {data.scene_name}")
        
        self.handlers['scenes'] = [on_scene_created, on_current_program_scene_changed]
        self.client.callback.register(self.handlers['scenes'])
        print("Scene monitoring enabled")
    
    def add_input_monitoring(self):
        """Add input-related event handlers."""
        def on_input_mute_state_changed(data):
            emoji = "🔇" if data.input_muted else "🔊"
            print(f"{emoji} {data.input_name} mute: {data.input_muted}")
            
        def on_input_volume_changed(data):
            print(f"🎚️ {data.input_name} volume: {data.input_volume_db:.1f}dB")
        
        self.handlers['inputs'] = [on_input_mute_state_changed, on_input_volume_changed]
        self.client.callback.register(self.handlers['inputs'])
        print("Input monitoring enabled")
    
    def remove_monitoring(self, category):
        """Remove specific category of event handlers."""
        if category in self.handlers:
            self.client.callback.deregister(self.handlers[category])
            del self.handlers[category]
            print(f"{category} monitoring disabled")
    
    def list_active_events(self):
        """Show currently registered events."""
        events = self.client.callback.get()
        if events:
            print("Active events:")
            for event in events:
                print(f"  - {event}")
        else:
            print("No events registered")
    
    def shutdown(self):
        """Clean shutdown."""
        self.client.disconnect()

# Usage
manager = OBSEventManager()

try:
    # Start with scene monitoring
    manager.add_scene_monitoring()
    manager.list_active_events()
    
    input("Press Enter to add input monitoring...\n")
    manager.add_input_monitoring()
    manager.list_active_events()
    
    input("Press Enter to remove scene monitoring...\n") 
    manager.remove_monitoring('scenes')
    manager.list_active_events()
    
    input("Press Enter to exit...\n")
    
finally:
    manager.shutdown()

Install with Tessl CLI