dmap.py

# src/dima/dmap.py
from __future__ import annotations
import json
from typing import Any, Dict, Optional, Tuple, Union

import numpy as np
import scipy.sparse as sp
from scipy.sparse.linalg import eigsh, LinearOperator, lobpcg
from huggingface_hub import HfApi, HfFolder, upload_file, hf_hub_download

from .ann import ANNBackend, make_ann
from .utils import median_eps_from_knn_d2


def k_ideal(d: int, N: int) -> int:
    """

    Heuristic for kNN graph size in diffusion maps.

    Stable default: grows slowly with N and linearly with d.

    """
    d = int(max(1, d))
    N = int(max(2, N))
    k = int(np.ceil(2.0 * d * np.log2(N)))
    return int(min(max(8, k), N - 1))


def _sqdist_ab(A: np.ndarray, B: np.ndarray) -> np.ndarray:
    """

    Squared Euclidean distances between rows:

      A: (a,d), B: (b,d) -> D2: (a,b)

    """
    A = np.asarray(A, dtype=np.float64)
    B = np.asarray(B, dtype=np.float64)
    A2 = np.sum(A * A, axis=1, keepdims=True)
    B2 = np.sum(B * B, axis=1, keepdims=True).T
    G = A @ B.T
    return np.maximum(A2 + B2 - 2.0 * G, 0.0)


