Skip to content

StarStreamer Documentation

Welcome to StarStreamer - a Python-based streaming automation framework that replaces visual scripting with pure Python code.

New in v0.6.1 ✨ NiceGUI Web Interface - Complete management interface with settings, modules, events, and real-time monitoring!

Why StarStreamer?

StarStreamer is built for developers who want to automate their streams with code, not GUI configurations.

Key Features

  • 🐍 Pure Python - Write actions as Python functions
  • ⚡ Event-Driven - React to platform events via plugin integrations
  • 🔌 Platform Plugins - Twitch, OBS, Discord integrations with consistent APIs
  • ⏰ Active Triggers - Schedule tasks and monitor conditions automatically
  • 🖥️ Web Interface - Modern NiceGUI-based management interface with real-time monitoring
  • 🔄 Hot Reload - Update code without restarting
  • 💉 Dependency Injection - Clean, testable code architecture
  • 🎯 Type Safe - Full type hints and IDE support
  • 🧪 Test Driven - Built with TDD principles

Quick Examples

Chat Commands (Event-Based)

from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger, CooldownTrigger
from starstreamer.plugins.twitch import TwitchClient

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hello"))
@trigger(CooldownTrigger(30, per_user=True))
async def hello_command(event, twitch: TwitchClient, logger):
    """Respond to !hello command with cooldown using Twitch plugin"""
    username = event.data['user']['display_name']
    await twitch.send_message(f"Hello @{username}! 👋")
    logger.info(f"Greeted {username}")

Active Triggers (Timer-Based) ✨ New!

from starstreamer import on_event
from starstreamer.triggers import TimerTrigger
from starstreamer.plugins.twitch import TwitchClient

@on_event("active_trigger.execute")
async def stream_uptime_announcer(event, twitch: TwitchClient, logger):
    """Announce stream uptime every 30 minutes using Twitch plugin"""
    if event.data.get("trigger_type") == "TimerTrigger":
        trigger = event.data.get("trigger")
        if trigger and trigger.name == "uptime_announce":
            uptime = get_stream_uptime()  # Your function
            await twitch.send_message(f"🕒 Stream uptime: {uptime}")

# Register timer: TimerTrigger(interval=1800, name="uptime_announce")

Architecture Overview

graph LR
    A[Platform Plugins] --> B[Event Bus]
    B --> C[Modules/Handlers]
    C --> E[Services]
    E --> R[Repositories]
    R --> D[Database]

    L[Active Triggers] --> M[ActiveTriggerManager]
    M --> C

    F[Twitch Plugin] --> A
    G[OBS Plugin] --> A
    H[Discord Plugin] --> A

    T[Timer Triggers] --> L
    S[State Triggers] --> L
    Q[Query Triggers] --> L

    subgraph "Data Layer"
        R --> U[UserRepository]
        R --> I[IdentityRepository]
        R --> EC[EconomyRepository]
        U --> D
        I --> D
        EC --> D
    end

    subgraph "Platform Integrations"
        F --> FE[EventSub WebSocket]
        F --> FA[Helix API]
        G --> GW[obs-websocket]
        H --> HG[Discord Gateway]
    end

Module System

StarStreamer uses a modular architecture where features are organized into self-contained modules:

  • Chat Module - Basic chat commands and interactions
  • Alerts Module - Stream event notifications (follows, subs, raids)
  • RPG Module - Economy system and game features
  • Custom Modules - Easy to create your own!

Getting Help

System Requirements

  • Python 3.11 or higher
  • Windows, macOS, or Linux
  • Twitch account with API credentials
  • OBS Studio (for OBS integration)