update

.dockerignore

@@ -0,0 +1,9 @@
+.git
+.venv
+.DS_Store
+.pytest_cache
+__pycache__
+*.pyc
+.embedding_cache
+neighbors.json
+neighbors.csv

.gitignore

@@ -0,0 +1,7 @@
+.DS_Store
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.embedding_cache/
+neighbors.json
+neighbors.csv

Dockerfile

@@ -1,11 +1,10 @@
 FROM python:3.11-slim
 
 # Системные зависимости
-RUN apt-get update && apt-get install -y \
+RUN apt-get update && apt-get install -y --no-install-recommends \
     ffmpeg \
     libsndfile1 \
-    gcc \
-    g++ \
+    curl \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app
@@ -14,14 +13,10 @@ WORKDIR /app
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
-# Копируем весь проект
 COPY . .
 
-# Создаем папку для треков
-RUN mkdir -p tracks
-
 # Порт Hugging Face Spaces
 EXPOSE 7860
 
 # Запуск FastAPI
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,5 +1,5 @@
 ---
-title: Track Embedding API
+title: OpenMusic Analysis API
 emoji: 🎵
 colorFrom: blue
 colorTo: purple
@@ -7,6 +7,281 @@ sdk: docker
 pinned: false
 ---
 
-# Track Embedding API
+# OpenMusic Music Analysis API
 
-FastAPI service for CLAP audio embeddings.
+Versioned server-side analysis for separate global audio, temporal audio and lyrics
+representations. The service deliberately does not fuse these embedding spaces.
+
+## Architecture
+
+```text
+multipart HTTP request
+        ↓
+validation + temporary-file lifetime
+        ↓
+MusicAnalysisService
+├── GlobalAudioAnalyzer ───┐
+├── TemporalAudioAnalyzer ─┼── shared CLAP audio encoder
+├── LyricsAnalyzer ─────────── BGE-M3 text encoder
+└── ModelRegistry
+        ↓
+versioned typed response
+```
+
+The HTTP layer validates, calls the application service and serializes. Analyzer
+implementations own preprocessing, inference, aggregation and normalization. Model
+objects are lazy singletons behind an inference semaphore; they are never created per
+request or exposed to the HTTP layer.
+
+`AnalysisContext` decodes an upload once to mono float32 PCM at 48 kHz and caches
+additional deterministic in-memory resamples by sample rate. Source-file decoding is
+centralized in ffmpeg and supports mp3, m4a, flac, wav, aac, ogg and opus.
+
+## Models
+
+| Representation | Model | Exact revision | Dimension | Normalization |
+|---|---|---|---:|---|
+| `audio.global` | `laion/clap-htsat-unfused` | `8fa0f1c6d0433df6e97c127f64b2a1d6c0dcda8a` | 512 | L2 per window, mean, L2 final |
+| `audio.temporal` | `laion/clap-htsat-unfused` | `8fa0f1c6d0433df6e97c127f64b2a1d6c0dcda8a` | 512 per segment | L2 per segment |
+| `lyrics.global` | `BAAI/bge-m3` | `5617a9f61b028005a4858fdac845db406aefb181` | 1024 | L2 per chunk, token-weighted mean, L2 final |
+
+CLAP is the existing backend's baseline, now pinned instead of following Hugging Face
+`main`. BGE-M3 was selected for its MIT license, English/Russian and broader
+multilingual support, semantic retrieval training, 1024-dimensional dense space and
+long-document support up to 8192 tokens. Lyrics still use structural chunks so section
+information is not discarded and section embeddings can be exposed later. Compared
+alternatives were multilingual E5 large (good multilingual quality but a 512-token
+limit), multilingual MPNet (older, 128-token sentence/paragraph setup), and GTE
+multilingual base (smaller, long-context, but requires repository remote code).
+
+Every registry item declares `model_id`, exact `model_version`, modality, dimension,
+dtype, normalization, license, preprocessing version and full representation config.
+The preprocessing version contains a deterministic hash of all result-affecting config,
+so changing window/hop/chunk settings changes cache identity.
+
+## Audio preprocessing
+
+The decoder runs ffprobe validation, then one deterministic ffmpeg decode (`-threads 1`)
+to mono little-endian float32 PCM at 48,000 Hz. Future model sample rates are produced
+from that request-scoped PCM with deterministic polyphase resampling.
+
+`audio.global` defaults:
+
+- 10-second analysis windows;
+- adaptive count: one for short tracks, up to four as duration grows;
+- window starts are uniformly distributed over 10%–90% of the valid start range;
+- short tracks are repeated deterministically to a full model window;
+- every CLAP vector is L2-normalized, vectors are averaged, then normalized again.
+
+This prevents a global track vector from effectively describing only the intro.
+Windows longer than CLAP's native 10 seconds are deterministically divided into evenly
+placed 10-second encoder subwindows and aggregated; CLAP never receives a long input
+from which its processor could choose a random crop.
+
+`audio.temporal` defaults:
+
+- 10-second window and 10-second hop;
+- explicit tail window so the ending is represented;
+- at most 24 segments, with deterministic uniform selection if the candidate count is
+  larger;
+- chronological millisecond timestamps and one normalized CLAP vector per segment.
+
+The temporal summary reports only representation geometry:
+`number_of_segments`, cosine `mean_adjacent_distance`, `max_adjacent_distance`,
+`trajectory_variance`, and `largest_transition_index`. It does not claim to measure
+emotion, climax or tension.
+
+## Lyrics preprocessing
+
+`LyricsPreprocessor` applies Unicode NFKC, normalizes line endings, removes control
+characters, LRC metadata/timestamps and pure technical URL lines, while preserving
+punctuation, paragraphs and repeated choruses. It recognizes English and Russian
+section markers such as Verse/Chorus/Bridge and Куплет/Припев/Бридж.
+
+Sections are the first chunk boundary. Oversized sections are split by lines and then by
+token IDs, never by arbitrary character count. The default chunk limit is 512 tokens.
+Repeated sections remain repeated and therefore retain their weight. BGE-M3 uses its
+document-side CLS dense representation; chunk vectors are normalized and aggregated by
+token count. The internal chunk/section pipeline is ready for a later `lyrics.sections`
+or `lyrics.temporal` response without changing `lyrics.global`.
+
+## Installation and deployment
+
+The production image uses Python 3.11. A system ffmpeg/ffprobe is required.
+
+```bash
+python3.11 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+uvicorn main:app --host 0.0.0.0 --port 7860
+```
+
+Or:
+
+```bash
+docker compose up --build
+```
+
+Docker Compose exposes `http://localhost:8000`; Hugging Face Spaces uses container port
+7860. Model files are cached in the `hf_cache` volume.
+
+Device selection is centralized. `OPENMUSIC_DEVICE=auto` chooses CUDA, then Apple MPS,
+then CPU. Set `OPENMUSIC_DEVICE=cpu` for a forced CPU fallback. Output embeddings remain
+float32 regardless of device. `OPENMUSIC_INFERENCE_CONCURRENCY=1` is the safe default
+that avoids duplicate loading and unbounded concurrent GPU work.
+
+## API
+
+### Models
+
+```bash
+curl http://localhost:7860/v1/models
+```
+
+`GET /v1/models` is the client source of truth. `loaded` distinguishes an available lazy
+analyzer from one whose model is currently resident.
+
+### Analyze a track
+
+```bash
+curl -X POST http://localhost:7860/v1/tracks/analyze \
+  -F 'audio=@song.flac' \
+  -F 'lyrics=[Verse]
+Hello world' \
+  -F 'track_id=local:42' \
+  -F 'content_identity=sha256:...' \
+  -F 'requested_representations=audio.global' \
+  -F 'requested_representations=lyrics.global'
+```
+
+Repeated fields, a comma-separated value, or a JSON array are accepted for
+`requested_representations`. With no field, both audio representations are calculated,
+plus `lyrics.global` only when non-empty lyrics are supplied. Explicitly requesting
+`lyrics.global` without lyrics is an error.
+
+Abbreviated response:
+
+```json
+{
+  "schema_version": "1",
+  "track": {"track_id": "local:42", "content_identity": "sha256:..."},
+  "representations": {
+    "audio.global": {
+      "representation": "audio.global",
+      "model_id": "laion/clap-htsat-unfused",
+      "model_version": "8fa0f1c6d0433df6e97c127f64b2a1d6c0dcda8a",
+      "preprocessing_version": "audio-clap-global-v1.<config-hash>",
+      "modality": "audio.global",
+      "dimension": 512,
+      "dtype": "float32",
+      "normalized": true,
+      "configuration": {},
+      "embedding": [0.01, -0.02],
+      "analysis": {"duration_ms": 213000, "windows_used": 4}
+    },
+    "audio.temporal": {
+      "representation": "audio.temporal",
+      "model_id": "laion/clap-htsat-unfused",
+      "model_version": "8fa0f1c6d0433df6e97c127f64b2a1d6c0dcda8a",
+      "preprocessing_version": "audio-clap-temporal-v1.<config-hash>",
+      "modality": "audio.temporal",
+      "dimension": 512,
+      "dtype": "float32",
+      "normalized": true,
+      "configuration": {},
+      "segments": [{"start_ms": 0, "end_ms": 10000, "embedding": [0.01]}],
+      "summary": {"number_of_segments": 1, "mean_adjacent_distance": 0.0,
+        "max_adjacent_distance": 0.0, "trajectory_variance": 0.0,
+        "largest_transition_index": null}
+    }
+  }
+}
+```
+
+Errors never contain stack traces:
+
+```json
+{"error": {"code": "AUDIO_DECODE_FAILED", "message": "The uploaded audio could not be decoded."}}
+```
+
+`GET /v1/status` reports the selected device and loaded/available representations without
+environment details. `GET /health` is a lightweight container probe.
+
+## Configuration
+
+| Variable | Default |
+|---|---:|
+| `OPENMUSIC_GLOBAL_WINDOW_SECONDS` | `10` |
+| `OPENMUSIC_GLOBAL_WINDOWS` | `4` |
+| `OPENMUSIC_TEMPORAL_WINDOW_SECONDS` | `10` |
+| `OPENMUSIC_TEMPORAL_HOP_SECONDS` | `10` |
+| `OPENMUSIC_TEMPORAL_MAX_SEGMENTS` | `24` |
+| `OPENMUSIC_CLAP_BATCH_SIZE` | `4` |
+| `OPENMUSIC_LYRICS_CHUNK_TOKENS` | `512` |
+| `OPENMUSIC_MAX_UPLOAD_BYTES` | `104857600` |
+| `OPENMUSIC_MAX_LYRICS_CHARACTERS` | `100000` |
+| `OPENMUSIC_MAX_AUDIO_SECONDS` | `1800` |
+| `OPENMUSIC_REQUEST_TIMEOUT_SECONDS` | `300` |
+
+Changing result-affecting representation configuration changes its generated
+`preprocessing_version` and benchmark cache key.
+
+## Benchmark tools
+
+Place optional UTF-8 lyrics beside audio using the same stem (`song.flac` + `song.txt`).
+
+```bash
+python tools/nearest_neighbors.py ./tracks \
+  --representation audio.global --top-k 10 --output reports/audio-global.json
+
+python tools/nearest_neighbors.py ./tracks \
+  --representation lyrics.global --output reports/lyrics-global.json
+
+python tools/nearest_neighbors.py ./tracks \
+  --representation audio.temporal --output reports/audio-temporal.json
+```
+
+The tool writes human-readable JSON and CSV. Its cache identity includes audio SHA-256,
+lyrics SHA-256, model ID, exact revision, preprocessing version, representation and full
+configuration. Unchanged embeddings are reused. Temporal reports contain two explicitly
+experimental metrics: normalized-time interpolation with mean cosine, and classic DTW
+over cosine distance. Neither is treated as the final similarity definition.
+
+Measure the current machine with a representative track:
+
+```bash
+python tools/profile_analysis.py song.flac --lyrics song.txt
+```
+
+It reports total model startup, each representation latency, peak process RSS and peak
+CUDA allocation when CUDA is available.
+
+## Tests
+
+```bash
+pip install -r requirements-dev.txt
+pytest
+```
+
+Tests use deterministic fake encoders and do not download model weights. They cover API
+selection/errors, global aggregation, temporal segmentation/limits, request-scoped
+decode reuse, English/Russian structured lyrics and long chunking, model registry
+consistency, inference failure sanitization, and both temporal similarity baselines.
+
+## Extension points and limitations
+
+A future MERT analyzer should implement the audio analyzer protocol, request its sample
+rate from `AnalysisContext`, declare a separate registry item and use explicit names such
+as `audio.mert.global` and `audio.mert.temporal`. It must not be averaged with CLAP. It is
+not enabled in V1 to avoid adding a second large model/dependency path before comparative
+benchmarks exist.
+
+- CLAP temporal vectors are a trajectory through CLAP's audio space, not a trained music
+  emotion representation.
+- BGE-M3 semantic lyrics similarity is not lyrics emotional similarity.
+- Embeddings from CLAP and BGE-M3 (and future MERT models) are different spaces and must
+  never be compared directly with cosine or added together.
+- Temporal interpolation and DTW are baselines for experiments, not validated music
+  structure metrics.
+- Exact revisions and preprocessing config make inference reproducible, but very small
+  floating-point differences can still occur across PyTorch/device/hardware versions.

deploy.sh

@@ -16,11 +16,11 @@ echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.
 apt-get update
 apt-get install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin
 
-echo "=== Клонирование / копирование проекта ==="
-mkdir -p /opt/audio-api/tracks
-cd /opt/audio-api
+echo "=== Подготовка директории OpenMusic Analysis API ==="
+mkdir -p /opt/openmusic-analysis
+cd /opt/openmusic-analysis
 
-# Скопируйте сюда ваши файлы: main.py, Dockerfile, docker-compose.yml, requirements.txt
+# Скопируйте в эту директорию весь репозиторий перед следующим шагом.
 
 echo "=== Сборка и запуск ==="
 docker compose build

docker-compose.yml

@@ -1,18 +1,19 @@
 services:
-  audio-api:
+  openmusic-analysis:
     build: .
-    container_name: audio-embedding-api
+    container_name: openmusic-analysis-api
     restart: unless-stopped        # автоперезапуск при падении
     ports:
-      - "8000:8000"
+      - "8000:7860"
     volumes:
-      - ./tracks:/app/tracks       # треки хранятся снаружи контейнера
-      - hf_cache:/root/.cache/huggingface  # кеш моделей между перезапусками
+      - hf_cache:/root/.cache/huggingface
     environment:
       - PYTHONUNBUFFERED=1
-      - TRANSFORMERS_CACHE=/root/.cache/huggingface
+      - HF_HOME=/root/.cache/huggingface
+      - OPENMUSIC_DEVICE=auto
+      - OPENMUSIC_INFERENCE_CONCURRENCY=1
     healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
+      test: ["CMD", "curl", "-f", "http://localhost:7860/health"]
       interval: 30s
       timeout: 10s
       retries: 3
@@ -20,7 +21,7 @@ services:
     deploy:
       resources:
         limits:
-          memory: 6G               # CLAP занимает ~2GB, оставляем запас
+          memory: 10G
     logging:
       driver: "json-file"
       options:
@@ -28,4 +29,4 @@ services:
         max-file: "3"
 
 volumes:
-  hf_cache:                        # постоянный том для моделей HuggingFace
+  hf_cache:

main.py

@@ -1,305 +1,17 @@
-import os
-import glob
-import shutil
-import tempfile
 import logging
-import traceback
-from contextlib import asynccontextmanager
 
-import numpy as np
-import torch
-from fastapi import FastAPI, File, UploadFile, HTTPException, Request
-from fastapi.responses import JSONResponse
-from sklearn.metrics.pairwise import cosine_similarity
+from openmusic_analysis.api import create_app
 
-# ─── Логирование ──────────────────────────────────────────────────────────────
 
 logging.basicConfig(
     level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(message)s",
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
 )
-log = logging.getLogger(__name__)
 
-# ─── ffmpeg ───────────────────────────────────────────────────────────────────
+app = create_app()
 
-try:
-    import imageio_ffmpeg
-    from pydub import AudioSegment
-
-    _ffmpeg_exe = imageio_ffmpeg.get_ffmpeg_exe()
-    AudioSegment.converter = _ffmpeg_exe
-    os.environ["PATH"] = os.path.dirname(_ffmpeg_exe) + os.pathsep + os.environ.get("PATH", "")
-    log.info(f"ffmpeg: {_ffmpeg_exe}")
-except Exception as e:
-    log.warning(f"ffmpeg не найден: {e}. MP3 могут не читаться.")
-
-# ─── Загрузка аудио ───────────────────────────────────────────────────────────
-
-def load_audio(file_path: str, sr: int = 48000) -> np.ndarray:
-    """
-    Надёжная загрузка аудио: сначала pydub (поддерживает MP3),
-    при ошибке — librosa.
-    """
-    try:
-        from pydub import AudioSegment
-        seg = AudioSegment.from_file(file_path)
-        seg = seg.set_frame_rate(sr).set_channels(1)
-        samples = np.array(seg.get_array_of_samples(), dtype=np.float32)
-        max_val = float(np.iinfo(seg.array_type).max)
-        return samples / (max_val + 1e-8)
-    except Exception as e1:
-        log.warning(f"pydub не смог прочитать {file_path}: {e1}. Пробуем librosa...")
-        try:
-            import librosa
-            y, _ = librosa.load(file_path, sr=sr, mono=True)
-            return y.astype(np.float32)
-        except Exception as e2:
-            raise RuntimeError(f"Не удалось прочитать аудио: pydub={e1} | librosa={e2}")
-
-# ─── Модели ───────────────────────────────────────────────────────────────────
-
-from transformers import (
-    ClapAudioModelWithProjection,
-    ClapTextModelWithProjection,
-    ClapProcessor,
-    AutoTokenizer,
-)
-
-log.info("Загрузка CLAP аудио модели...")
-clap_model = ClapAudioModelWithProjection.from_pretrained("laion/clap-htsat-unfused")
-clap_processor = ClapProcessor.from_pretrained("laion/clap-htsat-unfused")
-clap_model.eval()
-log.info("CLAP аудио готов")
-
-log.info("Загрузка CLAP текстовой модели...")
-text_model = ClapTextModelWithProjection.from_pretrained("laion/clap-htsat-unfused")
-tokenizer = AutoTokenizer.from_pretrained("laion/clap-htsat-unfused")
-text_model.eval()
-log.info("CLAP текст готов")
-
-# ─── Embedding ────────────────────────────────────────────────────────────────
-
-def extract_audio_features(file_path: str) -> np.ndarray:
-    """
-    CLAP embedding: анализирует 3 фрагмента трека (10%, 40%, 70%)
-    и возвращает усреднённый нормализованный вектор 512-dim.
-    """
-    audio_np = load_audio(file_path, sr=48000)
-    sr = 48000
-    chunk_size = 10 * sr
-    positions = [0.1, 0.4, 0.7]
-    chunks = []
-
-    for pos in positions:
-        start = int(pos * len(audio_np))
-        end = start + chunk_size
-        if end > len(audio_np):
-            end = len(audio_np)
-            start = max(0, end - chunk_size)
-        chunk = audio_np[start:end]
-        if len(chunk) > sr * 2:
-            chunks.append(chunk)
-
-    # Если трек короткий — берём весь целиком
-    if not chunks:
-        if len(audio_np) < sr * 1:
-            raise ValueError("Трек слишком короткий (менее 1 секунды)")
-        chunks = [audio_np]
-
-    embeddings = []
-    for chunk in chunks:
-        try:
-            inputs = clap_processor(
-    audios=chunk,
-    sampling_rate=48000,
-    return_tensors="pt"
-)
-            with torch.no_grad():
-                emb = clap_model(**inputs).audio_embeds.squeeze().numpy()
-            embeddings.append(emb)
-        except Exception as e:
-            log.warning(f"Ошибка на фрагменте: {e}")
-            continue
-
-    if not embeddings:
-        raise RuntimeError("Не удалось получить ни одного embedding")
-
-    final_emb = np.mean(embeddings, axis=0)
-    norm = np.linalg.norm(final_emb)
-    if norm > 0:
-        final_emb = final_emb / norm
-    return final_emb.astype(np.float32)
-
-
-def generate_caption(file_path: str) -> str:
-    """Описание трека на основе акустических характеристик."""
-    try:
-        import librosa
-        y = load_audio(file_path, sr=22050)
-
-        tempo, _ = librosa.beat.beat_track(y=y, sr=22050)
-        tempo = float(np.atleast_1d(tempo)[0])
-        rms = float(np.mean(librosa.feature.rms(y=y)))
-        zcr = float(np.mean(librosa.feature.zero_crossing_rate(y)))
-        chroma = librosa.feature.chroma_stft(y=y, sr=22050)
-        chroma_mean = np.mean(chroma, axis=1)
-        dominant_note = ["C","C#","D","D#","E","F","F#","G","G#","A","A#","B"][int(np.argmax(chroma_mean))]
-        harmonic, _ = librosa.effects.hpss(y)
-        harmonic_ratio = float(np.mean(np.abs(harmonic))) / (float(np.mean(np.abs(y))) + 1e-8)
-
-        tempo_desc   = "slow tempo" if tempo < 70 else "moderate tempo" if tempo < 100 else "upbeat tempo" if tempo < 140 else "fast tempo"
-        energy_desc  = "very quiet and calm" if rms < 0.02 else "soft and gentle" if rms < 0.05 else "moderate energy" if rms < 0.1 else "high energy and loud"
-        texture_desc = "with strong percussion and rhythmic elements" if zcr > 0.1 else "with moderate rhythmic elements" if zcr > 0.05 else "smooth and melodic without heavy percussion"
-        harmony_desc = "rich harmonic content" if harmonic_ratio > 0.6 else "balanced mix of harmony and rhythm" if harmonic_ratio > 0.3 else "rhythm-driven with minimal harmony"
-
-        return (
-            f"This music has a {tempo_desc} around {int(tempo)} BPM. "
-            f"It sounds {energy_desc} {texture_desc}. "
-            f"The track features {harmony_desc} in the key of {dominant_note}."
-        )
-    except Exception as e:
-        log.warning(f"generate_caption failed: {e}")
-        return "music audio track"  # fallback — не падаем
-
-
-def text_to_embedding(text: str) -> np.ndarray:
-    """CLAP embedding из текста."""
-    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True, max_length=77)
-    with torch.no_grad():
-        emb = text_model(**inputs).text_embeds.squeeze().numpy()
-    norm = np.linalg.norm(emb)
-    return (emb / norm if norm > 0 else emb).astype(np.float32)
-
-
-def save_temp_file(file: UploadFile) -> str:
-    """Сохраняет загруженный файл во временное место."""
-    suffix = os.path.splitext(file.filename or "audio.mp3")[1].lower() or ".mp3"
-    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
-        shutil.copyfileobj(file.file, tmp)
-        return tmp.name
-
-
-def safe_remove(path: str):
-    try:
-        os.remove(path)
-    except Exception:
-        pass
-
-# ─── Индексирование треков при старте ────────────────────────────────────────
-
-TRACK_FILES = glob.glob("tracks/**/*.*", recursive=True)
-TRACK_FILES = [f for f in TRACK_FILES if f.lower().endswith((".mp3", ".wav", ".flac", ".ogg", ".m4a"))]
-
-track_features: list[np.ndarray] = []
-available_tracks: list[str] = []
-
-for path in TRACK_FILES:
-    if not os.path.exists(path):
-        log.warning(f"[skip] не найден: {path}")
-        continue
-    try:
-        feat = extract_audio_features(path)
-        track_features.append(feat)
-        available_tracks.append(path)
-        log.info(f"[ok] проиндексирован: {path}")
-    except Exception as e:
-        log.error(f"[error] {path}: {e}")
-
-track_features_matrix = np.array(track_features) if track_features else np.empty((0, 512))
-log.info(f"Индекс готов: {len(available_tracks)} треков")
-
-# ─── FastAPI ──────────────────────────────────────────────────────────────────
-
-app = FastAPI(title="Audio Embedding API")
-
-# Глобальный обработчик всех необработанных исключений
-@app.exception_handler(Exception)
-async def global_exception_handler(request: Request, exc: Exception):
-    log.error(f"Необработанная ошибка [{request.url}]: {exc}\n{traceback.format_exc()}")
-    return JSONResponse(
-        status_code=500,
-        content={"detail": f"Внутренняя ошибка сервера: {type(exc).__name__}: {exc}"},
-    )
-
-# ─── Роуты ──────────────────���────────────────────────────────────────────────
-
-@app.post("/embedding/smart")
-async def get_smart_embedding(file: UploadFile = File(...)):
-    """
-    Возвращает embedding: 60% аудио CLAP + 40% текстового описания трека.
-    Также возвращает caption с описанием характеристик трека.
-    """
-    if not file.filename:
-        raise HTTPException(status_code=400, detail="Имя файла не указано")
-
-    tmp_path = save_temp_file(file)
-    try:
-        audio_emb = extract_audio_features(tmp_path)
-        caption   = generate_caption(tmp_path)
-        text_emb  = text_to_embedding(caption)
-
-        combined = 0.6 * audio_emb + 0.4 * text_emb
-        norm = np.linalg.norm(combined)
-        combined = (combined / norm if norm > 0 else combined).astype(np.float32)
-
-        log.info(f"[smart] {file.filename} → caption: {caption}")
-        return {"caption": caption, "embedding": combined.tolist()}
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        log.error(f"[smart] {file.filename}: {e}\n{traceback.format_exc()}")
-        raise HTTPException(status_code=500, detail=f"Ошибка обработки: {type(e).__name__}: {e}")
-    finally:
-        safe_remove(tmp_path)
-
-
-@app.post("/find_similar_tracks/")
-async def find_similar_tracks_route(file: UploadFile = File(...)):
-    """
-    Находит похожие треки из локальной библиотеки (папка tracks/).
-    """
-    if len(available_tracks) == 0:
-        raise HTTPException(status_code=503, detail="Библиотека треков пуста — добавьте файлы в папку tracks/")
-
-    tmp_path = save_temp_file(file)
-    try:
-        query_vec = extract_audio_features(tmp_path)
-        sims = cosine_similarity([query_vec], track_features_matrix)[0]
-        indices = np.argsort(sims)[::-1]
-
-        results = [
-            {
-                "rank": int(rank + 1),
-                "track": available_tracks[i],
-                "similarity": round(float(sims[i]), 4),
-            }
-            for rank, i in enumerate(indices)
-        ]
-        log.info(f"[search] {file.filename} → top={results[0]['track']} ({results[0]['similarity']})")
-        return {"query": file.filename, "results": results}
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        log.error(f"[search] {file.filename}: {e}\n{traceback.format_exc()}")
-        raise HTTPException(status_code=500, detail=f"Ошибка поиска: {type(e).__name__}: {e}")
-    finally:
-        safe_remove(tmp_path)
-
-
-@app.get("/health")
-async def health():
-    """Проверка состояния сервера."""
-    return {
-        "status": "ok",
-        "indexed_tracks": len(available_tracks),
-        "models": ["clap-audio", "clap-text"],
-    }
-
-
-# ─── Запуск ───────────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
     import uvicorn
+
     uvicorn.run(app, host="0.0.0.0", port=7860)

openmusic_analysis/__init__.py

@@ -0,0 +1,5 @@
+"""OpenMusic versioned music-analysis backend."""
+
+from .application import MusicAnalysisService, build_service
+
+__all__ = ["MusicAnalysisService", "build_service"]

openmusic_analysis/analyzers/__init__.py

@@ -0,0 +1,11 @@
+from .clap import ClapAudioEncoder, ClapGlobalAudioAnalyzer, ClapTemporalAudioAnalyzer
+from .lyrics import BGEM3LyricsAnalyzer, BGEM3TextEncoder, LyricsPreprocessor
+
+__all__ = [
+    "BGEM3LyricsAnalyzer",
+    "BGEM3TextEncoder",
+    "ClapAudioEncoder",
+    "ClapGlobalAudioAnalyzer",
+    "ClapTemporalAudioAnalyzer",
+    "LyricsPreprocessor",
+]

openmusic_analysis/analyzers/clap.py