class DMAP:
    """

    Diffusion Maps encoder with Nyström out-of-sample extension.



    Notation (arrays named by indices):

      R_iX: reference ambient data

      K_ij: kernel on graph edges (sparse CSR)

      q_i = Σ_j K_ij

      qα_i = (q_i)^α

      Kα_ij = K_ij / (qα_i qα_j)

      d_i = Σ_j Kα_ij

      A_ij = Kα_ij / sqrt(d_i d_j) (symmetric)

      eigsh(A) -> λ_x, u_ix

      ψ_ix = u_ix / sqrt(d_i)

      R_ix = (λ_x)^t ψ_ix



    Nyström OOS for novel ambient R_aX:

      K_ai = exp(-β * D2_ai / ε)

      q_a = Σ_i K_ai, qα_a = (q_a)^α

      Kα_ai = K_ai / (qα_a qα_i)

      d_a = Σ_i Kα_ai

      P_ai = Kα_ai / d_a

      R_ax = Σ_i P_ai * (R_ix / λ_x)



    This implementation uses a kNN graph, sparse eigensolver, and

    provides optional dense refinement via streaming LOBPCG.

    """

    def __init__(

        self,

        R_iX: np.ndarray,

        *,

        d: int = 32,

        beta: float = 1.0,

        alpha: float = 0.0,

        t: float = 1.0,

        k: Optional[int] = None,

        eps: Optional[float] = None,

        eps_use_kth: bool = True,

        eps_mul: float = 1.0,

        drop_trivial: bool = True,

        seed: int = 0,

        sym: str = "max",

        dtype: Any = np.float32,

        ann_backend: ANNBackend = "auto",

        ann_params: Optional[Dict[str, Any]] = None,

        n_jobs: int = -1,

        # dense refinement (O(N^2) compute, streaming memory)

        refine_dense: bool = False,

        stream_block: int = 4096,

        lobpcg_maxiter: int = 3,

        lobpcg_tol: float = 1e-6,

        use_symmetry: bool = True,

        # allow unicode kwargs (β, α, ε, ε_mul, ε_use_kth, ...)

        **kwargs: Any,

    ):
        # ---- map unicode kwargs -> ascii ----
        if "β" in kwargs:
            beta = kwargs.pop("β")
        if "α" in kwargs:
            alpha = kwargs.pop("α")
        if "ε" in kwargs:
            eps = kwargs.pop("ε")
        if "ε_use_kth" in kwargs:
            eps_use_kth = kwargs.pop("ε_use_kth")
        if "ε_mul" in kwargs:
            eps_mul = kwargs.pop("ε_mul")
        if "sym" in kwargs:
            sym = kwargs.pop("sym")
        if kwargs:
            raise TypeError(f"Unexpected kwargs: {sorted(kwargs.keys())}")

        self.d = int(d)
        self.k = int(k_ideal(self.d, int(np.asarray(R_iX).shape[0])) if k is None else int(k))

        self.beta = float(beta)
        self.alpha = float(alpha)
        self.t = float(t)
        self.drop_trivial = bool(drop_trivial)
        self.seed = int(seed)
        self.sym = str(sym)
        self.dtype = dtype

        # unicode aliases (so older code + packers can find them)
        self.β = self.beta
        self.α = self.alpha

        self.refine_dense = bool(refine_dense)
        self.stream_block = int(stream_block)
        self.lobpcg_maxiter = int(lobpcg_maxiter)
        self.lobpcg_tol = float(lobpcg_tol)
        self.use_symmetry = bool(use_symmetry)

        rng = np.random.default_rng(self.seed)

        # reference data
        R_iX = np.asarray(R_iX)
        if R_iX.ndim != 2:
            raise ValueError(f"R_iX must be 2D array, got shape {R_iX.shape}")
        self.R_iX = np.ascontiguousarray(R_iX.astype(self.dtype, copy=False))

        Nref, D = self.R_iX.shape
        self.Nref = int(Nref)
        self.D = int(D)

        if not (2 <= self.k < Nref):
            raise ValueError(f"Invalid k={self.k} for Nref={Nref}")

        # ANN index (ambient)
        self.ann, self.ann_backend = make_ann(ann_backend, ann_params=ann_params, n_jobs=n_jobs)
        self.ann.build(self.R_iX)

        # kNN of reference points
        j_iK, D2_iK = self.ann.search(self.R_iX, self.k)  # (Nref,k)

        # eps selection
        if eps is None:
            self.eps = float(median_eps_from_knn_d2(D2_iK, use_kth=bool(eps_use_kth)) * float(eps_mul))
        else:
            self.eps = float(eps)
        self.ε = self.eps  # unicode alias

        # -------------------------
        # 1) Build sparse graph and warm-start eigensolver
        # -------------------------
        K_iK = np.exp(-self.beta * (D2_iK / self.eps)).astype(np.float64, copy=False)

        indptr = (np.arange(Nref + 1, dtype=np.int64) * self.k)
        indices = j_iK.reshape(-1).astype(np.int64, copy=False)
        data = K_iK.reshape(-1)

        K_ij = sp.csr_matrix((data, indices, indptr), shape=(Nref, Nref), dtype=np.float64)

        # symmetrize
        if self.sym == "max":
            K_ij = K_ij.maximum(K_ij.T)
        elif self.sym == "mean":
            K_ij = (K_ij + K_ij.T) * 0.5
        else:
            raise ValueError(f"Unknown sym={self.sym!r}")

        # degrees q_i and qalpha_i (warm)
        q_i_warm = np.asarray(K_ij.sum(axis=1)).ravel()
        q_i_warm = np.maximum(q_i_warm, 1e-30)
        qalpha_i_warm = np.maximum(np.power(q_i_warm, self.alpha), 1e-30)

        Qinv = sp.diags(1.0 / qalpha_i_warm, format="csr")
        Kalpha_ij = Qinv @ K_ij @ Qinv

        d_i_warm = np.asarray(Kalpha_ij.sum(axis=1)).ravel()
        d_i_warm = np.maximum(d_i_warm, 1e-30)

        Dinv_sqrt = sp.diags(1.0 / np.sqrt(d_i_warm), format="csr")
        A_ij = Dinv_sqrt @ Kalpha_ij @ Dinv_sqrt

        nev = self.d + (1 if self.drop_trivial else 0)
        v0 = rng.normal(size=Nref).astype(np.float64)
        lam0, u0 = eigsh(A_ij, k=nev, which="LA", v0=v0)

        ord0 = np.argsort(lam0)[::-1]
        lam0 = lam0[ord0]
        u0 = u0[:, ord0]

        # LOBPCG warm-start block (orthonormalize)
        X0, _ = np.linalg.qr(u0.astype(np.float64, copy=False))

        # -------------------------
        # 2) Optional refinement: streaming dense LOBPCG on dense PSD operator
        # -------------------------
        if self.refine_dense:
            try:
                # Rebuild dense operator using streaming blocks without allocating full N^2 matrix
                X = self.R_iX.astype(np.float64, copy=False)
                X2 = np.sum(X * X, axis=1, keepdims=True)

                # compute q_i, qalpha_i, d_i for dense kernel operator:
                # q_i = Σ_j K_ij,  K_ij = exp(-β*||xi-xj||^2 / eps)
                # Kα_ij = K_ij/(qα_i qα_j)
                # d_i = Σ_j Kα_ij

                # Step A: q_i
                ones = np.ones((Nref, 1), dtype=np.float64)
                q_i = self._K_matmat_dense(X, X2, ones).ravel()
                q_i = np.maximum(q_i, 1e-30)
                qalpha_i = np.maximum(np.power(q_i, self.alpha), 1e-30)

                # Step B: d_i
                inv_qalpha = 1.0 / qalpha_i
                V = (ones * inv_qalpha[:, None])  # (N,1)  actually N x 1
                tmp = self._K_matmat_dense(X, X2, V).ravel()  # Σ_j K_ij * inv_qalpha_j
                d_i = inv_qalpha * tmp  # Σ_j K_ij/(qα_i qα_j)
                d_i = np.maximum(d_i, 1e-30)

                # Build symmetric operator A(v) = D^{-1/2} Q^{-1} K Q^{-1} D^{-1/2} v
                inv_sqrt_d = 1.0 / np.sqrt(d_i)
                inv_qalpha = 1.0 / qalpha_i

                def matvec(v: np.ndarray) -> np.ndarray:
                    v = v.astype(np.float64, copy=False).reshape(-1, 1)  # (N,1)
                    w = v * inv_sqrt_d[:, None]
                    w = w * inv_qalpha[:, None]
                    y = self._K_matmat_dense(X, X2, w)
                    y = y * inv_qalpha[:, None]
                    y = y * inv_sqrt_d[:, None]
                    return y.ravel()

                Aop = LinearOperator((Nref, Nref), matvec=matvec, dtype=np.float64)

                # LOBPCG refine using warm-start X0
                lam, u = lobpcg(Aop, X0, largest=True, maxiter=self.lobpcg_maxiter, tol=self.lobpcg_tol)

                ord1 = np.argsort(lam)[::-1]
                lam = lam[ord1]
                u = u[:, ord1]

                self.q_i = q_i.astype(np.float64, copy=False)
                self.qalpha_i = qalpha_i.astype(np.float64, copy=False)
                self.d_i = d_i.astype(np.float64, copy=False)
            except Exception:
                # fallback to warm start if refinement fails
                lam, u = lam0, u0
                self.q_i = q_i_warm.astype(np.float64, copy=False)
                self.qalpha_i = qalpha_i_warm.astype(np.float64, copy=False)
                self.d_i = d_i_warm.astype(np.float64, copy=False)
        else:
            lam, u = lam0, u0
            self.q_i = q_i_warm.astype(np.float64, copy=False)
            self.qalpha_i = qalpha_i_warm.astype(np.float64, copy=False)
            self.d_i = d_i_warm.astype(np.float64, copy=False)

        # provide unicode aliases for packers / older code
        self.qα_i = self.qalpha_i
        self.λ_x = lam.astype(np.float64, copy=False)

        # psi and drop trivial
        psi = u / np.sqrt(self.d_i)[:, None]

        if self.drop_trivial:
            lam = lam[1:]
            psi = psi[:, 1:]
            u = u[:, 1:]

        # diffusion coords
        R_ix = psi * (lam ** self.t)[None, :]

        # store
        self.λ_x = lam.astype(np.float64, copy=False)      # (d,)
        self.u_ix = u.astype(np.float64, copy=False)      # (Nref,d)
        self.ψ_ix = psi.astype(np.float64, copy=False)    # (Nref,d)
        self.R_ix = R_ix.astype(np.float64, copy=False)   # (Nref,d)
        self.π_i = (self.d_i / self.d_i.sum()).astype(np.float64, copy=False)

        # for Nyström: R_ix / λ_x
        self.R_over_λ_ix = (self.R_ix / self.λ_x[None, :]).astype(np.float64, copy=False)

    # --------- Dense kernel streaming utilities ----------

    def _rbf_block(self, Xi: np.ndarray, Xj: np.ndarray, X2i: np.ndarray, X2j: np.ndarray) -> np.ndarray:
        # ||xi-xj||^2 = xi^2 + xj^2 - 2 xi·xj
        G = Xi @ Xj.T
        D2 = np.maximum(X2i + X2j.T - 2.0 * G, 0.0)
        return np.exp(-self.beta * (D2 / self.eps))

    def _K_matmat_dense(self, X: np.ndarray, X2: np.ndarray, V: np.ndarray) -> np.ndarray:
        """

        Compute (dense) K @ V without forming K explicitly, using streaming blocks.

        X: (N,D), X2: (N,1), V: (N,m) -> out: (N,m)

        """
        X = np.asarray(X, dtype=np.float64)
        X2 = np.asarray(X2, dtype=np.float64)
        V = np.asarray(V, dtype=np.float64)
        N = X.shape[0]
        bs = self.stream_block

        out = np.zeros((N, V.shape[1]), dtype=np.float64)

        if not self.use_symmetry:
            for i0 in range(0, N, bs):
                i1 = min(N, i0 + bs)
                Xi = X[i0:i1]
                X2i = X2[i0:i1]
                acc = np.zeros((i1 - i0, V.shape[1]), dtype=np.float64)
                for j0 in range(0, N, bs):
                    j1 = min(N, j0 + bs)
                    Xj = X[j0:j1]
                    X2j = X2[j0:j1]
                    Kij = self._rbf_block(Xi, Xj, X2i, X2j)
                    acc += Kij @ V[j0:j1]
                out[i0:i1] = acc
            return out

        # symmetric tiling
        for i0 in range(0, N, bs):
            i1 = min(N, i0 + bs)
            Xi = X[i0:i1]
            X2i = X2[i0:i1]
            Vi = V[i0:i1]

            # diagonal tile
            Kii = self._rbf_block(Xi, Xi, X2i, X2i)
            out[i0:i1] += Kii @ Vi

            for j0 in range(i1, N, bs):
                j1 = min(N, j0 + bs)
                Xj = X[j0:j1]
                X2j = X2[j0:j1]
                Vj = V[j0:j1]

                Kij = self._rbf_block(Xi, Xj, X2i, X2j)
                out[i0:i1] += Kij @ Vj
                out[j0:j1] += Kij.T @ Vi

        return out

    # --------- Nyström embedding ----------

    def __call__(self, R_aX: Union[np.ndarray, list], *, batch_size: Optional[int] = None) -> np.ndarray:
        R_aX = np.asarray(R_aX)
        single = (R_aX.ndim == 1)
        if single:
            R_aX = R_aX[None, :]

        R_aX = np.ascontiguousarray(R_aX.astype(self.dtype, copy=False))

        if batch_size is None:
            R_ax = self._embed(R_aX)
        else:
            bs = int(batch_size)
            out = []
            for s in range(0, R_aX.shape[0], bs):
                out.append(self._embed(R_aX[s:s + bs]))
            R_ax = np.vstack(out)

        return R_ax[0] if single else R_ax

    def _embed(self, R_aX: np.ndarray) -> np.ndarray:
        # kNN for novel points
        j_aK, D2_aK = self.ann.search(R_aX, self.k)  # (a,k)

        K_ai = np.exp(-self.beta * (D2_aK.astype(np.float64) / self.eps))  # (a,k)

        q_a = np.maximum(K_ai.sum(axis=1), 1e-30)
        qalpha_a = np.maximum(np.power(q_a, self.alpha), 1e-30)

        qalpha_i = np.maximum(self.qalpha_i[j_aK], 1e-30)
        Kalpha_ai = K_ai / (qalpha_a[:, None] * qalpha_i)

        d_a = np.maximum(Kalpha_ai.sum(axis=1), 1e-30)
        P_ai = Kalpha_ai / d_a[:, None]

        R_over = self.R_over_λ_ix[j_aK, :]  # (a,k,d)
        R_ax = (P_ai[:, :, None] * R_over).sum(axis=1)
        return R_ax

    # -------------------------
    # Serialization helpers
    # -------------------------

    @staticmethod
    def _np_dtype_str(x) -> str:
        """Best-effort numpy dtype string for serialization."""
        try:
            return str(np.dtype(x))
        except Exception:
            return "float32"

    def state_dict(self) -> Dict[str, Any]:
        """

        Minimal, inference-sufficient state for Nyström out-of-sample embedding.



        The resulting dict is intentionally compatible with `FrozenDMAP` in `dima.py`

        (keys: R_iX, qalpha_i, R_over_lam_ix, k, beta, alpha, eps, dtype).

        """
        R_over = getattr(self, "R_over_λ_ix", None)
        if R_over is None:
            # Backward/alternate name safety
            R_over = getattr(self, "R_over_lam_ix", None)

        if R_over is None:
            raise AttributeError("DMAP object has no Nyström matrix `R_over_λ_ix` (did training finish?).")

        state: Dict[str, Any] = dict(
            k=int(self.k),
            beta=float(self.beta),
            alpha=float(self.alpha),
            eps=float(self.eps),
            dtype=self._np_dtype_str(getattr(self, "dtype", "float32")),
            # arrays
            R_iX=np.asarray(self.R_iX),
            qalpha_i=np.asarray(self.qalpha_i, dtype=np.float64),
            R_over_lam_ix=np.asarray(R_over, dtype=np.float64),
            # small meta (optional)
            meta=dict(
                Nref=int(np.asarray(self.R_iX).shape[0]),
                D=int(np.asarray(self.R_iX).shape[1]),
                d=int(np.asarray(R_over).shape[1]),
                t=float(getattr(self, "t", 1.0)),
                drop_trivial=bool(getattr(self, "drop_trivial", False)),
                sym=str(getattr(self, "sym", "max")),
            ),
        )
        return state

    def save_local(self, weights_file: str = "dmap.msgpack", config_file: Optional[str] = "dmap_config.json") -> None:
        """

        Save DMAP weights (and optionally a lightweight JSON config) locally.



        - `weights_file`: binary msgpack with arrays (via flax.serialization)

        - `config_file`: JSON with scalar metadata only (no large arrays)

        """
        # local import keeps `dima` usable without flax unless you call this
        from flax import serialization as flax_ser  # type: ignore
        import json as _json

        state = self.state_dict()
        blob = flax_ser.msgpack_serialize(state)

        with open(weights_file, "wb") as f:
            f.write(blob)

        if config_file is not None:
            cfg = dict(
                k=int(state["k"]),
                beta=float(state["beta"]),
                alpha=float(state["alpha"]),
                eps=float(state["eps"]),
                dtype=str(state.get("dtype", "float32")),
                **(state.get("meta", {}) or {}),
            )
            with open(config_file, "w") as f:
                _json.dump(cfg, f, indent=2)

        return None

    @classmethod
    def from_state(

        cls,

        state: Dict[str, Any],

        *,

        ann_backend: ANNBackend = "auto",

        ann_params: Optional[Dict[str, Any]] = None,

        n_jobs: int = -1,

    ) -> "DMAP":
        """

        Rehydrate a DMAP object from `state_dict()` output WITHOUT retraining.



        This bypasses `__init__` and rebuilds only what is needed for out-of-sample Nyström embedding:

        reference points, normalization factors, Nyström matrix, and ANN index.

        """
        # allow passing a full DIMA state dict
        if "encoder" in state and isinstance(state["encoder"], dict):
            state = state["encoder"]  # type: ignore[assignment]

        # tolerate unicode key variants
        k = int(state["k"])
        beta = float(state.get("beta", state.get("β")))
        alpha = float(state.get("alpha", state.get("α")))
        eps = float(state.get("eps", state.get("ε")))

        dtype = np.dtype(state.get("dtype", "float32"))
        R_iX = np.ascontiguousarray(np.asarray(state["R_iX"]).astype(dtype, copy=False))
        qalpha_i = np.asarray(state.get("qalpha_i", state.get("qα_i")), dtype=np.float64)

        R_over = state.get("R_over_lam_ix", state.get("R_over_λ_ix"))
        if R_over is None:
            raise KeyError("State must contain 'R_over_lam_ix' (or 'R_over_λ_ix').")
        R_over = np.asarray(R_over, dtype=np.float64)

        obj = cls.__new__(cls)  # bypass __init__

        # required for embedding
        obj.k = k
        obj.beta = beta
        obj.alpha = alpha
        obj.eps = eps
        obj.dtype = dtype
        obj.R_iX = R_iX
        obj.qalpha_i = qalpha_i
        obj.qα_i = obj.qalpha_i  # alias
        obj.R_over_λ_ix = R_over
        obj.R_over_lam_ix = obj.R_over_λ_ix  # alias for external code
        obj.β = obj.beta
        obj.α = obj.alpha
        obj.ε = obj.eps

        # optional metadata for introspection
        meta = state.get("meta", {}) if isinstance(state.get("meta", {}), dict) else {}
        obj.d = int(meta.get("d", R_over.shape[1]))
        obj.t = float(meta.get("t", 1.0))
        obj.drop_trivial = bool(meta.get("drop_trivial", False))
        obj.sym = str(meta.get("sym", "max"))

        # rebuild ANN index
        obj.ann, obj.ann_backend = make_ann(ann_backend, ann_params=ann_params, n_jobs=n_jobs)
        obj.ann.build(obj.R_iX)

        return obj

    @classmethod
    def load_local(

        cls,

        weights_file: str = "dmap.msgpack",

        *,

        ann_backend: ANNBackend = "auto",

        ann_params: Optional[Dict[str, Any]] = None,

        n_jobs: int = -1,

    ) -> "DMAP":
        """Load DMAP weights saved by `save_local()`."""
        from flax import serialization as flax_ser  # type: ignore

        with open(weights_file, "rb") as f:
            state = flax_ser.msgpack_restore(f.read())

        return cls.from_state(state, ann_backend=ann_backend, ann_params=ann_params, n_jobs=n_jobs)

    def upload_to_huggingface(

        self,

        repo_id: str,

        *,

        weights_file: str = "dmap.msgpack",

        config_file: str = "dmap_config.json",

        token: Optional[str] = None,

        repo_type: str = "model",

        revision: Optional[str] = None,

    ) -> None:
        """

        Upload DMAP weights to the Hugging Face Hub.



        This follows the same pattern as `DIMA.save_hf`:

        - saves locally first (msgpack + JSON)

        - creates the repo if needed

        - uploads both files

        """
        try:
            from huggingface_hub import HfApi, HfFolder, upload_file  # type: ignore
        except Exception as e:
            raise RuntimeError("huggingface_hub not installed. Install extras: `pip install dima[hf]`.") from e

        self.save_local(weights_file=weights_file, config_file=config_file)

        if token is None:
            token = HfFolder.get_token()
        if token is None:
            raise RuntimeError("No HF token found. Run `huggingface-cli login`, or pass `token=...`.")

        import os as _os

        api = HfApi()
        api.create_repo(repo_id=repo_id, repo_type=repo_type, exist_ok=True, token=token)

        wf = _os.path.basename(weights_file)
        cf = _os.path.basename(config_file)

        upload_file(
            path_or_fileobj=weights_file,
            path_in_repo=wf,
            repo_id=repo_id,
            token=token,
            repo_type=repo_type,
            revision=revision,
        )
        upload_file(
            path_or_fileobj=config_file,
            path_in_repo=cf,
            repo_id=repo_id,
            token=token,
            repo_type=repo_type,
            revision=revision,
        )

        return None

    @classmethod
    def download_from_huggingface(

        cls,

        repo_id: str,

        *,

        weights_file: str = "dmap.msgpack",

        ann_backend: ANNBackend = "auto",

        ann_params: Optional[Dict[str, Any]] = None,

        n_jobs: int = -1,

        token: Optional[str] = None,

        repo_type: str = "model",

        revision: Optional[str] = None,

    ) -> "DMAP":
        """

        Download DMAP weights from Hugging Face Hub and rehydrate a DMAP instance.



        Note: despite the name, this *downloads from* the Hub.

        """
        try:
            from huggingface_hub import hf_hub_download  # type: ignore
        except Exception as e:
            raise RuntimeError("huggingface_hub not installed. Install extras: `pip install dima[hf]`.") from e

        path = hf_hub_download(
            repo_id=repo_id,
            filename=weights_file,
            token=token,
            repo_type=repo_type,
            revision=revision,
        )
        return cls.load_local(path, ann_backend=ann_backend, ann_params=ann_params, n_jobs=n_jobs)



__all__ = ["DMAP", "k_ideal"]