# --- src/dima/ddpm.py # +++ src/dima/ddpm.py # @@ -368,4 +368,204 @@ # return self.reverse_from_T(noise) # src/dima/ddpm.py from __future__ import annotations from typing import Any, Optional import jax import jax.numpy as jnp from jax import random from flax import linen as nn from flax.training import train_state from flax import struct import optax from .ann import ANNBackend, make_ann from .ddpm import DDPM # +from .ddpm import DDPM, DDIM from .dmap import DMAP from .gplm import GPLM __all__ = ["DDPM", "EpsMLP", "cosine_schedule", "sinusoidal_embedding"] # ------------------------------ # DDIM (deterministic sampler; eta=0.0) # ------------------------------ class DDIM(DDPM): """ DDIM sampler for D-dimensional latents. Notes ----- * Training is identical to DDPM (epsilon prediction). * Sampling / refinement uses the DDIM update: x_{t_prev} = sqrt(a_bar_prev) * x0_hat + sqrt(1 - a_bar_prev - sigma^2) * eps_pred + sigma * z, where sigma = eta * sqrt( (1-a_bar_prev)/(1-a_bar_t) * (1 - a_bar_t/a_bar_prev) ). With eta=0.0 the reverse process is deterministic conditioned on x_t. Added (HF I/O): - state_dict / save_local / load_local / from_state - upload_to_huggingface / download_from_huggingface """ def __init__( self, Z_iX: jnp.ndarray, *, eta: float = 0.0, steps: Optional[int] = None, **ddpm_kwargs: Any, ): super().__init__(Z_iX, **ddpm_kwargs) self.eta = float(eta) self.steps = None if steps is None else int(steps) @staticmethod def _make_ddim_step(params_ema, apply_fn, alpha_bar_s, eps: float, eta: float): eta = float(eta) @jax.jit def step(carry, t_pair): key, x = carry # x: (B, D) t, t_prev = t_pair # scalar int32 # gather ā_t and ā_prev (supports skipping) a_bar_t = jnp.clip(alpha_bar_s[t], eps, 1.0) a_bar_prev = jnp.clip(alpha_bar_s[t_prev], eps, 1.0) B = x.shape[0] t_batch = jnp.full((B, 1), t, dtype=jnp.int32) eps_pred = apply_fn({"params": params_ema}, x, t_batch) # (B, D) # x0 estimate sqrt_one_minus = jnp.sqrt(jnp.clip(1.0 - a_bar_t, eps, 1.0)) x0_hat = (x - sqrt_one_minus * eps_pred) / jnp.sqrt(a_bar_t) # sigma_t (controls stochasticity). eta=0 => sigma=0 (deterministic) frac = (1.0 - a_bar_prev) / jnp.clip(1.0 - a_bar_t, eps, 1.0) inside = 1.0 - a_bar_t / jnp.clip(a_bar_prev, eps, 1.0) sigma = eta * jnp.sqrt(jnp.clip(frac * inside, 0.0, 1.0)) # direction term dir_coeff = jnp.sqrt(jnp.clip(1.0 - a_bar_prev - sigma**2, 0.0, 1.0)) x_prev = jnp.sqrt(a_bar_prev) * x0_hat + dir_coeff * eps_pred # optional noise injection (eta>0) if eta > 0.0: key, k = random.split(key) z = random.normal(k, x.shape) x_prev = x_prev + sigma * z return (key, x_prev), x_prev return step def _t_schedule(self, t_start: int, *, steps: Optional[int] = None) -> jnp.ndarray: """ Descending integer timesteps from t_start -> 0. If steps is provided (or self.steps), we use a reduced DDIM schedule by rounding a linear grid and making it unique. """ t_start = int(t_start) if t_start <= 0: return jnp.array([0], dtype=jnp.int32) S = self.steps if steps is None else int(steps) if S is None or S >= (t_start + 1): return jnp.arange(t_start, -1, -1, dtype=jnp.int32) grid = jnp.linspace(float(t_start), 0.0, int(S), dtype=jnp.float32) ts = jnp.unique(jnp.round(grid).astype(jnp.int32)) ts = ts[::-1] # descending # guarantee endpoints if ts[0] != t_start: ts = jnp.concatenate([jnp.array([t_start], dtype=jnp.int32), ts]) if ts[-1] != 0: ts = jnp.concatenate([ts, jnp.array([0], dtype=jnp.int32)]) return ts def refine_latents( self, z0: jnp.ndarray, t_start: int = 10, key: Optional[jax.Array] = None, add_noise: bool = True, *, eta: Optional[float] = None, steps: Optional[int] = None, ) -> jnp.ndarray: """ Refine latents by: (optional) forward-noise z0 to step t_start DDIM reverse from t_start -> 0 using EMA params. """ z0 = jnp.asarray(z0, dtype=jnp.float32) if z0.ndim != 2 or z0.shape[1] != self.D: raise ValueError(f"z0 must have shape (B,{self.D}).") if not (0 <= int(t_start) < self.T): raise ValueError(f"t_start must be in [0, {self.T-1}]") t_start = int(t_start) # RNG if key is None: self.key, key = random.split(self.key) else: # advance internal RNG too self.key, _ = random.split(key) # optional forward noise if add_noise and t_start > 0: key, k_eps = random.split(key) eps_noise = random.normal(k_eps, z0.shape) a_bar_t = jnp.clip(self.alpha_bar_s[t_start], self.eps, 1.0) # Corrected attribute name z_t = jnp.sqrt(a_bar_t) * z0 + jnp.sqrt(1.0 - a_bar_t) * eps_noise else: z_t = z0 # schedule ts = self._t_schedule(t_start, steps=steps) if ts.shape[0] == 1: return z_t t_pairs = jnp.stack([ts[:-1], ts[1:]], axis=1) # (K,2) step = self._make_ddim_step( self.state.ema_params, self.state.apply_fn, self.alpha_bar_s, # Corrected attribute name self.eps, self.eta if eta is None else float(eta), ) (final_key, _), trace = jax.lax.scan(step, (key, z_t), xs=t_pairs) self.key = final_key return trace[-1] def __call__( self, z0: jnp.ndarray, t_start: int = 10, key: Optional[jax.Array] = None, add_noise: bool = True, *, eta: Optional[float] = None, steps: Optional[int] = None, ) -> jnp.ndarray: return self.refine_latents( z0, t_start=t_start, key=key, add_noise=add_noise, eta=eta, steps=steps ) def reverse_from_T(self, x_T: jnp.ndarray, *, eta: Optional[float] = None, steps: Optional[int] = None) -> jnp.ndarray: x_T = jnp.asarray(x_T, dtype=jnp.float32) if x_T.ndim != 2 or x_T.shape[1] != self.D: raise ValueError(f"x_T must have shape (B,{self.D}).") self.key, k0 = random.split(self.key) ts = self._t_schedule(self.T - 1, steps=steps) t_pairs = jnp.stack([ts[:-1], ts[1:]], axis=1) step = self._make_ddim_step( self.state.ema_params, self.state.apply_fn, self.alpha_bar_s, # Corrected attribute name self.eps, self.eta if eta is None else float(eta), ) (_, _), trace = jax.lax.scan(step, (k0, x_T), xs=t_pairs) return trace[-1] def sample(self, N: int = 10_000, *, eta: Optional[float] = None, steps: Optional[int] = None) -> jnp.ndarray: self.key, k = random.split(self.key) noise = random.normal(k, (int(N), self.D)) return self.reverse_from_T(noise, eta=eta, steps=steps) # ------------------------------------------------------------------ # Serialization + Hugging Face Hub I/O (integrated) # ------------------------------------------------------------------ def state_dict(self): """ Msgpack-safe checkpoint dict (includes DDIM-specific eta/steps). Uses flax.serialization.to_state_dict for TrainStateEMA to avoid tuple issues (opt_state). """ from flax import serialization as flax_ser # type: ignore import numpy as np return dict( # DDPM core hyperparameters T=int(self.T), D=int(self.D), hidden_dim=int(getattr(self, "hidden_dim", 0) or self.hidden_dim), t_embed_dim=int(getattr(self, "t_embed_dim", 0) or self.t_embed_dim), learning_rate=float(self.learning_rate), ema_decay=float(self.ema_decay), beta_max=float(self.beta_max), batch_size=None if (self.batch_size is None) else int(self.batch_size), eps=float(self.eps), # DDIM specifics eta=float(self.eta), steps=None if (self.steps is None) else int(self.steps), # RNG key (portable) key=np.asarray(self.key), # TrainStateEMA (params, ema_params, opt_state, step) train_state=flax_ser.to_state_dict(self.state), ) def save_local(self, weights_file: str = "ddim.msgpack", config_file: Optional[str] = "ddim_config.json") -> None: from flax import serialization as flax_ser # type: ignore import json as _json ckpt = self.state_dict() blob = flax_ser.msgpack_serialize(ckpt) with open(weights_file, "wb") as f: f.write(blob) if config_file is not None: cfg = dict( T=int(ckpt["T"]), D=int(ckpt["D"]), hidden_dim=int(ckpt["hidden_dim"]), t_embed_dim=int(ckpt["t_embed_dim"]), learning_rate=float(ckpt["learning_rate"]), ema_decay=float(ckpt["ema_decay"]), beta_max=float(ckpt["beta_max"]), batch_size=ckpt["batch_size"], eps=float(ckpt["eps"]), eta=float(ckpt["eta"]), steps=ckpt["steps"], ) with open(config_file, "w") as f: _json.dump(cfg, f, indent=2) @classmethod def from_state(cls, ckpt, *, key: Optional[jax.Array] = None) -> "DDIM": """ Rehydrate a DDIM instance without retraining: - rebuilds a skeleton DDIM (which builds schedules/model/optimizer) - restores TrainStateEMA via flax.serialization.from_state_dict """ from flax import serialization as flax_ser # type: ignore import numpy as np T = int(ckpt["T"]) D = int(ckpt["D"]) hidden_dim = int(ckpt["hidden_dim"]) t_embed_dim = int(ckpt["t_embed_dim"]) learning_rate = float(ckpt["learning_rate"]) ema_decay = float(ckpt["ema_decay"]) beta_max = float(ckpt["beta_max"]) batch_size = ckpt.get("batch_size", None) eps = float(ckpt.get("eps", 1e-5)) eta = float(ckpt.get("eta", 0.0)) steps = ckpt.get("steps", None) steps = None if (steps is None) else int(steps) if key is None: if "key" in ckpt: key = jnp.asarray(np.asarray(ckpt["key"])) else: key = random.PRNGKey(0) # skeleton (no training) dummy = jnp.zeros((1, D), dtype=jnp.float32) obj = cls( dummy, T=T, hidden_dim=hidden_dim, t_embed_dim=t_embed_dim, learning_rate=learning_rate, n_iter=0, ema_decay=ema_decay, beta_max=beta_max, batch_size=batch_size, key=key, verbose_every=0, eps=eps, eta=eta, steps=steps, ) obj.state = flax_ser.from_state_dict(obj.state, ckpt["train_state"]) # restore RNG key and DDIM attrs if "key" in ckpt: obj.key = jnp.asarray(np.asarray(ckpt["key"])) obj.eta = eta obj.steps = steps return obj @classmethod def load_local(cls, weights_file: str = "ddim.msgpack", *, key: Optional[jax.Array] = None) -> "DDIM": from flax import serialization as flax_ser # type: ignore with open(weights_file, "rb") as f: ckpt = flax_ser.msgpack_restore(f.read()) return cls.from_state(ckpt, key=key) def upload_to_huggingface( self, repo_id: str, *, weights_file: str = "ddim.msgpack", config_file: str = "ddim_config.json", token: Optional[str] = None, repo_type: str = "model", revision: Optional[str] = None, ) -> None: """ Upload DDIM checkpoint to HF Hub (msgpack + JSON). Mirrors the DMAP/DDPM pattern. """ try: from huggingface_hub import HfApi, HfFolder, upload_file # type: ignore except Exception as e: raise RuntimeError("huggingface_hub not installed. Install it (or `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, ) @classmethod def download_from_huggingface( cls, repo_id: str, *, weights_file: str = "ddim.msgpack", token: Optional[str] = None, repo_type: str = "model", revision: Optional[str] = None, key: Optional[jax.Array] = None, ) -> "DDIM": """ Download DDIM checkpoint from HF Hub and return a rehydrated DDIM instance. """ try: from huggingface_hub import hf_hub_download # type: ignore except Exception as e: raise RuntimeError("huggingface_hub not installed. Install it (or `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, key=key) # optional alias for naming symmetry download_to_huggingface = download_from_huggingface __all__ = ["DDPM", "DDIM", "EpsMLP", "cosine_schedule", "sinusoidal_embedding"] # --- src/dima/dima.py # +++ src/dima/dima.py # @@ -15,7 +15,7 @@ # from flax import serialization as flax_ser # @@ -374,7 +374,13 @@ # ddpm_init["eps"] = float(ddpm_eps) # with jax.default_device(self.ddpm_device): # - self.dm = DDPM(Z_ix_j, **ddpm_init) # + sampler = str(ddpm_init.pop("sampler", "ddpm")).lower() # + if sampler == "ddim": # + eta = float(ddpm_init.pop("eta", 0.0)) # + steps = ddpm_init.pop("steps", None) # + self.dm = DDIM(Z_ix_j, eta=eta, steps=steps, **ddpm_init) # + else: # + self.dm = DDPM(Z_ix_j, **ddpm_init) # self.training_time = time.time() - t0