light-show-manager 1.0.0

Timeline-based show orchestration framework for coordinating time-synchronized commands

Light Show Manager

A pure Python framework for orchestrating time-synchronized commands across any hardware or software system.

Features

• Timeline-Based: Schedule events at precise timestamps
• Sync & Async Support: Separate methods for sync and async commands
• Batch Operations: Execute multiple commands simultaneously
• Device State Management: Automatic tracking and restoration of device states
• Unified Logging: Color-coded console output and timestamped log files
• Lifecycle Hooks: Pre-show setup, post-show cleanup, per-event callbacks
• Graceful Shutdown: Always runs cleanup on Ctrl+C or errors
• Pure Python: No dependencies (uses stdlib asyncio)
• Hardware Agnostic: Works with any devices/APIs/systems

Installation

pip install light-show-manager

Or for development:

git clone https://github.com/JimmyJammed/light-show-manager.git
cd light-show-manager
pip install -e .

Quick Start

from lightshow import LightShowManager, Show
import asyncio

# Create a show
show = Show(name="demo", duration=10.0)

# Add SYNC events (run in thread pool)
show.add_sync_event(0.0, lambda: print("Start!"))
show.add_sync_batch(5.0, [cmd1, cmd2, cmd3])

# Add ASYNC events (awaited)
show.add_async_event(2.5, async_function)

# Create manager with lifecycle hooks
manager = LightShowManager(
    shows=[show],
    pre_show=setup,      # Runs before show
    post_show=cleanup    # ALWAYS runs after (even on error/interrupt)
)

# Run show
asyncio.run(manager.run_show("demo"))

Core Concepts

Sync vs Async - Separate Methods

The package uses explicit separate methods to distinguish between sync and async commands:

Sync Methods (Thread Pool Execution)

# Single sync event
show.add_sync_event(timestamp, command, description)

# Batch of sync commands (all run concurrently in thread pool)
show.add_sync_batch(timestamp, [cmd1, cmd2, cmd3], description)

Async Methods (Direct Await)

# Single async event
show.add_async_event(timestamp, async_command, description)

# Batch of async commands (all awaited concurrently)
show.add_async_batch(timestamp, [async1, async2, async3], description)

This design makes it crystal clear which execution mode you're using - no guessing, no auto-detection ambiguity.

Complete API

Creating Shows

from lightshow import Show

show = Show(
    name="my_show",
    duration=180.0,  # 3 minutes
    description="My awesome show"
)

Adding Sync Events

# Sync functions (blocking I/O, GPIO, serial, etc.)
def turn_on_relay():
    GPIO.output(17, HIGH)
    time.sleep(0.1)

# Single event
show.add_sync_event(0.0, turn_on_relay, "Turn on relay")

# Batch (all fire at same time, run concurrently)
show.add_sync_batch(
    5.0,
    [
        lambda: lights1.on(),
        lambda: lights2.on(),
        lambda: motor.start()
    ],
    description="Turn on all devices"
)

# Bulk add
show.add_sync_events([
    (0.0, cmd1, "Event 1"),
    (5.0, cmd2, "Event 2"),
    (10.0, cmd3, "Event 3")
])

Adding Async Events

# Async functions
async def play_audio():
    await audio_player.play("song.mp3")

# Single event
show.add_async_event(0.0, play_audio, "Start music")

# Batch (all await concurrently)
show.add_async_batch(
    10.0,
    [
        async_lights1.on(),
        async_lights2.on(),
        async_audio.play("song.mp3")
    ],
    description="Async batch"
)

# Bulk add
show.add_async_events([
    (0.0, async1, "Event 1"),
    (5.0, async2, "Event 2"),
    (10.0, async3, "Event 3")
])

Lifecycle Hooks

def pre_show(show, context):
    """Setup before show starts."""
    print(f"Starting: {show.name}")
    # Turn off all lights
    # Play intro sound
    # Reset state

def post_show(show, context):
    """Cleanup after show (ALWAYS runs)."""
    print(f"Ending: {show.name}")
    # Turn off all lights
    # Stop motors
    # Restore GPIO states

def on_event(event, show, context):
    """Called after each event fires."""
    print(f"Event: {event.description}")

def on_error(error, event_or_show, context):
    """Called when errors occur."""
    print(f"Error: {error}")

