led-sync-studio/.planning/phases/02-app-core-audio/02-RESEARCH.md

Phase 2: App Core + Audio — Research

Researched: 2026-04-03 Domain: Python asyncio application — audio playback, real-time beat detection, choreography scheduling, UDP transport Confidence: HIGH (core stack verified via PyPI; ARM installation paths verified via piwheels and source; patterns verified via official docs)

---

<user_constraints>

User Constraints (from CONTEXT.md)

Locked Decisions

• D-01: Choreography files are Pydantic-validated JSON. Schema: {"song_path": str, "bpm": float|null, "events": [{"timestamp": float, "zone": str, "animation": str, "params": {...}, "loop": int|null, "duration": float|null}]}.
• D-02: Events are an ordered array sorted by timestamp. Each event specifies zone, animation name, params (matching ESP32 protocol), optional loop count, and optional duration.
• D-03: File extension: .choreo.json
• D-04: Two separate threads — miniaudio for song playback (with sample-accurate position tracking), sounddevice+aubio for system audio capture and beat detection. Neither thread touches the other's audio device.
• D-05: Beat detection thread uses call_from_thread() bridge to post beat events into the asyncio event loop. Beat events carry the detected timestamp.
• D-06: Audio capture uses PipeWire's PulseAudio compatibility layer via sounddevice. Monitor source for system audio loopback.
• D-07: Absolute monotonic timestamp scheduling. Record start_time = time.monotonic() at play start. Each event fires at start_time + event.timestamp. Recalculate remaining sleep each iteration — never accumulate relative delays.
• D-08: Drift tolerance: must stay under 20ms over 5 minutes of playback. Validate with test.
• D-09: Pause/resume adjusts start_time offset so event scheduling remains correct after resume.
• D-10: Asyncio DatagramProtocol wrapping UDP socket. Sends JSON commands to ESP32 on port 4210. Matches the protocol defined in docs/protocol.md exactly.
• D-11: Heartbeat: optional periodic ping to detect ESP32 availability (not blocking — fire-and-forget).
• D-12: Connection state tracked (connected/disconnected) based on ESP32 STATUS responses. Displayed in CLI.
• D-13: Simple asyncio REPL for testing. Commands: play <file>, pause, resume, seek <seconds>, stop, load <choreo.json>, add <timestamp> <zone> <animation> [params_json], save <file>, status, quit.
• D-14: Not a production interface — Phase 3 TUI replaces this. Minimal error handling, no fancy output.

Claude's Discretion

• Python package structure (src layout vs flat)
• asyncio task organization and cancellation patterns
• Pydantic model field validation details
• Error handling granularity in transport layer
• aubio configuration (buffer size, hop size, threshold)

Deferred Ideas (OUT OF SCOPE)

None — discussion stayed within phase scope. </user_constraints>

---

<phase_requirements>

Phase Requirements

ID	Description	Research Support
AUD-01	User can load and play local song files (MP3, FLAC, WAV)	miniaudio 1.61 via PlaybackDevice + stream_file; just_playback wrapper for pause/seek state
AUD-02	User can pause, resume, and seek to any position	miniaudio generator pattern: track frames_played for position; seek via stream_file(seek_frame=); pause by halting generator; monotonic offset for resume (D-09)
AUD-03	System audio captured via PipeWire/PulseAudio for beat detection	sounddevice 0.5.5 py3-none-any wheel; PortAudio + PipeWire PulseAudio compat layer; monitor source loopback
AUD-04	Beat/onset detection recognizes beats from audio stream in real-time	aubio 0.4.9 tempo object; 512-sample hop size at 44100 Hz = ~11ms latency; source build required on aarch64
CHR-04	Choreographies can be saved and loaded as JSON files	Pydantic v2 model_dump_json() / model_validate_json(); ChoreoFile + ChoreoEvent models
CHR-05	Choreography playback sends animation commands on-time to ESP32	asyncio scheduler task; absolute monotonic timestamps (D-07); DatagramProtocol sendto()
</phase_requirements>

---

Summary

Phase 2 builds the entire headless Pi-side application: song playback, beat detection, choreography data model, event scheduler, and UDP transport to the ESP32. The Phase 3 Textual TUI will wrap this layer directly — so all asyncio task and state design must be TUI-integration-ready from the start.

