apps/theme_lab/app.py

"""FastAPI app for browsing and generating musical themes."""

from __future__ import annotations

import argparse
from contextlib import asynccontextmanager
import os
import pickle
import random
import shutil
import sqlite3
import subprocess
import threading
import uuid
from pathlib import Path
from typing import Literal

from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from theme_generation.cli import load_generation_inputs
from theme_generation.common import COMMON_DURATION_BEATS, PITCH_CLASS, START_SYMBOL, Symbol
from theme_generation.constraints import make_theme_acceptor
from theme_generation.engines.markov import (
    ConstraintSet,
    LongestFeasiblePolicy,
    OrderStackModel,
    prepare_constrained_order_stack,
    require_vo_regular,
)
from theme_generation.engines.transformer import TransformerConfig, generate_transformer
from theme_generation.engines.transformer import load_transformer_checkpoint, sample_transformer_checkpoint
from theme_generation.io import ensure_muses_import_path, sequence_to_pitches, write_samples


ROOT = Path(__file__).resolve().parents[2]
DB_PATH = ROOT / "audit" / "themes_audit.sqlite"
STATIC_DIR = Path(__file__).resolve().parent / "static"
GENERATED_ROOT = ROOT / "outputs" / "web_app_runs"
CATALOG_SCORE_CACHE_VERSION = "v2"
CATALOG_SCORE_ROOT = GENERATED_ROOT / f"catalog_scores_{CATALOG_SCORE_CACHE_VERSION}"
DEFAULT_TRANSFORMER_CHECKPOINT = ROOT / "models" / "theme_transformer_default.pt"
MARKOV_CACHE_DIR = ROOT / "models" / "theme_lab_markov_cache"
DEFAULT_MARKOV_CACHE = MARKOV_CACHE_DIR / "default.pkl"
MARKOV_CACHE_FORMAT_VERSION = 1
MARKOV_PRECOMPUTED_WARM_SAMPLES = 4
SCORE_PREVIEW_RENDER_VERSION = "verovio-resources-accidentals-v3"
GENERATED_ROOT.mkdir(parents=True, exist_ok=True)
CATALOG_SCORE_ROOT.mkdir(parents=True, exist_ok=True)


_markov_cache_lock = threading.Lock()
_markov_cache: dict[tuple[object, ...], dict[str, object]] = {}
_transformer_lock = threading.Lock()
_transformer_checkpoint = None


@asynccontextmanager
async def lifespan(_: FastAPI):
    start_background_warmup()
    yield


app = FastAPI(title="Theme Lab", version="0.1.0", lifespan=lifespan)
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
app.mount("/svgs", StaticFiles(directory=ROOT / "svgs"), name="svgs")
app.mount("/midis", StaticFiles(directory=ROOT / "midis"), name="midis")
app.mount("/abcs", StaticFiles(directory=ROOT / "abcs"), name="abcs")
app.mount("/generated", StaticFiles(directory=GENERATED_ROOT), name="generated")


class GenerateRequest(BaseModel):
    engine: Literal["markov", "transformer"] = "markov"
    samples: int = Field(2, ge=1, le=12)
    length: int = Field(24, ge=4, le=64)
    key: str = "C"
    seed: int = Field(42, ge=0)
    endpoint_strength: float = Field(1.0, ge=0.0, le=5.0)
    min_duration: str = "16th"
    duration_grid: str = "16th"
    no_triplets: bool = False
    loose_triplets: bool = False
    max_order: int = Field(3, ge=1, le=3)
    transformer_steps: int = Field(80, ge=10, le=1000)
    transformer_top_k: int = Field(16, ge=0, le=64)
    transformer_temperature: float = Field(1.0, ge=0.05, le=2.0)
    transformer_device: str = "auto"


def connect_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def clean_key(raw: str | None) -> str | None:
    if not raw:
        return None
    return raw.split("%", 1)[0].strip()


def theme_urls(theme_id: str) -> dict[str, str]:
    return {
        "svg_url": f"/api/themes/{theme_id}/score.svg?v={CATALOG_SCORE_CACHE_VERSION}",
        "static_svg_url": f"/svgs/{theme_id}.svg",
        "midi_url": f"/midis/{theme_id}.mid",
        "abc_url": f"/abcs/{theme_id}.abc",
        "notes_url": f"/api/themes/{theme_id}/notes",
    }


