From d339ba588d11de65d10f23e61afaed27b2e42ed2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 7 Apr 2026 11:40:29 +0000 Subject: [PATCH] feat(06-02): add feature extraction and segmentation backend modules + API - Create lightsync/audio/features.py with extract_features and extract_features_async - Create lightsync/audio/segments.py with detect_segments via agglomerative clustering - Add /api/audio/segments and /api/audio/features endpoints to audio.py - Initialize app.state.segments cache in main.py lifespan --- lightsync/api/audio.py | 38 ++++++++++++++++++ lightsync/audio/features.py | 78 +++++++++++++++++++++++++++++++++++++ lightsync/audio/segments.py | 67 +++++++++++++++++++++++++++++++ lightsync/main.py | 2 + 4 files changed, 185 insertions(+) create mode 100644 lightsync/audio/features.py create mode 100644 lightsync/audio/segments.py diff --git a/lightsync/api/audio.py b/lightsync/api/audio.py index 3543c7e..64e9cee 100644 --- a/lightsync/api/audio.py +++ b/lightsync/api/audio.py @@ -10,6 +10,8 @@ from fastapi.responses import StreamingResponse from pydantic import BaseModel from lightsync.audio.waveform import extract_peaks from lightsync.audio.beats import analyze_async +from lightsync.audio.segments import detect_segments_async +from lightsync.audio.features import extract_features_async router = APIRouter() @@ -190,3 +192,39 @@ async def fetch_youtube(request: Request, body: YouTubeFetchRequest): yield f"data: {json.dumps({'type': 'error', 'msg': str(e)})}\n\n" return StreamingResponse(event_stream(), media_type="text/event-stream") + + +@router.get("/segments") +async def get_segments(request: Request, path: str, n_segments: int | None = None): + """Return structural segments for an audio file. Cached in app.state.segments.""" + segments_cache: dict = getattr(request.app.state, "segments", None) or {} + + cache_key = f"{path}:{n_segments or 'auto'}" + if cache_key not in segments_cache: + p = Path(path) + if not p.exists(): + raise HTTPException(status_code=404, detail=f"File not found: {path}") + try: + segments = await detect_segments_async(path, n_segments) + if not hasattr(request.app.state, "segments") or request.app.state.segments is None: + request.app.state.segments = {} + request.app.state.segments[cache_key] = segments + except Exception as exc: + raise HTTPException(status_code=500, detail=f"Segmentation failed: {exc}") from exc + else: + segments = segments_cache[cache_key] + + return {"segments": segments, "count": len(segments), "file": path} + + +@router.get("/features") +async def get_features(request: Request, path: str): + """Return extracted audio features (onsets, chroma, RMS) for a file.""" + p = Path(path) + if not p.exists(): + raise HTTPException(status_code=404, detail=f"File not found: {path}") + try: + features = await extract_features_async(path) + except Exception as exc: + raise HTTPException(status_code=500, detail=f"Feature extraction failed: {exc}") from exc + return {**features, "file": path} diff --git a/lightsync/audio/features.py b/lightsync/audio/features.py new file mode 100644 index 0000000..bb34435 --- /dev/null +++ b/lightsync/audio/features.py @@ -0,0 +1,78 @@ +"""Audio feature extraction: onsets, chroma, RMS energy (Phase 6). + +extract_features(path) -> { + "onset_times": [float, ...], + "rms_1s": [float, ...], + "chroma_1s": [int, ...], + "duration": float, + "sr": int, +} +""" +from __future__ import annotations +import asyncio +from pathlib import Path +from typing import Any + + +def extract_features(path: str) -> dict[str, Any]: + """Extract onsets, chroma, RMS from audio file. + + Args: + path: Absolute path to audio file. + + Returns: + Dictionary with onset_times (seconds), rms_1s (per-second mean RMS), + chroma_1s (per-second dominant pitch class 0-11), duration, sr. + """ + import librosa + import numpy as np + + p = Path(path) + if not p.exists(): + raise FileNotFoundError(f"Audio file not found: {path}") + + y, sr = librosa.load(path, sr=None, mono=True) + + # Onsets — note attacks / transients + onset_frames = librosa.onset.onset_detect(y=y, sr=sr, units='frames') + onset_times = librosa.frames_to_time(onset_frames, sr=sr).tolist() + + # Chroma — tonal/harmonic content (CQT-based) + chroma = librosa.feature.chroma_cqt(y=y, sr=sr) # shape (12, T) + + # RMS energy — loudness envelope + rms = librosa.feature.rms(y=y)[0] # shape (T,) + + # Reduce to 1s resolution for downstream consumers (heuristic + LLM) + hop_length = 512 + frame_rate = sr / hop_length + duration = len(y) / sr + + n_seconds = int(duration) + rms_1s = [] + chroma_1s = [] + for i in range(n_seconds): + start_frame = int(i * frame_rate) + end_frame = int((i + 1) * frame_rate) + # RMS mean for this second + rms_slice = rms[start_frame:min(end_frame, len(rms))] + rms_1s.append(float(np.mean(rms_slice)) if len(rms_slice) > 0 else 0.0) + # Dominant pitch class for this second + chroma_slice = chroma[:, start_frame:min(end_frame, chroma.shape[1])] + if chroma_slice.shape[1] > 0: + chroma_1s.append(int(np.argmax(np.mean(chroma_slice, axis=1)))) + else: + chroma_1s.append(0) + + return { + "onset_times": onset_times, + "rms_1s": rms_1s, + "chroma_1s": chroma_1s, + "duration": duration, + "sr": sr, + } + + +async def extract_features_async(path: str) -> dict[str, Any]: + """Async wrapper — runs in thread pool.""" + return await asyncio.to_thread(extract_features, path) diff --git a/lightsync/audio/segments.py b/lightsync/audio/segments.py new file mode 100644 index 0000000..b53a128 --- /dev/null +++ b/lightsync/audio/segments.py @@ -0,0 +1,67 @@ +"""Structural segmentation via librosa agglomerative clustering (Phase 6). + +detect_segments(path, n_segments=None) -> [{"start": float, "end": float, "label": str}, ...] + +Uses chroma features + recurrence matrix + agglomerative clustering. +Target 4-12 sections for a typical 3-5 minute song. +""" +from __future__ import annotations +import asyncio +from pathlib import Path +from typing import Any + + +def detect_segments(path: str, n_segments: int | None = None) -> list[dict[str, Any]]: + """Detect structural sections using agglomerative clustering on chroma features. + + Args: + path: Absolute path to audio file. + n_segments: Number of segments (default: auto based on duration, ~1 per 30s). + + Returns: + List of dicts: [{"start": float, "end": float, "label": "SEC 1"}, ...] + """ + import librosa + import numpy as np + + p = Path(path) + if not p.exists(): + raise FileNotFoundError(f"Audio file not found: {path}") + + y, sr = librosa.load(path, sr=None, mono=True) + duration = len(y) / sr + + # Auto-select segment count if not specified: ~1 per 30s, clamped to 4-12 + if n_segments is None: + n_segments = max(4, min(12, int(duration / 30))) + + # Build chroma feature matrix + chroma = librosa.feature.chroma_cqt(y=y, sr=sr, bins_per_octave=36) + + # Recurrence matrix — similarity structure + R = librosa.segment.recurrence_matrix(chroma, width=3, mode='affinity', sym=True) + + # Filter recurrence matrix to enhance diagonal consistency + R_filtered = librosa.segment.path_enhance(R, 15) + + # Agglomerative clustering — returns boundary frame indices + bounds = librosa.segment.agglomerative(R_filtered, k=n_segments) + bound_times = librosa.frames_to_time(bounds, sr=sr).tolist() + + # Build section list: [{start, end, label}, ...] + # Pitfall 2 from RESEARCH: agglomerative returns k boundaries, yielding k+1 sections + all_times = [0.0] + bound_times + [duration] + sections = [] + for i in range(len(all_times) - 1): + sections.append({ + "start": all_times[i], + "end": all_times[i + 1], + "label": f"SEC {i + 1}", + }) + + return sections + + +async def detect_segments_async(path: str, n_segments: int | None = None) -> list[dict[str, Any]]: + """Async wrapper — runs in thread pool.""" + return await asyncio.to_thread(detect_segments, path, n_segments) diff --git a/lightsync/main.py b/lightsync/main.py index 279c0de..5ac1280 100644 --- a/lightsync/main.py +++ b/lightsync/main.py @@ -25,6 +25,8 @@ async def lifespan(app: FastAPI): # Beat analysis cache: filepath -> {"tempo": float, "beats": [float, ...]} app.state.beats = {} + # Segments cache: "filepath:n_segments" -> [{start, end, label}, ...] + app.state.segments = {} # UDP sender for device communication (Phase 3) udp_sender = UDPSender()