docs(phase-02): research app core + audio phase

Research covers miniaudio playback, sounddevice+aubio beat detection,
asyncio DatagramProtocol UDP transport, Pydantic v2 choreography models,
absolute monotonic scheduler pattern, and ARM install paths for all deps.
Flags critical aarch64 wheel gaps (aubio, miniaudio) requiring Wave 0
Pi validation before implementation begins.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-04-03 14:04:12 +02:00
parent 2f642daf0c
commit 0cc9990fa7

View File

@@ -0,0 +1,727 @@
# 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):**
```bash
# 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.
```python
# 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.
```python
# 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.
```python
# 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
```python
# 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.
```python
# 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)
```python
# 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):**
```python
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 | 060 (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:**
```bash
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
```python
# 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
```python
# 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
```python
# 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)