def row_to_theme(row: sqlite3.Row) -> dict[str, object]:
    result = dict(row)
    result["key_root"] = clean_key(result.get("abc_key"))
    result.update(theme_urls(result["id"]))
    return result


@app.get("/")
def index() -> FileResponse:
    return FileResponse(STATIC_DIR / "index.html")


@app.get("/api/health")
def health() -> dict[str, object]:
    verovio_cli = shutil.which("verovio") is not None
    verovio_python = has_python_verovio_renderer()
    verovio_resource_path = None
    if verovio_python:
        try:
            import verovio

            verovio_resource_path = configure_verovio_resources(verovio)
        except Exception:
            verovio_resource_path = None
    return {
        "ok": True,
        "database": DB_PATH.exists(),
        "verovio": verovio_cli or verovio_python,
        "verovio_cli": verovio_cli,
        "verovio_python": verovio_python,
        "verovio_resource_path": verovio_resource_path,
        "score_preview_render_version": SCORE_PREVIEW_RENDER_VERSION,
        "markov_cache_entries": len(_markov_cache),
        "default_markov_cache": DEFAULT_MARKOV_CACHE.exists(),
        "transformer_checkpoint": DEFAULT_TRANSFORMER_CHECKPOINT.exists(),
        "transformer_loaded": _transformer_checkpoint is not None,
    }


@app.get("/api/stats")
def stats() -> dict[str, object]:
    with connect_db() as conn:
        theme_count = conn.execute("SELECT COUNT(*) FROM themes WHERE parse_error IS NULL").fetchone()[0]
        note_count = conn.execute("SELECT COUNT(*) FROM notes").fetchone()[0]
        composers = conn.execute(
            "SELECT composer, COUNT(*) AS count FROM themes "
            "WHERE parse_error IS NULL AND composer IS NOT NULL "
            "GROUP BY composer ORDER BY count DESC, composer LIMIT 24"
        ).fetchall()
        keys = conn.execute(
            "SELECT abc_key, COUNT(*) AS count FROM themes "
            "WHERE parse_error IS NULL AND abc_key IS NOT NULL "
            "GROUP BY abc_key ORDER BY count DESC LIMIT 24"
        ).fetchall()
    return {
        "themes": theme_count,
        "notes": note_count,
        "top_composers": [dict(row) for row in composers],
        "top_keys": [{"key": clean_key(row["abc_key"]), "count": row["count"]} for row in keys],
    }


@app.get("/api/composers")
def list_composers(
    q: str = "",
    limit: int = Query(12, ge=1, le=40),
) -> dict[str, object]:
    params: list[object] = []
    where = ["parse_error IS NULL", "composer IS NOT NULL", "composer != ''"]
    if q.strip():
        words = [word for word in q.strip().split() if word]
        for word in words:
            where.append("composer LIKE ?")
            params.append(f"%{word}%")
    params.append(limit)
    sql = f"""
        SELECT composer, COUNT(*) AS count
        FROM themes
        WHERE {' AND '.join(where)}
        GROUP BY composer
        ORDER BY
            CASE WHEN composer LIKE ? THEN 0 ELSE 1 END,
            count DESC,
            composer
        LIMIT ?
    """
    starts_with = f"{q.strip()}%" if q.strip() else "%"
    query_params = [*params[:-1], starts_with, params[-1]]
    with connect_db() as conn:
        rows = conn.execute(sql, query_params).fetchall()
    return {"items": [dict(row) for row in rows]}


