diff --git a/BACKLOG.md b/BACKLOG.md index c3db5a56..4ae1389a 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -30,3 +30,6 @@ All eight detectors emit fields visible in [`apps/backend/tests/test_analyze.py` ## Open - πŸ”² `services/patchSmith.ts` β€” generates Vital/Operator patch parameters from detected features. Slot: Phase 3 (not yet built); unique differentiator (download-ready preset output). Evaluate against `PURPOSE.md` decision framework before scheduling. + +## In Progress +- 🟑 **Audition samples (Phase 3 β€” branch `claude/gemini-sample-generation-U753E`).** Heuristic WAV/MIDI clips derived from Phase 1 measurements (and Phase 2 context when available) so producers can ear-check the measurement chain. PyTheory generates the musical plan, FluidSynth (with sine-additive fallback) renders audio, NumPy synthesizes drum one-shots, manifest carries citations. Design doc: [`docs/SAMPLE_GENERATION.md`](docs/SAMPLE_GENERATION.md). Endpoints: `POST/GET /api/analysis-runs/{run_id}/samples`. Adjacent to `patchSmith.ts` but solves a different problem β€” audition validates the measurements rather than producing a saveable preset. diff --git a/apps/backend/requirements.txt b/apps/backend/requirements.txt index 54696e71..5a27a597 100644 --- a/apps/backend/requirements.txt +++ b/apps/backend/requirements.txt @@ -70,3 +70,12 @@ typing_extensions==4.15.0 urllib3==2.6.3 google-genai>=1.14.0,<2.0.0 uvicorn==0.41.0 +# Phase 3 (audition samples). Both deps are optional: sample_theory.py has a +# pure-Python music-theory fallback and sample_synthesis.py has a NumPy +# sine-additive fallback. Tests exercise the fallbacks so CI doesn't require +# either of these to be importable, but installing them upgrades audition +# quality from "in-tune" to "musical." pyfluidsynth additionally needs the +# system `libfluidsynth` library and a General MIDI soundfont β€” see +# docs/SAMPLE_GENERATION.md. +pytheory>=0.42,<1.0 +pyfluidsynth>=1.3,<2.0 diff --git a/apps/backend/sample_drums.py b/apps/backend/sample_drums.py new file mode 100644 index 00000000..03fcf0e8 --- /dev/null +++ b/apps/backend/sample_drums.py @@ -0,0 +1,145 @@ +"""NumPy-based drum one-shot synthesis for audition samples. + +Phase 1 measures the kick's fundamental and decay (`kickDetail.fundamentalHz`, +`kickDetail.decayTimeMs`). We turn those into a sub-sine + transient click. +Snare and hi-hat are not directly measured β€” we render plausible defaults and +the manifest is explicit that those are heuristic, not measurement-grounded. + +PyTheory is not involved here; drums are not harmonic. Synthesis lives outside +the music-theory layer. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np + + +SAMPLE_RATE = 44_100 +_RNG = np.random.default_rng(seed=42) # deterministic noise so tests are stable + + +@dataclass(frozen=True) +class DrumOneShot: + samples: np.ndarray # float32, mono, range [-1, 1] + sample_rate: int + duration_seconds: float + label: str + + +def synth_kick( + *, + fundamental_hz: float = 55.0, + decay_time_ms: float = 250.0, + click_amount: float = 0.35, + duration_seconds: float = 0.6, +) -> DrumOneShot: + """Synthesize a kick: pitch-modulated sub-sine plus transient click. + + The pitch starts an octave above `fundamental_hz` and exponentially decays + to the fundamental over the first 30 ms. This mimics the body-and-thump + shape of an electronic kick well enough for an audition. + + `decay_time_ms` controls how long the body sustains. Phase 1 typically + measures values in the 150–400 ms range for electronic kicks. + """ + if fundamental_hz <= 0: + raise ValueError(f"fundamental_hz must be positive; got {fundamental_hz}") + if duration_seconds <= 0: + raise ValueError(f"duration_seconds must be positive; got {duration_seconds}") + + num_samples = int(round(SAMPLE_RATE * duration_seconds)) + t = np.arange(num_samples, dtype=np.float64) / SAMPLE_RATE + + # Pitch envelope: octave drop over the first 30 ms. + pitch_decay_s = 0.030 + pitch_env = np.exp(-t / pitch_decay_s) * fundamental_hz + fundamental_hz + phase = 2.0 * np.pi * np.cumsum(pitch_env) / SAMPLE_RATE + body = np.sin(phase) + + # Amplitude envelope: exponential decay with measured time constant. + decay_s = max(decay_time_ms, 50.0) / 1000.0 + amp_env = np.exp(-t / (decay_s * 0.4)) # slightly tighter than reported decay + body *= amp_env + + # Click: short high-frequency burst with its own fast envelope. + click_env = np.exp(-t / 0.004) # 4 ms decay + click_noise = _RNG.standard_normal(num_samples).astype(np.float64) + click = click_noise * click_env * click_amount + + samples = body * 0.85 + click + samples = _normalize(samples) + + return DrumOneShot( + samples=samples.astype(np.float32), + sample_rate=SAMPLE_RATE, + duration_seconds=duration_seconds, + label=f"Kick at {fundamental_hz:.0f} Hz", + ) + + +def synth_snare( + *, + body_hz: float = 200.0, + duration_seconds: float = 0.4, +) -> DrumOneShot: + """Snare: tone body + noise crack. Heuristic only β€” no Phase 1 ground truth.""" + num_samples = int(round(SAMPLE_RATE * duration_seconds)) + t = np.arange(num_samples, dtype=np.float64) / SAMPLE_RATE + + body_env = np.exp(-t / 0.08) + body = np.sin(2.0 * np.pi * body_hz * t) * body_env + + noise_env = np.exp(-t / 0.12) + noise = _RNG.standard_normal(num_samples) * noise_env + + # Simple high-shelf: take the running difference to emphasize highs. + noise_hp = np.empty_like(noise) + noise_hp[0] = noise[0] + noise_hp[1:] = noise[1:] - 0.97 * noise[:-1] + + samples = body * 0.4 + noise_hp * 0.85 + samples = _normalize(samples) + return DrumOneShot( + samples=samples.astype(np.float32), + sample_rate=SAMPLE_RATE, + duration_seconds=duration_seconds, + label="Snare (heuristic)", + ) + + +def synth_hat(*, duration_seconds: float = 0.18) -> DrumOneShot: + """Closed hi-hat: short filtered noise. Heuristic β€” no Phase 1 ground truth.""" + num_samples = int(round(SAMPLE_RATE * duration_seconds)) + t = np.arange(num_samples, dtype=np.float64) / SAMPLE_RATE + + env = np.exp(-t / 0.035) + noise = _RNG.standard_normal(num_samples) + + # Two-tap high-pass (same idea as snare, vectorized to match snare's style). + # The filter has no feedback, so it's an FIR we can express as array slices. + hp = np.empty_like(noise) + hp[0] = noise[0] + if num_samples > 1: + hp[1] = noise[1] - 0.98 * noise[0] + if num_samples > 2: + hp[2:] = noise[2:] - 1.96 * noise[1:-1] + 0.97 * noise[:-2] + + samples = hp * env + samples = _normalize(samples) * 0.7 + return DrumOneShot( + samples=samples.astype(np.float32), + sample_rate=SAMPLE_RATE, + duration_seconds=duration_seconds, + label="Closed hat (heuristic)", + ) + + +def _normalize(samples: np.ndarray, *, headroom_db: float = 1.5) -> np.ndarray: + """Peak-normalize to leave a small headroom margin.""" + peak = float(np.max(np.abs(samples))) if samples.size else 0.0 + if peak <= 1e-9: + return samples + target = 10 ** (-headroom_db / 20.0) + return samples * (target / peak) diff --git a/apps/backend/sample_generation.py b/apps/backend/sample_generation.py new file mode 100644 index 00000000..69e97982 --- /dev/null +++ b/apps/backend/sample_generation.py @@ -0,0 +1,403 @@ +"""Phase 3 audition sample generation orchestrator. + +Reads a Phase 1 result + optional Phase 2 result, produces audition WAV/MIDI +artifacts, and emits a manifest with citation metadata so every generated +sample traces back to a Phase 1 field. The manifest *is* the chain of custody +for this stage β€” it must remain the source of truth for "what justifies this +audio?" + +Shape of the manifest is documented in `docs/SAMPLE_GENERATION.md`. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import numpy as np +import soundfile as sf + +import sample_drums +import sample_synthesis +import sample_theory + +logger = logging.getLogger(__name__) + +SCHEMA_VERSION = "samples.v1" +FRAMING_NOTE = ( + "Heuristic audition. Verifies tonal/rhythmic foundation derived from " + "Phase 1 measurements. Not an Ableton-accurate reconstruction β€” follow " + "Phase 2 in Ableton for production character." +) + + +@dataclass(frozen=True) +class GenerationResult: + manifest: dict[str, Any] + artifact_paths: dict[str, Path] # sample_id -> WAV path + midi_paths: dict[str, Path] # sample_id -> MIDI path (subset of samples) + manifest_path: Path + + +def generate_samples( + *, + run_id: str, + phase1: dict[str, Any], + phase2: dict[str, Any] | None, + output_dir: Path, + pitch_note_hints: list[int] | None = None, + prefer_fluidsynth: bool = True, +) -> GenerationResult: + """Build the full audition set for a single run. + + `pitch_note_hints` may carry scale degrees pulled from the stem-summary + pitch/note translation output. When absent we render the default ascent. + + `prefer_fluidsynth` is plumbed through so tests can force the fallback + even if FluidSynth is installed on the dev machine. + """ + output_dir.mkdir(parents=True, exist_ok=True) + artifact_paths: dict[str, Path] = {} + midi_paths: dict[str, Path] = {} + samples: list[dict[str, Any]] = [] + selected_backend: sample_synthesis.Backend | None = None + selected_soundfont: str | None = None + + key_string = _safe_get(phase1, "key") + key_confidence = _safe_float(_safe_get(phase1, "keyConfidence")) + bpm = _safe_float(_safe_get(phase1, "bpm")) or 120.0 + + # --- Tonal samples (chord progression + bass) -------------------------- # + ctx: sample_theory.TheoryContext | None = None + if key_string: + try: + ctx = sample_theory.build_context( + key=key_string, bpm=bpm, key_confidence=key_confidence + ) + except ValueError as exc: + logger.info("Skipping tonal samples; unparsed key %r (%s)", key_string, exc) + + if ctx is not None: + low_confidence = bool(key_confidence is not None and key_confidence < 0.5) + + chord_plan = sample_theory.plan_chord_progression(ctx, bars=8) + chord_result = sample_synthesis.render_clip( + chord_plan, prefer_fluidsynth=prefer_fluidsynth + ) + selected_backend = chord_result.backend + selected_soundfont = chord_result.soundfont_path + chord_wav = output_dir / "tonal_chord_progression.wav" + chord_mid = output_dir / "tonal_chord_progression.mid" + sample_synthesis.write_wav(chord_result.samples, path=chord_wav) + sample_synthesis.write_midi(chord_plan, path=chord_mid) + artifact_paths["tonal_chord_progression"] = chord_wav + midi_paths["tonal_chord_progression"] = chord_mid + samples.append( + _sample_record( + sample_id="tonal_chord_progression", + label=_compose_chord_label(ctx, low_confidence), + category="tonal", + filename=chord_wav.name, + midi_filename=chord_mid.name, + duration_seconds=chord_result.duration_seconds, + confidence=_confidence_band(key_confidence), + low_confidence=low_confidence, + phase1_fields=["key", "keyConfidence", "bpm"], + phase2_recommendations=_phase2_tonal_citations(phase2), + rationale=( + f"8-bar diatonic progression in {ctx.root_name} {ctx.mode} at " + f"{ctx.tempo_bpm:.0f} BPM (key confidence " + f"{_format_confidence(key_confidence)})." + ), + ) + ) + + bass_plan = sample_theory.plan_bass_root(ctx, bars=8) + bass_result = sample_synthesis.render_clip( + bass_plan, prefer_fluidsynth=prefer_fluidsynth + ) + # `selected_backend` is already pinned by the chord render above β€” + # `render_clip` always returns a concrete backend string, so the bass + # render can never narrow or widen the choice. + bass_wav = output_dir / "tonal_bass_root.wav" + bass_mid = output_dir / "tonal_bass_root.mid" + sample_synthesis.write_wav(bass_result.samples, path=bass_wav) + sample_synthesis.write_midi(bass_plan, path=bass_mid) + artifact_paths["tonal_bass_root"] = bass_wav + midi_paths["tonal_bass_root"] = bass_mid + samples.append( + _sample_record( + sample_id="tonal_bass_root", + label=f"Bass root on {ctx.root_name}", + category="tonal", + filename=bass_wav.name, + midi_filename=bass_mid.name, + duration_seconds=bass_result.duration_seconds, + confidence=_confidence_band(key_confidence), + low_confidence=low_confidence, + phase1_fields=["key", "keyConfidence", "bpm", "bassDetail"] + if _safe_get(phase1, "bassDetail") + else ["key", "keyConfidence", "bpm"], + phase2_recommendations=_phase2_tonal_citations(phase2), + rationale=( + f"Sustained root tone at MIDI bass register. Tonic " + f"derived from Phase 1 key ({ctx.root_name} {ctx.mode})." + ), + ) + ) + + # --- Drum one-shots ---------------------------------------------------- # + kick_detail = _safe_get(phase1, "kickDetail") or {} + kick_fundamental = _safe_float(kick_detail.get("fundamentalHz")) or 55.0 + kick_decay = _safe_float(kick_detail.get("decayTimeMs")) or 250.0 + kick_confidence_raw = _safe_float(kick_detail.get("confidence")) + + kick = sample_drums.synth_kick( + fundamental_hz=kick_fundamental, + decay_time_ms=kick_decay, + ) + kick_path = output_dir / "drum_kick.wav" + sf.write(str(kick_path), kick.samples, kick.sample_rate, subtype="PCM_16") + artifact_paths["drum_kick"] = kick_path + samples.append( + _sample_record( + sample_id="drum_kick", + label=kick.label, + category="drums", + filename=kick_path.name, + midi_filename=None, + duration_seconds=kick.duration_seconds, + confidence=_confidence_band(kick_confidence_raw), + low_confidence=bool( + kick_confidence_raw is not None and kick_confidence_raw < 0.5 + ), + phase1_fields=["kickDetail.fundamentalHz", "kickDetail.decayTimeMs"] + if kick_detail + else [], + phase2_recommendations=_phase2_kick_citations(phase2), + rationale=( + f"Sub-sine at {kick_fundamental:.0f} Hz with {kick_decay:.0f} ms " + "decay, derived from measured kickDetail." + if kick_detail + else "Default kick β€” Phase 1 did not surface kickDetail." + ), + ) + ) + + snare = sample_drums.synth_snare() + snare_path = output_dir / "drum_snare.wav" + sf.write(str(snare_path), snare.samples, snare.sample_rate, subtype="PCM_16") + artifact_paths["drum_snare"] = snare_path + samples.append( + _sample_record( + sample_id="drum_snare", + label=snare.label, + category="drums", + filename=snare_path.name, + midi_filename=None, + duration_seconds=snare.duration_seconds, + confidence="LOW", + low_confidence=True, + phase1_fields=[], + phase2_recommendations=[], + rationale=( + "Heuristic snare β€” Phase 1 does not measure a snare fundamental. " + "Provided for kit completeness; do not treat as measurement-grounded." + ), + ) + ) + + hat = sample_drums.synth_hat() + hat_path = output_dir / "drum_hat.wav" + sf.write(str(hat_path), hat.samples, hat.sample_rate, subtype="PCM_16") + artifact_paths["drum_hat"] = hat_path + samples.append( + _sample_record( + sample_id="drum_hat", + label=hat.label, + category="drums", + filename=hat_path.name, + midi_filename=None, + duration_seconds=hat.duration_seconds, + confidence="LOW", + low_confidence=True, + phase1_fields=[], + phase2_recommendations=[], + rationale=( + "Heuristic closed hi-hat β€” Phase 1 does not measure a hat. " + "Provided for kit completeness; do not treat as measurement-grounded." + ), + ) + ) + + # --- Melody lead phrase ------------------------------------------------ # + if ctx is not None: + melody_detail = _safe_get(phase1, "melodyDetail") + melody_plan = sample_theory.plan_melody_phrase( + ctx, scale_degrees=pitch_note_hints, bars=4 + ) + if melody_plan is not None: + melody_result = sample_synthesis.render_clip( + melody_plan, prefer_fluidsynth=prefer_fluidsynth + ) + # See bass-render note above β€” `selected_backend` is pinned at the + # first chord render and `render_clip` never returns falsy. + melody_wav = output_dir / "melody_lead.wav" + melody_mid = output_dir / "melody_lead.mid" + sample_synthesis.write_wav(melody_result.samples, path=melody_wav) + sample_synthesis.write_midi(melody_plan, path=melody_mid) + artifact_paths["melody_lead"] = melody_wav + midi_paths["melody_lead"] = melody_mid + + phase1_fields = ["key", "keyConfidence", "bpm"] + if melody_detail: + phase1_fields.append("melodyDetail") + hints_used = pitch_note_hints is not None and len(pitch_note_hints) > 0 + samples.append( + _sample_record( + sample_id="melody_lead", + label="Lead phrase in detected key", + category="melody", + filename=melody_wav.name, + midi_filename=melody_mid.name, + duration_seconds=melody_result.duration_seconds, + confidence=_confidence_band(key_confidence), + low_confidence=bool( + key_confidence is not None and key_confidence < 0.5 + ), + phase1_fields=phase1_fields, + phase2_recommendations=[], + rationale=( + "Scale-degree sequence from pitch/note translation hints, " + f"rendered on a square-lead voice in {ctx.root_name} {ctx.mode}." + if hints_used + else ( + f"Default 1-2-3-5 ascent in {ctx.root_name} {ctx.mode} β€” " + "no pitch/note translation hints available." + ) + ), + ) + ) + + manifest: dict[str, Any] = { + "schemaVersion": SCHEMA_VERSION, + "runId": run_id, + "generatedAt": datetime.now(UTC).isoformat(), + "synthesisBackend": selected_backend or "sine_fallback", + "soundfont": selected_soundfont, + "framing": FRAMING_NOTE, + "theoryBackend": ( + "pytheory" if sample_theory.pytheory_available() else "fallback" + ), + "samples": samples, + } + manifest_path = output_dir / "samples_manifest.json" + manifest_path.write_text(json.dumps(manifest, indent=2)) + + return GenerationResult( + manifest=manifest, + artifact_paths=artifact_paths, + midi_paths=midi_paths, + manifest_path=manifest_path, + ) + + +# --- Helpers ---------------------------------------------------------------- # + + +def _sample_record( + *, + sample_id: str, + label: str, + category: str, + filename: str, + midi_filename: str | None, + duration_seconds: float, + confidence: str, + low_confidence: bool, + phase1_fields: list[str], + phase2_recommendations: list[str], + rationale: str, +) -> dict[str, Any]: + record: dict[str, Any] = { + "id": sample_id, + "label": label, + "category": category, + "filename": filename, + "mimeType": "audio/wav", + "durationSeconds": round(duration_seconds, 3), + "confidence": confidence, + "lowConfidence": low_confidence, + "cites": { + "phase1Fields": phase1_fields, + "phase2Recommendations": phase2_recommendations, + "rationale": rationale, + }, + } + if midi_filename is not None: + record["midiFilename"] = midi_filename + return record + + +def _safe_get(payload: dict[str, Any] | None, key: str) -> Any: + if not isinstance(payload, dict): + return None + return payload.get(key) + + +def _safe_float(value: Any) -> float | None: + if value is None: + return None + try: + result = float(value) + except (TypeError, ValueError): + return None + if result != result: # NaN check + return None + return result + + +def _confidence_band(value: float | None) -> str: + if value is None: + return "MED" + if value >= 0.75: + return "HIGH" + if value >= 0.5: + return "MED" + return "LOW" + + +def _format_confidence(value: float | None) -> str: + return "n/a" if value is None else f"{value:.2f}" + + +def _compose_chord_label( + ctx: sample_theory.TheoryContext, low_confidence: bool +) -> str: + base = f"Chord progression in {ctx.root_name} {ctx.mode}" + return f"{base} (low-confidence key)" if low_confidence else base + + +def _phase2_tonal_citations(phase2: dict[str, Any] | None) -> list[str]: + if not isinstance(phase2, dict): + return [] + cites: list[str] = [] + if phase2.get("styleProfile"): + cites.append("styleProfile.authoritativeMeasurements.key") + if isinstance(phase2.get("sonicElements"), dict): + if phase2["sonicElements"].get("harmonicContent"): + cites.append("sonicElements.harmonicContent") + return cites + + +def _phase2_kick_citations(phase2: dict[str, Any] | None) -> list[str]: + if not isinstance(phase2, dict): + return [] + sonic = phase2.get("sonicElements") + if isinstance(sonic, dict) and sonic.get("kick"): + return ["sonicElements.kick"] + return [] diff --git a/apps/backend/sample_synthesis.py b/apps/backend/sample_synthesis.py new file mode 100644 index 00000000..b81a1547 --- /dev/null +++ b/apps/backend/sample_synthesis.py @@ -0,0 +1,288 @@ +"""MIDI plan β†’ WAV rendering for audition samples. + +Two render paths: + +1. **FluidSynth** (`pyfluidsynth` + a General MIDI soundfont) β€” high-quality. + Used when both the Python binding and a soundfont file are available. +2. **Sine-additive fallback** β€” pure NumPy. Always available; in-tune; raw. + +Both paths produce the same float32 mono 44.1 kHz numpy array, so callers +don't need to branch on which backend was selected. The selected backend is +recorded on `RenderResult.backend` so it can flow into the manifest. + +MIDI artifacts are emitted via `pretty_midi`, which is already a hard dep, so +the user can drop a `.mid` into Ableton even if the audio render is rough. +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Literal + +import numpy as np +import pretty_midi # type: ignore[import-untyped] +import soundfile as sf + +from sample_theory import ClipPlan + +logger = logging.getLogger(__name__) + +SAMPLE_RATE = 44_100 + +# --- FluidSynth probe ------------------------------------------------------- # + +try: # pragma: no cover - import probe + import fluidsynth # type: ignore[import-untyped] + + _FLUIDSYNTH_IMPORTABLE = True +except Exception as exc: # pragma: no cover - exercised when pyfluidsynth absent + fluidsynth = None + _FLUIDSYNTH_IMPORTABLE = False + logger.info("pyfluidsynth not importable (%s); using sine fallback.", exc) + + +_DEFAULT_SOUNDFONT_CANDIDATES: tuple[Path, ...] = ( + Path(__file__).parent / "assets" / "soundfonts" / "default.sf2", + Path("/usr/share/sounds/sf2/FluidR3_GM.sf2"), + Path("/usr/share/sounds/sf2/default-GM.sf2"), + Path("/usr/share/sounds/sf3/default-GM.sf3"), + Path("/opt/homebrew/share/fluid-synth/sf2/FluidR3_GM.sf2"), +) + + +Backend = Literal["fluidsynth", "sine_fallback"] + + +@dataclass(frozen=True) +class RenderResult: + samples: np.ndarray # float32 mono, range ~[-1, 1] + sample_rate: int + backend: Backend + soundfont_path: str | None + duration_seconds: float + + +def locate_soundfont(explicit: Path | str | None = None) -> Path | None: + """Resolve a soundfont path from explicit input β†’ env var β†’ known locations.""" + if explicit is not None: + path = Path(explicit) + return path if path.is_file() else None + env_path = os.environ.get("SONIC_ANALYZER_SOUNDFONT") + if env_path and Path(env_path).is_file(): + return Path(env_path) + for candidate in _DEFAULT_SOUNDFONT_CANDIDATES: + if candidate.is_file(): + return candidate + return None + + +def render_clip( + plan: ClipPlan, + *, + soundfont: Path | str | None = None, + prefer_fluidsynth: bool = True, +) -> RenderResult: + """Render a ClipPlan to an in-memory float32 audio buffer. + + Picks the best available backend. Caller can force the fallback by passing + `prefer_fluidsynth=False`; tests use that to exercise the fallback path + even when FluidSynth happens to be installed. + """ + if prefer_fluidsynth and _FLUIDSYNTH_IMPORTABLE: + sf_path = locate_soundfont(soundfont) + if sf_path is not None: + try: + return _render_with_fluidsynth(plan, sf_path) + except Exception as exc: # pragma: no cover - hard to trigger in CI + logger.warning( + "FluidSynth render failed (%s); falling back to sine synth.", exc + ) + return _render_with_sine_fallback(plan) + + +def write_wav(samples: np.ndarray, *, path: Path, sample_rate: int = SAMPLE_RATE) -> None: + """Write float32 samples to a 16-bit PCM WAV at the conventional bit depth.""" + path.parent.mkdir(parents=True, exist_ok=True) + clipped = np.clip(samples, -1.0, 1.0).astype(np.float32) + sf.write(str(path), clipped, sample_rate, subtype="PCM_16") + + +def write_midi(plan: ClipPlan, *, path: Path) -> None: + """Emit a MIDI file from a ClipPlan so users can audition in Ableton.""" + pm = pretty_midi.PrettyMIDI(initial_tempo=plan.tempo_bpm) + inst = pretty_midi.Instrument(program=plan.program) + beats_per_second = plan.tempo_bpm / 60.0 + for note in plan.notes: + start_seconds = note.start_beat / beats_per_second + end_seconds = (note.start_beat + note.duration_beats) / beats_per_second + inst.notes.append( + pretty_midi.Note( + velocity=int(np.clip(note.velocity, 1, 127)), + pitch=int(np.clip(note.pitch_midi, 0, 127)), + start=float(start_seconds), + end=float(end_seconds), + ) + ) + pm.instruments.append(inst) + path.parent.mkdir(parents=True, exist_ok=True) + pm.write(str(path)) + + +# --- FluidSynth path -------------------------------------------------------- # + + +def _render_with_fluidsynth(plan: ClipPlan, soundfont_path: Path) -> RenderResult: # pragma: no cover - exercised only when FluidSynth + a soundfont are available locally + """Offline-render a ClipPlan via pyfluidsynth. + + We bypass an audio driver entirely (`driver=None`) and use `get_samples()` + to pull rendered audio out in fixed-size blocks. That keeps the call site + headless-server-safe. + """ + if fluidsynth is None: + raise RuntimeError("pyfluidsynth missing despite probe passing") + + seconds_total = (plan.duration_beats / plan.tempo_bpm) * 60.0 + total_samples = int(round(SAMPLE_RATE * seconds_total)) + + synth = fluidsynth.Synth(samplerate=float(SAMPLE_RATE)) + sfid = synth.sfload(str(soundfont_path)) + # Channel 0, bank 0, preset = clip's GM program. + synth.program_select(0, sfid, 0, plan.program) + + # Build an event timeline: each note becomes a noteon and a noteoff at the + # right sample index. Sorted so we can iterate forward. + beats_per_second = plan.tempo_bpm / 60.0 + events: list[tuple[int, str, int, int]] = [] + for note in plan.notes: + on_sample = int(round(note.start_beat / beats_per_second * SAMPLE_RATE)) + off_sample = int( + round( + (note.start_beat + note.duration_beats) / beats_per_second * SAMPLE_RATE + ) + ) + events.append((on_sample, "on", note.pitch_midi, int(note.velocity))) + events.append((off_sample, "off", note.pitch_midi, 0)) + events.sort(key=lambda e: (e[0], 0 if e[1] == "off" else 1)) + + block = 1024 + buffer = np.zeros(total_samples, dtype=np.float32) + event_idx = 0 + cursor = 0 + while cursor < total_samples: + while event_idx < len(events) and events[event_idx][0] <= cursor: + _, kind, pitch, velocity = events[event_idx] + if kind == "on": + synth.noteon(0, pitch, velocity) + else: + synth.noteoff(0, pitch) + event_idx += 1 + chunk_size = min(block, total_samples - cursor) + # get_samples returns interleaved stereo int16. Mix to mono float32. + raw = synth.get_samples(chunk_size) + stereo = np.asarray(raw, dtype=np.int16).reshape(-1, 2) + mono = stereo.mean(axis=1).astype(np.float32) / 32768.0 + buffer[cursor : cursor + chunk_size] = mono[:chunk_size] + cursor += chunk_size + + synth.delete() + return RenderResult( + samples=buffer, + sample_rate=SAMPLE_RATE, + backend="fluidsynth", + soundfont_path=str(soundfont_path), + duration_seconds=seconds_total, + ) + + +# --- Sine fallback ---------------------------------------------------------- # + + +def _render_with_sine_fallback(plan: ClipPlan) -> RenderResult: + """Sine-additive synth with a simple ADSR envelope. + + Three harmonics (1f, 2f, 3f) with decreasing amplitude give a slightly + less plain sound than a pure sine, without straying into territory we'd + have to defend musically. + """ + seconds_total = (plan.duration_beats / plan.tempo_bpm) * 60.0 + total_samples = int(round(SAMPLE_RATE * seconds_total)) + buffer = np.zeros(total_samples, dtype=np.float64) + + beats_per_second = plan.tempo_bpm / 60.0 + for note in plan.notes: + start_sample = int(round(note.start_beat / beats_per_second * SAMPLE_RATE)) + note_samples = int( + round(note.duration_beats / beats_per_second * SAMPLE_RATE) + ) + if note_samples <= 0: + continue + end_sample = min(start_sample + note_samples, total_samples) + if start_sample >= total_samples: + continue + + actual_samples = end_sample - start_sample + t = np.arange(actual_samples, dtype=np.float64) / SAMPLE_RATE + freq = 440.0 * (2.0 ** ((note.pitch_midi - 69) / 12.0)) + + voice = ( + np.sin(2.0 * np.pi * freq * t) * 0.6 + + np.sin(2.0 * np.pi * (2.0 * freq) * t) * 0.25 + + np.sin(2.0 * np.pi * (3.0 * freq) * t) * 0.12 + ) + envelope = _adsr_envelope(actual_samples, sample_rate=SAMPLE_RATE) + velocity_scale = note.velocity / 127.0 + buffer[start_sample:end_sample] += voice * envelope * velocity_scale + + # Soft normalization to leave headroom. + peak = float(np.max(np.abs(buffer))) if buffer.size else 0.0 + if peak > 1e-9: + buffer *= 0.8 / peak + + return RenderResult( + samples=buffer.astype(np.float32), + sample_rate=SAMPLE_RATE, + backend="sine_fallback", + soundfont_path=None, + duration_seconds=seconds_total, + ) + + +def _adsr_envelope( + num_samples: int, + *, + sample_rate: int, + attack_s: float = 0.01, + decay_s: float = 0.04, + sustain_level: float = 0.7, + release_s: float = 0.12, +) -> np.ndarray: + """Standard 4-stage envelope. Length matches the note; trims if too short.""" + env = np.ones(num_samples, dtype=np.float64) * sustain_level + + attack_samples = min(int(attack_s * sample_rate), num_samples) + if attack_samples > 0: + env[:attack_samples] = np.linspace(0.0, 1.0, attack_samples, dtype=np.float64) + + decay_start = attack_samples + decay_samples = min(int(decay_s * sample_rate), num_samples - decay_start) + if decay_samples > 0: + env[decay_start : decay_start + decay_samples] = np.linspace( + 1.0, sustain_level, decay_samples, dtype=np.float64 + ) + + release_samples = min(int(release_s * sample_rate), num_samples) + if release_samples > 0: + env[-release_samples:] *= np.linspace( + 1.0, 0.0, release_samples, dtype=np.float64 + ) + return env + + +def fluidsynth_available() -> bool: + """True iff both the python binding and a soundfont file are reachable.""" + if not _FLUIDSYNTH_IMPORTABLE: + return False + return locate_soundfont() is not None diff --git a/apps/backend/sample_theory.py b/apps/backend/sample_theory.py new file mode 100644 index 00000000..9dd785d6 --- /dev/null +++ b/apps/backend/sample_theory.py @@ -0,0 +1,337 @@ +"""Music-theory adapter for Phase 3 audition sample generation. + +Takes Phase 1 measurement output (key, bpm, optional melody/stem hints) and +produces structured MIDI plans that downstream synthesis can render to audio. + +Primary path uses PyTheory (https://github.com/kennethreitz/pytheory). When +that import fails we fall back to a self-contained Western music-theory +implementation so the audition feature remains functional in lean environments +and so tests do not depend on pytheory being importable. + +Contract: both paths return the same dataclass shapes with the same MIDI note +numbers for the same input. Test coverage exercises both paths. +""" + +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass, field +from typing import Literal + +logger = logging.getLogger(__name__) + +# --- Pytheory probe (graceful, no-op if absent) ----------------------------- # + +try: # pragma: no cover - import probe; behavior covered by both branches + import pytheory # type: ignore[import-untyped] + + _PYTHEORY_AVAILABLE = True +except Exception as exc: # pragma: no cover - exercised when pytheory absent + pytheory = None + _PYTHEORY_AVAILABLE = False + logger.info("pytheory not importable (%s); using pure-Python fallback.", exc) + + +def pytheory_available() -> bool: + """Expose probe so callers and tests can branch deterministically.""" + return _PYTHEORY_AVAILABLE + + +# --- Pitch-class constants -------------------------------------------------- # + +_PITCH_CLASS: dict[str, int] = { + "C": 0, "C#": 1, "Db": 1, "D": 2, "D#": 3, "Eb": 3, + "E": 4, "F": 5, "F#": 6, "Gb": 6, "G": 7, "G#": 8, "Ab": 8, + "A": 9, "A#": 10, "Bb": 10, "B": 11, +} + +# Intervals are semitones from tonic. Restricted to the modes Phase 1 actually +# emits (major / minor) plus the common modal options the melody detector may +# surface. Add more if/when needed. +_SCALE_INTERVALS: dict[str, tuple[int, ...]] = { + "major": (0, 2, 4, 5, 7, 9, 11), + "minor": (0, 2, 3, 5, 7, 8, 10), + "dorian": (0, 2, 3, 5, 7, 9, 10), + "phrygian": (0, 1, 3, 5, 7, 8, 10), + "lydian": (0, 2, 4, 6, 7, 9, 11), + "mixolydian": (0, 2, 4, 5, 7, 9, 10), +} + +# Diatonic chord progressions in scale-degree notation. We pick genre-neutral +# loops that survive being played as plain triads; the audition is not the +# place to be clever with secondary dominants. +_DIATONIC_PROGRESSIONS: dict[str, tuple[int, ...]] = { + "major": (1, 6, 4, 5), # I vi IV V + "minor": (1, 6, 7, 5), # i VI VII V +} + + +Mode = Literal["major", "minor", "dorian", "phrygian", "lydian", "mixolydian"] + + +@dataclass(frozen=True) +class NoteEvent: + """A single MIDI-style note event with absolute beat timing.""" + + pitch_midi: int + start_beat: float + duration_beats: float + velocity: int = 96 + + +@dataclass(frozen=True) +class ClipPlan: + """Render-ready plan: a sequence of note events at a known tempo.""" + + tempo_bpm: float + duration_beats: float + notes: list[NoteEvent] + program: int = 0 # General MIDI program number (0 = Acoustic Grand Piano) + + +@dataclass(frozen=True) +class TheoryContext: + """Parsed key + tempo, plus the source-of-truth confidence carried through.""" + + root_pc: int + mode: Mode + root_name: str # canonical "F#" / "Bb" β€” what we display + tempo_bpm: float + key_confidence: float | None + backend: Literal["pytheory", "fallback"] + + +# --- Public API ------------------------------------------------------------- # + + +def parse_key(key_string: str) -> tuple[int, Mode, str]: + """Parse a Phase 1-style key string into (pitch_class, mode, display_root). + + Accepts inputs like "F# minor", "C major", "Bb dorian", or just "C" + (mode defaults to major to match the Phase 1 convention). + """ + if not key_string or not isinstance(key_string, str): + raise ValueError(f"empty or non-string key: {key_string!r}") + + cleaned = key_string.strip() + # Tolerate trailing qualifiers like "(uncertain)"; strip parenthetical noise. + cleaned = re.sub(r"\s*\(.*?\)\s*$", "", cleaned).strip() + if not cleaned: + raise ValueError(f"empty key after cleaning: {key_string!r}") + + parts = cleaned.split() + root_token = parts[0] + mode_token = parts[1].lower() if len(parts) > 1 else "major" + + # Normalize root: "f#" -> "F#"; reject unknown roots loudly so the caller + # can degrade gracefully rather than silently mis-tuning. + normalized = root_token[:1].upper() + root_token[1:] + if normalized not in _PITCH_CLASS: + raise ValueError(f"unrecognized root note: {root_token!r} in {key_string!r}") + + if mode_token not in _SCALE_INTERVALS: + # Phase 1 occasionally emits "Minor" capitalized or "min"/"maj" abbreviations. + # Map a few common variants before giving up. + mode_token = {"min": "minor", "maj": "major"}.get(mode_token, mode_token) + if mode_token not in _SCALE_INTERVALS: + raise ValueError(f"unsupported mode: {mode_token!r} in {key_string!r}") + + return _PITCH_CLASS[normalized], mode_token, normalized # type: ignore[return-value] + + +def build_context( + *, key: str, bpm: float, key_confidence: float | None = None +) -> TheoryContext: + """Build a TheoryContext from Phase 1 outputs.""" + root_pc, mode, display = parse_key(key) + backend: Literal["pytheory", "fallback"] = ( + "pytheory" if _PYTHEORY_AVAILABLE else "fallback" + ) + return TheoryContext( + root_pc=root_pc, + mode=mode, + root_name=display, + tempo_bpm=float(bpm), + key_confidence=key_confidence, + backend=backend, + ) + + +def plan_chord_progression( + ctx: TheoryContext, *, bars: int = 8, voicing_octave: int = 4 +) -> ClipPlan: + """Build a diatonic chord-progression plan for audition. + + Two bars per chord; we cycle through a 4-chord loop. At 8 bars that's two + full repeats. Voicing places the chord root near the requested octave. + """ + if bars < 2 or bars % 2 != 0: + raise ValueError(f"bars must be an even integer >= 2; got {bars}") + + # The diatonic progression dict only covers strict major/minor today. + # For other modes, treat them as their parent (major-ish or minor-ish). + progression_key = "major" if ctx.mode in {"major", "lydian", "mixolydian"} else "minor" + degrees = _DIATONIC_PROGRESSIONS[progression_key] + + notes: list[NoteEvent] = [] + beats_per_chord = 8.0 # two 4/4 bars per chord + for chord_index in range(bars // 2): + degree = degrees[chord_index % len(degrees)] + chord_pitches = _diatonic_triad_midi( + ctx.root_pc, ctx.mode, degree, base_octave=voicing_octave + ) + start = chord_index * beats_per_chord + for pitch in chord_pitches: + notes.append( + NoteEvent( + pitch_midi=pitch, + start_beat=start, + duration_beats=beats_per_chord, + velocity=88, + ) + ) + + return ClipPlan( + tempo_bpm=ctx.tempo_bpm, + duration_beats=bars * 4.0, + notes=notes, + program=0, # Acoustic Grand Piano (GM #1, zero-indexed) + ) + + +def plan_bass_root(ctx: TheoryContext, *, bars: int = 8) -> ClipPlan: + """Sustained bass note on the tonic, two MIDI octaves below voicing range.""" + bass_pitch = _midi_from_pc(ctx.root_pc, octave=2) + # Two-bar sustains; let the user actually hear pitch. + sustained_beats = 8.0 + notes: list[NoteEvent] = [] + for chord_index in range(bars // 2): + notes.append( + NoteEvent( + pitch_midi=bass_pitch, + start_beat=chord_index * sustained_beats, + duration_beats=sustained_beats, + velocity=100, + ) + ) + return ClipPlan( + tempo_bpm=ctx.tempo_bpm, + duration_beats=bars * 4.0, + notes=notes, + program=33, # Electric Bass (finger), GM #34 + ) + + +def plan_melody_phrase( + ctx: TheoryContext, + *, + scale_degrees: list[int] | None = None, + bars: int = 4, +) -> ClipPlan | None: + """Render a short lead phrase from scale-degree hints. + + If no scale degrees are supplied we fabricate a simple ascent (1-2-3-5) so + the user at least hears the key context. Returns None if a sensible plan + can't be built (defensive β€” calling code will then omit the artifact). + """ + if scale_degrees is None or not scale_degrees: + # Default ascent: tonic β†’ 2nd β†’ 3rd β†’ 5th β†’ 3rd. Familiar enough to be + # recognizable as "in the key" without sounding like a real melody. + scale_degrees = [1, 2, 3, 5, 3, 1] + + # Reject obviously bad inputs rather than emitting garbage. + cleaned: list[int] = [d for d in scale_degrees if isinstance(d, int) and 1 <= d <= 7] + if not cleaned: + return None + + notes: list[NoteEvent] = [] + beats_per_note = (bars * 4.0) / max(len(cleaned), 1) + for index, degree in enumerate(cleaned): + pitch = _diatonic_scale_pitch(ctx.root_pc, ctx.mode, degree, octave=5) + notes.append( + NoteEvent( + pitch_midi=pitch, + start_beat=index * beats_per_note, + duration_beats=beats_per_note * 0.85, + velocity=92, + ) + ) + return ClipPlan( + tempo_bpm=ctx.tempo_bpm, + duration_beats=bars * 4.0, + notes=notes, + program=80, # Lead 1 (square), GM #81 β€” sounds synthetic, fits an EDM frame + ) + + +# --- Internal helpers ------------------------------------------------------- # + + +def _midi_from_pc(pitch_class: int, *, octave: int) -> int: + """C4 = 60. Map a pitch class + octave number to a MIDI note.""" + return (octave + 1) * 12 + (pitch_class % 12) + + +def _diatonic_scale_pitch( + root_pc: int, mode: str, scale_degree: int, *, octave: int +) -> int: + """1-indexed scale degree β†’ MIDI note. Degree 8 wraps to next octave.""" + intervals = _SCALE_INTERVALS[mode] + if scale_degree < 1: + raise ValueError(f"scale_degree must be >= 1; got {scale_degree}") + degree_index = (scale_degree - 1) % 7 + octave_offset = (scale_degree - 1) // 7 + semitones = intervals[degree_index] + 12 * octave_offset + return _midi_from_pc(root_pc, octave=octave) + semitones + + +def _diatonic_triad_midi( + root_pc: int, mode: str, scale_degree: int, *, base_octave: int +) -> tuple[int, int, int]: + """Stack thirds within the scale to build a triad on the given degree. + + Non-tonic chords are dropped by an octave so the progression voice-leads + within a narrow range around the tonic rather than ascending forever. The + tonic stays at the requested `base_octave` and acts as the anchor. + """ + intervals = _SCALE_INTERVALS[mode] + # Build a 14-note scale (two octaves) so stacking-by-third never wraps off. + extended = [intervals[i % 7] + 12 * (i // 7) for i in range(14)] + base_midi = _midi_from_pc(root_pc, octave=base_octave) + degree_index = (scale_degree - 1) % 7 + + root = base_midi + extended[degree_index] + third = base_midi + extended[degree_index + 2] + fifth = base_midi + extended[degree_index + 4] + if extended[degree_index] > 0: + # Anchor non-tonic chords below the tonic so vi / IV / V don't stack + # an octave above the I β€” that sounds wrong and obscures the cadence. + root -= 12 + third -= 12 + fifth -= 12 + return root, third, fifth + + +def cite_pytheory_if_used(ctx: TheoryContext) -> dict[str, object]: + """Optional helper: probe pytheory at runtime for a fingerprint we can log. + + Best-effort. If pytheory is importable but exposes a different surface than + we expect, we don't fail β€” we just return what we know. + """ + if not _PYTHEORY_AVAILABLE: + return {"backend": "fallback"} + info: dict[str, object] = {"backend": "pytheory"} + try: # pragma: no cover - tiny defensive probe + version = getattr(pytheory, "__version__", None) + if version: + info["version"] = str(version) + # PyTheory exposes a Tone class; use it as a soft sanity check that the + # library is healthy. We don't actually need its output β€” fallback + # tables are authoritative for MIDI numbers. + tone_cls = getattr(pytheory, "Tone", None) + if tone_cls is not None and hasattr(tone_cls, "from_string"): + info["toneClass"] = True + except Exception as exc: # pragma: no cover + info["probeError"] = str(exc) + return info diff --git a/apps/backend/server.py b/apps/backend/server.py index 9281154a..181576d1 100644 --- a/apps/backend/server.py +++ b/apps/backend/server.py @@ -143,6 +143,8 @@ _validate_phase2_semantics, ) +import server_samples + app = FastAPI(title="Sonic Analyzer Local API") @@ -2366,6 +2368,77 @@ async def get_run_source_audio( ) +# ── Audition samples (Phase 3) ─────────────────────────────────────────────── +# +# Heuristic reconstructions of the track's tonal foundation + drum kit, derived +# from Phase 1 measurements (and enriched by Phase 2 when available). Used by +# the UI to let producers ear-check the measurement chain. See +# `docs/SAMPLE_GENERATION.md` for the chain-of-custody framing. + +@app.post("/api/analysis-runs/{run_id}/samples") +async def create_run_samples( + run_id: str, + force: bool = Query(False, description="Regenerate even if a manifest exists"), + x_asa_user_id: str | None = Header(None), + x_asa_user_email: str | None = Header(None), +) -> JSONResponse: + user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email) + if isinstance(user_context, JSONResponse): + return user_context + runtime = get_analysis_runtime() + try: + snapshot = runtime.get_run(run_id, owner_user_id=user_context.user_id) + except (KeyError, PermissionError): + return _run_not_found_response(run_id) + + try: + manifest = await asyncio.to_thread( + server_samples.generate_and_register_samples, + runtime=runtime, + run_id=run_id, + snapshot=snapshot, + force=force, + ) + except server_samples.SamplesPreconditionError as exc: + return JSONResponse( + status_code=exc.status_code, + content={"error": {"code": exc.code, "message": exc.message}}, + ) + return JSONResponse(status_code=201, content=manifest) + + +@app.get("/api/analysis-runs/{run_id}/samples") +async def get_run_samples( + run_id: str, + x_asa_user_id: str | None = Header(None), + x_asa_user_email: str | None = Header(None), +) -> JSONResponse: + user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email) + if isinstance(user_context, JSONResponse): + return user_context + runtime = get_analysis_runtime() + try: + runtime.get_run(run_id, owner_user_id=user_context.user_id) + except (KeyError, PermissionError): + return _run_not_found_response(run_id) + + manifest = server_samples.fetch_existing_manifest(runtime=runtime, run_id=run_id) + if manifest is None: + return JSONResponse( + status_code=404, + content={ + "error": { + "code": "SAMPLES_NOT_GENERATED", + "message": ( + f"No audition samples have been generated for run '{run_id}'. " + "POST to this URL to create them." + ), + } + }, + ) + return JSONResponse(content=manifest) + + @app.get( "/api/analysis-runs/{run_id}/export/csv/{field_path}", response_model=None, diff --git a/apps/backend/server_samples.py b/apps/backend/server_samples.py new file mode 100644 index 00000000..a6e05c61 --- /dev/null +++ b/apps/backend/server_samples.py @@ -0,0 +1,217 @@ +"""Helpers for the Phase 3 audition-sample HTTP routes. + +The thin `@app.post / @app.get` wrappers live in `server.py` to match the +existing house style. All business logic β€” payload extraction, generation, +artifact registration, manifest decoration β€” lives here. + +These endpoints are on-demand: nothing in the staged-execution loop runs them +automatically. The user (or UI) explicitly POSTs to +`/api/analysis-runs/{run_id}/samples` after Phase 2 is complete. +""" + +from __future__ import annotations + +import json +import logging +import tempfile +from pathlib import Path +from typing import Any + +import sample_generation +from analysis_runtime import AnalysisRuntime + +logger = logging.getLogger(__name__) + +SAMPLE_AUDIO_KIND_PREFIX = "sample_audio" +SAMPLE_MIDI_KIND_PREFIX = "sample_midi" +SAMPLE_MANIFEST_KIND = "sample_manifest" + + +class SamplesPreconditionError(Exception): + """The run isn't in a state where samples can be generated yet.""" + + def __init__(self, code: str, message: str, status_code: int = 409): + super().__init__(message) + self.code = code + self.message = message + self.status_code = status_code + + +def _extract_phase1_from_snapshot(snapshot: dict[str, Any]) -> dict[str, Any] | None: + stages = snapshot.get("stages") or {} + measurement = stages.get("measurement") or {} + if measurement.get("status") != "completed": + return None + result = measurement.get("result") + return result if isinstance(result, dict) else None + + +def _extract_phase2_from_snapshot(snapshot: dict[str, Any]) -> dict[str, Any] | None: + """Pull the preferred Phase 2 result if one has completed. + + Returns None if interpretation never ran or didn't complete β€” sample + generation still proceeds from Phase 1 alone in that case. Any non- + "completed" status reaches that path through the isinstance gate below, + since stages without a completed attempt carry a `null` result anyway. + """ + stages = snapshot.get("stages") or {} + interpretation = stages.get("interpretation") or {} + if interpretation.get("status") != "completed": + return None + result = interpretation.get("result") + return result if isinstance(result, dict) else None + + +def generate_and_register_samples( + *, + runtime: AnalysisRuntime, + run_id: str, + snapshot: dict[str, Any], + force: bool = False, + prefer_fluidsynth: bool = True, +) -> dict[str, Any]: + """Run the orchestrator, persist artifacts, return a decorated manifest. + + The decorated manifest is the same shape the orchestrator emits, with + each sample augmented by `artifactId` so the frontend can construct + download URLs without a second round-trip. + """ + phase1 = _extract_phase1_from_snapshot(snapshot) + if phase1 is None: + raise SamplesPreconditionError( + code="MEASUREMENT_NOT_COMPLETED", + message=( + "Audition samples require a completed Phase 1 measurement. " + "Wait for the measurement stage to finish before requesting samples." + ), + status_code=409, + ) + + phase2 = _extract_phase2_from_snapshot(snapshot) + + if not force: + existing = runtime.get_internal_artifacts_by_kind(run_id, SAMPLE_MANIFEST_KIND) + if existing: + raise SamplesPreconditionError( + code="SAMPLES_ALREADY_GENERATED", + message=( + "An audition-sample manifest already exists for this run. " + "Pass ?force=true to regenerate." + ), + status_code=409, + ) + + with tempfile.TemporaryDirectory(prefix=f"asa-samples-{run_id}-") as tmp_root: + tmp_dir = Path(tmp_root) + result = sample_generation.generate_samples( + run_id=run_id, + phase1=phase1, + phase2=phase2, + output_dir=tmp_dir, + pitch_note_hints=None, # Pitch/note translation hints are a follow-up. + prefer_fluidsynth=prefer_fluidsynth, + ) + + # Persist each WAV/MIDI as a run artifact. The artifact kind names the + # sample so the GET-by-kind endpoint can filter just sample artifacts + # without including spectral or stem outputs. + sample_artifact_ids: dict[str, str] = {} + midi_artifact_ids: dict[str, str] = {} + for sample_id, wav_path in result.artifact_paths.items(): + record = runtime.record_artifact( + run_id, + kind=f"{SAMPLE_AUDIO_KIND_PREFIX}:{sample_id}", + source_path=str(wav_path), + filename=wav_path.name, + mime_type="audio/wav", + provenance={ + "sampleId": sample_id, + "schemaVersion": result.manifest["schemaVersion"], + }, + ) + sample_artifact_ids[sample_id] = record["artifactId"] + for sample_id, mid_path in result.midi_paths.items(): + record = runtime.record_artifact( + run_id, + kind=f"{SAMPLE_MIDI_KIND_PREFIX}:{sample_id}", + source_path=str(mid_path), + filename=mid_path.name, + mime_type="audio/midi", + provenance={ + "sampleId": sample_id, + "schemaVersion": result.manifest["schemaVersion"], + }, + ) + midi_artifact_ids[sample_id] = record["artifactId"] + + # And persist the manifest itself. Future GETs read this back to + # decorate the response identically. + manifest_record = runtime.record_artifact( + run_id, + kind=SAMPLE_MANIFEST_KIND, + source_path=str(result.manifest_path), + filename=result.manifest_path.name, + mime_type="application/json", + provenance={ + "schemaVersion": result.manifest["schemaVersion"], + "sampleArtifactIds": sample_artifact_ids, + "midiArtifactIds": midi_artifact_ids, + }, + ) + + decorated = _decorate_manifest( + result.manifest, sample_artifact_ids, midi_artifact_ids + ) + decorated["manifestArtifactId"] = manifest_record["artifactId"] + return decorated + + +def fetch_existing_manifest( + *, + runtime: AnalysisRuntime, + run_id: str, +) -> dict[str, Any] | None: + """Reconstruct the decorated manifest from previously-persisted artifacts. + + Returns None if no manifest has been generated yet β€” caller can decide + whether that's a 404 or a different signal. + """ + manifests = runtime.get_internal_artifacts_by_kind(run_id, SAMPLE_MANIFEST_KIND) + if not manifests: + return None + # get_internal_artifacts_by_kind returns rows ordered by created_at ASC, + # so the latest is the tail. (force=true paths create multiple rows.) + latest = manifests[-1] + + manifest_path = runtime.resolve_artifact_local_path(latest.get("path")) + if manifest_path is None or not manifest_path.is_file(): + return None + raw = json.loads(manifest_path.read_text()) + provenance = latest.get("provenance") or {} + decorated = _decorate_manifest( + raw, + provenance.get("sampleArtifactIds") or {}, + provenance.get("midiArtifactIds") or {}, + ) + decorated["manifestArtifactId"] = latest["artifactId"] + return decorated + + +def _decorate_manifest( + manifest: dict[str, Any], + sample_artifact_ids: dict[str, str], + midi_artifact_ids: dict[str, str], +) -> dict[str, Any]: + """Attach artifactId fields to each sample so the UI can build URLs.""" + decorated = dict(manifest) + decorated_samples: list[dict[str, Any]] = [] + for sample in manifest.get("samples", []): + copy = dict(sample) + sample_id = sample.get("id") + if sample_id in sample_artifact_ids: + copy["artifactId"] = sample_artifact_ids[sample_id] + if sample_id in midi_artifact_ids: + copy["midiArtifactId"] = midi_artifact_ids[sample_id] + decorated_samples.append(copy) + decorated["samples"] = decorated_samples + return decorated diff --git a/apps/backend/tests/test_sample_drums.py b/apps/backend/tests/test_sample_drums.py new file mode 100644 index 00000000..24855c24 --- /dev/null +++ b/apps/backend/tests/test_sample_drums.py @@ -0,0 +1,73 @@ +"""Tests for the NumPy drum-synthesis layer. + +The kick test is the load-bearing one: it verifies that a measured +`fundamentalHz` actually shows up as the dominant spectral energy in the +rendered audio. If this regresses, the audition lies about what the +measurement says. +""" + +import sys +import unittest +from pathlib import Path + +import numpy as np + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + +import sample_drums # noqa: E402 + + +class KickTests(unittest.TestCase): + def test_kick_fft_peak_lands_near_fundamental(self) -> None: + # Use 80 Hz so the peak sits in a region with enough FFT resolution. + kick = sample_drums.synth_kick(fundamental_hz=80.0, decay_time_ms=250.0) + # Look at the steady-state portion (after the initial pitch sweep). + steady_start = int(0.05 * kick.sample_rate) + steady_segment = kick.samples[steady_start:].astype(np.float64) + spectrum = np.abs(np.fft.rfft(steady_segment)) + freqs = np.fft.rfftfreq(steady_segment.size, d=1.0 / kick.sample_rate) + peak_freq = float(freqs[int(np.argmax(spectrum))]) + # Generous tolerance: the pitch envelope means peak energy can sit a + # little above the fundamental in the first ~30 ms. + self.assertAlmostEqual(peak_freq, 80.0, delta=20.0) + + def test_kick_obeys_decay_envelope(self) -> None: + kick = sample_drums.synth_kick(fundamental_hz=55.0, decay_time_ms=150.0) + # Tail amplitude should be substantially lower than head amplitude. + head_rms = float(np.sqrt(np.mean(kick.samples[:1000] ** 2))) + tail_rms = float( + np.sqrt(np.mean(kick.samples[-1000:] ** 2)) + ) + self.assertGreater(head_rms, tail_rms * 5.0) + + def test_kick_rejects_negative_fundamental(self) -> None: + with self.assertRaises(ValueError): + sample_drums.synth_kick(fundamental_hz=-1.0) + + +class SnareTests(unittest.TestCase): + def test_snare_is_well_formed(self) -> None: + snare = sample_drums.synth_snare() + self.assertEqual(snare.sample_rate, sample_drums.SAMPLE_RATE) + self.assertEqual(snare.samples.dtype, np.float32) + # Within [-1, 1] after normalization. + self.assertLessEqual(float(np.max(np.abs(snare.samples))), 1.0) + # Non-silent. + self.assertGreater( + float(np.sqrt(np.mean(snare.samples.astype(np.float64) ** 2))), 0.01 + ) + + +class HatTests(unittest.TestCase): + def test_hat_is_short_and_non_silent(self) -> None: + hat = sample_drums.synth_hat() + self.assertLess(hat.duration_seconds, 0.3) + self.assertGreater( + float(np.sqrt(np.mean(hat.samples.astype(np.float64) ** 2))), 0.005 + ) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/apps/backend/tests/test_sample_generation.py b/apps/backend/tests/test_sample_generation.py new file mode 100644 index 00000000..04019b6a --- /dev/null +++ b/apps/backend/tests/test_sample_generation.py @@ -0,0 +1,176 @@ +"""End-to-end orchestrator tests. + +These tests feed synthetic Phase 1 / Phase 2 dicts into `generate_samples` +and assert on the manifest contract documented in +`docs/SAMPLE_GENERATION.md`. The citation array is the chain of custody for +this stage β€” if a generated sample isn't tied back to a Phase 1 field, the +audition has nothing to justify it. +""" + +import json +import sys +import tempfile +import unittest +import wave +from pathlib import Path + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + +import sample_generation # noqa: E402 + + +def _baseline_phase1() -> dict: + return { + "bpm": 124.0, + "bpmConfidence": 0.91, + "key": "F# minor", + "keyConfidence": 0.83, + "kickDetail": { + "fundamentalHz": 55.0, + "decayTimeMs": 220.0, + "confidence": 0.82, + }, + "melodyDetail": {"placeholder": True}, + } + + +def _baseline_phase2() -> dict: + return { + "trackCharacter": "deep house", + "styleProfile": { + "genre": "Deep House", + "authoritativeMeasurements": {"bpm": 124.0, "key": "F# minor"}, + }, + "sonicElements": { + "kick": "Punchy sub-heavy kick around 55 Hz.", + "harmonicContent": "Minor 7th pads, layered piano.", + }, + } + + +class OrchestratorTests(unittest.TestCase): + def test_full_input_produces_all_sample_categories(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + result = sample_generation.generate_samples( + run_id="run-1", + phase1=_baseline_phase1(), + phase2=_baseline_phase2(), + output_dir=Path(tmp), + pitch_note_hints=[1, 2, 3, 5], + prefer_fluidsynth=False, + ) + + sample_ids = {s["id"] for s in result.manifest["samples"]} + self.assertIn("tonal_chord_progression", sample_ids) + self.assertIn("tonal_bass_root", sample_ids) + self.assertIn("drum_kick", sample_ids) + self.assertIn("drum_snare", sample_ids) + self.assertIn("drum_hat", sample_ids) + self.assertIn("melody_lead", sample_ids) + + # Every WAV file on disk. + for sample in result.manifest["samples"]: + wav_path = Path(tmp) / sample["filename"] + self.assertTrue(wav_path.is_file(), f"missing {wav_path}") + with wave.open(str(wav_path), "rb") as wav: + self.assertGreater(wav.getnframes(), 0) + + # Manifest file written. + manifest_on_disk = json.loads(result.manifest_path.read_text()) + self.assertEqual(manifest_on_disk["runId"], "run-1") + self.assertEqual(manifest_on_disk["schemaVersion"], "samples.v1") + self.assertEqual(manifest_on_disk["synthesisBackend"], "sine_fallback") + + def test_tonal_samples_skipped_when_key_missing(self) -> None: + phase1 = _baseline_phase1() + phase1.pop("key") + with tempfile.TemporaryDirectory() as tmp: + result = sample_generation.generate_samples( + run_id="run-no-key", + phase1=phase1, + phase2=None, + output_dir=Path(tmp), + prefer_fluidsynth=False, + ) + sample_ids = {s["id"] for s in result.manifest["samples"]} + self.assertNotIn("tonal_chord_progression", sample_ids) + self.assertNotIn("tonal_bass_root", sample_ids) + self.assertNotIn("melody_lead", sample_ids) + # Drums still emitted. + self.assertIn("drum_kick", sample_ids) + + def test_low_confidence_key_flags_tonal_samples(self) -> None: + phase1 = _baseline_phase1() + phase1["keyConfidence"] = 0.2 + with tempfile.TemporaryDirectory() as tmp: + result = sample_generation.generate_samples( + run_id="run-low-confidence", + phase1=phase1, + phase2=None, + output_dir=Path(tmp), + prefer_fluidsynth=False, + ) + tonal = [s for s in result.manifest["samples"] if s["category"] == "tonal"] + self.assertTrue(tonal, "expected tonal samples even at low confidence") + for sample in tonal: + self.assertTrue(sample["lowConfidence"]) + self.assertEqual(sample["confidence"], "LOW") + # Drums shouldn't be flagged off the back of key confidence. + kick = next(s for s in result.manifest["samples"] if s["id"] == "drum_kick") + self.assertEqual(kick["confidence"], "HIGH") + + def test_every_sample_cites_or_explains_absence(self) -> None: + # Chain-of-custody invariant: a sample must either cite a Phase 1 + # field or carry a rationale that explicitly names it as heuristic. + with tempfile.TemporaryDirectory() as tmp: + result = sample_generation.generate_samples( + run_id="run-cite", + phase1=_baseline_phase1(), + phase2=_baseline_phase2(), + output_dir=Path(tmp), + prefer_fluidsynth=False, + ) + for sample in result.manifest["samples"]: + cites = sample["cites"] + has_phase1 = len(cites["phase1Fields"]) > 0 + rationale = cites["rationale"].lower() + mentions_heuristic = ( + "heuristic" in rationale or "default" in rationale + ) + self.assertTrue( + has_phase1 or mentions_heuristic, + f"{sample['id']} cites nothing and isn't labeled heuristic: {sample}", + ) + + def test_kick_uses_measured_fundamental_when_present(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + result = sample_generation.generate_samples( + run_id="run-kick", + phase1=_baseline_phase1(), + phase2=None, + output_dir=Path(tmp), + prefer_fluidsynth=False, + ) + kick = next(s for s in result.manifest["samples"] if s["id"] == "drum_kick") + self.assertEqual(kick["cites"]["phase1Fields"], [ + "kickDetail.fundamentalHz", + "kickDetail.decayTimeMs", + ]) + self.assertIn("55", kick["label"]) # "Kick at 55 Hz" + + def test_manifest_records_theory_backend(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + result = sample_generation.generate_samples( + run_id="run-backend", + phase1=_baseline_phase1(), + phase2=None, + output_dir=Path(tmp), + prefer_fluidsynth=False, + ) + self.assertIn(result.manifest["theoryBackend"], {"pytheory", "fallback"}) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/apps/backend/tests/test_sample_synthesis.py b/apps/backend/tests/test_sample_synthesis.py new file mode 100644 index 00000000..917e0639 --- /dev/null +++ b/apps/backend/tests/test_sample_synthesis.py @@ -0,0 +1,115 @@ +"""Tests for the MIDI-plan β†’ WAV/MIDI synthesis layer. + +The sine fallback path is the one exercised by these tests; the FluidSynth +path requires a system library and a soundfont, which we don't assume are +present in CI. The fallback is the everywhere-available backstop and must +stay correct. +""" + +import sys +import tempfile +import unittest +import wave +from pathlib import Path + +import numpy as np + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + +import sample_synthesis # noqa: E402 +import sample_theory # noqa: E402 + + +def _single_note_plan(pitch: int = 69, bpm: float = 120.0, duration_beats: float = 4.0) -> sample_theory.ClipPlan: + """Minimal plan: one note for `duration_beats`.""" + return sample_theory.ClipPlan( + tempo_bpm=bpm, + duration_beats=duration_beats, + notes=[ + sample_theory.NoteEvent( + pitch_midi=pitch, start_beat=0.0, duration_beats=duration_beats + ) + ], + program=0, + ) + + +class SineFallbackTests(unittest.TestCase): + def test_render_returns_expected_shape(self) -> None: + plan = _single_note_plan(pitch=69, bpm=120.0, duration_beats=4.0) + result = sample_synthesis.render_clip(plan, prefer_fluidsynth=False) + expected_samples = int(round(2.0 * sample_synthesis.SAMPLE_RATE)) # 4 beats @ 120 = 2 s + self.assertEqual(result.samples.shape, (expected_samples,)) + self.assertEqual(result.samples.dtype, np.float32) + self.assertEqual(result.backend, "sine_fallback") + self.assertIsNone(result.soundfont_path) + + def test_render_contains_energy_at_target_frequency(self) -> None: + # A4 = MIDI 69 = 440 Hz exactly. + plan = _single_note_plan(pitch=69, bpm=120.0, duration_beats=4.0) + result = sample_synthesis.render_clip(plan, prefer_fluidsynth=False) + + # Look at the middle of the note to avoid envelope transients. + mid_start = result.samples.size // 4 + mid_end = result.samples.size - result.samples.size // 4 + segment = result.samples[mid_start:mid_end].astype(np.float64) + spectrum = np.abs(np.fft.rfft(segment)) + freqs = np.fft.rfftfreq(segment.size, d=1.0 / sample_synthesis.SAMPLE_RATE) + peak_freq = float(freqs[int(np.argmax(spectrum))]) + # Allow a few Hz of bin-resolution slop. + self.assertAlmostEqual(peak_freq, 440.0, delta=5.0) + + def test_render_silent_when_no_notes(self) -> None: + plan = sample_theory.ClipPlan( + tempo_bpm=120.0, duration_beats=4.0, notes=[], program=0 + ) + result = sample_synthesis.render_clip(plan, prefer_fluidsynth=False) + peak = float(np.max(np.abs(result.samples))) + self.assertEqual(peak, 0.0) + + +class WriteWavTests(unittest.TestCase): + def test_write_wav_round_trips(self) -> None: + plan = _single_note_plan(pitch=60, bpm=120.0, duration_beats=2.0) + result = sample_synthesis.render_clip(plan, prefer_fluidsynth=False) + + with tempfile.TemporaryDirectory() as tmp: + wav_path = Path(tmp) / "out.wav" + sample_synthesis.write_wav(result.samples, path=wav_path) + self.assertTrue(wav_path.is_file()) + with wave.open(str(wav_path), "rb") as wav: + self.assertEqual(wav.getframerate(), sample_synthesis.SAMPLE_RATE) + self.assertEqual(wav.getnchannels(), 1) + # 16-bit subtype = 2 bytes per sample. + self.assertEqual(wav.getsampwidth(), 2) + self.assertGreater(wav.getnframes(), 0) + + +class WriteMidiTests(unittest.TestCase): + def test_write_midi_emits_expected_note(self) -> None: + import pretty_midi # local import β€” only test needs it + + plan = sample_theory.ClipPlan( + tempo_bpm=120.0, + duration_beats=4.0, + notes=[ + sample_theory.NoteEvent(pitch_midi=60, start_beat=0.0, duration_beats=2.0), + sample_theory.NoteEvent(pitch_midi=64, start_beat=2.0, duration_beats=2.0), + ], + program=0, + ) + with tempfile.TemporaryDirectory() as tmp: + mid_path = Path(tmp) / "out.mid" + sample_synthesis.write_midi(plan, path=mid_path) + self.assertTrue(mid_path.is_file()) + pm = pretty_midi.PrettyMIDI(str(mid_path)) + self.assertEqual(len(pm.instruments), 1) + self.assertEqual(len(pm.instruments[0].notes), 2) + pitches = sorted(n.pitch for n in pm.instruments[0].notes) + self.assertEqual(pitches, [60, 64]) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/apps/backend/tests/test_sample_theory.py b/apps/backend/tests/test_sample_theory.py new file mode 100644 index 00000000..a80d47c2 --- /dev/null +++ b/apps/backend/tests/test_sample_theory.py @@ -0,0 +1,150 @@ +"""Tests for the Phase 3 music-theory adapter. + +The MIDI numbers asserted here are the canonical Western-music values for the +given key/mode/degree combinations. Both the PyTheory and fallback paths must +agree with these reference values β€” if a render starts sounding out-of-key, +this is the first place to look. +""" + +import sys +import unittest +from pathlib import Path + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + +import sample_theory # noqa: E402 + + +class ParseKeyTests(unittest.TestCase): + def test_parses_canonical_keys(self) -> None: + self.assertEqual(sample_theory.parse_key("C major"), (0, "major", "C")) + self.assertEqual(sample_theory.parse_key("A minor"), (9, "minor", "A")) + self.assertEqual(sample_theory.parse_key("F# minor"), (6, "minor", "F#")) + self.assertEqual(sample_theory.parse_key("Bb major"), (10, "major", "Bb")) + + def test_defaults_to_major_when_mode_missing(self) -> None: + self.assertEqual(sample_theory.parse_key("D"), (2, "major", "D")) + + def test_tolerates_parenthetical_suffix(self) -> None: + # Phase 1 occasionally tacks on confidence qualifiers. + self.assertEqual( + sample_theory.parse_key("F# minor (low confidence)"), + (6, "minor", "F#"), + ) + + def test_rejects_unknown_root(self) -> None: + with self.assertRaises(ValueError): + sample_theory.parse_key("H minor") + + def test_rejects_unknown_mode(self) -> None: + with self.assertRaises(ValueError): + sample_theory.parse_key("C ionian-augmented-fancy") + + def test_rejects_empty_input(self) -> None: + with self.assertRaises(ValueError): + sample_theory.parse_key("") + + +class BuildContextTests(unittest.TestCase): + def test_carries_confidence_through(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=128.0, key_confidence=0.83) + self.assertEqual(ctx.root_pc, 0) + self.assertEqual(ctx.mode, "major") + self.assertEqual(ctx.root_name, "C") + self.assertEqual(ctx.tempo_bpm, 128.0) + self.assertEqual(ctx.key_confidence, 0.83) + self.assertIn(ctx.backend, {"pytheory", "fallback"}) + + +class ChordProgressionTests(unittest.TestCase): + def test_c_major_progression_voicings(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=120.0) + plan = sample_theory.plan_chord_progression(ctx, bars=8, voicing_octave=4) + + # 4 chords Γ— 3 notes each = 12 events. + self.assertEqual(len(plan.notes), 12) + + # Chord 1 (C major): C-E-G at octave 4. + chord_one_pitches = sorted(n.pitch_midi for n in plan.notes if n.start_beat == 0.0) + self.assertEqual(chord_one_pitches, [60, 64, 67]) + + # Chord 2 (A minor, vi): A-C-E. Starts at beat 8. + chord_two_pitches = sorted( + n.pitch_midi for n in plan.notes if n.start_beat == 8.0 + ) + self.assertEqual(chord_two_pitches, [57, 60, 64]) + + # Chord 3 (F major, IV): F-A-C. Starts at beat 16. + chord_three_pitches = sorted( + n.pitch_midi for n in plan.notes if n.start_beat == 16.0 + ) + self.assertEqual(chord_three_pitches, [53, 57, 60]) + + # Chord 4 (G major, V): G-B-D. Starts at beat 24. + chord_four_pitches = sorted( + n.pitch_midi for n in plan.notes if n.start_beat == 24.0 + ) + self.assertEqual(chord_four_pitches, [55, 59, 62]) + + def test_a_minor_progression_voicings(self) -> None: + ctx = sample_theory.build_context(key="A minor", bpm=120.0) + plan = sample_theory.plan_chord_progression(ctx, bars=8, voicing_octave=4) + + # i (A minor) at the tonic octave: A4-C5-E5. + chord_one_pitches = sorted(n.pitch_midi for n in plan.notes if n.start_beat == 0.0) + self.assertEqual(chord_one_pitches, [69, 72, 76]) + + # VI (F major), dropped to the octave below the tonic anchor: + # F4-A4-C5. Distance from tonic chord: a stepwise descent in the bass. + chord_two_pitches = sorted( + n.pitch_midi for n in plan.notes if n.start_beat == 8.0 + ) + self.assertEqual(chord_two_pitches, [65, 69, 72]) + + def test_rejects_odd_bar_counts(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=120.0) + with self.assertRaises(ValueError): + sample_theory.plan_chord_progression(ctx, bars=7) + + +class BassRootTests(unittest.TestCase): + def test_bass_lives_two_octaves_below_voicing(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=120.0) + plan = sample_theory.plan_bass_root(ctx, bars=8) + # C at octave 2 = MIDI 36. + for note in plan.notes: + self.assertEqual(note.pitch_midi, 36) + + def test_bass_count_matches_bars(self) -> None: + ctx = sample_theory.build_context(key="A minor", bpm=120.0) + plan = sample_theory.plan_bass_root(ctx, bars=8) + self.assertEqual(len(plan.notes), 4) # one per two bars + self.assertEqual(plan.duration_beats, 32.0) + + +class MelodyPlanTests(unittest.TestCase): + def test_default_phrase_uses_in_key_pitches(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=120.0) + plan = sample_theory.plan_melody_phrase(ctx) + self.assertIsNotNone(plan) + assert plan is not None # narrowing for type checkers + # Default ascent 1-2-3-5-3-1 in C major at octave 5 -> C5 D5 E5 G5 E5 C5. + pitches = [n.pitch_midi for n in plan.notes] + self.assertEqual(pitches, [72, 74, 76, 79, 76, 72]) + + def test_respects_supplied_hints(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=120.0) + plan = sample_theory.plan_melody_phrase(ctx, scale_degrees=[1, 3, 5]) + assert plan is not None + self.assertEqual([n.pitch_midi for n in plan.notes], [72, 76, 79]) + + def test_returns_none_for_empty_clean_input(self) -> None: + ctx = sample_theory.build_context(key="C major", bpm=120.0) + # All ineligible degrees should produce no plan rather than fabricate one. + self.assertIsNone(sample_theory.plan_melody_phrase(ctx, scale_degrees=[0, 99])) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/apps/backend/tests/test_server_samples.py b/apps/backend/tests/test_server_samples.py new file mode 100644 index 00000000..b0cc92f1 --- /dev/null +++ b/apps/backend/tests/test_server_samples.py @@ -0,0 +1,165 @@ +"""Contract tests for the audition-sample HTTP helper layer. + +We test the helper module directly (rather than spinning up a full FastAPI +TestClient) because the layered server is heavy to boot and the route +handlers themselves are thin wrappers. The helpers are where the precondition +logic, manifest decoration, and artifact persistence happen β€” and those are +the places a bug would slip through. +""" + +import sys +import tempfile +import unittest +from pathlib import Path + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + +from analysis_runtime import AnalysisRuntime # noqa: E402 +import server_samples # noqa: E402 + + +def _baseline_snapshot(*, phase2_completed: bool = True) -> dict: + interpretation = {"status": "completed", "result": {"trackCharacter": "fixture"}} + if not phase2_completed: + interpretation = {"status": "ready", "result": None} + return { + "stages": { + "measurement": { + "status": "completed", + "result": { + "bpm": 124.0, + "bpmConfidence": 0.9, + "key": "F# minor", + "keyConfidence": 0.78, + "kickDetail": { + "fundamentalHz": 55.0, + "decayTimeMs": 240.0, + "confidence": 0.8, + }, + }, + }, + "interpretation": interpretation, + } + } + + +class ServerSamplesTests(unittest.TestCase): + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory(prefix="asa_server_samples_test_") + self.runtime = AnalysisRuntime(Path(self.tempdir.name) / "runtime") + result = self.runtime.create_run( + filename="test.wav", + content=b"\x00" * 256, + mime_type="audio/wav", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="default", + interpretation_model=None, + ) + self.run_id = result["runId"] + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def test_generation_succeeds_with_phase1_only(self) -> None: + # Phase 2 absent is acceptable; we still emit tonal/drum samples. + snapshot = _baseline_snapshot(phase2_completed=False) + manifest = server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + force=False, + prefer_fluidsynth=False, + ) + self.assertEqual(manifest["schemaVersion"], "samples.v1") + self.assertIn("manifestArtifactId", manifest) + # Every sample carries an artifactId for the frontend to dereference. + for sample in manifest["samples"]: + self.assertIn( + "artifactId", + sample, + f"sample {sample['id']} missing artifactId", + ) + + def test_rejects_when_measurement_not_completed(self) -> None: + snapshot = _baseline_snapshot() + snapshot["stages"]["measurement"]["status"] = "running" + with self.assertRaises(server_samples.SamplesPreconditionError) as ctx: + server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + prefer_fluidsynth=False, + ) + self.assertEqual(ctx.exception.code, "MEASUREMENT_NOT_COMPLETED") + self.assertEqual(ctx.exception.status_code, 409) + + def test_rejects_regeneration_unless_force(self) -> None: + snapshot = _baseline_snapshot() + server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + prefer_fluidsynth=False, + ) + with self.assertRaises(server_samples.SamplesPreconditionError) as ctx: + server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + force=False, + prefer_fluidsynth=False, + ) + self.assertEqual(ctx.exception.code, "SAMPLES_ALREADY_GENERATED") + + def test_force_regenerates(self) -> None: + snapshot = _baseline_snapshot() + first = server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + prefer_fluidsynth=False, + ) + second = server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + force=True, + prefer_fluidsynth=False, + ) + # New manifest artifact id; the SQLite ids are UUIDs so they will differ. + self.assertNotEqual(first["manifestArtifactId"], second["manifestArtifactId"]) + + def test_fetch_existing_returns_none_before_generation(self) -> None: + self.assertIsNone( + server_samples.fetch_existing_manifest( + runtime=self.runtime, run_id=self.run_id + ) + ) + + def test_fetch_existing_returns_decorated_manifest_after_generation(self) -> None: + snapshot = _baseline_snapshot() + created = server_samples.generate_and_register_samples( + runtime=self.runtime, + run_id=self.run_id, + snapshot=snapshot, + prefer_fluidsynth=False, + ) + fetched = server_samples.fetch_existing_manifest( + runtime=self.runtime, run_id=self.run_id + ) + self.assertIsNotNone(fetched) + assert fetched is not None # type narrow for the static checker + # Same sample IDs, each with an artifactId. + created_ids = {s["id"] for s in created["samples"]} + fetched_ids = {s["id"] for s in fetched["samples"]} + self.assertEqual(created_ids, fetched_ids) + for sample in fetched["samples"]: + self.assertIn("artifactId", sample) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/apps/ui/src/components/AnalysisResults.tsx b/apps/ui/src/components/AnalysisResults.tsx index 98b006b7..47ce64af 100644 --- a/apps/ui/src/components/AnalysisResults.tsx +++ b/apps/ui/src/components/AnalysisResults.tsx @@ -28,6 +28,7 @@ import { motion } from 'motion/react'; import { downloadFile, generateMarkdown } from '../utils/exportUtils'; import { INTERPRETATION_LABEL } from '../services/phaseLabels'; import { MeasurementDashboard } from './MeasurementDashboard'; +import { SamplePlayback } from './SamplePlayback'; import { SessionMusicianPanel } from './SessionMusicianPanel'; import { StemListeningNotesPanel } from './StemListeningNotesPanel'; import { hasStemListeningNotesContent } from '../services/sessionMusician'; @@ -2290,6 +2291,13 @@ export function AnalysisResults({ runId={runId} /> + {apiBaseUrl && runId && ( + + )} ); } diff --git a/apps/ui/src/components/SamplePlayback.tsx b/apps/ui/src/components/SamplePlayback.tsx new file mode 100644 index 00000000..ffe57133 --- /dev/null +++ b/apps/ui/src/components/SamplePlayback.tsx @@ -0,0 +1,300 @@ +import React, { useCallback, useEffect, useMemo, useState } from 'react'; + +import { SampleRecord, SamplesManifest } from '../types/samples'; +import { + artifactStreamUrl, + fetchExistingManifest, + generateSamples, +} from '../services/sampleGenerationClient'; +import { BackendClientError } from '../services/backendPhase1Client'; + +interface SamplePlaybackProps { + runId: string | null | undefined; + apiBaseUrl: string; + /** + * If false, the panel renders an explanatory placeholder rather than the + * generate button. The caller passes false when measurement isn't complete + * yet β€” there's nothing to audition against. + */ + measurementCompleted: boolean; +} + +type PanelStatus = + | { kind: 'idle' } + | { kind: 'loading' } + | { kind: 'generating' } + | { kind: 'error'; message: string }; + +const CATEGORY_LABELS: Record = { + tonal: 'Tonal β€” key & chord foundation', + drums: 'Drum kit', + melody: 'Melody / lead phrase', +}; + +/** + * Audition panel that renders generated audio clips for the current run. + * + * Honest-uncertainty framing is non-negotiable per `PURPOSE.md`: these clips + * are heuristic reconstructions of the measurement layer, not Ableton-accurate + * renderings. The panel says so up top, every time. + */ +export function SamplePlayback({ + runId, + apiBaseUrl, + measurementCompleted, +}: SamplePlaybackProps): React.ReactElement | null { + const [manifest, setManifest] = useState(null); + const [status, setStatus] = useState({ kind: 'idle' }); + + // On mount, see if a manifest already exists for this run. + useEffect(() => { + if (!runId) return; + let cancelled = false; + const abort = new AbortController(); + setStatus({ kind: 'loading' }); + fetchExistingManifest(runId, { apiBaseUrl, signal: abort.signal }) + .then((existing) => { + if (cancelled) return; + setManifest(existing); + setStatus({ kind: 'idle' }); + }) + .catch((err) => { + if (cancelled) return; + setStatus({ kind: 'error', message: friendlyError(err) }); + }); + return () => { + cancelled = true; + abort.abort(); + }; + }, [runId, apiBaseUrl]); + + const handleGenerate = useCallback( + async (force: boolean) => { + if (!runId) return; + setStatus({ kind: 'generating' }); + try { + const next = await generateSamples(runId, { apiBaseUrl, force }); + setManifest(next); + setStatus({ kind: 'idle' }); + } catch (err) { + setStatus({ kind: 'error', message: friendlyError(err) }); + } + }, + [runId, apiBaseUrl], + ); + + const groupedSamples = useMemo< + Array<[SampleRecord['category'], SampleRecord[]]> + >(() => { + if (!manifest) return []; + const groups = new Map(); + for (const sample of manifest.samples) { + const list = groups.get(sample.category) ?? []; + list.push(sample); + groups.set(sample.category, list); + } + return Array.from(groups.entries()); + }, [manifest]); + + if (!runId) return null; + + return ( +
+
+
+

+ Audition samples (Phase 3 β€” heuristic) +

+

+ Short clips derived from Phase 1 measurements (and Phase 2 context when + available) so you can ear-check the measurement chain. These are not + Ableton-accurate reconstructions β€” follow Phase 2 in Live for the + production character. +

+
+ {manifest && ( + + )} +
+ + {!measurementCompleted && ( +

+ Measurements still running β€” audition samples become available once Phase 1 + completes. +

+ )} + + {measurementCompleted && !manifest && status.kind !== 'generating' && ( + + )} + + {status.kind === 'generating' && ( +

Rendering audition clips…

+ )} + + {status.kind === 'error' && ( +

+ {status.message} +

+ )} + + {manifest && groupedSamples.length > 0 && ( +
+ + {groupedSamples.map(([category, samples]) => ( + + + + ))} +
+ )} +
+ ); +} + +function ManifestMeta({ manifest }: { manifest: SamplesManifest }) { + const synthesisLabel = + manifest.synthesisBackend === 'fluidsynth' + ? 'FluidSynth + GM SoundFont' + : 'NumPy sine-additive fallback'; + const theoryLabel = + manifest.theoryBackend === 'pytheory' + ? 'PyTheory' + : 'Pure-Python theory fallback'; + return ( +
+ + Music theory: {theoryLabel} + + + Synthesis: {synthesisLabel} + +
+ ); +} + +interface SampleGroupProps { + category: SampleRecord['category']; + samples: SampleRecord[]; + runId: string; + apiBaseUrl: string; +} + +function SampleGroup({ category, samples, runId, apiBaseUrl }: SampleGroupProps) { + return ( +
+

+ {CATEGORY_LABELS[category]} +

+
    + {samples.map((sample) => ( + + + + ))} +
+
+ ); +} + +interface SampleCardProps { + sample: SampleRecord; + runId: string; + apiBaseUrl: string; +} + +function SampleCard({ sample, runId, apiBaseUrl }: SampleCardProps) { + const audioUrl = sample.artifactId + ? artifactStreamUrl(runId, sample.artifactId, apiBaseUrl) + : null; + const midiUrl = sample.midiArtifactId + ? artifactStreamUrl(runId, sample.midiArtifactId, apiBaseUrl) + : null; + + return ( +
  • +
    + {sample.label} + + {sample.lowConfidence ? 'Low confidence' : `${sample.confidence} confidence`} + +
    + {audioUrl ? ( +
    • + ); +} + +function confidenceClassFor( + confidence: SampleRecord['confidence'], + lowConfidence: boolean, +): string { + if (lowConfidence || confidence === 'LOW') return 'bg-amber-900/40 text-amber-300'; + if (confidence === 'MED') return 'bg-zinc-700 text-zinc-200'; + return 'bg-emerald-900/40 text-emerald-300'; +} + +function friendlyError(err: unknown): string { + if (err instanceof BackendClientError) return err.message; + if (err instanceof Error) return err.message; + return 'Unknown error while talking to the sample generation service.'; +} diff --git a/apps/ui/src/services/sampleGenerationClient.ts b/apps/ui/src/services/sampleGenerationClient.ts new file mode 100644 index 00000000..40afcc32 --- /dev/null +++ b/apps/ui/src/services/sampleGenerationClient.ts @@ -0,0 +1,159 @@ +/** + * Typed HTTP client for the Phase 3 audition-sample endpoints. + * + * Endpoints: + * - POST /api/analysis-runs/{run_id}/samples β€” generate (returns manifest) + * - GET /api/analysis-runs/{run_id}/samples β€” fetch existing manifest + * - GET /api/analysis-runs/{run_id}/artifacts/{id} β€” stream a WAV/MIDI + * + * The first two return the same manifest shape; the third is a binary stream + * we expose as a URL the consumer can hand to an