The key architectural constraint is the audio thread boundary. sounddevice and miniaudio run in OS-level threads; neither can touch asyncio directly. Beat events must travel through loop.call_soon_threadsafe() into the asyncio loop. The playback clock runs as a pure asyncio task using absolute monotonic timestamps — never relative sleeps.

One critical installation risk: aubio 0.4.9 has no pre-built wheels on PyPI or piwheels for aarch64 (64-bit ARM) Raspberry Pi OS. It requires source build with libaubio-dev. This must be validated in Wave 0 before any beat detection code is written. miniaudio 1.61 similarly has no aarch64 wheels on PyPI but piwheels covers 32-bit (armv7l) Pi OS only. For 64-bit Pi OS, miniaudio also requires source build. The Wave 0 plan must include an environment validation step on the target Pi.

Primary recommendation: Build in four sequential sub-layers — (1) data model + file I/O, (2) UDP transport, (3) playback clock + scheduler, (4) audio threads + beat bridge — each testable independently via the asyncio REPL before the next layer begins.

---

Project Constraints (from CLAUDE.md)

This project runs on a VPS with Docker (groll.cloud). The Pi-side application is a separate context — it does NOT run inside Docker on the VPS. The CLAUDE.md describes the VPS deployment environment, not the Raspberry Pi target. No Docker constraints apply to this phase.

Relevant CLAUDE.md directives for the overall project:

• Git-hosted at git.groll.cloud (Gitea, GITEA_API_TOKEN in ~/.env)
• Deploy script: ~/bin/deploy.sh /home/claude/<slug> — irrelevant to Pi app development

---

Standard Stack

Core

Library	Version	Purpose	Why Standard
Python	3.11+	Runtime	Ships on Raspberry Pi OS bookworm; required by Textual (Phase 3)
asyncio	stdlib	Event loop, task coordination, UDP transport	No dependency; core primitive for Textual integration
miniaudio	1.61	MP3/FLAC/WAV playback + position tracking	Pure C backend, no SDL/ALSA dep, exposes frame position, supports seek
sounddevice	0.5.5	PipeWire/PulseAudio audio capture via PortAudio	Pure-Python wheel, PipeWire compat layer, NumPy arrays in callback
aubio	0.4.9	Real-time beat and onset detection	C-core, causal/streaming design, ~11ms latency at 44100Hz/512 hop
numpy	1.26+	Audio buffer type for aubio and sounddevice	Required by both; already present in Pi OS
pydantic	2.12.5	Choreography file schema validation + serialization	Rust-backed v2, model_validate_json/model_dump_json, free JSON schema
uv	latest	Python package manager	Fast, lockfile-based, works on ARM Pi OS

Supporting

Library	Version	Purpose	When to Use
just_playback	0.1.8	High-level miniaudio wrapper with pause/seek/curr_pos	Reference for position tracking pattern — may use directly or mirror its approach
libportaudio2	system	PortAudio shared lib required by sounddevice	Install via apt before sounddevice
libaubio-dev / libsndfile-dev	system	Build dependencies for aubio source compilation	Required on aarch64 Pi OS where no wheel exists

Alternatives Considered

Instead of	Could Use	Tradeoff
miniaudio	pygame.mixer	pygame conflicts with sounddevice for device access; no precise position
miniaudio	gstreamer (gi.repository)	Heavyweight, fragile Python bindings on Pi
sounddevice	pyaudio	Requires ALSA headers, no asyncio, harder ARM install
aubio	librosa	Batch/offline design, not causal streaming; heavy scipy dep
pydantic v2	dataclasses	No JSON schema, no model_validate_json, no field validators
asyncio DatagramProtocol	asyncio-dgram	Extra dep; stdlib is sufficient for fire-and-forget UDP

Installation (on target Raspberry Pi):

# System dependencies
sudo apt install -y libportaudio2 libaubio-dev libsndfile-dev python3-dev gcc

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create project environment
cd /path/to/led-sync-studio
uv init .
uv add miniaudio sounddevice numpy pydantic

# aubio: source build (no aarch64 wheel on PyPI or piwheels)
uv add aubio   # will compile from source — requires libaubio-dev and gcc
# OR install directly from git if 0.4.9 fails to build on Python 3.12+:
uv add "aubio @ git+https://github.com/aubio/aubio"

Version verification (confirmed 2026-04-03 via PyPI API):

• miniaudio: 1.61 (July 2024)
• sounddevice: 0.5.5 (March 2026)
• aubio: 0.4.9 (February 2019 — last stable release, no newer version)
• pydantic: 2.12.5

