35 KiB
35 KiB
phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, must_haves
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | must_haves | |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 06-ai-sync | 03 | execute | 3 |
|
|
true |
|
|
Purpose: Fulfills SYNC-03 — auto-fill the timeline with animation blocks from audio analysis. The heuristic base always works; the LLM path provides a richer alternative when configured. This is the capstone feature of Phase 6.
Output: /api/sync/generate endpoint, AI Sync panel in the UI, working heuristic show generator, optional LLM show generator.
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/phases/06-ai-sync/06-CONTEXT.md @.planning/phases/06-ai-sync/06-RESEARCH.md @.planning/phases/06-ai-sync/06-01-SUMMARY.md @.planning/phases/06-ai-sync/06-02-SUMMARY.mdFrom lightsync/audio/features.py (created in 06-02):
def extract_features(path: str) -> dict:
"""Returns {"onset_times": [...], "rms_1s": [...], "chroma_1s": [...], "duration": float, "sr": int}"""
async def extract_features_async(path: str) -> dict:
...
From lightsync/audio/segments.py (created in 06-02):
def detect_segments(path: str, n_segments: int | None = None) -> list[dict]:
"""Returns [{"start": float, "end": float, "label": "SEC 1"}, ...]"""
async def detect_segments_async(path: str, n_segments: int | None = None) -> list[dict]:
...
From lightsync/audio/beats.py:
def analyze(path: str) -> dict:
"""Returns {"tempo": float, "beats": [float, ...]}"""
async def analyze_async(path: str) -> dict:
...
From lightsync/models/show.py:
class CueModel(BaseModel):
id: UUID = Field(default_factory=uuid4)
timestamp: float
duration: float = 4.0
mode: Literal["animation", "frame_sequence"] = "animation"
animation: str | None = None
params: dict[str, Any] = Field(default_factory=dict)
From lightsync/frontend/timeline/timeline.js:
class TimelineCanvas {
tracks: [{device_id, device_name, strip_type, cues: []}]
history: CommandHistory
segmentOverlay: SegmentOverlay // from 06-02
}
From lightsync/frontend/timeline/segments.js (created in 06-02):
class SegmentOverlay {
getSelected() -> {start, end, label} | null
}
From lightsync/frontend/timeline/commands.js:
class PlaceBlockCommand { ... }
// Used for undo/redo — but AI sync replaces all blocks, so we need a bulk replace approach
From lightsync/frontend/app.js:
const client = new LightSyncClient(); // WebSocket client
let timeline = null; // TimelineCanvas instance
// Animation types available: chase, pulse, rainbow, strobe, color_wipe, fire, solid
From lightsync/main.py:
def create_app() -> FastAPI:
from lightsync.api import shows, devices, ws, audio, timeline
app.include_router(audio.router, prefix="/api/audio")
# Add: app.include_router(sync.router, prefix="/api/sync")
"""AI-assisted show generation endpoint (Phase 6, SYNC-03).
POST /api/sync/generate — generates animation blocks from audio analysis.
Two modes: heuristic (always available) and LLM (requires ANTHROPIC_API_KEY).
"""
from __future__ import annotations
import asyncio
import colorsys
import logging
import os
from typing import Any
from uuid import uuid4
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel
from lightsync.audio.beats import analyze_async
from lightsync.audio.features import extract_features_async
from lightsync.audio.segments import detect_segments_async
logger = logging.getLogger(__name__)
router = APIRouter()
ANIMATION_TYPES = ['chase', 'pulse', 'rainbow', 'strobe', 'color_wipe', 'fire', 'solid']
class SyncGenerateRequest(BaseModel):
path: str # audio file path
device_ids: list[str] # device UUIDs to generate for
use_llm: bool = False # optional LLM generation
section_range: list[float] | None = None # [start, end] to fill only a section (D-11)
class GeneratedBlock(BaseModel):
id: str
device_id: str
timestamp: float
duration: float
animation: str
params: dict[str, Any]
def _chroma_to_rgb(pitch_class: int) -> str:
"""Convert pitch class (0-11) to hex RGB color via HSL mapping.
Pitch class 0 (C) = 0deg hue, each class advances 30deg (360/12).
"""
hue = pitch_class / 12.0
r, g, b = colorsys.hls_to_rgb(hue, 0.5, 0.9)
return f"#{int(r*255):02x}{int(g*255):02x}{int(b*255):02x}"
def _intensity_to_animation(rms_mean: float, rms_max: float) -> str:
"""Map RMS intensity level to animation type.
High energy -> fast/flashy, low energy -> slow/ambient.
"""
if rms_max == 0:
return 'solid'
ratio = rms_mean / rms_max
if ratio > 0.7:
# High energy — fast animations
return 'strobe' if ratio > 0.85 else 'chase'
elif ratio > 0.4:
# Medium energy
return 'pulse' if ratio > 0.55 else 'color_wipe'
else:
# Low energy — ambient
return 'rainbow' if ratio > 0.2 else 'solid'
def _speed_from_tempo(tempo: float, intensity: float) -> float:
"""Derive animation speed from tempo and intensity. Range: 0.5 - 5.0."""
base = tempo / 60.0 # beats per second
return max(0.5, min(5.0, base * (0.5 + intensity)))
def heuristic_generate(
beats: list[float],
tempo: float,
features: dict[str, Any],
segments: list[dict[str, Any]],
device_ids: list[str],
section_range: list[float] | None = None,
) -> list[dict[str, Any]]:
"""Generate animation blocks using heuristic rules.
Algorithm per RESEARCH.md Pattern 5:
1. For each section: compute mean RMS -> intensity, dominant chroma -> color
2. Intensity maps to animation type (high=strobe/chase, medium=pulse/wipe, low=solid/rainbow)
3. Duration: snap to beat grid — each block spans from beat[i] to beat[i+N]
4. Color: chroma pitch class -> HSL hue -> RGB
Args:
beats: beat timestamps in seconds
tempo: BPM
features: {"onset_times", "rms_1s", "chroma_1s", "duration"}
segments: [{"start", "end", "label"}, ...]
device_ids: list of device UUID strings
section_range: optional [start, end] to restrict generation
Returns:
List of block dicts: [{"id", "device_id", "timestamp", "duration", "animation", "params"}, ...]
"""
rms_1s = features.get('rms_1s', [])
chroma_1s = features.get('chroma_1s', [])
duration = features.get('duration', 0)
rms_max = max(rms_1s) if rms_1s else 1.0
# Filter beats and segments by section_range if specified
if section_range and len(section_range) == 2:
range_start, range_end = section_range
beats_in_range = [b for b in beats if range_start <= b < range_end]
segments_in_range = [s for s in segments if s['end'] > range_start and s['start'] < range_end]
else:
range_start, range_end = 0.0, duration
beats_in_range = beats
segments_in_range = segments
blocks = []
for seg in segments_in_range:
seg_start = max(seg['start'], range_start)
seg_end = min(seg['end'], range_end)
# Compute section-level features
sec_start_idx = int(seg_start)
sec_end_idx = min(int(seg_end), len(rms_1s))
if sec_end_idx <= sec_start_idx:
continue
rms_section = rms_1s[sec_start_idx:sec_end_idx]
chroma_section = chroma_1s[sec_start_idx:sec_end_idx] if chroma_1s else [0]
rms_mean = sum(rms_section) / len(rms_section) if rms_section else 0
dominant_chroma = max(set(chroma_section), key=chroma_section.count) if chroma_section else 0
# Derive animation type and color
anim_type = _intensity_to_animation(rms_mean, rms_max)
color = _chroma_to_rgb(dominant_chroma)
intensity = rms_mean / rms_max if rms_max > 0 else 0.5
speed = _speed_from_tempo(tempo, intensity)
# Get beats within this section
section_beats = [b for b in beats_in_range if seg_start <= b < seg_end]
if len(section_beats) < 2:
# Too few beats — place one block spanning the section
for device_id in device_ids:
blocks.append({
"id": str(uuid4()),
"device_id": device_id,
"timestamp": seg_start,
"duration": seg_end - seg_start,
"animation": anim_type,
"params": {"color": color, "speed": round(speed, 2)},
})
continue
# Group beats into blocks — N beats per block based on density
# Dense sections (>3 beats/sec) = shorter blocks, sparse = longer blocks
density = len(section_beats) / (seg_end - seg_start) if seg_end > seg_start else 1
beats_per_block = max(2, min(8, int(8 / max(density, 0.5))))
i = 0
while i < len(section_beats):
block_start = section_beats[i]
end_idx = min(i + beats_per_block, len(section_beats) - 1)
block_end = section_beats[end_idx] if end_idx > i else block_start + 2.0
block_duration = max(0.5, block_end - block_start)
for device_id in device_ids:
blocks.append({
"id": str(uuid4()),
"device_id": device_id,
"timestamp": round(block_start, 3),
"duration": round(block_duration, 3),
"animation": anim_type,
"params": {"color": color, "speed": round(speed, 2)},
})
i += beats_per_block
return blocks
async def llm_generate(
beats: list[float],
tempo: float,
features: dict[str, Any],
segments: list[dict[str, Any]],
device_ids: list[str],
section_range: list[float] | None = None,
) -> list[dict[str, Any]]:
"""Generate animation blocks via Anthropic LLM (D-06: generates from scratch, not from heuristic).
Lazy-initializes the Anthropic client to avoid startup crash if API key is missing (Pitfall 4).
"""
import json as json_mod
key = os.environ.get('ANTHROPIC_API_KEY')
if not key:
raise HTTPException(status_code=503, detail='ANTHROPIC_API_KEY not configured')
try:
import anthropic
except ImportError:
raise HTTPException(status_code=503, detail='anthropic package not installed')
client = anthropic.Anthropic(api_key=key)
# Build compact feature payload for LLM (D-06)
rms_1s = features.get('rms_1s', [])
chroma_1s = features.get('chroma_1s', [])
# Filter by section range if specified
if section_range and len(section_range) == 2:
range_start, range_end = section_range
filtered_segments = [s for s in segments if s['end'] > range_start and s['start'] < range_end]
filtered_beats = [b for b in beats if range_start <= b < range_end]
else:
range_start = 0.0
range_end = features.get('duration', 0)
filtered_segments = segments
filtered_beats = beats
feature_payload = {
"tempo_bpm": tempo,
"duration_seconds": range_end - range_start,
"time_range": {"start": range_start, "end": range_end},
"beat_times": filtered_beats[:200], # cap for token budget
"segments": filtered_segments,
"rms_1s": rms_1s[int(range_start):int(range_end)],
"chroma_1s": chroma_1s[int(range_start):int(range_end)],
"device_ids": device_ids,
"available_animations": ANIMATION_TYPES,
}
prompt = f"""You are a music-to-light show generator. Given audio analysis features, generate a list of LED animation blocks that sync with the music.
INPUT FEATURES:
{json_mod.dumps(feature_payload, indent=2)}
RULES:
- Output a JSON array of block objects
- Each block: {{"device_id": "<from device_ids>", "timestamp": <seconds>, "duration": <seconds>, "animation": "<from available_animations>", "params": {{"color": "#rrggbb", "speed": <0.5-5.0>}}}}
- High-energy sections (high RMS) should use strobe, chase (fast)
- Low-energy sections should use solid, rainbow (slow)
- Blocks should align to beat timestamps where possible
- Generate blocks for ALL devices in device_ids
- Colors should reflect the harmonic content (chroma values)
- Ensure full coverage of the time range with no large gaps
OUTPUT: Only the JSON array, no explanation."""
def _call_api():
return client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=4096,
messages=[{"role": "user", "content": prompt}],
)
response = await asyncio.to_thread(_call_api)
# Parse LLM response
text = response.content[0].text.strip()
# Strip markdown code fences if present
if text.startswith('```'):
text = text.split('\n', 1)[1]
if text.endswith('```'):
text = text.rsplit('```', 1)[0]
text = text.strip()
try:
raw_blocks = json_mod.loads(text)
except json_mod.JSONDecodeError as e:
raise HTTPException(status_code=500, detail=f'LLM returned invalid JSON: {e}')
# Validate and normalize blocks
blocks = []
for rb in raw_blocks:
if not isinstance(rb, dict):
continue
if rb.get('device_id') not in device_ids:
continue
if rb.get('animation') not in ANIMATION_TYPES:
rb['animation'] = 'solid'
blocks.append({
"id": str(uuid4()),
"device_id": rb.get('device_id', device_ids[0]),
"timestamp": float(rb.get('timestamp', 0)),
"duration": float(rb.get('duration', 2.0)),
"animation": rb.get('animation', 'solid'),
"params": rb.get('params', {"color": "#00ffff", "speed": 1.0}),
})
return blocks
@router.post("/generate")
async def generate_sync(request: Request, body: SyncGenerateRequest):
"""Generate animation blocks from audio analysis.
D-04: Heuristic base always runs. LLM is optional (use_llm flag).
D-06: LLM generates from scratch, not from heuristic output.
D-11: section_range limits generation to a time window.
"""
from pathlib import Path
p = Path(body.path)
if not p.exists():
raise HTTPException(status_code=404, detail=f"File not found: {body.path}")
if not body.device_ids:
raise HTTPException(status_code=400, detail="No device_ids provided")
# Gather all analysis data (cached or computed)
try:
beats_data = await analyze_async(body.path)
features = await extract_features_async(body.path)
segments = await detect_segments_async(body.path)
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Audio analysis failed: {exc}") from exc
beats = beats_data.get('beats', [])
tempo = beats_data.get('tempo', 120.0)
if body.use_llm:
blocks = await llm_generate(
beats, tempo, features, segments,
body.device_ids, body.section_range,
)
else:
blocks = heuristic_generate(
beats, tempo, features, segments,
body.device_ids, body.section_range,
)
return {"blocks": blocks, "count": len(blocks), "mode": "llm" if body.use_llm else "heuristic"}
@router.get("/llm-available")
async def llm_available():
"""Check if LLM generation is available (API key configured)."""
key = os.environ.get('ANTHROPIC_API_KEY', '')
return {"available": bool(key)}
Edit lightsync/main.py — Register the sync router. Add import and router registration in create_app().
After the existing from lightsync.api import shows, devices, ws, audio, timeline line (around line 45), add:
from lightsync.api import sync
After the existing app.include_router(timeline.router, prefix="/api") line (around line 50), add:
app.include_router(sync.router, prefix="/api/sync")
- In the
#timeline-toolbardiv (around line 53-66), add the AI SYNC button before the closing</div>. Insert after the SNAP button:
<div class="toolbar-sep"></div>
<button id="btn-ai-sync-open" class="transport-btn" title="AI-assisted show generation">AI SYNC</button>
- Add the AI Sync panel as a new div AFTER the
#timeline-toolbardiv and BEFORE the#timeline-canvascanvas element. This panel is shown/hidden by the button:
<div id="ai-sync-panel" class="ai-sync-panel" style="display:none;">
<div class="ai-sync-header">AI SYNC</div>
<div class="ai-sync-body">
<div class="ai-sync-info" id="ai-sync-info">SELECT AUDIO FILE FIRST</div>
<div class="ai-sync-toggles">
<label class="ai-sync-toggle"><input type="checkbox" id="ai-use-beats" checked> BEATS</label>
<label class="ai-sync-toggle"><input type="checkbox" id="ai-use-onsets" checked> ONSETS</label>
<label class="ai-sync-toggle"><input type="checkbox" id="ai-use-chroma" checked> CHROMA</label>
<label class="ai-sync-toggle"><input type="checkbox" id="ai-use-rms" checked> RMS</label>
</div>
<div class="ai-sync-actions">
<button id="btn-ai-generate" class="transport-btn ai-sync-generate" disabled>GENERATE</button>
<label class="ai-sync-toggle ai-sync-llm-toggle">
<input type="checkbox" id="ai-use-llm" disabled>
<span id="ai-llm-label">USE LLM</span>
</label>
</div>
</div>
</div>
style.css — Add AI Sync panel styles at the end. Terminal aesthetic per D-08 and context specifics.
/* AI Sync panel (Phase 6) */
.ai-sync-panel {
background: var(--bg-panel-dark);
border: 1px solid var(--border-dim);
margin: 0 4px;
padding: 0;
font-family: var(--font-mono);
}
.ai-sync-header {
background: var(--bg-panel);
padding: 4px 8px;
font-size: 11px;
color: var(--accent);
text-transform: uppercase;
letter-spacing: 1px;
border-bottom: 1px solid var(--border-dim);
}
.ai-sync-body {
padding: 6px 8px;
display: flex;
align-items: center;
gap: 12px;
}
.ai-sync-info {
font-size: 11px;
color: var(--text-dim);
min-width: 140px;
}
.ai-sync-toggles {
display: flex;
gap: 8px;
}
.ai-sync-toggle {
font-size: 11px;
color: var(--text-primary);
cursor: pointer;
display: flex;
align-items: center;
gap: 3px;
}
.ai-sync-toggle input[type="checkbox"] {
accent-color: var(--accent);
}
.ai-sync-actions {
display: flex;
align-items: center;
gap: 8px;
margin-left: auto;
}
.ai-sync-generate {
background: var(--accent);
color: var(--bg-primary);
font-weight: bold;
padding: 3px 12px;
}
.ai-sync-generate:disabled {
opacity: 0.3;
cursor: not-allowed;
}
.ai-sync-llm-toggle {
opacity: 0.4;
}
.ai-sync-llm-toggle:not(.disabled) {
opacity: 1;
}
app.js — Wire AI Sync panel interactions. Add after the initTimeline() function call (end of file, before or after the show selector wiring).
- Add the AI Sync panel open/close toggle:
// ── AI Sync panel ───────────────────────────────────────────────────────────
document.getElementById('btn-ai-sync-open')?.addEventListener('click', () => {
const panel = document.getElementById('ai-sync-panel');
if (!panel) return;
const visible = panel.style.display !== 'none';
panel.style.display = visible ? 'none' : 'block';
});
- Check LLM availability on page load and enable/disable the toggle:
fetch('/api/sync/llm-available')
.then(r => r.json())
.then(data => {
const checkbox = document.getElementById('ai-use-llm');
const label = document.getElementById('ai-llm-label');
const container = checkbox?.closest('.ai-sync-llm-toggle');
if (data.available) {
checkbox.disabled = false;
if (container) container.classList.remove('disabled');
} else {
checkbox.disabled = true;
if (label) label.title = 'ANTHROPIC_API_KEY not configured';
if (container) container.classList.add('disabled');
}
})
.catch(() => {});
- Enable GENERATE button when audio is loaded. In the existing
audio.addEventListener('loadedmetadata', ...)insideinitTimeline(), add:
// Enable AI Sync generate button
const genBtn = document.getElementById('btn-ai-generate');
if (genBtn) genBtn.disabled = false;
const infoEl = document.getElementById('ai-sync-info');
if (infoEl) {
const seg = timeline.segmentOverlay?.getSelected();
infoEl.textContent = seg ? `SECTION: ${seg.label}` : 'FULL SHOW';
}
- Update info text when segment selection changes. Add a periodic check or hook into the segment overlay click. The simplest approach: add a
requestAnimationFramebased checker in initTimeline, or wire it into the canvas click. Add after the timeline.startRenderLoop() call:
// Update AI Sync info when segment selection changes
let _lastSegIdx = -1;
setInterval(() => {
if (!timeline?.segmentOverlay) return;
const idx = timeline.segmentOverlay.selectedIndex;
if (idx !== _lastSegIdx) {
_lastSegIdx = idx;
const infoEl = document.getElementById('ai-sync-info');
if (infoEl) {
const seg = timeline.segmentOverlay.getSelected();
infoEl.textContent = seg ? `SECTION: ${seg.label}` : 'FULL SHOW';
}
}
}, 200);
- Wire the GENERATE button. This is the main interaction:
document.getElementById('btn-ai-generate')?.addEventListener('click', async () => {
if (!timeline) return;
const audio = document.getElementById('player');
if (!audio?.src) return;
// D-07: Confirm replacement if existing blocks
const hasBlocks = timeline.tracks.some(t => t.cues.length > 0);
if (hasBlocks) {
if (!confirm('Replace all existing blocks?')) return;
}
const btn = document.getElementById('btn-ai-generate');
const infoEl = document.getElementById('ai-sync-info');
btn.disabled = true;
btn.textContent = 'GENERATING...';
if (infoEl) infoEl.textContent = 'ANALYZING...';
try {
const filename = audio.src.split('/').pop();
const path = `/app/shows/${filename}`;
// D-11: If a section is selected, only fill that range
const selectedSeg = timeline.segmentOverlay?.getSelected();
const sectionRange = selectedSeg ? [selectedSeg.start, selectedSeg.end] : null;
// Collect device IDs from timeline tracks
const deviceIds = timeline.tracks.map(t => t.device_id);
const useLlm = document.getElementById('ai-use-llm')?.checked || false;
const res = await fetch('/api/sync/generate', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
path: path,
device_ids: deviceIds,
use_llm: useLlm,
section_range: sectionRange,
}),
});
if (!res.ok) {
const err = await res.json();
throw new Error(err.detail || `HTTP ${res.status}`);
}
const data = await res.json();
// Clear existing blocks (or section range only)
if (sectionRange) {
// Clear only blocks in the section range
for (const track of timeline.tracks) {
track.cues = track.cues.filter(c =>
c.timestamp + (c.duration || 0) < sectionRange[0] ||
c.timestamp >= sectionRange[1]
);
}
} else {
// Clear all blocks
for (const track of timeline.tracks) {
track.cues = [];
}
}
// Place generated blocks onto tracks
for (const block of data.blocks) {
const track = timeline.tracks.find(t => String(t.device_id) === String(block.device_id));
if (track) {
track.cues.push({
id: block.id,
timestamp: block.timestamp,
duration: block.duration,
animation: block.animation,
params: block.params,
mode: 'animation',
});
}
}
if (infoEl) infoEl.textContent = `GENERATED ${data.count} BLOCKS (${data.mode.toUpperCase()})`;
console.log(`[ai-sync] Generated ${data.count} blocks via ${data.mode}`);
} catch (e) {
console.error('[ai-sync]', e);
if (infoEl) infoEl.textContent = `ERROR: ${e.message}`;
} finally {
btn.disabled = false;
btn.textContent = 'GENERATE';
}
});
<success_criteria>
- Heuristic show generation produces a playable show from any audio file
- Blocks are musically meaningful: tempo-aligned, energy-matched, harmonically colored
- LLM path works when ANTHROPIC_API_KEY is configured; gracefully disabled otherwise
- Per-section fill works with segment overlay selection
- Confirmation dialog prevents accidental block replacement </success_criteria>