@@ -0,0 +1,303 @@
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Protocol
+
+import numpy as np
+
+from openmusic_analysis.audio import AnalysisContext
+from openmusic_analysis.audio.windows import (
+    fixed_window,
+    global_window_starts,
+    temporal_window_starts,
+)
+from openmusic_analysis.domain import (
+    GlobalEmbeddingResult,
+    ModelMetadata,
+    TemporalEmbeddingResult,
+    TemporalSegment,
+    TemporalSummary,
+)
+from openmusic_analysis.errors import AnalysisError, ModelInferenceError
+from openmusic_analysis.runtime import InferenceGate
+from openmusic_analysis.settings import (
+    CLAP_MODEL_ID,
+    CLAP_MODEL_REVISION,
+    GlobalAudioConfig,
+    TemporalAudioConfig,
+    config_dict,
+    preprocessing_version,
+)
+
+from .math import finite_float_list, l2_normalize, normalize_rows
+
+
+log = logging.getLogger(__name__)
+
+
+class AudioEmbeddingEncoder(Protocol):
+    dimension: int
+    loaded: bool
+
+    async def encode(self, windows: list[np.ndarray]) -> np.ndarray: ...
+
+    async def ready(self) -> None: ...
+
+
+class ClapAudioEncoder:
+    """One shared, lazy CLAP audio encoder used by global and temporal analyzers."""
+
+    dimension = 512
+    sample_rate = 48_000
+
+    def __init__(self, device: str, gate: InferenceGate, batch_size: int = 4) -> None:
+        self.device = device
+        self.gate = gate
+        self.batch_size = batch_size
+        self.loaded = False
+        self._load_lock = threading.Lock()
+        self._processor = None
+        self._model = None
+
+    async def ready(self) -> None:
+        await self.gate.run(self._ensure_loaded)
+
+    def _ensure_loaded(self) -> None:
+        if self.loaded:
+            return
+        with self._load_lock:
+            if self.loaded:
+                return
+            import torch
+            from transformers import ClapAudioModelWithProjection, ClapProcessor
+
+            log.info("Loading %s at %s on %s", CLAP_MODEL_ID, CLAP_MODEL_REVISION, self.device)
+            processor = ClapProcessor.from_pretrained(
+                CLAP_MODEL_ID, revision=CLAP_MODEL_REVISION
+            )
+            model = ClapAudioModelWithProjection.from_pretrained(
+                CLAP_MODEL_ID, revision=CLAP_MODEL_REVISION
+            )
+            if int(model.config.projection_dim) != self.dimension:
+                raise RuntimeError("Pinned CLAP projection dimension does not match registry")
+            if int(processor.feature_extractor.sampling_rate) != self.sample_rate:
+                raise RuntimeError("Pinned CLAP sample rate does not match preprocessing")
+            model.to(torch.device(self.device))
+            model.eval()
+            self._processor = processor
+            self._model = model
+            self.loaded = True
+
+    async def encode(self, windows: list[np.ndarray]) -> np.ndarray:
+        return await self.gate.run(self._encode_sync, windows)
+
+    def _encode_sync(self, windows: list[np.ndarray]) -> np.ndarray:
+        self._ensure_loaded()
+        import torch
+
+        assert self._processor is not None and self._model is not None
+        encoder_windows, groups = _clap_encoder_windows(
+            windows, self.sample_rate * 10
+        )
+        outputs: list[np.ndarray] = []
+        for index in range(0, len(encoder_windows), self.batch_size):
+            inputs = self._processor(
+                audios=encoder_windows[index : index + self.batch_size],
+                sampling_rate=self.sample_rate,
+                max_length=self.sample_rate * 10,
+                truncation="rand_trunc",
+                padding="repeatpad",
+                return_tensors="pt",
+            )
+            inputs = {name: value.to(self.device) for name, value in inputs.items()}
+            with torch.inference_mode():
+                output = self._model(**inputs).audio_embeds
+            outputs.append(output.detach().to(dtype=torch.float32, device="cpu").numpy())
+        raw = np.concatenate(outputs, axis=0)
+        normalized = normalize_rows(raw)
+        aggregated = [
+            l2_normalize(np.mean(normalized[start:end], axis=0, dtype=np.float64))
+            for start, end in groups
+        ]
+        return np.stack(aggregated)
+
+
+class ClapGlobalAudioAnalyzer:
+    representation = "audio.global"
+    input_kind = "audio"
+
+    def __init__(self, encoder: AudioEmbeddingEncoder, config: GlobalAudioConfig) -> None:
+        self.encoder = encoder
+        self.config = config
+
+    @property
+    def metadata(self) -> ModelMetadata:
+        return ModelMetadata(
+            representation=self.representation,
+            model_id=CLAP_MODEL_ID,
+            model_version=CLAP_MODEL_REVISION,
+            modality=self.representation,
+            dimension=self.encoder.dimension,
+            dtype="float32",
+            normalized=True,
+            preprocessing_version=preprocessing_version("audio-clap-global-v1", self.config),
+            license="Apache-2.0",
+            configuration=config_dict(self.config),
+            loaded=self.encoder.loaded,
+        )
+
+    async def analyze(self, context: AnalysisContext) -> GlobalEmbeddingResult:
+        try:
+            waveform = await context.waveform(self.config.sample_rate)
+            duration_seconds = waveform.size / self.config.sample_rate
+            if duration_seconds < self.config.minimum_audio_seconds:
+                raise AnalysisError(
+                    "AUDIO_TOO_SHORT",
+                    f"Audio must be at least {self.config.minimum_audio_seconds:g} second(s).",
+                    status_code=422,
+                )
+            window_samples = int(round(self.config.window_seconds * self.config.sample_rate))
+            starts = global_window_starts(
+                waveform.size,
+                window_samples,
+                self.config.target_windows,
+                self.config.coverage_start,
+                self.config.coverage_end,
+            )
+            windows = [fixed_window(waveform, start, window_samples) for start in starts]
+            embeddings = normalize_rows(await self.encoder.encode(windows))
+            aggregate = l2_normalize(np.mean(embeddings, axis=0, dtype=np.float64))
+            metadata = self.metadata
+            return GlobalEmbeddingResult(
+                **metadata.result_metadata(),
+                embedding=finite_float_list(aggregate),
+                analysis={
+                    "duration_ms": int(round(duration_seconds * 1000)),
+                    "window_start_ms": [
+                        int(round(start * 1000 / self.config.sample_rate)) for start in starts
+                    ],
+                    "windows_used": len(starts),
+                },
+            )
+        except AnalysisError:
+            raise
+        except Exception as exc:
+            log.exception("Global CLAP inference failed")
+            raise ModelInferenceError(self.representation) from exc
+
+
+class ClapTemporalAudioAnalyzer:
+    representation = "audio.temporal"
+    input_kind = "audio"
+
+    def __init__(self, encoder: AudioEmbeddingEncoder, config: TemporalAudioConfig) -> None:
+        self.encoder = encoder
+        self.config = config
+
+    @property
+    def metadata(self) -> ModelMetadata:
+        return ModelMetadata(
+            representation=self.representation,
+            model_id=CLAP_MODEL_ID,
+            model_version=CLAP_MODEL_REVISION,
+            modality=self.representation,
+            dimension=self.encoder.dimension,
+            dtype="float32",
+            normalized=True,
+            preprocessing_version=preprocessing_version(
+                "audio-clap-temporal-v1", self.config
+            ),
+            license="Apache-2.0",
+            configuration=config_dict(self.config),
+            loaded=self.encoder.loaded,
+        )
+
+    async def analyze(self, context: AnalysisContext) -> TemporalEmbeddingResult:
+        try:
+            waveform = await context.waveform(self.config.sample_rate)
+            duration_seconds = waveform.size / self.config.sample_rate
+            if duration_seconds < self.config.minimum_audio_seconds:
+                raise AnalysisError(
+                    "AUDIO_TOO_SHORT",
+                    f"Audio must be at least {self.config.minimum_audio_seconds:g} second(s).",
+                    status_code=422,
+                )
+            window_samples = int(round(self.config.window_seconds * self.config.sample_rate))
+            hop_samples = int(round(self.config.hop_seconds * self.config.sample_rate))
+            starts = temporal_window_starts(
+                waveform.size,
+                window_samples,
+                hop_samples,
+                self.config.max_segments,
+            )
+            windows = [fixed_window(waveform, start, window_samples) for start in starts]
+            embeddings = normalize_rows(await self.encoder.encode(windows))
+            segments = [
+                TemporalSegment(
+                    start_ms=int(round(start * 1000 / self.config.sample_rate)),
+                    end_ms=int(
+                        round(
+                            min(start + window_samples, waveform.size)
+                            * 1000
+                            / self.config.sample_rate
+                        )
+                    ),
+                    embedding=finite_float_list(embedding),
+                )
+                for start, embedding in zip(starts, embeddings, strict=True)
+            ]
+            summary = _trajectory_summary(embeddings)
+            metadata = self.metadata
+            return TemporalEmbeddingResult(
+                **metadata.result_metadata(), segments=segments, summary=summary
+            )
+        except AnalysisError:
+            raise
+        except Exception as exc:
+            log.exception("Temporal CLAP inference failed")
+            raise ModelInferenceError(self.representation) from exc
+
+
+def _trajectory_summary(embeddings: np.ndarray) -> TemporalSummary:
+    if embeddings.shape[0] > 1:
+        adjacent = 1.0 - np.sum(embeddings[:-1] * embeddings[1:], axis=1)
+        adjacent = np.clip(adjacent, 0.0, 2.0)
+        largest_index: int | None = int(np.argmax(adjacent)) + 1
+        mean_adjacent = float(np.mean(adjacent))
+        max_adjacent = float(np.max(adjacent))
+    else:
+        largest_index = None
+        mean_adjacent = 0.0
+        max_adjacent = 0.0
+    centroid = np.mean(embeddings, axis=0, dtype=np.float64)
+    variance = float(np.mean(np.sum((embeddings - centroid) ** 2, axis=1)))
+    return TemporalSummary(
+        number_of_segments=int(embeddings.shape[0]),
+        mean_adjacent_distance=mean_adjacent,
+        max_adjacent_distance=max_adjacent,
+        trajectory_variance=variance,
+        largest_transition_index=largest_index,
+    )
+
+
+def _clap_encoder_windows(
+    windows: list[np.ndarray], encoder_samples: int
+) -> tuple[list[np.ndarray], list[tuple[int, int]]]:
+    """Map every analysis window to deterministic fixed-size CLAP inputs."""
+    encoded: list[np.ndarray] = []
+    groups: list[tuple[int, int]] = []
+    for window in windows:
+        group_start = len(encoded)
+        if window.size <= encoder_samples:
+            encoded.append(fixed_window(window, 0, encoder_samples))
+        else:
+            count = max(2, int(np.ceil(window.size / encoder_samples)))
+            last_start = window.size - encoder_samples
+            starts = np.linspace(0, last_start, count, dtype=np.float64)
+            encoded.extend(
+                fixed_window(window, int(round(float(start))), encoder_samples)
+                for start in starts
+            )
+        groups.append((group_start, len(encoded)))
+    return encoded, groups

openmusic_analysis/analyzers/interfaces.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import Protocol
+
+from openmusic_analysis.audio import AnalysisContext
+from openmusic_analysis.domain import (
+    GlobalEmbeddingResult,
+    ModelMetadata,
+    TemporalEmbeddingResult,
+)
+
+
+class GlobalAudioAnalyzer(Protocol):
+    input_kind: str
+
+    @property
+    def metadata(self) -> ModelMetadata: ...
+
+    async def analyze(self, context: AnalysisContext) -> GlobalEmbeddingResult: ...
+
+
+class TemporalAudioAnalyzer(Protocol):
+    input_kind: str
+
+    @property
+    def metadata(self) -> ModelMetadata: ...
+
+    async def analyze(self, context: AnalysisContext) -> TemporalEmbeddingResult: ...
+
+
+class LyricsAnalyzer(Protocol):
+    input_kind: str
+
+    @property
+    def metadata(self) -> ModelMetadata: ...
+
+    async def analyze(self, lyrics: str) -> GlobalEmbeddingResult: ...

openmusic_analysis/analyzers/lyrics.py