---

Architecture Patterns

Recommended Project Structure

led-sync-studio/
├── pyproject.toml
├── uv.lock
└── src/
    └── led_sync/
        ├── __init__.py
        ├── main.py              # asyncio.run() entry point, REPL loop
        ├── models.py            # Pydantic ChoreoFile, ChoreoEvent models
        ├── scheduler.py         # Playback clock + event scheduler asyncio task
        ├── audio/
        │   ├── __init__.py
        │   ├── player.py        # miniaudio PlaybackDevice wrapper
        │   └── beat_detector.py # sounddevice+aubio thread + call_soon_threadsafe bridge
        └── transport/
            ├── __init__.py
            └── udp_client.py    # asyncio DatagramProtocol to ESP32

Why src layout: Prevents accidental import of package from project root during testing; required by uv project template.

Pattern 1: Asyncio DatagramProtocol UDP Client

The transport layer uses loop.create_datagram_endpoint() with a custom protocol subclass. Once created, transport.sendto() is synchronous and non-blocking from the asyncio thread.

# Source: https://docs.python.org/3/library/asyncio-protocol.html#udp-echo-client
import asyncio
import json

class ESP32Transport(asyncio.DatagramProtocol):
    def __init__(self):
        self.transport = None
        self.connected = False

    def connection_made(self, transport):
        self.transport = transport
        self.connected = True

    def datagram_received(self, data, addr):
        # Parse STATUS responses from ESP32
        try:
            msg = json.loads(data.decode())
            if msg.get("status") == "ok":
                self.connected = True
        except Exception:
            pass

    def error_received(self, exc):
        pass  # UDP errors are non-fatal; fire-and-forget

    def connection_lost(self, exc):
        self.connected = False

    def send_command(self, cmd: dict) -> None:
        """Fire-and-forget: non-blocking from asyncio thread."""
        if self.transport:
            self.transport.sendto(json.dumps(cmd).encode())


async def create_esp32_transport(host: str, port: int = 4210):
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        ESP32Transport,
        remote_addr=(host, port),
    )
    return transport, protocol

Pattern 2: Beat Detection Thread with call_soon_threadsafe Bridge

sounddevice callbacks run on an OS audio thread. Beat events must be marshalled into the asyncio loop without blocking the audio thread.

# Source: sounddevice docs + architecture pattern from ARCHITECTURE.md
import sounddevice as sd
import aubio
import asyncio

class BeatDetector:
    def __init__(self, loop: asyncio.AbstractEventLoop, on_beat):
        self.loop = loop
        self.on_beat = on_beat  # async coroutine or callback
        self.sample_rate = 44100
        self.hop_size = 512
        self.tempo = aubio.tempo(
            method="default",
            buf_size=1024,     # window for FFT analysis
            hop_size=self.hop_size,
            samplerate=self.sample_rate,
        )

    def _audio_callback(self, indata, frames, time, status):
        """Called by sounddevice on audio thread — must NOT touch asyncio."""
        samples = indata[:, 0].astype("float32")  # mono
        is_beat = self.tempo(samples)
        if is_beat[0]:
            beat_time = time.inputBufferAdcTime
            # Bridge to asyncio: non-blocking, thread-safe
            self.loop.call_soon_threadsafe(
                self.loop.create_task,
                self.on_beat(beat_time)
            )

    def start(self, device=None):
        self.stream = sd.InputStream(
            samplerate=self.sample_rate,
            blocksize=self.hop_size,
            channels=1,
            dtype="float32",
            callback=self._audio_callback,
            device=device,  # None = PipeWire default; monitor source for loopback
        )
        self.stream.start()

    def stop(self):
        self.stream.stop()
        self.stream.close()