@app.get("/api/themes")
def list_themes(
    q: str = "",
    composer: str = "",
    key: str = "",
    limit: int = Query(24, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> dict[str, object]:
    params: list[object] = []
    where = ["t.parse_error IS NULL"]
    rank = "t.id"

    if q.strip():
        words = [word for word in q.strip().split() if word]
        like_parts = []
        for word in words:
            pattern = f"%{word}%"
            like_parts.append("(t.title LIKE ? OR t.composer LIKE ? OR d.keywords LIKE ?)")
            params.extend([pattern, pattern, pattern])
        where.append("(" + " AND ".join(like_parts) + ")")
        rank = "t.composer, t.title"

    if composer.strip():
        where.append("t.composer LIKE ?")
        params.append(f"%{composer.strip()}%")

    if key.strip():
        where.append("t.abc_key LIKE ?")
        params.append(f"{key.strip()}%")

    params.extend([limit, offset])
    sql = f"""
        SELECT t.id, t.title, t.composer, t.abc_key, t.abc_meter, t.note_count,
               t.active_span_bars, t.pitch_range, d.keywords
        FROM themes t
        LEFT JOIN theme_descriptions d ON d.id = t.id
        WHERE {' AND '.join(where)}
        ORDER BY {rank}
        LIMIT ? OFFSET ?
    """
    with connect_db() as conn:
        rows = conn.execute(sql, params).fetchall()
    return {"items": [row_to_theme(row) for row in rows], "limit": limit, "offset": offset}


@app.get("/api/themes/{theme_id}")
def get_theme(theme_id: str) -> dict[str, object]:
    with connect_db() as conn:
        row = conn.execute(
            """
            SELECT t.*, d.description, d.keywords
            FROM themes t
            LEFT JOIN theme_descriptions d ON d.id = t.id
            WHERE t.id = ?
            """,
            (theme_id,),
        ).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="Theme not found")
    return row_to_theme(row)


@app.get("/api/themes/{theme_id}/notes")
def theme_notes(theme_id: str) -> dict[str, object]:
    with connect_db() as conn:
        theme = conn.execute("SELECT id, title, composer, bpm FROM themes WHERE id = ?", (theme_id,)).fetchone()
        if theme is None:
            raise HTTPException(status_code=404, detail="Theme not found")
        rows = conn.execute(
            """
            SELECT pitch, start_beat, duration_beats, duration_value, velocity
            FROM notes
            WHERE theme_id = ?
            ORDER BY start_tick, note_index
            """,
            (theme_id,),
        ).fetchall()
    return {
        "id": theme_id,
        "title": theme["title"],
        "composer": theme["composer"],
        "bpm": theme["bpm"] or 120,
        "notes": [dict(row) for row in rows],
    }


def catalog_score_paths(theme_id: str) -> tuple[Path, Path]:
    safe_id = "".join(char for char in theme_id if char.isalnum() or char in ("-", "_"))
    stem = safe_id or "theme"
    return CATALOG_SCORE_ROOT / f"{stem}.musicxml", CATALOG_SCORE_ROOT / f"{stem}.svg"


def render_catalog_theme_score(theme_id: str) -> Path | None:
    musicxml_path, svg_path = catalog_score_paths(theme_id)
    if svg_path.exists():
        return svg_path

    ensure_muses_import_path()
    try:
        from muses.base.temporals import Piece, TemporalCollection
        from muses.io.musicxml import write_musicxml
    except ImportError:
        return None

    with connect_db() as conn:
        theme = conn.execute(
            """
            SELECT id, title, composer, abc_key, time_signature, abc_meter, tempo_us_per_beat
            FROM themes
            WHERE id = ? AND parse_error IS NULL
            """,
            (theme_id,),
        ).fetchone()
        if theme is None:
            raise HTTPException(status_code=404, detail="Theme not found")
        notes = conn.execute(
            """
            SELECT pitch, start_beat, duration_beats, velocity, channel
            FROM notes
            WHERE theme_id = ?
            ORDER BY start_tick, note_index
            """,
            (theme_id,),
        ).fetchall()

    melody = TemporalCollection(name="theme", instrument="piano", program_change=0)
    for note in notes:
        melody.insert_note(
            int(note["pitch"]),
            float(note["start_beat"]),
            float(note["duration_beats"]),
            velocity=int(note["velocity"]),
            midi_channel=int(note["channel"]),
        )

    piece = Piece(
        name=theme["title"] or theme_id,
        title=f"{theme_id}. {theme['title']}" if theme["title"] else theme_id,
        composer=theme["composer"] or "",
        melodies=[melody],
        time_signature=theme["time_signature"] or theme["abc_meter"] or "4/4",
        key_signature=clean_key(theme["abc_key"]) or "C",
        tempo=theme["tempo_us_per_beat"] or 500000,
    )

    try:
        write_musicxml(piece, musicxml_path, quantization_tolerance=0.1)
        return render_musicxml_to_svg(musicxml_path)
    except (RuntimeError, ValueError, OSError, subprocess.SubprocessError):
        return None


@app.get("/api/themes/{theme_id}/score.svg")
def theme_score_svg(theme_id: str) -> FileResponse:
    svg_path = render_catalog_theme_score(theme_id)
    if svg_path is None:
        fallback = ROOT / "svgs" / f"{theme_id}.svg"
        if fallback.exists():
            return FileResponse(fallback, media_type="image/svg+xml")
        raise HTTPException(status_code=404, detail="Theme score not available")
    return FileResponse(svg_path, media_type="image/svg+xml")


def sequence_note_events(sequence: tuple[Symbol, ...], key_name: str) -> list[dict[str, object]]:
    pitches = sequence_to_pitches(sequence, PITCH_CLASS[key_name])
    start = 0.0
    events = []
    for pitch, symbol in zip(pitches, sequence):
        duration = COMMON_DURATION_BEATS[symbol.duration]
        events.append(
            {
                "pitch": pitch,
                "start_beat": start,
                "duration_beats": duration,
                "duration_value": symbol.duration,
                "velocity": 72,
            }
        )
        start += duration
    return events


def has_verovio_renderer() -> bool:
    if shutil.which("verovio") is not None:
        return True
    return has_python_verovio_renderer()


def has_python_verovio_renderer() -> bool:
    try:
        import verovio  # noqa: F401
    except ImportError:
        return False
    return True


def packaged_verovio_resource_path(verovio_module) -> Path | None:
    module_file = getattr(verovio_module, "__file__", None)
    if not module_file:
        return None
    data_path = Path(module_file).resolve().parent / "data"
    if not data_path.exists():
        return None
    required = ["Bravura.xml", "Leipzig.xml"]
    if not all((data_path / name).exists() for name in required):
        return None
    return data_path


def configure_verovio_resources(verovio_module, toolkit=None) -> str | None:
    data_path = packaged_verovio_resource_path(verovio_module)
    if data_path is None:
        return None
    resource_path = str(data_path)
    try:
        if hasattr(verovio_module, "setDefaultResourcePath"):
            verovio_module.setDefaultResourcePath(resource_path)
        if toolkit is not None and hasattr(toolkit, "setResourcePath"):
            toolkit.setResourcePath(resource_path)
    except Exception:
        return None
    return resource_path


def render_musicxml_with_python_verovio(musicxml_path: Path, svg_path: Path) -> Path | None:
    try:
        import verovio

        configure_verovio_resources(verovio)
        musicxml_data = musicxml_path.read_text(encoding="utf-8")
        toolkit = verovio.toolkit()
        configure_verovio_resources(verovio, toolkit)
        toolkit.setOptions({"scale": 42})
        svg = toolkit.renderData(musicxml_data, {}) if hasattr(toolkit, "renderData") else ""
        if not svg:
            loaded = toolkit.loadData(musicxml_data) if hasattr(toolkit, "loadData") else toolkit.loadFile(str(musicxml_path))
            if not loaded:
                return None
            svg = toolkit.renderToSVG(1)
        if "<svg" not in svg:
            return None
        svg_path.write_text(svg, encoding="utf-8")
    except Exception:
        return None
    return svg_path if svg_path.exists() else None


def render_musicxml_to_svg(musicxml_path: Path) -> Path | None:
    svg_path = musicxml_path.with_suffix(".svg")
    python_svg = render_musicxml_with_python_verovio(musicxml_path, svg_path)
    if python_svg is not None:
        return python_svg
    verovio = shutil.which("verovio")
    if verovio is None:
        return None
    try:
        subprocess.run(
            [verovio, "-s", "42", "-o", str(svg_path), str(musicxml_path)],
            cwd=ROOT,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=20,
        )
    except (subprocess.SubprocessError, OSError):
        return None
    return svg_path if svg_path.exists() else None


def markov_cache_key(request: GenerateRequest) -> tuple[object, ...]:
    return (
        request.length,
        request.max_order,
        request.endpoint_strength,
        request.min_duration,
        request.duration_grid,
        request.no_triplets,
        request.loose_triplets,
    )


def cached_markov_backend(request: GenerateRequest, args: argparse.Namespace) -> dict[str, object]:
    key = markov_cache_key(request)
    with _markov_cache_lock:
        cached = _markov_cache.get(key)
        if cached is not None:
            return cached

        require_vo_regular()
        allowed_durations, sequences, stats_data, priors = load_generation_inputs(
            args,
            min_len=max(6, request.max_order + 1),
        )
        start_weights, end_weights = priors
        model = OrderStackModel.from_sequences(
            sequences,
            max_order=request.max_order,
            start_symbol=START_SYMBOL,
        )
        acceptor = make_theme_acceptor(
            length=request.length,
            alphabet=model.alphabet,
            start_weights=start_weights,
            end_weights=end_weights,
            strength=request.endpoint_strength,
            enforce_triplet_groups=not request.loose_triplets,
        )
        backend = prepare_constrained_order_stack(
            model,
            ConstraintSet(regular_acceptors=(acceptor,)),
            length=request.length,
            prefix=(START_SYMBOL,),
            policy=LongestFeasiblePolicy(),
        )

        # vo_regular_bp does expensive lazy work on the first sample. Do it once
        # while caching the backend instead of making the user's first click pay.
        backend.sample(rng=random.Random(0))
        cached = {
            "backend": backend,
            "allowed_durations": allowed_durations,
            "stats": stats_data,
            "diagnostics": {
                "constrained success mass": f"{backend.diagnostics.success_mass:.6g}",
                "engine": "vo_regular variable-order Markov",
                "cache": "warm",
            },
        }
        _markov_cache[key] = cached
    return cached


def default_markov_request() -> GenerateRequest:
    return GenerateRequest()


def load_precomputed_markov_backend(cache_path: Path = DEFAULT_MARKOV_CACHE) -> bool:
    if not cache_path.exists():
        return False

    request = default_markov_request()
    key = markov_cache_key(request)
    try:
        with cache_path.open("rb") as handle:
            payload = pickle.load(handle)
    except (OSError, pickle.PickleError, EOFError, AttributeError, TypeError, ValueError):
        return False

    if payload.get("format_version") != MARKOV_CACHE_FORMAT_VERSION or payload.get("key") != key:
        return False

    cached = payload.get("cached")
    if not isinstance(cached, dict) or "backend" not in cached:
        return False

    try:
        for seed in range(MARKOV_PRECOMPUTED_WARM_SAMPLES):
            cached["backend"].sample(rng=random.Random(seed))
    except Exception:
        return False

    diagnostics = dict(cached.get("diagnostics", {}))
    diagnostics["cache"] = "precomputed"
    cached["diagnostics"] = diagnostics
    with _markov_cache_lock:
        _markov_cache[key] = cached
    return True


def write_precomputed_markov_backend(cache_path: Path = DEFAULT_MARKOV_CACHE) -> Path:
    request = default_markov_request()
    key = markov_cache_key(request)
    cached = cached_markov_backend(request, args_from_request(request, GENERATED_ROOT / "_precompute"))
    payload = {
        "format_version": MARKOV_CACHE_FORMAT_VERSION,
        "key": key,
        "cached": cached,
    }
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = cache_path.with_suffix(".tmp")
    with tmp_path.open("wb") as handle:
        pickle.dump(payload, handle, protocol=pickle.HIGHEST_PROTOCOL)
    tmp_path.replace(cache_path)
    return cache_path


def generate_cached_markov(request: GenerateRequest, args: argparse.Namespace):
    cached = cached_markov_backend(request, args)
    backend = cached["backend"]
    rng = random.Random(request.seed)
    with _markov_cache_lock:
        generated = [backend.sample(rng=rng) for _ in range(request.samples)]
    return (
        generated,
        dict(cached["diagnostics"]),
        cached["stats"],
        cached["allowed_durations"],
    )


def warm_default_markov_backend() -> None:
    request = default_markov_request()
    try:
        cached_markov_backend(request, args_from_request(request, GENERATED_ROOT / "_warmup"))
    except Exception:
        # Keep startup resilient; /api/generate will return the concrete error.
        return


def load_default_transformer_checkpoint() -> None:
    global _transformer_checkpoint
    if not DEFAULT_TRANSFORMER_CHECKPOINT.exists():
        return
    try:
        checkpoint = load_transformer_checkpoint(DEFAULT_TRANSFORMER_CHECKPOINT, requested_device="auto")
    except Exception:
        return
    with _transformer_lock:
        _transformer_checkpoint = checkpoint


def start_background_warmup() -> None:
    loaded_precomputed = load_precomputed_markov_backend()
    if not loaded_precomputed and os.environ.get("THEME_LAB_WARM_MARKOV", "").lower() in {"1", "true", "yes"}:
        threading.Thread(target=warm_default_markov_backend, daemon=True).start()
    threading.Thread(target=load_default_transformer_checkpoint, daemon=True).start()


def args_from_request(request: GenerateRequest, output_dir: Path) -> argparse.Namespace:
    return argparse.Namespace(
        db=DB_PATH,
        output_dir=output_dir,
        length=request.length,
        samples=request.samples,
        key=request.key,
        endpoint_strength=request.endpoint_strength,
        seed=request.seed,
        min_duration=request.min_duration,
        duration_grid=request.duration_grid,
        no_triplets=request.no_triplets,
        loose_triplets=request.loose_triplets,
        write_abc=True,
        write_musicxml=True,
        max_order=request.max_order,
    )


@app.post("/api/generate")
def generate(request: GenerateRequest) -> dict[str, object]:
    if request.key not in PITCH_CLASS:
        raise HTTPException(status_code=400, detail=f"Unsupported key {request.key!r}")

    run_id = uuid.uuid4().hex[:12]
    output_dir = GENERATED_ROOT / run_id
    output_dir.mkdir(parents=True, exist_ok=True)
    args = args_from_request(request, output_dir)

    try:
        if request.engine == "markov":
            generated, diagnostics, stats_data, allowed_durations = generate_cached_markov(request, args)
        else:
            allowed_durations, sequences, stats_data, priors = load_generation_inputs(
                args,
                min_len=max(6, TransformerConfig().block_size // 4),
            )
            start_weights, end_weights = priors
            with _transformer_lock:
                checkpoint = _transformer_checkpoint
            if checkpoint is not None:
                generated, diagnostics = sample_transformer_checkpoint(
                    checkpoint=checkpoint,
                    length=request.length,
                    samples=request.samples,
                    start_weights=start_weights,
                    end_weights=end_weights,
                    endpoint_strength=request.endpoint_strength,
                    enforce_triplet_groups=not request.loose_triplets,
                    seed=request.seed,
                    temperature=request.transformer_temperature,
                    top_k=request.transformer_top_k,
                    max_retries=max(100, request.samples * 30),
                )
                diagnostics["checkpoint mode"] = "pretrained"
            else:
                cfg = TransformerConfig(
                    steps=request.transformer_steps,
                    top_k=request.transformer_top_k,
                    temperature=request.transformer_temperature,
                    max_retries=max(100, request.samples * 30),
                )
                generated, diagnostics = generate_transformer(
                    sequences=sequences,
                    length=request.length,
                    samples=request.samples,
                    start_weights=start_weights,
                    end_weights=end_weights,
                    endpoint_strength=request.endpoint_strength,
                    enforce_triplet_groups=not request.loose_triplets,
                    seed=request.seed,
                    cfg=cfg,
                    device=request.transformer_device,
                )
                diagnostics["checkpoint mode"] = "on-demand training"
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    write_samples(
        generated,
        output_dir=output_dir,
        key_name=request.key,
        engine_name=f"{request.engine} web",
        write_abc=True,
        write_musicxml_files=True,
    )

    samples = []
    for index, sequence in enumerate(generated, start=1):
        stem = f"generated_{index:02d}"
        musicxml_path = output_dir / f"{stem}.musicxml"
        notes = sequence_note_events(sequence, request.key)
        svg_path = render_musicxml_to_svg(musicxml_path)
        samples.append(
            {
                "index": index,
                "title": f"{request.engine} sample {index:02d}",
                "relative_pcs": [symbol.rpc for symbol in sequence],
                "durations": [symbol.duration for symbol in sequence],
                "notes": notes,
                "midi_url": f"/generated/{run_id}/{stem}.mid",
                "abc_url": f"/generated/{run_id}/{stem}.abc",
                "musicxml_url": f"/generated/{run_id}/{stem}.musicxml",
                "svg_url": f"/generated/{run_id}/{stem}.svg" if svg_path else None,
            }
        )

    return {
        "run_id": run_id,
        "engine": request.engine,
        "diagnostics": diagnostics,
        "stats": {
            "sequences": stats_data["sequence_count"],
            "events": stats_data["event_count"],
            "vocabulary_size": stats_data["vocab_size"],
            "allowed_durations": sorted(allowed_durations, key=COMMON_DURATION_BEATS.get),
        },
        "samples": samples,
    }


def main() -> None:
    import uvicorn

    port = int(os.environ.get("PORT", "7860"))
    uvicorn.run("apps.theme_lab.app:app", host="0.0.0.0", port=port, reload=False)


if __name__ == "__main__":
    main()