@@ -0,0 +1,331 @@
+from __future__ import annotations
+
+import logging
+import re
+import threading
+import unicodedata
+from dataclasses import dataclass
+from typing import Protocol
+
+import numpy as np
+
+from openmusic_analysis.domain import GlobalEmbeddingResult, ModelMetadata
+from openmusic_analysis.errors import AnalysisError, ModelInferenceError
+from openmusic_analysis.runtime import InferenceGate
+from openmusic_analysis.settings import (
+    BGE_MODEL_ID,
+    BGE_MODEL_REVISION,
+    LyricsConfig,
+    config_dict,
+    preprocessing_version,
+)
+
+from .math import finite_float_list, l2_normalize, normalize_rows
+
+
+log = logging.getLogger(__name__)
+
+_SECTION_MARKER = re.compile(
+    r"^\s*(?:\[|\()?\s*(?P<label>"
+    r"verse|chorus|pre[- ]?chorus|bridge|intro|outro|hook|refrain|interlude|"
+    r"куплет|припев|предприпев|бридж|вступление|проигрыш|финал"
+    r")(?:\s+[^\]\)]*)?\s*(?:\]|\))?\s*:?\s*$",
+    re.IGNORECASE,
+)
+_LRC_METADATA = re.compile(r"^\s*\[(?:ar|al|ti|by|offset|re|ve|length):.*\]\s*$", re.I)
+_LRC_TIMESTAMP = re.compile(r"^\s*(?:\[\d{1,3}:\d{2}(?:[.:]\d{1,3})?\])+\s*")
+_PURE_URL = re.compile(r"^\s*https?://\S+\s*$", re.I)
+
+
+@dataclass(frozen=True)
+class LyricsSection:
+    index: int
+    label: str | None
+    text: str
+
+
+@dataclass(frozen=True)
+class PreparedLyrics:
+    normalized_text: str
+    sections: list[LyricsSection]
+
+
+@dataclass(frozen=True)
+class LyricsChunk:
+    section_index: int
+    label: str | None
+    text: str
+    token_count: int
+
+
+class TextEmbeddingEncoder(Protocol):
+    dimension: int
+    loaded: bool
+
+    async def ready(self) -> None: ...
+
+    def count_tokens(self, text: str) -> int: ...
+
+    def split_tokens(self, text: str, max_tokens: int) -> list[str]: ...
+
+    async def encode(self, texts: list[str], batch_size: int) -> np.ndarray: ...
+
+
+class LyricsPreprocessor:
+    """Conservative cleanup that preserves punctuation, lines and repetitions."""
+
+    def normalize(self, lyrics: str) -> str:
+        value = unicodedata.normalize("NFKC", lyrics)
+        value = value.replace("\r\n", "\n").replace("\r", "\n").replace("\ufeff", "")
+        value = "".join(
+            character
+            for character in value
+            if character in {"\n", "\t"} or unicodedata.category(character) != "Cc"
+        )
+        cleaned: list[str] = []
+        for raw_line in value.split("\n"):
+            line = raw_line.rstrip()
+            if _LRC_METADATA.match(line) or _PURE_URL.match(line):
+                continue
+            line = _LRC_TIMESTAMP.sub("", line)
+            cleaned.append(line)
+        value = "\n".join(cleaned).strip()
+        value = re.sub(r"\n{3,}", "\n\n", value)
+        return value
+
+    def prepare(self, lyrics: str) -> PreparedLyrics:
+        normalized = self.normalize(lyrics)
+        if not normalized.strip():
+            raise AnalysisError(
+                "INVALID_LYRICS", "Lyrics must contain non-whitespace text.", status_code=422
+            )
+        sections: list[LyricsSection] = []
+        current_label: str | None = None
+        current_lines: list[str] = []
+
+        def flush() -> None:
+            nonlocal current_lines
+            text = "\n".join(current_lines).strip()
+            if text:
+                sections.append(
+                    LyricsSection(index=len(sections), label=current_label, text=text)
+                )
+            current_lines = []
+
+        for line in normalized.split("\n"):
+            marker = _SECTION_MARKER.match(line)
+            if marker:
+                flush()
+                current_label = marker.group("label").strip()
+                continue
+            if not line.strip() and current_lines:
+                flush()
+                current_label = None
+                continue
+            if line.strip():
+                current_lines.append(line)
+        flush()
+        if not sections:
+            sections = [LyricsSection(index=0, label=None, text=normalized)]
+        return PreparedLyrics(normalized_text=normalized, sections=sections)
+
+    def chunks(
+        self,
+        prepared: PreparedLyrics,
+        encoder: TextEmbeddingEncoder,
+        max_tokens: int,
+    ) -> list[LyricsChunk]:
+        chunks: list[LyricsChunk] = []
+        for section in prepared.sections:
+            prefix = f"[{section.label}]\n" if section.label else ""
+            candidate = prefix + section.text
+            if encoder.count_tokens(candidate) <= max_tokens:
+                chunks.append(
+                    LyricsChunk(
+                        section_index=section.index,
+                        label=section.label,
+                        text=candidate,
+                        token_count=max(1, encoder.count_tokens(candidate)),
+                    )
+                )
+                continue
+            for piece in _split_section_lines(section.text, encoder, max_tokens, prefix):
+                count = max(1, encoder.count_tokens(piece))
+                chunks.append(
+                    LyricsChunk(
+                        section_index=section.index,
+                        label=section.label,
+                        text=piece,
+                        token_count=count,
+                    )
+                )
+        return chunks
+
+
+def _split_section_lines(
+    text: str,
+    encoder: TextEmbeddingEncoder,
+    max_tokens: int,
+    prefix: str,
+) -> list[str]:
+    pieces: list[str] = []
+    current: list[str] = []
+    for line in text.split("\n"):
+        candidate = prefix + "\n".join([*current, line])
+        if current and encoder.count_tokens(candidate) > max_tokens:
+            pieces.append(prefix + "\n".join(current))
+            current = []
+        line_candidate = prefix + line
+        if encoder.count_tokens(line_candidate) > max_tokens:
+            prefix_content_tokens = max(0, encoder.count_tokens(prefix) - 2) if prefix else 0
+            piece_limit = max(4, max_tokens - prefix_content_tokens)
+            pieces.extend(
+                prefix + piece
+                for piece in encoder.split_tokens(line, piece_limit)
+                if piece.strip()
+            )
+        else:
+            current.append(line)
+    if current:
+        pieces.append(prefix + "\n".join(current))
+    return [piece for piece in pieces if piece.strip()]
+
+
+class BGEM3TextEncoder:
+    dimension = 1024
+
+    def __init__(self, device: str, gate: InferenceGate) -> None:
+        self.device = device
+        self.gate = gate
+        self.loaded = False
+        self._load_lock = threading.Lock()
+        self._tokenizer = None
+        self._model = None
+
+    async def ready(self) -> None:
+        await self.gate.run(self._ensure_loaded)
+
+    def _ensure_loaded(self) -> None:
+        if self.loaded:
+            return
+        with self._load_lock:
+            if self.loaded:
+                return
+            import torch
+            from transformers import AutoModel, AutoTokenizer
+
+            log.info("Loading %s at %s on %s", BGE_MODEL_ID, BGE_MODEL_REVISION, self.device)
+            tokenizer = AutoTokenizer.from_pretrained(
+                BGE_MODEL_ID, revision=BGE_MODEL_REVISION
+            )
+            model = AutoModel.from_pretrained(BGE_MODEL_ID, revision=BGE_MODEL_REVISION)
+            if int(model.config.hidden_size) != self.dimension:
+                raise RuntimeError("Pinned BGE-M3 dimension does not match registry")
+            model.to(torch.device(self.device))
+            model.eval()
+            self._tokenizer = tokenizer
+            self._model = model
+            self.loaded = True
+
+    def count_tokens(self, text: str) -> int:
+        self._ensure_loaded()
+        assert self._tokenizer is not None
+        return len(self._tokenizer.encode(text, add_special_tokens=True))
+
+    def split_tokens(self, text: str, max_tokens: int) -> list[str]:
+        self._ensure_loaded()
+        assert self._tokenizer is not None
+        usable_tokens = max(1, max_tokens - 2)
+        token_ids = self._tokenizer.encode(text, add_special_tokens=False)
+        return [
+            self._tokenizer.decode(
+                token_ids[index : index + usable_tokens], skip_special_tokens=True
+            )
+            for index in range(0, len(token_ids), usable_tokens)
+        ]
+
+    async def encode(self, texts: list[str], batch_size: int) -> np.ndarray:
+        return await self.gate.run(self._encode_sync, texts, batch_size)
+
+    def _encode_sync(self, texts: list[str], batch_size: int) -> np.ndarray:
+        self._ensure_loaded()
+        import torch
+
+        assert self._tokenizer is not None and self._model is not None
+        outputs: list[np.ndarray] = []
+        for index in range(0, len(texts), batch_size):
+            batch = texts[index : index + batch_size]
+            tokens = self._tokenizer(
+                batch,
+                padding=True,
+                truncation=True,
+                max_length=8192,
+                return_tensors="pt",
+            )
+            tokens = {name: value.to(self.device) for name, value in tokens.items()}
+            with torch.inference_mode():
+                hidden = self._model(**tokens).last_hidden_state[:, 0]
+                hidden = torch.nn.functional.normalize(hidden, p=2, dim=1)
+            outputs.append(hidden.detach().to(dtype=torch.float32, device="cpu").numpy())
+        return np.concatenate(outputs, axis=0)
+
+
+class BGEM3LyricsAnalyzer:
+    representation = "lyrics.global"
+    input_kind = "lyrics"
+
+    def __init__(
+        self,
+        encoder: TextEmbeddingEncoder,
+        preprocessor: LyricsPreprocessor,
+        config: LyricsConfig,
+    ) -> None:
+        self.encoder = encoder
+        self.preprocessor = preprocessor
+        self.config = config
+
+    @property
+    def metadata(self) -> ModelMetadata:
+        return ModelMetadata(
+            representation=self.representation,
+            model_id=BGE_MODEL_ID,
+            model_version=BGE_MODEL_REVISION,
+            modality=self.representation,
+            dimension=self.encoder.dimension,
+            dtype="float32",
+            normalized=True,
+            preprocessing_version=preprocessing_version("lyrics-bge-m3-v1", self.config),
+            license="MIT",
+            configuration=config_dict(self.config),
+            loaded=self.encoder.loaded,
+        )
+
+    async def analyze(self, lyrics: str) -> GlobalEmbeddingResult:
+        try:
+            prepared = self.preprocessor.prepare(lyrics)
+            await self.encoder.ready()
+            chunks = self.preprocessor.chunks(
+                prepared, self.encoder, self.config.max_chunk_tokens
+            )
+            vectors = normalize_rows(
+                await self.encoder.encode(
+                    [chunk.text for chunk in chunks], self.config.batch_size
+                )
+            )
+            weights = np.asarray([chunk.token_count for chunk in chunks], dtype=np.float64)
+            aggregate = l2_normalize(np.average(vectors, axis=0, weights=weights))
+            metadata = self.metadata
+            return GlobalEmbeddingResult(
+                **metadata.result_metadata(),
+                embedding=finite_float_list(aggregate),
+                analysis={
+                    "section_count": len(prepared.sections),
+                    "chunk_count": len(chunks),
+                    "chunk_token_counts": [chunk.token_count for chunk in chunks],
+                },
+            )
+        except AnalysisError:
+            raise
+        except Exception as exc:
+            log.exception("BGE-M3 lyrics inference failed")
+            raise ModelInferenceError(self.representation) from exc

openmusic_analysis/analyzers/math.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import numpy as np
+
+
+def l2_normalize(vector: np.ndarray) -> np.ndarray:
+    value = np.asarray(vector, dtype=np.float32)
+    norm = float(np.linalg.norm(value))
+    if not np.isfinite(norm) or norm <= 0.0:
+        raise ValueError("Cannot normalize a non-finite or zero embedding")
+    return np.ascontiguousarray(value / norm, dtype=np.float32)
+
+
+def normalize_rows(matrix: np.ndarray) -> np.ndarray:
+    value = np.asarray(matrix, dtype=np.float32)
+    norms = np.linalg.norm(value, axis=1, keepdims=True)
+    if not np.isfinite(norms).all() or np.any(norms <= 0.0):
+        raise ValueError("Cannot normalize non-finite or zero embeddings")
+    return np.ascontiguousarray(value / norms, dtype=np.float32)
+
+
+def finite_float_list(vector: np.ndarray) -> list[float]:
+    value = np.asarray(vector, dtype=np.float32)
+    if not np.isfinite(value).all():
+        raise ValueError("Embedding contains non-finite values")
+    return value.tolist()

openmusic_analysis/api.py