aubio configuration guidance (Claude's Discretion):

• buf_size=1024, hop_size=512 — recommended starting point per PITFALLS.md and maxhaesslein.de article. Latency = hop_size / sample_rate = ~11ms.
• For 120 BPM music, minimum beat interval = 500ms. The "beat hold" period should default to 200ms to suppress double-triggers.
• The tempo.get_bpm() method returns current estimated BPM after detection.

Pattern 3: Absolute Monotonic Scheduler

The playback clock recalculates remaining sleep on every iteration — never accumulates relative delays.

# Source: PITFALLS.md Pattern 3 + CONTEXT.md D-07/D-08/D-09
import asyncio
import time

class ChoreographyScheduler:
    def __init__(self, transport):
        self.transport = transport
        self.start_time: float | None = None
        self.pause_offset: float = 0.0   # accumulated pause time
        self.paused_at: float | None = None
        self._task: asyncio.Task | None = None

    @property
    def current_position(self) -> float:
        """Playback position in seconds. Thread-safe read."""
        if self.start_time is None:
            return 0.0
        if self.paused_at is not None:
            return self.paused_at - self.start_time - self.pause_offset
        return time.monotonic() - self.start_time - self.pause_offset

    def play(self, events: list, song_path: str) -> None:
        self.start_time = time.monotonic()
        self.pause_offset = 0.0
        self.paused_at = None
        self._task = asyncio.create_task(self._run(events))

    def pause(self) -> None:
        self.paused_at = time.monotonic()

    def resume(self) -> None:
        if self.paused_at is not None:
            self.pause_offset += time.monotonic() - self.paused_at
            self.paused_at = None

    async def _run(self, events: list) -> None:
        sorted_events = sorted(events, key=lambda e: e.timestamp)
        for event in sorted_events:
            while True:
                if self.paused_at is not None:
                    await asyncio.sleep(0.005)  # 5ms poll while paused
                    continue
                # Absolute fire time; recalculated fresh each iteration
                fire_at = self.start_time + self.pause_offset + event.timestamp
                remaining = fire_at - time.monotonic()
                if remaining <= 0:
                    break
                await asyncio.sleep(min(remaining, 0.005))
            self._dispatch(event)

    def _dispatch(self, event) -> None:
        cmd = {
            "v": 1,
            "zone": event.zone,
            "animation": event.animation,
            "params": event.params,
        }
        self.transport.send_command(cmd)

Pattern 4: Pydantic v2 Choreography Model

# Source: https://docs.pydantic.dev/2.11/concepts/serialization/
from pydantic import BaseModel, field_validator
from typing import Optional
import json

class ChoreoEvent(BaseModel):
    timestamp: float             # seconds from song start
    zone: str                    # "schrank", "wand", or "all"
    animation: str               # must match ESP32 animation names
    params: dict                 # speed, intensity, colors, direction
    loop: Optional[int] = None   # None = play once; N = repeat N times
    duration: Optional[float] = None  # seconds; None = run until next event

    @field_validator("zone")
    @classmethod
    def validate_zone(cls, v):
        if v not in ("schrank", "wand", "all"):
            raise ValueError(f"Invalid zone: {v}")
        return v

    @field_validator("animation")
    @classmethod
    def validate_animation(cls, v):
        valid = {"chase","pulse","rainbow","strobe","color_wash","breathe","sparkle","gradient_sweep"}
        if v not in valid:
            raise ValueError(f"Unknown animation: {v}")
        return v

class ChoreoFile(BaseModel):
    song_path: str
    bpm: Optional[float] = None
    events: list[ChoreoEvent] = []

    def save(self, path: str) -> None:
        with open(path, "w") as f:
            f.write(self.model_dump_json(indent=2))

    @classmethod
    def load(cls, path: str) -> "ChoreoFile":
        with open(path) as f:
            return cls.model_validate_json(f.read())

Pattern 5: miniaudio Playback with Position Tracking

miniaudio PlaybackDevice uses a generator pattern. Position is tracked by counting frames yielded.

# Source: https://github.com/irmen/pyminiaudio/blob/master/README.md
import miniaudio
import time
import threading

class AudioPlayer:
    def __init__(self):
        self._device: miniaudio.PlaybackDevice | None = None
        self._frames_played: int = 0
        self._sample_rate: int = 44100
        self._lock = threading.Lock()

    def _make_stream(self, file_path: str, seek_frame: int = 0):
        """Generator that tracks position and allows pause."""
        stream = miniaudio.stream_file(
            file_path,
            output_format=miniaudio.SampleFormat.SIGNED16,
            nchannels=2,
            sample_rate=self._sample_rate,
            seek_frame=seek_frame,
        )
        frame_count = seek_frame
        required = yield b""  # prime the generator
        for chunk in stream:
            with self._lock:
                self._frames_played = frame_count
            frame_count += len(chunk) // 4  # 2 channels * 2 bytes
            required = yield chunk

    @property
    def position_seconds(self) -> float:
        with self._lock:
            return self._frames_played / self._sample_rate

    def play(self, file_path: str, seek_seconds: float = 0.0) -> None:
        seek_frame = int(seek_seconds * self._sample_rate)
        self._device = miniaudio.PlaybackDevice(
            sample_rate=self._sample_rate,
            nchannels=2,
            output_format=miniaudio.SampleFormat.SIGNED16,
        )
        stream = self._make_stream(file_path, seek_frame)
        next(stream)  # prime
        self._device.start(stream)

    def stop(self) -> None:
        if self._device:
            self._device.stop()
            self._device.close()
            self._device = None

Note on pause/resume: miniaudio PlaybackDevice has no built-in pause. To pause: call device.stop() and record position_seconds. To resume: call play(file_path, seek_seconds=saved_position). This is the approach used by the just_playback wrapper library. The scheduler compensates via D-09 (pause_offset).

Pattern 6: Asyncio REPL (CLI Testing Interface)

# Source: asyncio docs + D-13
import asyncio
import sys

async def repl(app_state):
    print("LED Sync Studio CLI. Type 'help' for commands.")
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await asyncio.get_event_loop().connect_read_pipe(lambda: protocol, sys.stdin)

    while True:
        sys.stdout.write("> ")
        sys.stdout.flush()
        line = await reader.readline()
        if not line:
            break
        cmd = line.decode().strip()
        await handle_command(cmd, app_state)

Alternative (simpler, works for non-interactive SSH):

async def repl(app_state):
    loop = asyncio.get_running_loop()
    while True:
        line = await loop.run_in_executor(None, input, "> ")
        await handle_command(line.strip(), app_state)

Anti-Patterns to Avoid

• Blocking asyncio with audio: Never call aubio.tempo() or sd.InputStream.read() inside an async def. These are blocking C extensions. Always thread.
• Relative sleep accumulation: Never await asyncio.sleep(event.timestamp - prev_event.timestamp). Each sleep must recalculate against time.monotonic() absolute.
• Direct widget calls from audio thread: In Phase 3, never call Textual methods from the beat detection thread. Use call_from_thread() or call_soon_threadsafe() exclusively.
• time.time() for scheduling: Use time.monotonic() only. time.time() is affected by NTP adjustments and can jump.
• Pause by sleeping generator: Inserting silence (zero bytes) into the playback generator for pause creates desync between audio position and scheduler. Stop+seek is cleaner.

---

Don't Hand-Roll

Problem	Don't Build	Use Instead	Why
Beat detection algorithm	Custom FFT threshold	aubio.tempo object	Onset detection with history, phase confidence, BPM estimation — 10+ years of research
Audio thread/asyncio bridge	Custom shared state	loop.call_soon_threadsafe()	Thread-safe by spec; hand-rolled queues have subtle lock ordering bugs
JSON schema validation	Manual dict key checking	Pydantic v2 models	Field validators, type coercion, nested model validation, free JSON schema
MP3/FLAC decoding	Custom decoder	miniaudio	Handles codec differences, sample rate conversion, file format detection
UDP protocol framing	Custom packet format	stdlib json + asyncio DatagramProtocol	The protocol is already defined in docs/protocol.md — implement it exactly

Key insight: The audio pipeline complexity (sample-rate conversion, format decoding, beat tracking state machine, PipeWire device enumeration) is enormous if built from scratch. Every component in the standard stack represents years of edge-case handling.

---

ESP32 Protocol Reference (LOCKED — do not deviate)

The UDP transport client MUST implement docs/protocol.md exactly. Key facts:

Property	Value
Port	4210
Protocol version	"v": 1 — required in every command
Max payload	512 bytes
Zone values	"schrank", "wand", "all"
Color format	[R, G, B, W] arrays (W=0 for RGB-only animations)
Animation command	{"v":1, "zone":..., "animation":..., "params":{...}}
Stop command	{"v":1, "cmd":"stop", "zone":...}
Status query	{"v":1, "cmd":"status"}
Status response	{"v":1, "status":"ok", "wand":"...", "schrank":"...", "brightness":N}
Brightness cap	0–60 (firmware enforces max 60% regardless of sent value)
Animation names	chase, pulse, rainbow, strobe, color_wash, breathe, sparkle, gradient_sweep
ESP32 error handling	Silently ignores unknown commands; no UDP error responses
Animation start latency	<50ms from UDP receive (queue drain on next 20ms tick)

Heartbeat implementation (D-11): Send {"v":1,"cmd":"status"} every 5 seconds. Parse response to update connected state. No ACK mechanism on animation commands.

---

Common Pitfalls

Pitfall 1: aubio Wheel Unavailable on aarch64 (64-bit Pi OS)

What goes wrong: uv add aubio fails with build error, or installs 0-byte module, on 64-bit Raspberry Pi OS Bookworm (aarch64). PyPI has no wheels for aubio 0.4.9. Piwheels provides armv6l/armv7l wheels (32-bit) but NOT aarch64. Why it happens: aubio 0.4.9 is from 2019; no one has cut modern ARM64 CI wheels. The project appears unmaintained (last PyPI release: 2019-02-08). How to avoid: Install libaubio-dev and python3-dev via apt first, then uv add aubio to trigger source build. If Python version is 3.12+ and the source build fails, use uv add "aubio @ git+https://github.com/aubio/aubio" for the latest unreleased code. Wave 0 action required: Validate on actual target Pi before writing any beat detection code.

Pitfall 2: miniaudio Has No aarch64 Wheels Either

What goes wrong: uv add miniaudio on 64-bit Pi OS triggers source compilation. This usually succeeds (it's a CFFI package) but requires gcc and python3-dev. Why it happens: miniaudio 1.61 on PyPI provides manylinux wheels for x86_64 only (59 wheels listed, none for aarch64). Piwheels provides armv7l (32-bit). How to avoid: Pre-install gcc python3-dev via apt. The CFFI build is straightforward — the real risk is forgetting the system deps.

Pitfall 3: sounddevice Monitor Source Selection

What goes wrong: sd.InputStream() with default device captures microphone input, not system audio output. Beat detection fires on voice/ambient noise rather than music. Why it happens: PipeWire's "monitor" source (loopback of sink output) has a different device name/index than the default capture device. How to avoid: Enumerate devices with sd.query_devices() and find the monitor source. On PipeWire, it typically appears as "Monitor of [sink name]". Pass this as device= to sd.InputStream. For Phase 2 CLI, print available devices in the status command output.

Pitfall 4: miniaudio Pause Implementation

What goes wrong: Trying to pause miniaudio by suspending the generator (e.g., setting a flag and yielding silence) causes audio artifacts and position desync. Why it happens: The PlaybackDevice generator protocol does not support mid-stream suspension cleanly. How to avoid: Implement pause as device.stop() + record position_seconds. Implement resume as play(file_path, seek_seconds=saved_position). This causes a brief audio gap (~50ms) which is acceptable for a pause operation. The scheduler compensates via pause_offset (D-09).

Pitfall 5: Drift Under asyncio Event Loop Load

What goes wrong: The asyncio event loop also handles UDP I/O, REPL input, and (in Phase 3) TUI rendering. A blocking operation anywhere causes scheduler drift above the 20ms budget (D-08). Why it happens: asyncio.sleep(0.005) is a minimum sleep — it sleeps longer if the loop is busy. How to avoid: Keep all asyncio tasks non-blocking (no time.sleep(), no blocking file I/O). Use loop.run_in_executor() for any blocking calls. The scheduler must re-check remaining on every wake rather than sleeping to exact fire time.

Pitfall 6: Protocol Version Mismatch at Runtime

What goes wrong: Commands silently do nothing on ESP32 if "v" field is missing, wrong, or commands use wrong zone/animation names. Why it happens: ESP32 error handling is silent (no UDP error responses). Bugs look like network problems. How to avoid: Validate zone and animation names in the Pydantic ChoreoEvent model before they reach the transport layer. Include "v": 1 as a constant in the transport layer — never configurable.

---

Runtime State Inventory

This is a greenfield phase — no data migration or renamed runtime state. No persistent databases, no OS-registered services, no build artifacts from prior phases to reconcile.

The only runtime state from Phase 1 is the ESP32 firmware (flashed). Phase 2 does not touch it.

---

Environment Availability

This development environment (Ubuntu 24.04, x86_64) is NOT the target (Raspberry Pi, aarch64/armv7l). The plan must include a Wave 0 validation step on the actual Pi.

Dependency	Required By	Available (dev)	Available (Pi)	Version	Fallback
Python 3.11+	Runtime	Python 3.12.3	Pi OS bookworm: 3.11	3.12.3	—
asyncio	Event loop	stdlib	stdlib	—	—
miniaudio	AUD-01/02	No wheel/no pip	Source build (gcc needed)	1.61	—
sounddevice	AUD-03	No wheel/no pip	py3-none-any + libportaudio2	0.5.5	—
aubio	AUD-04	No wheel/no pip	Source build (libaubio-dev needed)	0.4.9	See note below
pydantic	CHR-04	Not installed	pip install	2.12.5	—
uv	Package mgr	Not installed	Install via curl	latest	pip
libportaudio2	sounddevice	✗ (dev machine)	apt install	—	—
libaubio-dev	aubio build	✗ (dev machine)	apt install	—	—
PipeWire	AUD-03/06	✗ (dev machine)	Pi OS bookworm default	—	PulseAudio fallback
gcc / python3-dev	source builds	✓ system	apt install (usually present)	—	—

aubio fallback note: If aubio 0.4.9 source build fails on Python 3.12 (Pi OS noble variants), alternatives are:

1. pip install git+https://github.com/aubio/aubio (dev branch, more Python 3.11+ compat)
2. apt install python3-aubio (system package, may be older but ARM-compiled)

Missing dependencies with no fallback (blocking for Pi execution):

• libportaudio2, libaubio-dev must be installed via apt before running the app on Pi
• PipeWire monitor source must be available (present by default on bookworm)

Wave 0 validation required on Pi:

sudo apt install -y libportaudio2 libaubio-dev libsndfile-dev python3-dev gcc
uv add miniaudio sounddevice aubio pydantic
python3 -c "import miniaudio; import sounddevice; import aubio; print('all ok')"

---

State of the Art

Old Approach	Current Approach	When Changed	Impact
pyaudio for capture	sounddevice	~2018	NumPy arrays, asyncio-compatible, no ALSA headers
pygame for playback	miniaudio	~2020	No SDL dep, no device conflict, precise position
pydantic v1 parse_raw	pydantic v2 model_validate_json	v2 release 2023	Faster (Rust), new API — use v2 exclusively
loop.call_soon_threadsafe for beat bridge	Same pattern, still current	—	No change needed
asyncio.DatagramProtocol UDP	Same pattern, still current	—	Low-level but correct for fire-and-forget

Deprecated/outdated:

• pydantic.parse_raw() / pydantic.parse_obj(): Removed in v2. Use model_validate_json() and model_validate().
• asyncio.get_event_loop() (Python 3.10+): Deprecated in favor of asyncio.get_running_loop() inside coroutines.
• time.clock(): Removed in Python 3.8. Use time.monotonic().

---

Open Questions

1. aubio Python 3.12 compatibility

   • What we know: aubio 0.4.9 was released 2019; Python 3.12 introduced breaking C API changes
   • What's unclear: Whether source build succeeds on Pi OS bookworm (Python 3.11) vs. noble (Python 3.12)
   • Recommendation: Test uv add aubio as the very first Wave 0 task on target Pi. If it fails, use python3-aubio apt package.

2. PipeWire monitor source device name

   • What we know: sd.query_devices() will list all devices; monitor source is a PortAudio-visible source
   • What's unclear: Exact device name/index on a specific Pi OS bookworm install
   • Recommendation: The status REPL command should print available audio devices for self-documentation.

3. miniaudio seek accuracy for MP3 files

   • What we know: stream_file(seek_frame=N) does coarse seeking for VBR MP3 (variable bitrate). FLAC/WAV are sample-accurate.
   • What's unclear: Whether seek accuracy is sufficient for choreography resume use case
   • Recommendation: Test seek precision with a 3-minute MP3 in Wave 0. If inaccurate, track frames in generator and seek from start when needed.

4. sounddevice and miniaudio simultaneous PipeWire access

   • What we know: Both request PortAudio/PipeWire simultaneously. PipeWire supports multiple clients.
   • What's unclear: Whether simultaneous playback+capture on the same Pi PipeWire instance causes issues.
   • Recommendation: Test both initialized simultaneously in Wave 0 before building the full audio pipeline.

---

Code Examples

Build a valid ESP32 animation command

# Source: docs/protocol.md
def build_animation_cmd(zone: str, animation: str, speed: float = 0.5,
                        intensity: float = 1.0, colors: list | None = None) -> dict:
    params = {"speed": speed, "intensity": intensity}
    if colors:
        params["colors"] = colors  # [[R,G,B,W], ...]
    return {"v": 1, "zone": zone, "animation": animation, "params": params}

# Example: chase on schrank at half speed, red
cmd = build_animation_cmd("schrank", "chase", speed=0.5, colors=[[255, 0, 0, 0]])
# {"v": 1, "zone": "schrank", "animation": "chase", "params": {"speed": 0.5, "intensity": 1.0, "colors": [[255, 0, 0, 0]]}}

Load and validate a .choreo.json file

# Source: Pydantic v2 docs
choreo = ChoreoFile.load("my_show.choreo.json")  # validates on load
# Add an event:
choreo.events.append(ChoreoEvent(
    timestamp=12.5,
    zone="wand",
    animation="pulse",
    params={"speed": 0.8, "intensity": 1.0, "colors": [[0, 100, 255, 0]]},
))
choreo.events.sort(key=lambda e: e.timestamp)
choreo.save("my_show.choreo.json")

Asyncio scheduler drift validation test

# Source: PITFALLS.md Pitfall 3 + D-08
import asyncio, time

async def validate_drift(test_events_s: list[float], tolerance_ms: float = 20.0):
    """Fire fake events and measure drift."""
    errors = []
    start = time.monotonic()
    for t in test_events_s:
        while True:
            remaining = (start + t) - time.monotonic()
            if remaining <= 0:
                break
            await asyncio.sleep(min(remaining, 0.005))
        actual = time.monotonic() - start
        drift_ms = abs(actual - t) * 1000
        errors.append(drift_ms)
        assert drift_ms < tolerance_ms, f"Drift {drift_ms:.1f}ms at t={t}s"
    print(f"Max drift: {max(errors):.2f}ms over {max(test_events_s)}s")

---

Sources

Primary (HIGH confidence)

• docs/protocol.md — complete ESP32 UDP protocol spec, zones, animation names, command format
• firmware/src/protocol.h — LedCommand, CmdType, Zone, AnimParams structures
• https://docs.python.org/3/library/asyncio-protocol.html — DatagramProtocol, create_datagram_endpoint, sendto
• https://docs.pydantic.dev/2.11/concepts/serialization/ — model_dump_json, model_validate_json
• PyPI API (queried 2026-04-03) — miniaudio 1.61, sounddevice 0.5.5, aubio 0.4.9, pydantic 2.12.5

Secondary (MEDIUM confidence)

• https://www.piwheels.org/project/miniaudio/ — ARM wheel availability (armv6l/armv7l only, not aarch64)
• https://www.piwheels.org/project/aubio/ — ARM wheel availability (armv6l/armv7l for cp37/cp39, not aarch64 or cp311)
• https://github.com/irmen/pyminiaudio/blob/master/README.md — PlaybackDevice API, stream_file, seek_frame
• https://github.com/cheofusi/just_playback — pause/resume/curr_pos pattern using miniaudio
• https://www.maxhaesslein.de/notes/real-time-beat-prediction-with-aubio/ — aubio.tempo config: buf_size=1024, hop_size=512
• https://python-sounddevice.readthedocs.io/ — InputStream callback, call_soon_threadsafe pattern
• .planning/research/STACK.md — library selection rationale (HIGH confidence, pre-verified)
• .planning/research/ARCHITECTURE.md — component boundaries, threading model (HIGH confidence)
• .planning/research/PITFALLS.md — timeline drift, audio thread safety, pitfall catalogue (HIGH confidence)

Tertiary (LOW confidence — flag for Pi validation)

• aarch64 build success for aubio 0.4.9 on Python 3.11 Pi OS bookworm: UNVERIFIED — must test on hardware
• sounddevice + miniaudio simultaneous PipeWire access: UNVERIFIED — must test on hardware
• PipeWire monitor source device name on bookworm: UNVERIFIED — enumerate at runtime

---

Metadata

Confidence breakdown:

• Standard stack (library versions): HIGH — verified via PyPI API 2026-04-03
• Architecture patterns: HIGH — asyncio DatagramProtocol from official docs; beat bridge from sounddevice official examples
• miniaudio position tracking: MEDIUM — pause/seek pattern inferred from just_playback source + README; no direct PlaybackDevice.current_frame property
• aubio configuration: MEDIUM — hop_size/buf_size from maxhaesslein.de article (2024) + aubio docs
• ARM installation paths: MEDIUM — piwheels confirmed armv7l only; aarch64 requires source build (inferred, not tested)
• Pitfalls: HIGH — drawn from PITFALLS.md which cites official issue trackers

Research date: 2026-04-03 Valid until: 2026-07-03 (90 days — libraries in this stack are stable; aubio unchanged since 2019)