manager = LightShowManager(
    shows=[show],
    pre_show=pre_show,
    post_show=post_show,  # ALWAYS runs
    on_event=on_event,
    on_error=on_error
)

Note: Hooks can be sync or async functions - the manager handles both automatically!

Running Shows

# Run single show
await manager.run_show("my_show")

# Run with context (passed to all hooks)
await manager.run_show("my_show", context={"volume": 75})

# Run rotation
await manager.run_rotation(["show1", "show2", "show3"])

# Run forever
await manager.run_rotation(["show1", "show2"], repeat=True)

Graceful Shutdown

The manager automatically handles Ctrl+C and termination signals:

manager = LightShowManager(
    shows=[show],
    post_show=cleanup  # ALWAYS runs, even on Ctrl+C
)

# User presses Ctrl+C during show
# → Show stops gracefully
# → post_show() runs
# → Clean exit

Examples

Example 1: Simple Sync Show

import asyncio
from lightshow import LightShowManager, Show

# Define sync functions
def lights_on():
    print("Lights ON")

def lights_off():
    print("Lights OFF")

# Create show
show = Show("demo", duration=10.0)
show.add_sync_event(0.0, lights_on)
show.add_sync_event(5.0, lights_off)

# Run
manager = LightShowManager(shows=[show])
asyncio.run(manager.run_show("demo"))

Example 2: Mixed Sync/Async

import asyncio
from lightshow import LightShowManager, Show

# Sync function
def gpio_on():
    GPIO.output(17, HIGH)

# Async function
async def play_audio():
    await audio.play("song.mp3")

# Create show
show = Show("mixed", duration=20.0)
show.add_sync_event(0.0, gpio_on)      # Sync
show.add_async_event(1.0, play_audio)  # Async

# Both work together seamlessly!
manager = LightShowManager(shows=[show])
asyncio.run(manager.run_show("mixed"))

Example 3: Device State Management

Automatically track and restore device states around shows:

import asyncio
from lightshow import LightShowManager, Show, with_device_state_management
from govee import GoveeClient

# Initialize device client
govee_client = GoveeClient(api_key="your-key", prefer_lan=True)

# Create manager
manager = LightShowManager()

# Register state management hooks (once, in main.py)
def save_device_states(devices, context):
    """Save state using your device library."""
    device_client = context.get('device_client')
    if device_client:
        device_client.save_state(devices)

def restore_device_states(devices, context):
    """Restore state using your device library."""
    device_client = context.get('device_client')
    if device_client:
        device_client.restore_state(devices)

manager.hooks.save_device_states = save_device_states
manager.hooks.restore_device_states = restore_device_states

# Use decorator on show builders (automatically tracks device usage)
@with_device_state_management(
    govee_client,
    spotlight_devices=[spotlight1, spotlight2]  # Turn off before show
)
def build_my_show(show_manager, govee_client, config):
    show = show_manager.create_show("my_show", duration=120.0)

    # All device operations are tracked automatically
    await govee_client.apply_scene(garage_lights, scene1)
    await govee_client.set_music_mode(neon_lights, mode_value=1, sensitivity=50)

    return show  # Decorator handles state save/restore!

What it does:

• Automatically tracks all devices used during show construction
• Saves their state before show starts (via your hook)
• Restores state after show ends (via your hook)
• No manual device lists needed!

Example 4: Unified Logging

The package includes a color-coded logging system with file output:

from lightshow import Logger, configure_stdlib_logging
import logging

# Initialize logger with file logging
logger = Logger(log_dir="logs")

# Route Python's stdlib logging through lightshow Logger
# (catches logs from third-party libraries like govee-python)
configure_stdlib_logging(logger, level=logging.INFO)

# Use the logger
logger.info("Show system starting...")
logger.warn("Device not responding")
logger.error("Failed to initialize hardware")
logger.debug("Detailed debugging info")

# Console output is color-coded:
# INFO  = White
# WARN  = Yellow
# ERROR = Red
# DEBUG = Cyan

# File output goes to: logs/log_2025_11_13_20_00_00.log
# with caller info for easy debugging:
# 2025-11-13 20:00:00 [INFO] Show starting [main.py:123 in main()]

Benefits:

• Color-coded console for quick visual scanning
• Timestamped log files for debugging
• Automatic caller information (file, line, function)
• Third-party library integration via stdlib logging bridge
• Respects NO_COLOR environment variable

Example 5: StrangerCourt Integration

from lightshow import LightShowManager, Show
from govee import GoveeClient
import asyncio

# Initialize hardware
govee = GoveeClient(api_key="...", prefer_lan=True)
govee.load_devices("govee_devices.json")
audio = AudioPlayer()
motor = MotorController()

# Create show
starcourt = Show("starcourt", duration=180.0)

# Add events (all SYNC in this case)
starcourt.add_sync_event(0.0, lambda: audio.play("starcourt.mp3"))

starcourt.add_sync_batch(2.5, [
    lambda: govee.set_color(govee.get_device("Garage Left"), (255, 0, 0)),
    lambda: govee.set_color(govee.get_device("Garage Right"), (0, 0, 255)),
    lambda: govee.set_brightness_all(govee.get_all_devices(), 100)
])

starcourt.add_sync_event(5.0, lambda: motor.set_speed(50))

# Lifecycle hooks
def pre_show(show, context):
    audio.play("powerdown.mp3")
    govee.power_all(govee.get_all_devices(), on=False)

def post_show(show, context):
    govee.power_all(govee.get_all_devices(), on=False)
    motor.stop()
    audio.stop()

# Run
manager = LightShowManager(
    shows=[starcourt],
    pre_show=pre_show,
    post_show=post_show
)

asyncio.run(manager.run_show("starcourt"))

Why Separate Methods?

We chose explicit separate methods over auto-detection for clarity:

✅ Benefits

1. Crystal Clear: No confusion about execution mode
2. Self-Documenting: Code shows intent explicitly
3. IDE Friendly: Better autocomplete and type hints
4. No Surprises: Behavior is predictable
5. Easy to Learn: Clear mental model

❌ Alternatives Considered

• Auto-detection: Requires understanding async introspection
• Single method with flag: Easy to forget the flag
• Wrappers: Extra boilerplate

Architecture

User Code
    ↓
LightShowManager
    ↓
┌────────────┬──────────────┬───────────────┐
│  Timeline  │   Executor   │   Lifecycle   │
│  (When)    │   (How)      │   (Hooks)     │
└────────────┴──────────────┴───────────────┘
    ↓              ↓                ↓
Commands    Sync: Thread     Pre/Post Show
(Callables)  Async: Await    Event Callbacks

Package Structure

light-show-manager/
├── lightshow/
│   ├── __init__.py
│   ├── manager.py          # LightShowManager
│   ├── show.py             # Show class
│   ├── timeline.py         # Timeline, TimelineEvent
│   ├── executor.py         # Async/sync executor
│   └── exceptions.py       # Custom exceptions
│
├── examples/
│   ├── simple_show.py      # Sync example
│   ├── async_show.py       # Async example
│   └── mixed_show.py       # Mixed sync/async
│
└── tests/

Error Handling

from lightshow.exceptions import (
    ShowNotFoundError,
    EventExecutionError,
    ShowInterruptedError
)

try:
    await manager.run_show("my_show")
except ShowNotFoundError:
    print("Show not found")
except EventExecutionError as e:
    print(f"Event failed: {e.event_description}")
except ShowInterruptedError:
    print("Show interrupted by user")

Testing

pip install -e ".[dev]"
pytest

Contributing

Contributions welcome! Please feel free to submit a Pull Request.

License

MIT License - see LICENSE file for details.

Acknowledgments

• Inspired by the StrangerCourt Halloween project
• Built for the maker/DIY automation community

Changelog

See CHANGELOG.md for detailed release notes and version history.

Recent Highlights

• Logger Module (Unreleased) - Color-coded console and file logging system
• Logging Bridge (Unreleased) - Integrate Python's standard logging with lightshow Logger
• Device State Management (0.2.0) - Automatic tracking and restoration decorator
• Timeline-Based Shows (0.1.0) - Precise timestamp synchronization

Support

• GitHub Issues: https://github.com/JimmyJammed/light-show-manager/issues
• GitHub Discussions: https://github.com/JimmyJammed/light-show-manager/discussions
• Documentation: https://github.com/JimmyJammed/light-show-manager#readme