b6aacad9ce
- Create lightsync/api/sync.py with POST /api/sync/generate and GET /api/sync/llm-available - heuristic_generate: intensity-to-animation mapping, chroma-to-color, beat-snapped block placement - llm_generate: Anthropic claude-3-haiku with lazy init, JSON parse + validation - section_range filtering for per-section generation (D-11) - Register sync.router with prefix /api/sync in main.py
344 lines
12 KiB
Python
344 lines
12 KiB
Python
"""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)}
|