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
This commit is contained in:
78
lightsync/audio/features.py
Normal file
78
lightsync/audio/features.py
Normal file
@@ -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)
|
||||
67
lightsync/audio/segments.py
Normal file
67
lightsync/audio/segments.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user