@@ -0,0 +1,227 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import tempfile
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import Annotated, Any
+
+from fastapi import FastAPI, File, Form, Request, UploadFile
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+
+from openmusic_analysis.application import MusicAnalysisService, build_service
+from openmusic_analysis.domain import AnalysisResponse, ModelsResponse
+from openmusic_analysis.errors import AnalysisError
+from openmusic_analysis.settings import Settings
+
+
+log = logging.getLogger(__name__)
+SUPPORTED_EXTENSIONS = {".mp3", ".m4a", ".flac", ".wav", ".aac", ".ogg", ".opus"}
+
+
+def create_app(
+    *,
+    service: MusicAnalysisService | None = None,
+    settings: Settings | None = None,
+) -> FastAPI:
+    settings = settings or Settings.from_env()
+    service = service or build_service(settings)
+
+    @asynccontextmanager
+    async def lifespan(_: FastAPI):
+        if settings.eager_model_loading:
+            await service.load_models()
+        yield
+
+    app = FastAPI(
+        title="OpenMusic Music Analysis API",
+        version="1.0.0",
+        lifespan=lifespan,
+    )
+    app.state.analysis_service = service
+    app.state.settings = settings
+
+    @app.middleware("http")
+    async def request_limits(request: Request, call_next: Any):
+        content_length = request.headers.get("content-length")
+        request_limit = (
+            settings.limits.max_upload_bytes
+            + settings.limits.max_lyrics_characters * 4
+            + 1024 * 1024
+        )
+        if content_length:
+            try:
+                if int(content_length) > request_limit:
+                    return _error_response(
+                        "REQUEST_TOO_LARGE", "Request body exceeds the configured limit.", 413
+                    )
+            except ValueError:
+                return _error_response("INVALID_CONTENT_LENGTH", "Invalid Content-Length.", 400)
+        try:
+            async with asyncio.timeout(settings.limits.request_timeout_seconds):
+                return await call_next(request)
+        except TimeoutError:
+            return _error_response("REQUEST_TIMEOUT", "Analysis timed out.", 504)
+
+    @app.exception_handler(AnalysisError)
+    async def analysis_error_handler(_: Request, exc: AnalysisError):
+        return _error_response(exc.code, exc.message, exc.status_code, exc.details)
+
+    @app.exception_handler(RequestValidationError)
+    async def validation_error_handler(_: Request, exc: RequestValidationError):
+        missing_audio = any(
+            error.get("type") == "missing" and tuple(error.get("loc", ())) == ("body", "audio")
+            for error in exc.errors()
+        )
+        if missing_audio:
+            return _error_response("MISSING_AUDIO", "Multipart field 'audio' is required.", 422)
+        return _error_response(
+            "INVALID_REQUEST",
+            "Request validation failed.",
+            422,
+            {"errors": _safe_validation_errors(exc.errors())},
+        )
+
+    @app.exception_handler(Exception)
+    async def unhandled_error_handler(request: Request, exc: Exception):
+        log.exception("Unhandled error for %s", request.url.path, exc_info=exc)
+        return _error_response("INTERNAL_ERROR", "Internal server error.", 500)
+
+    @app.get("/v1/models", response_model=ModelsResponse)
+    async def models() -> ModelsResponse:
+        return ModelsResponse(models=service.registry.models())
+
+    @app.get("/v1/status")
+    async def status() -> dict[str, Any]:
+        return {
+            "status": "ok",
+            "device": service.device,
+            "loaded_models": service.registry.loaded_representations(),
+            "available_representations": list(service.registry.representations),
+        }
+
+    @app.get("/health")
+    async def health() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.post("/v1/tracks/analyze", response_model=AnalysisResponse)
+    async def analyze_track(
+        audio: Annotated[UploadFile, File(description="Audio file")],
+        lyrics: Annotated[str | None, Form()] = None,
+        track_id: Annotated[str | None, Form()] = None,
+        content_identity: Annotated[str | None, Form()] = None,
+        requested_representations: Annotated[list[str] | None, Form()] = None,
+    ) -> AnalysisResponse:
+        if lyrics is not None and len(lyrics) > settings.limits.max_lyrics_characters:
+            raise AnalysisError(
+                "LYRICS_TOO_LARGE",
+                "Lyrics exceed the configured character limit.",
+                status_code=413,
+            )
+        suffix = Path(audio.filename or "").suffix.lower()
+        if suffix not in SUPPORTED_EXTENSIONS:
+            raise AnalysisError(
+                "UNSUPPORTED_AUDIO_FORMAT",
+                f"Supported extensions: {', '.join(sorted(SUPPORTED_EXTENSIONS))}.",
+                status_code=415,
+            )
+        representations = _parse_representations(requested_representations)
+        temp_path = await _store_upload(audio, suffix, settings.limits.max_upload_bytes)
+        try:
+            return await service.analyze(
+                temp_path,
+                lyrics=lyrics,
+                requested_representations=representations,
+                track_id=track_id,
+                content_identity=content_identity,
+            )
+        finally:
+            try:
+                os.unlink(temp_path)
+            except FileNotFoundError:
+                pass
+
+    return app
+
+
+async def _store_upload(upload: UploadFile, suffix: str, max_bytes: int) -> str:
+    size = 0
+    path: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(prefix="openmusic-", suffix=suffix, delete=False) as file:
+            path = file.name
+            while chunk := await upload.read(1024 * 1024):
+                size += len(chunk)
+                if size > max_bytes:
+                    raise AnalysisError(
+                        "AUDIO_TOO_LARGE",
+                        "Audio upload exceeds the configured byte limit.",
+                        status_code=413,
+                    )
+                file.write(chunk)
+        if size == 0:
+            raise AnalysisError("EMPTY_AUDIO", "Uploaded audio is empty.", status_code=422)
+        return path
+    except Exception:
+        if path:
+            try:
+                os.unlink(path)
+            except FileNotFoundError:
+                pass
+        raise
+    finally:
+        await upload.close()
+
+
+def _parse_representations(values: list[str] | None) -> list[str] | None:
+    if values is None:
+        return None
+    parsed: list[str] = []
+    for value in values:
+        stripped = value.strip()
+        if stripped.startswith("["):
+            try:
+                decoded = json.loads(stripped)
+            except json.JSONDecodeError as exc:
+                raise AnalysisError(
+                    "INVALID_REPRESENTATIONS",
+                    "requested_representations contains invalid JSON.",
+                    status_code=422,
+                ) from exc
+            if not isinstance(decoded, list) or not all(isinstance(item, str) for item in decoded):
+                raise AnalysisError(
+                    "INVALID_REPRESENTATIONS",
+                    "requested_representations JSON must be an array of strings.",
+                    status_code=422,
+                )
+            parsed.extend(decoded)
+        else:
+            parsed.extend(part.strip() for part in stripped.split(",") if part.strip())
+    return parsed
+
+
+def _error_response(
+    code: str,
+    message: str,
+    status_code: int,
+    details: dict[str, Any] | None = None,
+) -> JSONResponse:
+    body: dict[str, Any] = {"error": {"code": code, "message": message}}
+    if details is not None:
+        body["error"]["details"] = details
+    return JSONResponse(status_code=status_code, content=body)
+
+
+def _safe_validation_errors(errors: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [
+        {
+            "type": error.get("type"),
+            "location": list(error.get("loc", ())),
+            "message": error.get("msg"),
+        }
+        for error in errors
+    ]

openmusic_analysis/application.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from openmusic_analysis.analyzers import (
+    BGEM3LyricsAnalyzer,
+    BGEM3TextEncoder,
+    ClapAudioEncoder,
+    ClapGlobalAudioAnalyzer,
+    ClapTemporalAudioAnalyzer,
+    LyricsPreprocessor,
+)
+from openmusic_analysis.audio import AnalysisContext, AudioDecoder
+from openmusic_analysis.domain import AnalysisResponse, TrackReference
+from openmusic_analysis.errors import AnalysisError
+from openmusic_analysis.registry import ModelRegistry
+from openmusic_analysis.runtime import DeviceManager, InferenceGate
+from openmusic_analysis.settings import Settings
+
+
+DEFAULT_AUDIO_REPRESENTATIONS = ("audio.global", "audio.temporal")
+
+
+class MusicAnalysisService:
+    def __init__(
+        self,
+        *,
+        registry: ModelRegistry,
+        decoder: AudioDecoder,
+        device: str,
+    ) -> None:
+        self.registry = registry
+        self.decoder = decoder
+        self.device = device
+
+    async def analyze(
+        self,
+        source_path: str | Path,
+        *,
+        lyrics: str | None,
+        requested_representations: list[str] | None,
+        track_id: str | None,
+        content_identity: str | None,
+    ) -> AnalysisResponse:
+        if lyrics is not None and not lyrics.strip():
+            raise AnalysisError(
+                "INVALID_LYRICS", "Lyrics must contain non-whitespace text.", status_code=422
+            )
+        requested = self._resolve_representations(requested_representations, lyrics)
+        context = AnalysisContext(source_path, self.decoder)
+        results = {}
+        for representation in requested:
+            analyzer = self.registry.analyzer(representation)
+            if analyzer.input_kind == "audio":
+                result = await analyzer.analyze(context)
+            elif analyzer.input_kind == "lyrics":
+                if lyrics is None:
+                    raise AnalysisError(
+                        "LYRICS_REQUIRED",
+                        f"Representation '{representation}' requires lyrics.",
+                        status_code=422,
+                    )
+                result = await analyzer.analyze(lyrics)
+            else:
+                raise RuntimeError(f"Unknown analyzer input kind: {analyzer.input_kind}")
+            results[representation] = result
+        return AnalysisResponse(
+            track=TrackReference(track_id=track_id, content_identity=content_identity),
+            representations=results,
+        )
+
+    def _resolve_representations(
+        self, requested: list[str] | None, lyrics: str | None
+    ) -> list[str]:
+        if requested is None:
+            values = list(DEFAULT_AUDIO_REPRESENTATIONS)
+            if lyrics is not None:
+                values.append("lyrics.global")
+        else:
+            values = list(dict.fromkeys(requested))
+            if not values:
+                raise AnalysisError(
+                    "INVALID_REPRESENTATIONS",
+                    "requested_representations must not be empty.",
+                    status_code=422,
+                )
+        unsupported = [value for value in values if value not in self.registry.representations]
+        if unsupported:
+            raise AnalysisError(
+                "UNSUPPORTED_REPRESENTATION",
+                f"Unsupported representation(s): {', '.join(unsupported)}.",
+                status_code=422,
+                details={"supported": list(self.registry.representations)},
+            )
+        return values
+
+    async def load_models(self) -> None:
+        seen: set[int] = set()
+        for representation in self.registry.representations:
+            analyzer = self.registry.analyzer(representation)
+            encoder = getattr(analyzer, "encoder", None)
+            if encoder is not None and id(encoder) not in seen:
+                seen.add(id(encoder))
+                await encoder.ready()
+
+
+def build_service(settings: Settings | None = None) -> MusicAnalysisService:
+    settings = settings or Settings.from_env()
+    device = DeviceManager.select(settings.device)
+    gate = InferenceGate(settings.inference_concurrency)
+    clap_encoder = ClapAudioEncoder(
+        device, gate, batch_size=settings.global_audio.inference_batch_size
+    )
+    text_encoder = BGEM3TextEncoder(device, gate)
+    analyzers = [
+        ClapGlobalAudioAnalyzer(clap_encoder, settings.global_audio),
+        ClapTemporalAudioAnalyzer(clap_encoder, settings.temporal_audio),
+        BGEM3LyricsAnalyzer(text_encoder, LyricsPreprocessor(), settings.lyrics),
+    ]
+    registry = ModelRegistry(analyzers)
+    decoder = AudioDecoder(
+        ffmpeg_binary=settings.ffmpeg_binary,
+        ffprobe_binary=settings.ffprobe_binary,
+        canonical_sample_rate=settings.global_audio.sample_rate,
+        max_audio_seconds=settings.limits.max_audio_seconds,
+        timeout_seconds=settings.limits.decode_timeout_seconds,
+    )
+    return MusicAnalysisService(registry=registry, decoder=decoder, device=device)

openmusic_analysis/audio/__init__.py

@@ -0,0 +1,4 @@
+from .context import AnalysisContext
+from .decoder import AudioDecoder, AudioMetadata, DecodedAudio
+
+__all__ = ["AnalysisContext", "AudioDecoder", "AudioMetadata", "DecodedAudio"]

openmusic_analysis/audio/context.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+import math
+from pathlib import Path
+
+import numpy as np
+from scipy.signal import resample_poly
+
+from .decoder import AudioDecoder, AudioMetadata, DecodedAudio
+
+
+class AnalysisContext:
+    """Request-scoped decoded-audio and resampling cache."""
+
+    def __init__(self, source_path: str | Path, decoder: AudioDecoder) -> None:
+        self.source_path = Path(source_path)
+        self.decoder = decoder
+        self._decoded: DecodedAudio | None = None
+        self._waveforms: dict[int, np.ndarray] = {}
+        self._lock = asyncio.Lock()
+
+    @property
+    def metadata(self) -> AudioMetadata | None:
+        return self._decoded.metadata if self._decoded else None
+
+    async def waveform(self, sample_rate: int) -> np.ndarray:
+        cached = self._waveforms.get(sample_rate)
+        if cached is not None:
+            return cached
+        async with self._lock:
+            cached = self._waveforms.get(sample_rate)
+            if cached is not None:
+                return cached
+            if self._decoded is None:
+                self._decoded = await asyncio.to_thread(self.decoder.decode, self.source_path)
+                self._waveforms[self._decoded.sample_rate] = self._decoded.waveform
+            if sample_rate not in self._waveforms:
+                source = self._decoded.waveform
+                source_rate = self._decoded.sample_rate
+                waveform = await asyncio.to_thread(
+                    _resample_deterministic, source, source_rate, sample_rate
+                )
+                self._waveforms[sample_rate] = waveform
+            return self._waveforms[sample_rate]
+
+
+def _resample_deterministic(
+    waveform: np.ndarray, source_rate: int, target_rate: int
+) -> np.ndarray:
+    if source_rate == target_rate:
+        return waveform
+    divisor = math.gcd(source_rate, target_rate)
+    result = resample_poly(
+        waveform.astype(np.float64, copy=False),
+        target_rate // divisor,
+        source_rate // divisor,
+        window=("kaiser", 5.0),
+        padtype="constant",
+    )
+    return np.ascontiguousarray(result, dtype=np.float32)

openmusic_analysis/audio/decoder.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+
+from openmusic_analysis.errors import AudioDecodeError
+
+
+@dataclass(frozen=True)
+class AudioMetadata:
+    duration_ms: int
+    source_sample_rate: int | None
+    source_channels: int | None
+    source_format: str | None
+
+
+@dataclass(frozen=True)
+class DecodedAudio:
+    waveform: np.ndarray
+    sample_rate: int
+    metadata: AudioMetadata
+
+
+class AudioDecoder:
+    """The only source-file decoder; analyzers receive canonical in-memory PCM."""
+
+    def __init__(
+        self,
+        *,
+        ffmpeg_binary: str = "ffmpeg",
+        ffprobe_binary: str = "ffprobe",
+        canonical_sample_rate: int = 48_000,
+        max_audio_seconds: float = 1800.0,
+        timeout_seconds: float = 120.0,
+    ) -> None:
+        self.ffmpeg_binary = ffmpeg_binary
+        self.ffprobe_binary = ffprobe_binary
+        self.canonical_sample_rate = canonical_sample_rate
+        self.max_audio_seconds = max_audio_seconds
+        self.timeout_seconds = timeout_seconds
+
+    def decode(self, source_path: str | Path) -> DecodedAudio:
+        path = str(source_path)
+        probe = self._probe(path)
+        if probe["duration"] is not None and probe["duration"] > self.max_audio_seconds:
+            raise AudioDecodeError(
+                f"Audio duration exceeds the {self.max_audio_seconds:g} second limit."
+            )
+
+        command = [
+            self.ffmpeg_binary,
+            "-v",
+            "error",
+            "-nostdin",
+            "-threads",
+            "1",
+            "-i",
+            path,
+            "-map",
+            "0:a:0",
+            "-vn",
+            "-ac",
+            "1",
+            "-ar",
+            str(self.canonical_sample_rate),
+            "-acodec",
+            "pcm_f32le",
+            "-f",
+            "f32le",
+            "pipe:1",
+        ]
+        try:
+            completed = subprocess.run(
+                command,
+                check=False,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                timeout=self.timeout_seconds,
+            )
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            raise AudioDecodeError() from exc
+        if completed.returncode != 0 or not completed.stdout:
+            raise AudioDecodeError()
+
+        waveform = np.frombuffer(completed.stdout, dtype="<f4").copy()
+        if waveform.size == 0 or not np.isfinite(waveform).all():
+            raise AudioDecodeError()
+        actual_duration = waveform.size / self.canonical_sample_rate
+        if actual_duration > self.max_audio_seconds:
+            raise AudioDecodeError(
+                f"Audio duration exceeds the {self.max_audio_seconds:g} second limit."
+            )
+        metadata = AudioMetadata(
+            duration_ms=int(round(actual_duration * 1000)),
+            source_sample_rate=probe["sample_rate"],
+            source_channels=probe["channels"],
+            source_format=probe["format"],
+        )
+        return DecodedAudio(
+            waveform=np.ascontiguousarray(waveform, dtype=np.float32),
+            sample_rate=self.canonical_sample_rate,
+            metadata=metadata,
+        )
+
+    def _probe(self, path: str) -> dict[str, int | float | str | None]:
+        command = [
+            self.ffprobe_binary,
+            "-v",
+            "error",
+            "-select_streams",
+            "a:0",
+            "-show_entries",
+            "stream=sample_rate,channels,duration:format=format_name,duration",
+            "-of",
+            "json",
+            path,
+        ]
+        try:
+            completed = subprocess.run(
+                command,
+                check=False,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                timeout=min(30.0, self.timeout_seconds),
+                text=True,
+            )
+            payload = json.loads(completed.stdout) if completed.returncode == 0 else {}
+        except (OSError, subprocess.TimeoutExpired, json.JSONDecodeError) as exc:
+            raise AudioDecodeError() from exc
+        streams = payload.get("streams") or []
+        if not streams:
+            raise AudioDecodeError()
+        stream = streams[0]
+        container = payload.get("format") or {}
+        duration_value = stream.get("duration") or container.get("duration")
+        try:
+            duration = float(duration_value) if duration_value is not None else None
+        except (TypeError, ValueError):
+            duration = None
+        return {
+            "duration": duration,
+            "sample_rate": _optional_int(stream.get("sample_rate")),
+            "channels": _optional_int(stream.get("channels")),
+            "format": container.get("format_name"),
+        }
+
+
+def _optional_int(value: object) -> int | None:
+    try:
+        return int(value) if value is not None else None
+    except (TypeError, ValueError):
+        return None

openmusic_analysis/audio/windows.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import math
+
+import numpy as np
+
+
+def global_window_starts(
+    duration_samples: int,
+    window_samples: int,
+    target_windows: int,
+    coverage_start: float,
+    coverage_end: float,
+) -> list[int]:
+    if duration_samples <= window_samples:
+        return [0]
+    duration_in_windows = duration_samples / window_samples
+    count = min(target_windows, max(1, int(math.floor(duration_in_windows + 0.5))))
+    if count == 1:
+        return [int(round(0.5 * (duration_samples - window_samples)))]
+    valid_start = duration_samples - window_samples
+    fractions = np.linspace(coverage_start, coverage_end, count, dtype=np.float64)
+    return [int(round(float(fraction) * valid_start)) for fraction in fractions]
+
+
+def temporal_window_starts(
+    duration_samples: int,
+    window_samples: int,
+    hop_samples: int,
+    max_segments: int,
+) -> list[int]:
+    if duration_samples <= window_samples:
+        return [0]
+    last_start = duration_samples - window_samples
+    starts = list(range(0, last_start + 1, hop_samples))
+    if starts[-1] != last_start:
+        starts.append(last_start)
+    if len(starts) <= max_segments:
+        return starts
+    indices = np.linspace(0, len(starts) - 1, max_segments, dtype=np.float64)
+    selected = sorted({starts[int(round(float(index)))] for index in indices})
+    return selected
+
+
+def fixed_window(waveform: np.ndarray, start: int, length: int) -> np.ndarray:
+    chunk = waveform[start : start + length]
+    if chunk.size == length:
+        return np.ascontiguousarray(chunk, dtype=np.float32)
+    if chunk.size == 0:
+        return np.zeros(length, dtype=np.float32)
+    repeats = math.ceil(length / chunk.size)
+    padded = np.tile(chunk, repeats)[:length]
+    return np.ascontiguousarray(padded, dtype=np.float32)

openmusic_analysis/domain.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class ModelMetadata(StrictModel):
+    representation: str
+    model_id: str
+    model_version: str
+    modality: str
+    dimension: int = Field(gt=0)
+    dtype: str = "float32"
+    normalized: bool
+    preprocessing_version: str
+    license: str
+    configuration: dict[str, Any] = Field(default_factory=dict)
+    loaded: bool = False
+
+    def result_metadata(self) -> dict[str, Any]:
+        return self.model_dump(exclude={"loaded", "license"})
+
+
+class GlobalEmbeddingResult(StrictModel):
+    representation: str
+    model_id: str
+    model_version: str
+    modality: str
+    preprocessing_version: str
+    dimension: int
+    dtype: str
+    normalized: bool
+    configuration: dict[str, Any]
+    embedding: list[float]
+    analysis: dict[str, Any] = Field(default_factory=dict)
+
+
+class TemporalSegment(StrictModel):
+    start_ms: int = Field(ge=0)
+    end_ms: int = Field(gt=0)
+    embedding: list[float]
+
+
+class TemporalSummary(StrictModel):
+    number_of_segments: int = Field(ge=0)
+    mean_adjacent_distance: float
+    max_adjacent_distance: float
+    trajectory_variance: float
+    largest_transition_index: int | None
+
+
+class TemporalEmbeddingResult(StrictModel):
+    representation: str
+    model_id: str
+    model_version: str
+    modality: str
+    preprocessing_version: str
+    dimension: int
+    dtype: str
+    normalized: bool
+    configuration: dict[str, Any]
+    segments: list[TemporalSegment]
+    summary: TemporalSummary
+
+
+RepresentationResult = GlobalEmbeddingResult | TemporalEmbeddingResult
+
+
+class TrackReference(StrictModel):
+    track_id: str | None = None
+    content_identity: str | None = None
+
+
+class AnalysisResponse(StrictModel):
+    schema_version: str = "1"
+    track: TrackReference
+    representations: dict[str, RepresentationResult]
+
+
+class ModelsResponse(StrictModel):
+    schema_version: str = "1"
+    models: list[ModelMetadata]
+
+
+class ErrorBody(StrictModel):
+    code: str
+    message: str
+    details: dict[str, Any] | None = None
+
+
+class ErrorResponse(StrictModel):
+    error: ErrorBody

openmusic_analysis/errors.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class AnalysisError(Exception):
+    def __init__(
+        self,
+        code: str,
+        message: str,
+        *,
+        status_code: int = 422,
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.status_code = status_code
+        self.details = details
+
+
+class AudioDecodeError(AnalysisError):
+    def __init__(self, message: str = "The uploaded audio could not be decoded.") -> None:
+        super().__init__("AUDIO_DECODE_FAILED", message, status_code=422)
+
+
+class ModelInferenceError(AnalysisError):
+    def __init__(self, representation: str) -> None:
+        super().__init__(
+            "MODEL_INFERENCE_FAILED",
+            f"Inference failed for representation '{representation}'.",
+            status_code=500,
+        )

openmusic_analysis/experiments/__init__.py

@@ -0,0 +1,3 @@
+from .temporal_similarity import aligned_interpolation_similarity, dtw_cosine_similarity
+
+__all__ = ["aligned_interpolation_similarity", "dtw_cosine_similarity"]

openmusic_analysis/experiments/temporal_similarity.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import numpy as np
+
+from openmusic_analysis.analyzers.math import normalize_rows
+
+
+def aligned_interpolation_similarity(
+    first: np.ndarray, second: np.ndarray, *, points: int = 32
+) -> float:
+    """Interpolate both trajectories to normalized time and average cosine similarity."""
+    if points < 2:
+        raise ValueError("points must be at least 2")
+    first_value = _validate_trajectory(first)
+    second_value = _validate_trajectory(second)
+    first_aligned = _interpolate(first_value, points)
+    second_aligned = _interpolate(second_value, points)
+    return float(np.mean(np.sum(first_aligned * second_aligned, axis=1)))
+
+
+def dtw_cosine_similarity(first: np.ndarray, second: np.ndarray) -> float:
+    """Classic DTW over cosine distance, converted to an average-path similarity."""
+    first_value = _validate_trajectory(first)
+    second_value = _validate_trajectory(second)
+    distances = np.clip(1.0 - first_value @ second_value.T, 0.0, 2.0)
+    rows, columns = distances.shape
+    costs = np.full((rows + 1, columns + 1), np.inf, dtype=np.float64)
+    lengths = np.zeros((rows + 1, columns + 1), dtype=np.int32)
+    costs[0, 0] = 0.0
+    for row in range(1, rows + 1):
+        for column in range(1, columns + 1):
+            candidates = (
+                (costs[row - 1, column], lengths[row - 1, column]),
+                (costs[row, column - 1], lengths[row, column - 1]),
+                (costs[row - 1, column - 1], lengths[row - 1, column - 1]),
+            )
+            previous_cost, previous_length = min(candidates, key=lambda item: item[0])
+            costs[row, column] = previous_cost + distances[row - 1, column - 1]
+            lengths[row, column] = previous_length + 1
+    average_distance = costs[rows, columns] / max(1, int(lengths[rows, columns]))
+    return float(1.0 - average_distance)
+
+
+def _validate_trajectory(value: np.ndarray) -> np.ndarray:
+    array = np.asarray(value, dtype=np.float32)
+    if array.ndim != 2 or array.shape[0] == 0 or array.shape[1] == 0:
+        raise ValueError("trajectory must be a non-empty [segments, dimension] matrix")
+    if not np.isfinite(array).all():
+        raise ValueError("trajectory contains non-finite values")
+    return normalize_rows(array)
+
+
+def _interpolate(trajectory: np.ndarray, points: int) -> np.ndarray:
+    if trajectory.shape[0] == 1:
+        return np.repeat(trajectory, points, axis=0)
+    source_time = np.linspace(0.0, 1.0, trajectory.shape[0])
+    target_time = np.linspace(0.0, 1.0, points)
+    result = np.stack(
+        [np.interp(target_time, source_time, trajectory[:, dim]) for dim in range(trajectory.shape[1])],
+        axis=1,
+    )
+    return normalize_rows(result)

openmusic_analysis/registry.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+from openmusic_analysis.domain import ModelMetadata
+
+
+class ModelRegistry:
+    """Source of truth built directly from the configured analyzer instances."""
+
+    def __init__(self, analyzers: Iterable[Any]) -> None:
+        self._analyzers: dict[str, Any] = {}
+        for analyzer in analyzers:
+            representation = analyzer.metadata.representation
+            if representation in self._analyzers:
+                raise ValueError(f"Duplicate analyzer for {representation}")
+            self._analyzers[representation] = analyzer
+
+    @property
+    def representations(self) -> tuple[str, ...]:
+        return tuple(self._analyzers)
+
+    def analyzer(self, representation: str) -> Any:
+        return self._analyzers[representation]
+
+    def models(self) -> list[ModelMetadata]:
+        return [analyzer.metadata for analyzer in self._analyzers.values()]
+
+    def loaded_representations(self) -> list[str]:
+        return [model.representation for model in self.models() if model.loaded]

openmusic_analysis/runtime.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from typing import Any, TypeVar
+
+
+T = TypeVar("T")
+
+
+class DeviceManager:
+    @staticmethod
+    def select(configured: str = "auto") -> str:
+        if configured != "auto":
+            return configured
+        try:
+            import torch
+
+            if torch.cuda.is_available():
+                return "cuda"
+            if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
+                return "mps"
+        except ImportError:
+            pass
+        return "cpu"
+
+
+class InferenceGate:
+    """One process-wide bound on model loading and inference concurrency."""
+
+    def __init__(self, concurrency: int) -> None:
+        self._semaphore = asyncio.Semaphore(max(1, concurrency))
+
+    async def run(self, function: Callable[..., T], *args: Any) -> T:
+        task = asyncio.create_task(self._guarded(function, *args))
+        try:
+            return await asyncio.shield(task)
+        except asyncio.CancelledError:
+            task.add_done_callback(_consume_background_exception)
+            raise
+
+    async def _guarded(self, function: Callable[..., T], *args: Any) -> T:
+        async with self._semaphore:
+            return await asyncio.to_thread(function, *args)
+
+
+def _consume_background_exception(task: asyncio.Task[Any]) -> None:
+    if not task.cancelled():
+        task.exception()

openmusic_analysis/settings.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from dataclasses import asdict, dataclass
+from typing import Any
+
+
+CLAP_MODEL_ID = "laion/clap-htsat-unfused"
+CLAP_MODEL_REVISION = "8fa0f1c6d0433df6e97c127f64b2a1d6c0dcda8a"
+BGE_MODEL_ID = "BAAI/bge-m3"
+BGE_MODEL_REVISION = "5617a9f61b028005a4858fdac845db406aefb181"
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass(frozen=True)
+class GlobalAudioConfig:
+    sample_rate: int = 48_000
+    window_seconds: float = 10.0
+    target_windows: int = 4
+    coverage_start: float = 0.10
+    coverage_end: float = 0.90
+    minimum_audio_seconds: float = 1.0
+    short_window_padding: str = "repeat_to_window_length"
+    selection: str = "uniform_over_valid_start_range"
+    aggregation: str = "l2_each_then_mean_then_l2"
+    encoder_window_seconds: float = 10.0
+    oversize_window_encoding: str = "uniform_10s_subwindows"
+    encoder_subwindow_aggregation: str = "l2_each_then_mean_then_l2"
+    decoding_version: str = "ffmpeg-mono-f32le-v1"
+    resampling: str = "scipy-resample-poly-kaiser5-v1"
+    feature_extractor: str = "transformers-clap-4.44.2-fixed-input"
+    inference_batch_size: int = 4
+
+    def __post_init__(self) -> None:
+        if (
+            self.sample_rate <= 0
+            or self.window_seconds <= 0
+            or self.target_windows <= 0
+            or self.inference_batch_size <= 0
+        ):
+            raise ValueError("Global audio rates, window and count must be positive")
+        if not 0 <= self.coverage_start <= self.coverage_end <= 1:
+            raise ValueError("Global coverage must satisfy 0 <= start <= end <= 1")
+
+
+@dataclass(frozen=True)
+class TemporalAudioConfig:
+    sample_rate: int = 48_000
+    window_seconds: float = 10.0
+    hop_seconds: float = 10.0
+    max_segments: int = 24
+    minimum_audio_seconds: float = 1.0
+    short_window_padding: str = "repeat_to_window_length"
+    segmentation: str = "sliding_with_tail_coverage_and_uniform_limit"
+    encoder_window_seconds: float = 10.0
+    oversize_window_encoding: str = "uniform_10s_subwindows"
+    encoder_subwindow_aggregation: str = "l2_each_then_mean_then_l2"
+    decoding_version: str = "ffmpeg-mono-f32le-v1"
+    resampling: str = "scipy-resample-poly-kaiser5-v1"
+    feature_extractor: str = "transformers-clap-4.44.2-fixed-input"
+    inference_batch_size: int = 4
+
+    def __post_init__(self) -> None:
+        if (
+            self.sample_rate <= 0
+            or self.window_seconds <= 0
+            or self.hop_seconds <= 0
+            or self.max_segments <= 0
+            or self.inference_batch_size <= 0
+        ):
+            raise ValueError("Temporal audio rates, windows, hop and limit must be positive")
+
+
+@dataclass(frozen=True)
+class LyricsConfig:
+    max_chunk_tokens: int = 512
+    batch_size: int = 8
+    chunking: str = "section_first_then_token_split"
+    aggregation: str = "token_weighted_mean_of_l2_chunks_then_l2"
+    unicode_normalization: str = "NFKC"
+    pooling: str = "bge-m3-cls"
+    model_context_tokens: int = 8192
+
+    def __post_init__(self) -> None:
+        if self.max_chunk_tokens < 4 or self.max_chunk_tokens > self.model_context_tokens:
+            raise ValueError("Lyrics chunk tokens must be between 4 and model context")
+        if self.batch_size <= 0:
+            raise ValueError("Lyrics batch size must be positive")
+
+
+@dataclass(frozen=True)
+class LimitConfig:
+    max_upload_bytes: int = 100 * 1024 * 1024
+    max_lyrics_characters: int = 100_000
+    max_audio_seconds: float = 30 * 60
+    request_timeout_seconds: float = 300.0
+    decode_timeout_seconds: float = 120.0
+
+    def __post_init__(self) -> None:
+        if min(
+            self.max_upload_bytes,
+            self.max_lyrics_characters,
+            self.max_audio_seconds,
+            self.request_timeout_seconds,
+            self.decode_timeout_seconds,
+        ) <= 0:
+            raise ValueError("All service limits must be positive")
+
+
+@dataclass(frozen=True)
+class Settings:
+    device: str = "auto"
+    inference_concurrency: int = 1
+    ffmpeg_binary: str = "ffmpeg"
+    ffprobe_binary: str = "ffprobe"
+    eager_model_loading: bool = False
+    global_audio: GlobalAudioConfig = GlobalAudioConfig()
+    temporal_audio: TemporalAudioConfig = TemporalAudioConfig()
+    lyrics: LyricsConfig = LyricsConfig()
+    limits: LimitConfig = LimitConfig()
+
+    def __post_init__(self) -> None:
+        if self.inference_concurrency <= 0:
+            raise ValueError("Inference concurrency must be positive")
+        if self.global_audio.inference_batch_size != self.temporal_audio.inference_batch_size:
+            raise ValueError("Global and temporal CLAP analyzers must share a batch size")
+
+    @classmethod
+    def from_env(cls) -> "Settings":
+        clap_batch_size = int(os.getenv("OPENMUSIC_CLAP_BATCH_SIZE", "4"))
+        global_audio = GlobalAudioConfig(
+            window_seconds=float(os.getenv("OPENMUSIC_GLOBAL_WINDOW_SECONDS", "10")),
+            target_windows=int(os.getenv("OPENMUSIC_GLOBAL_WINDOWS", "4")),
+            inference_batch_size=clap_batch_size,
+        )
+        temporal_audio = TemporalAudioConfig(
+            window_seconds=float(os.getenv("OPENMUSIC_TEMPORAL_WINDOW_SECONDS", "10")),
+            hop_seconds=float(os.getenv("OPENMUSIC_TEMPORAL_HOP_SECONDS", "10")),
+            max_segments=int(os.getenv("OPENMUSIC_TEMPORAL_MAX_SEGMENTS", "24")),
+            inference_batch_size=clap_batch_size,
+        )
+        lyrics = LyricsConfig(
+            max_chunk_tokens=int(os.getenv("OPENMUSIC_LYRICS_CHUNK_TOKENS", "512")),
+            batch_size=int(os.getenv("OPENMUSIC_LYRICS_BATCH_SIZE", "8")),
+        )
+        limits = LimitConfig(
+            max_upload_bytes=int(os.getenv("OPENMUSIC_MAX_UPLOAD_BYTES", str(100 * 1024 * 1024))),
+            max_lyrics_characters=int(os.getenv("OPENMUSIC_MAX_LYRICS_CHARACTERS", "100000")),
+            max_audio_seconds=float(os.getenv("OPENMUSIC_MAX_AUDIO_SECONDS", "1800")),
+            request_timeout_seconds=float(os.getenv("OPENMUSIC_REQUEST_TIMEOUT_SECONDS", "300")),
+            decode_timeout_seconds=float(os.getenv("OPENMUSIC_DECODE_TIMEOUT_SECONDS", "120")),
+        )
+        return cls(
+            device=os.getenv("OPENMUSIC_DEVICE", "auto"),
+            inference_concurrency=max(1, int(os.getenv("OPENMUSIC_INFERENCE_CONCURRENCY", "1"))),
+            ffmpeg_binary=os.getenv("OPENMUSIC_FFMPEG", "ffmpeg"),
+            ffprobe_binary=os.getenv("OPENMUSIC_FFPROBE", "ffprobe"),
+            eager_model_loading=_env_bool("OPENMUSIC_EAGER_MODELS", False),
+            global_audio=global_audio,
+            temporal_audio=temporal_audio,
+            lyrics=lyrics,
+            limits=limits,
+        )
+
+
+def config_dict(config: Any) -> dict[str, Any]:
+    return asdict(config)
+
+
+def preprocessing_version(base: str, config: Any) -> str:
+    """Tie a preprocessing version to every result-affecting configuration value."""
+    payload = json.dumps(config_dict(config), sort_keys=True, separators=(",", ":"))
+    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]
+    return f"{base}.{digest}"

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_default_fixture_loop_scope = function
+testpaths = tests

requirements-dev.txt

@@ -0,0 +1,4 @@
+-r requirements.txt
+httpx==0.27.2
+pytest==8.3.3
+pytest-asyncio==0.24.0

requirements.txt

@@ -1,13 +1,11 @@
 fastapi==0.115.0
 uvicorn[standard]==0.30.6
-python-multipart==0.0.9
+python-multipart==0.0.32
+numpy==1.26.4
+scipy==1.14.1
 torch==2.3.1
-torchaudio==2.3.1
 transformers==4.44.2
-scikit-learn==1.5.1
-numpy==1.26.4
-librosa==0.10.2
-pydub==0.25.1
-imageio-ffmpeg==0.5.1
-soundfile==0.12.1
-accelerate==0.33.0
+tokenizers==0.19.1
+huggingface-hub==0.24.6
+safetensors==0.4.5
+sentencepiece==0.2.0

tests/conftest.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from openmusic_analysis.analyzers.clap import (
+    ClapGlobalAudioAnalyzer,
+    ClapTemporalAudioAnalyzer,
+)
+from openmusic_analysis.analyzers.lyrics import BGEM3LyricsAnalyzer, LyricsPreprocessor
+from openmusic_analysis.application import MusicAnalysisService
+from openmusic_analysis.audio.decoder import AudioMetadata, DecodedAudio
+from openmusic_analysis.errors import AudioDecodeError
+from openmusic_analysis.registry import ModelRegistry
+from openmusic_analysis.settings import GlobalAudioConfig, LyricsConfig, TemporalAudioConfig
+
+
+class FakeAudioEncoder:
+    dimension = 4
+    loaded = True
+
+    def __init__(self) -> None:
+        self.calls: list[list[np.ndarray]] = []
+
+    async def ready(self) -> None:
+        return None
+
+    async def encode(self, windows: list[np.ndarray]) -> np.ndarray:
+        self.calls.append(windows)
+        rows = []
+        for window in windows:
+            rows.append(
+                [
+                    float(np.mean(window)),
+                    float(np.std(window)),
+                    float(window[0]),
+                    float(window[-1]) + 1.0,
+                ]
+            )
+        return np.asarray(rows, dtype=np.float32)
+
+
+class FailingAudioEncoder(FakeAudioEncoder):
+    async def encode(self, windows: list[np.ndarray]) -> np.ndarray:
+        raise RuntimeError("internal model detail must not leak")
+
+
+class FakeTextEncoder:
+    dimension = 6
+    loaded = True
+
+    async def ready(self) -> None:
+        return None
+
+    def count_tokens(self, text: str) -> int:
+        return len(text.split()) + 2
+
+    def split_tokens(self, text: str, max_tokens: int) -> list[str]:
+        words = text.split()
+        size = max(1, max_tokens - 2)
+        return [" ".join(words[index : index + size]) for index in range(0, len(words), size)]
+
+    async def encode(self, texts: list[str], batch_size: int) -> np.ndarray:
+        rows = []
+        for text in texts:
+            digest = hashlib.sha256(text.encode("utf-8")).digest()
+            rows.append([float(value + 1) for value in digest[: self.dimension]])
+        return np.asarray(rows, dtype=np.float32)
+
+
+class FakeDecoder:
+    canonical_sample_rate = 10
+
+    def __init__(self, waveform: np.ndarray | None = None) -> None:
+        self.waveform = (
+            np.asarray(waveform, dtype=np.float32)
+            if waveform is not None
+            else np.linspace(-1.0, 1.0, 95, dtype=np.float32)
+        )
+        self.calls = 0
+
+    def decode(self, source_path: str | Path) -> DecodedAudio:
+        self.calls += 1
+        if Path(source_path).read_bytes().startswith(b"bad"):
+            raise AudioDecodeError()
+        return DecodedAudio(
+            waveform=self.waveform,
+            sample_rate=self.canonical_sample_rate,
+            metadata=AudioMetadata(
+                duration_ms=int(round(self.waveform.size * 1000 / self.canonical_sample_rate)),
+                source_sample_rate=self.canonical_sample_rate,
+                source_channels=1,
+                source_format="fake",
+            ),
+        )
+
+
+def make_service(
+    *,
+    waveform: np.ndarray | None = None,
+    audio_encoder: FakeAudioEncoder | None = None,
+) -> tuple[MusicAnalysisService, FakeDecoder, FakeAudioEncoder]:
+    audio_encoder = audio_encoder or FakeAudioEncoder()
+    decoder = FakeDecoder(waveform)
+    global_config = GlobalAudioConfig(
+        sample_rate=10,
+        window_seconds=2.0,
+        target_windows=4,
+        minimum_audio_seconds=1.0,
+    )
+    temporal_config = TemporalAudioConfig(
+        sample_rate=10,
+        window_seconds=2.0,
+        hop_seconds=2.0,
+        max_segments=4,
+        minimum_audio_seconds=1.0,
+    )
+    analyzers = [
+        ClapGlobalAudioAnalyzer(audio_encoder, global_config),
+        ClapTemporalAudioAnalyzer(audio_encoder, temporal_config),
+        BGEM3LyricsAnalyzer(
+            FakeTextEncoder(),
+            LyricsPreprocessor(),
+            LyricsConfig(max_chunk_tokens=12, batch_size=4),
+        ),
+    ]
+    service = MusicAnalysisService(
+        registry=ModelRegistry(analyzers), decoder=decoder, device="cpu"
+    )
+    return service, decoder, audio_encoder
+
+
+@pytest.fixture
+def service_bundle():
+    return make_service()

tests/test_api.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from openmusic_analysis.api import create_app
+from openmusic_analysis.settings import Settings
+
+from .conftest import FailingAudioEncoder, make_service
+
+
+def client_for(service) -> TestClient:
+    return TestClient(create_app(service=service, settings=Settings()))
+
+
+def upload(content: bytes = b"valid fake audio"):
+    return {"audio": ("track.mp3", content, "audio/mpeg")}
+
+
+def test_valid_audio_only_request(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post("/v1/tracks/analyze", files=upload())
+    assert response.status_code == 200
+    assert set(response.json()["representations"]) == {"audio.global", "audio.temporal"}
+
+
+def test_audio_and_multilingual_lyrics(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post(
+        "/v1/tracks/analyze",
+        files=upload(),
+        data={"lyrics": "[Verse]\nHello world\n\n[Припев]\nПривет, мир!"},
+    )
+    assert response.status_code == 200
+    assert "lyrics.global" in response.json()["representations"]
+
+
+def test_requested_representations_avoid_unrequested_work(service_bundle):
+    service, decoder, encoder = service_bundle
+    response = client_for(service).post(
+        "/v1/tracks/analyze",
+        files=upload(),
+        data={
+            "lyrics": "lyrics that must not be encoded",
+            "requested_representations": "audio.global",
+        },
+    )
+    assert response.status_code == 200
+    assert list(response.json()["representations"]) == ["audio.global"]
+    assert decoder.calls == 1
+    assert len(encoder.calls) == 1
+
+
+def test_invalid_file_returns_structured_decode_error(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post("/v1/tracks/analyze", files=upload(b"bad bytes"))
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "AUDIO_DECODE_FAILED"
+
+
+def test_missing_audio_is_structured(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post("/v1/tracks/analyze", data={"lyrics": "text"})
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "MISSING_AUDIO"
+
+
+def test_unsupported_representation(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post(
+        "/v1/tracks/analyze",
+        files=upload(),
+        data={"requested_representations": "audio.emotion"},
+    )
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "UNSUPPORTED_REPRESENTATION"
+
+
+def test_unsupported_extension(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post(
+        "/v1/tracks/analyze", files={"audio": ("track.txt", b"data", "text/plain")}
+    )
+    assert response.status_code == 415
+    assert response.json()["error"]["code"] == "UNSUPPORTED_AUDIO_FORMAT"
+
+
+def test_explicit_lyrics_representation_requires_lyrics(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post(
+        "/v1/tracks/analyze",
+        files=upload(),
+        data={"requested_representations": "lyrics.global"},
+    )
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "LYRICS_REQUIRED"
+
+
+def test_invalid_empty_lyrics(service_bundle):
+    service, _, _ = service_bundle
+    response = client_for(service).post(
+        "/v1/tracks/analyze", files=upload(), data={"lyrics": "   \n "}
+    )
+    assert response.status_code == 422
+    assert response.json()["error"]["code"] == "INVALID_LYRICS"
+
+
+def test_model_failure_does_not_leak_internal_exception():
+    service, _, _ = make_service(audio_encoder=FailingAudioEncoder())
+    response = client_for(service).post(
+        "/v1/tracks/analyze",
+        files=upload(),
+        data={"requested_representations": "audio.global"},
+    )
+    assert response.status_code == 500
+    assert response.json()["error"]["code"] == "MODEL_INFERENCE_FAILED"
+    assert "internal model detail" not in response.text

tests/test_audio_analyzers.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from openmusic_analysis.analyzers.clap import (
+    ClapGlobalAudioAnalyzer,
+    ClapTemporalAudioAnalyzer,
+)
+from openmusic_analysis.audio import AnalysisContext
+from openmusic_analysis.settings import GlobalAudioConfig, TemporalAudioConfig
+
+from .conftest import FakeAudioEncoder, FakeDecoder
+
+
+def context(tmp_path: Path, waveform: np.ndarray) -> AnalysisContext:
+    source = tmp_path / "audio.wav"
+    source.write_bytes(b"valid")
+    return AnalysisContext(source, FakeDecoder(waveform))
+
+
+@pytest.mark.asyncio
+async def test_global_is_deterministic_finite_normalized_and_multiwindow(tmp_path):
+    waveform = np.linspace(-1, 1, 100, dtype=np.float32)
+    encoder = FakeAudioEncoder()
+    config = GlobalAudioConfig(
+        sample_rate=10, window_seconds=2, target_windows=4, minimum_audio_seconds=1
+    )
+    analyzer = ClapGlobalAudioAnalyzer(encoder, config)
+    first = await analyzer.analyze(context(tmp_path, waveform))
+    second = await analyzer.analyze(context(tmp_path, waveform))
+    first_vector = np.asarray(first.embedding)
+    second_vector = np.asarray(second.embedding)
+    assert first.dimension == 4
+    assert np.isfinite(first_vector).all()
+    assert np.linalg.norm(first_vector) == pytest.approx(1.0, abs=1e-6)
+    assert first_vector == pytest.approx(second_vector, abs=1e-7)
+    assert first.analysis["windows_used"] == 4
+    assert len(encoder.calls[0]) == 4
+
+
+@pytest.mark.asyncio
+async def test_global_short_track_uses_one_padded_window(tmp_path):
+    encoder = FakeAudioEncoder()
+    config = GlobalAudioConfig(
+        sample_rate=10, window_seconds=2, target_windows=4, minimum_audio_seconds=0.2
+    )
+    result = await ClapGlobalAudioAnalyzer(encoder, config).analyze(
+        context(tmp_path, np.arange(5, dtype=np.float32))
+    )
+    assert result.analysis["windows_used"] == 1
+    assert encoder.calls[0][0].size == 20
+
+
+@pytest.mark.asyncio
+async def test_temporal_segments_are_ordered_and_deterministic(tmp_path):
+    waveform = np.linspace(-1, 1, 95, dtype=np.float32)
+    config = TemporalAudioConfig(
+        sample_rate=10,
+        window_seconds=2,
+        hop_seconds=2,
+        max_segments=20,
+        minimum_audio_seconds=1,
+    )
+    analyzer = ClapTemporalAudioAnalyzer(FakeAudioEncoder(), config)
+    first = await analyzer.analyze(context(tmp_path, waveform))
+    second = await analyzer.analyze(context(tmp_path, waveform))
+    starts = [segment.start_ms for segment in first.segments]
+    assert starts == sorted(starts)
+    assert starts == [segment.start_ms for segment in second.segments]
+    assert first.model_dump() == second.model_dump()
+    assert all(segment.start_ms < segment.end_ms for segment in first.segments)
+
+
+@pytest.mark.asyncio
+async def test_temporal_short_track_has_one_segment(tmp_path):
+    config = TemporalAudioConfig(
+        sample_rate=10,
+        window_seconds=2,
+        hop_seconds=1,
+        max_segments=4,
+        minimum_audio_seconds=0.2,
+    )
+    result = await ClapTemporalAudioAnalyzer(FakeAudioEncoder(), config).analyze(
+        context(tmp_path, np.arange(5, dtype=np.float32))
+    )
+    assert len(result.segments) == 1
+    assert result.segments[0].start_ms == 0
+    assert result.segments[0].end_ms == 500
+    assert result.summary.largest_transition_index is None
+
+
+@pytest.mark.asyncio
+async def test_temporal_long_track_respects_max_segments_and_tail(tmp_path):
+    config = TemporalAudioConfig(
+        sample_rate=10,
+        window_seconds=2,
+        hop_seconds=1,
+        max_segments=5,
+        minimum_audio_seconds=1,
+    )
+    result = await ClapTemporalAudioAnalyzer(FakeAudioEncoder(), config).analyze(
+        context(tmp_path, np.arange(300, dtype=np.float32))
+    )
+    assert len(result.segments) == 5
+    assert result.segments[0].start_ms == 0
+    assert result.segments[-1].end_ms == 30_000
+    assert result.summary.number_of_segments == 5
+    assert np.isfinite(result.summary.trajectory_variance)
+
+
+@pytest.mark.asyncio
+async def test_request_context_decodes_only_once_for_two_analyzers(tmp_path):
+    source = tmp_path / "audio.wav"
+    source.write_bytes(b"valid")
+    decoder = FakeDecoder(np.arange(100, dtype=np.float32))
+    analysis_context = AnalysisContext(source, decoder)
+    encoder = FakeAudioEncoder()
+    await ClapGlobalAudioAnalyzer(
+        encoder,
+        GlobalAudioConfig(sample_rate=10, window_seconds=2, minimum_audio_seconds=1),
+    ).analyze(analysis_context)
+    await ClapTemporalAudioAnalyzer(
+        encoder,
+        TemporalAudioConfig(sample_rate=10, window_seconds=2, minimum_audio_seconds=1),
+    ).analyze(analysis_context)
+    assert decoder.calls == 1

tests/test_benchmark.py

@@ -0,0 +1,44 @@
+from pathlib import Path
+
+from tools.nearest_neighbors import _cache_key, _neighbor_report
+
+
+def test_benchmark_cache_key_covers_content_model_and_configuration(tmp_path: Path):
+    audio = tmp_path / "track.wav"
+    audio.write_bytes(b"first content")
+    model = {
+        "model_id": "model",
+        "model_version": "revision",
+        "preprocessing_version": "prep",
+        "representation": "audio.global",
+        "configuration": {"windows": 4},
+    }
+    first = _cache_key(audio, None, model)
+    changed_model = {**model, "configuration": {"windows": 5}}
+    assert _cache_key(audio, None, changed_model) != first
+    audio.write_bytes(b"different content")
+    assert _cache_key(audio, None, model) != first
+
+
+def test_temporal_neighbor_report_contains_both_experimental_metrics():
+    records = [
+        {
+            "path": "a.wav",
+            "representation": {
+                "segments": [{"embedding": [1.0, 0.0]}, {"embedding": [0.0, 1.0]}]
+            },
+        },
+        {
+            "path": "b.wav",
+            "representation": {
+                "segments": [
+                    {"embedding": [1.0, 0.0]},
+                    {"embedding": [0.7, 0.7]},
+                    {"embedding": [0.0, 1.0]},
+                ]
+            },
+        },
+    ]
+    report = _neighbor_report(records, "audio.temporal", 1)
+    assert set(report["tracks"][0]["neighbors"]) == {"aligned", "dtw"}
+    assert report["tracks"][0]["neighbors"]["aligned"][0]["track"] == "b.wav"

tests/test_decoder.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+import wave
+
+import numpy as np
+import pytest
+
+from openmusic_analysis.audio.decoder import AudioDecoder
+from openmusic_analysis.errors import AudioDecodeError
+
+
+def test_decoder_rejects_invalid_audio(tmp_path: Path):
+    source = tmp_path / "invalid.mp3"
+    source.write_bytes(b"not an audio file")
+    with pytest.raises(AudioDecodeError):
+        AudioDecoder(timeout_seconds=5).decode(source)
+
+
+def test_decoder_produces_canonical_mono_float32_pcm(tmp_path: Path):
+    source = tmp_path / "valid.wav"
+    samples = (np.sin(np.linspace(0, 10, 8_000)) * 20_000).astype("<i2")
+    with wave.open(str(source), "wb") as output:
+        output.setnchannels(1)
+        output.setsampwidth(2)
+        output.setframerate(8_000)
+        output.writeframes(samples.tobytes())
+    decoded = AudioDecoder(canonical_sample_rate=16_000, timeout_seconds=5).decode(source)
+    assert decoded.sample_rate == 16_000
+    assert decoded.waveform.dtype == np.float32
+    assert decoded.waveform.ndim == 1
+    assert decoded.waveform.size == pytest.approx(16_000, abs=2)
+    assert decoded.metadata.source_sample_rate == 8_000

tests/test_lyrics.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from openmusic_analysis.analyzers.lyrics import BGEM3LyricsAnalyzer, LyricsPreprocessor
+from openmusic_analysis.errors import AnalysisError
+from openmusic_analysis.settings import LyricsConfig
+
+from .conftest import FakeTextEncoder
+
+
+def analyzer(max_tokens: int = 12) -> BGEM3LyricsAnalyzer:
+    return BGEM3LyricsAnalyzer(
+        FakeTextEncoder(),
+        LyricsPreprocessor(),
+        LyricsConfig(max_chunk_tokens=max_tokens, batch_size=4),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "lyrics",
+    [
+        "[Verse]\nHello, world!\nI remember you.\n\n[Chorus]\nCome home, come home.",
+        "[Куплет 1]\nЯ помню этот день.\n\n[Припев]\nВернись, вернись ко мне!",
+    ],
+)
+async def test_english_and_russian_are_finite_and_normalized(lyrics):
+    result = await analyzer().analyze(lyrics)
+    vector = np.asarray(result.embedding)
+    assert result.dimension == 6
+    assert np.isfinite(vector).all()
+    assert np.linalg.norm(vector) == pytest.approx(1.0, abs=1e-6)
+
+
+def test_preprocessor_preserves_multiline_sections_punctuation_and_repeated_chorus():
+    lyrics = "[ar:metadata]\r\n[Verse]\r\nHello, world!\r\n\r\n[Chorus]\r\nAgain!\r\n[Chorus]\r\nAgain!"
+    prepared = LyricsPreprocessor().prepare(lyrics)
+    assert "[ar:metadata]" not in prepared.normalized_text
+    assert "Hello, world!" in prepared.normalized_text
+    assert [section.label.lower() for section in prepared.sections] == [
+        "verse",
+        "chorus",
+        "chorus",
+    ]
+    assert [section.text for section in prepared.sections].count("Again!") == 2
+
+
+@pytest.mark.asyncio
+async def test_long_lyrics_use_token_chunks_and_deterministic_aggregation():
+    lyrics = "[Verse]\n" + " ".join(f"word{index}" for index in range(80))
+    first = await analyzer(max_tokens=10).analyze(lyrics)
+    second = await analyzer(max_tokens=10).analyze(lyrics)
+    assert first.analysis["chunk_count"] > 1
+    assert max(first.analysis["chunk_token_counts"]) <= 10
+    assert first.embedding == pytest.approx(second.embedding, abs=1e-7)
+
+
+@pytest.mark.asyncio
+async def test_empty_lyrics_are_rejected():
+    with pytest.raises(AnalysisError) as raised:
+        await analyzer().analyze(" \n\t ")
+    assert raised.value.code == "INVALID_LYRICS"

tests/test_registry.py

@@ -0,0 +1,19 @@
+from fastapi.testclient import TestClient
+
+from openmusic_analysis.api import create_app
+from openmusic_analysis.settings import Settings
+
+
+def test_models_endpoint_matches_actual_analyzers(service_bundle):
+    service, _, _ = service_bundle
+    response = TestClient(create_app(service=service, settings=Settings())).get("/v1/models")
+    assert response.status_code == 200
+    endpoint_models = response.json()["models"]
+    actual = [model.model_dump(mode="json") for model in service.registry.models()]
+    assert endpoint_models == actual
+    assert {item["representation"] for item in endpoint_models} == {
+        "audio.global",
+        "audio.temporal",
+        "lyrics.global",
+    }
+    assert all(len(item["model_version"]) == 40 for item in endpoint_models)

tests/test_temporal_similarity.py

@@ -0,0 +1,18 @@
+import numpy as np
+import pytest
+
+from openmusic_analysis.experiments import (
+    aligned_interpolation_similarity,
+    dtw_cosine_similarity,
+)
+
+
+def test_temporal_similarity_identical_and_length_invariant_baselines():
+    trajectory = np.asarray([[1, 0], [0.7, 0.7], [0, 1]], dtype=np.float32)
+    stretched = np.asarray(
+        [[1, 0], [0.9, 0.3], [0.7, 0.7], [0.3, 0.9], [0, 1]], dtype=np.float32
+    )
+    assert aligned_interpolation_similarity(trajectory, trajectory) == pytest.approx(1.0)
+    assert dtw_cosine_similarity(trajectory, trajectory) == pytest.approx(1.0)
+    assert aligned_interpolation_similarity(trajectory, stretched) > 0.95
+    assert dtw_cosine_similarity(trajectory, stretched) > 0.9

tools/__init__.py

@@ -0,0 +1 @@
+"""Developer and benchmarking tools."""

tools/nearest_neighbors.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import asyncio
+import csv
+import hashlib
+import json
+import sys
+from itertools import combinations
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from openmusic_analysis.application import build_service
+from openmusic_analysis.experiments import (
+    aligned_interpolation_similarity,
+    dtw_cosine_similarity,
+)
+
+
+AUDIO_EXTENSIONS = {".mp3", ".m4a", ".flac", ".wav", ".aac", ".ogg", ".opus"}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Benchmark OpenMusic representation neighbors")
+    parser.add_argument("folder", type=Path)
+    parser.add_argument(
+        "--representation",
+        choices=["audio.global", "audio.temporal", "lyrics.global"],
+        default="audio.global",
+    )
+    parser.add_argument("--top-k", type=int, default=5)
+    parser.add_argument("--cache-dir", type=Path, default=Path(".embedding_cache"))
+    parser.add_argument("--output", type=Path, default=Path("neighbors.json"))
+    parser.add_argument(
+        "--lyrics-suffix",
+        default=".txt",
+        help="Sidecar suffix used by lyrics.global (for example song.mp3 + song.txt)",
+    )
+    return parser.parse_args()
+
+
+async def main() -> None:
+    args = parse_args()
+    if args.top_k < 1:
+        raise SystemExit("--top-k must be at least 1")
+    service = build_service()
+    metadata = service.registry.analyzer(args.representation).metadata
+    files = sorted(
+        path for path in args.folder.rglob("*") if path.suffix.lower() in AUDIO_EXTENSIONS
+    )
+    if not files:
+        raise SystemExit(f"No supported audio files found under {args.folder}")
+    args.cache_dir.mkdir(parents=True, exist_ok=True)
+    records: list[dict[str, Any]] = []
+    for audio_path in files:
+        lyrics = (
+            _read_lyrics(audio_path, args.lyrics_suffix)
+            if args.representation == "lyrics.global"
+            else None
+        )
+        if args.representation == "lyrics.global" and lyrics is None:
+            print(f"skip {audio_path}: missing lyrics sidecar", file=sys.stderr)
+            continue
+        key = _cache_key(audio_path, lyrics, metadata.model_dump(exclude={"loaded"}))
+        cache_path = args.cache_dir / f"{key}.json"
+        if cache_path.exists():
+            representation = json.loads(cache_path.read_text(encoding="utf-8"))["representation"]
+        else:
+            response = await service.analyze(
+                audio_path,
+                lyrics=lyrics,
+                requested_representations=[args.representation],
+                track_id=str(audio_path.relative_to(args.folder)),
+                content_identity=f"sha256:{_file_hash(audio_path)}",
+            )
+            representation = response.representations[args.representation].model_dump()
+            _atomic_json_write(
+                cache_path,
+                {"cache_key": key, "source": str(audio_path), "representation": representation},
+            )
+        records.append({"path": str(audio_path), "representation": representation})
+    if len(records) < 2:
+        raise SystemExit("At least two analyzable tracks are required")
+    report = _neighbor_report(records, args.representation, args.top_k)
+    report["representation"] = args.representation
+    report["model"] = metadata.model_dump(exclude={"loaded"})
+    _atomic_json_write(args.output, report)
+    _write_csv(args.output.with_suffix(".csv"), report)
+    print(f"wrote {args.output} and {args.output.with_suffix('.csv')}")
+
+
+def _neighbor_report(
+    records: list[dict[str, Any]], representation: str, top_k: int
+) -> dict[str, Any]:
+    metrics = ["cosine"] if representation != "audio.temporal" else ["aligned", "dtw"]
+    scores: dict[str, dict[tuple[int, int], float]] = {metric: {} for metric in metrics}
+    for first_index, second_index in combinations(range(len(records)), 2):
+        first = records[first_index]["representation"]
+        second = records[second_index]["representation"]
+        if representation == "audio.temporal":
+            first_value = np.asarray([item["embedding"] for item in first["segments"]])
+            second_value = np.asarray([item["embedding"] for item in second["segments"]])
+            scores["aligned"][(first_index, second_index)] = aligned_interpolation_similarity(
+                first_value, second_value
+            )
+            scores["dtw"][(first_index, second_index)] = dtw_cosine_similarity(
+                first_value, second_value
+            )
+        else:
+            first_value = np.asarray(first["embedding"], dtype=np.float32)
+            second_value = np.asarray(second["embedding"], dtype=np.float32)
+            scores["cosine"][(first_index, second_index)] = float(first_value @ second_value)
+    tracks = []
+    for index, record in enumerate(records):
+        neighbors: dict[str, list[dict[str, Any]]] = {}
+        for metric in metrics:
+            ranked = []
+            for other_index, other in enumerate(records):
+                if other_index == index:
+                    continue
+                pair = tuple(sorted((index, other_index)))
+                ranked.append(
+                    {"track": other["path"], "similarity": scores[metric][pair]}
+                )
+            neighbors[metric] = sorted(
+                ranked, key=lambda item: item["similarity"], reverse=True
+            )[:top_k]
+        tracks.append({"track": record["path"], "neighbors": neighbors})
+    return {"tracks": tracks}
+
+
+def _cache_key(audio_path: Path, lyrics: str | None, model: dict[str, Any]) -> str:
+    payload = {
+        "content_hash": _file_hash(audio_path),
+        "lyrics_hash": hashlib.sha256((lyrics or "").encode("utf-8")).hexdigest(),
+        "model_id": model["model_id"],
+        "model_version": model["model_version"],
+        "preprocessing_version": model["preprocessing_version"],
+        "representation": model["representation"],
+        "configuration": model["configuration"],
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _file_hash(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as file:
+        while chunk := file.read(1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _read_lyrics(audio_path: Path, suffix: str) -> str | None:
+    sidecar = audio_path.with_suffix(suffix)
+    return sidecar.read_text(encoding="utf-8") if sidecar.exists() else None
+
+
+def _atomic_json_write(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
+    )
+    temporary.replace(path)
+
+
+def _write_csv(path: Path, report: dict[str, Any]) -> None:
+    with path.open("w", encoding="utf-8", newline="") as file:
+        writer = csv.DictWriter(file, fieldnames=["track", "metric", "rank", "neighbor", "similarity"])
+        writer.writeheader()
+        for track in report["tracks"]:
+            for metric, neighbors in track["neighbors"].items():
+                for rank, neighbor in enumerate(neighbors, start=1):
+                    writer.writerow(
+                        {
+                            "track": track["track"],
+                            "metric": metric,
+                            "rank": rank,
+                            "neighbor": neighbor["track"],
+                            "similarity": neighbor["similarity"],
+                        }
+                    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tools/profile_analysis.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import resource
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from openmusic_analysis.application import build_service
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(description="Measure model load and per-representation latency")
+    parser.add_argument("audio", type=Path)
+    parser.add_argument("--lyrics", type=Path)
+    args = parser.parse_args()
+    service = build_service()
+    lyrics = args.lyrics.read_text(encoding="utf-8") if args.lyrics else None
+    measurements = {}
+    started = time.perf_counter()
+    await service.load_models()
+    measurements["model_startup_seconds"] = time.perf_counter() - started
+    representations = ["audio.global", "audio.temporal"]
+    if lyrics is not None:
+        representations.append("lyrics.global")
+    for representation in representations:
+        started = time.perf_counter()
+        await service.analyze(
+            args.audio,
+            lyrics=lyrics,
+            requested_representations=[representation],
+            track_id=None,
+            content_identity=None,
+        )
+        measurements[f"{representation}_seconds"] = time.perf_counter() - started
+    measurements["peak_process_rss_platform_units"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+    try:
+        import torch
+
+        if torch.cuda.is_available():
+            measurements["peak_cuda_memory_bytes"] = torch.cuda.max_memory_allocated()
+    except ImportError:
+        pass
+    print(json.dumps(measurements, indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())