From b6aacad9ce4effb4ac25d91e9395e89c10e9a25d Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Apr 2026 11:45:49 +0000 Subject: [PATCH] =?UTF-8?q?feat(06-03):=20add=20AI=20sync=20backend=20?= =?UTF-8?q?=E2=80=94=20heuristic=20mapper,=20LLM=20generator,=20API=20endp?= =?UTF-8?q?oint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- lightsync/api/sync.py | 343 ++++++++++++++++++++++++++++++++++++++++++ lightsync/main.py | 3 +- 2 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 lightsync/api/sync.py diff --git a/lightsync/api/sync.py b/lightsync/api/sync.py new file mode 100644 index 0000000..2e862bb --- /dev/null +++ b/lightsync/api/sync.py @@ -0,0 +1,343 @@ +"""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": "", "timestamp": , "duration": , "animation": "", "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)} diff --git a/lightsync/main.py b/lightsync/main.py index 5ac1280..0a93116 100644 --- a/lightsync/main.py +++ b/lightsync/main.py @@ -44,12 +44,13 @@ def create_app() -> FastAPI: app = FastAPI(title="LightSync", lifespan=lifespan, redirect_slashes=False) # API routes (registered before static mount — see Pattern 7) - from lightsync.api import shows, devices, ws, audio, timeline + from lightsync.api import shows, devices, ws, audio, timeline, sync app.include_router(shows.router, prefix="/api/shows") app.include_router(devices.router, prefix="/api/devices") app.include_router(ws.router) app.include_router(audio.router, prefix="/api/audio") app.include_router(timeline.router, prefix="/api") + app.include_router(sync.router, prefix="/api/sync") # Serve uploaded audio files so