raaga_id/model.py

"""Model: architecture + load/save (PRD §17).

Framework is PyTorch (D15). The v0 floor (D16 step 1) is an XGBoost classifier over
the fixed-length frame vector — cheap, strong, trains in minutes on CPU. The CNN /
TDMS / transformer rungs come later; `RaagaCNN` is a stub marking that seam.
"""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

import numpy as np

from .config import MODELS_DIR, TOP_K


@dataclass
class Prediction:
    raaga: str
    confidence: float


class RaagaXGB:
    """XGBoost floor over frame vectors. Wraps label<->index + top-k decode."""

    def __init__(self, classes: list[str], booster=None, temperature: float = 1.0):
        self.classes = list(classes)
        self._booster = booster
        self.temperature = temperature   # calibration (D25); 1.0 = uncalibrated identity

    def fit(self, X: np.ndarray, y_idx: np.ndarray, **kwargs):
        from xgboost import XGBClassifier

        self._booster = XGBClassifier(
            objective="multi:softprob",
            num_class=len(self.classes),
            n_estimators=kwargs.pop("n_estimators", 400),
            max_depth=kwargs.pop("max_depth", 6),
            learning_rate=kwargs.pop("learning_rate", 0.1),
            n_jobs=-1,
            **kwargs,
        )
        self._booster.fit(X, y_idx)
        return self

    def predict_proba(self, X: np.ndarray) -> np.ndarray:
        if self._booster is None:
            raise RuntimeError("model is not trained/loaded")
        return self._booster.predict_proba(X)

    def top_k(self, x: np.ndarray, k: int = TOP_K) -> list[Prediction]:
        """Top-k raagas + confidence for one frame vector (D6: always top-3)."""
        proba = self.predict_proba(x.reshape(1, -1))[0]
        return self._decode(proba, k)

    def aggregate_top_k(self, X: np.ndarray, k: int = TOP_K) -> list[Prediction]:
        """Top-k for a clip given its per-window vectors: mean the window
        probabilities, then decode (D7: aggregate predictions across a long clip)."""
        X = np.atleast_2d(X)
        proba = self.predict_proba(X).mean(axis=0)
        return self._decode(proba, k)

    def _decode(self, proba: np.ndarray, k: int) -> list[Prediction]:
        from .calibrate import apply_temperature
        proba = apply_temperature(proba, self.temperature)   # D25; argmax-preserving
        order = np.argsort(proba)[::-1][:k]
        return [Prediction(self.classes[i], float(proba[i])) for i in order]

    def save(self, path: str | Path | None = None) -> Path:
        path = Path(path) if path else MODELS_DIR / "raaga_xgb.json"
        path.parent.mkdir(parents=True, exist_ok=True)
        self._booster.save_model(str(path))
        path.with_suffix(".classes.json").write_text(json.dumps(self.classes))
        return path

    @classmethod
    def load(cls, path: str | Path) -> "RaagaXGB":
        from xgboost import XGBClassifier

        from .calibrate import load_temperature

        path = Path(path)
        booster = XGBClassifier()
        booster.load_model(str(path))
        classes = json.loads(path.with_suffix(".classes.json").read_text())
        return cls(classes, booster, temperature=load_temperature(path))


class RaagaCNN:
    """PHASE 2 SEAM (D16). CNN on tonic-normalized mel/CQT, in torch/Lightning.
    Benchmark target: compIAM `DEEPSRGM`."""

    def __init__(self, *_, **__):
        raise NotImplementedError("CNN rung — implement after the XGBoost floor lands")