# how to use # ddpm = DDPM(Z_train, T=100, n_iter=20_000, key=random.PRNGKey(0)) # # Fast unconditional samples (try 15–30 first) # z = ddpm.sample_dpmpp(N=4096, num_steps=20) # # Fast “refine latents” starting from an intermediate noise level #z_ref = ddpm.refine_latents_dpmpp(z0, t_start=30, num_steps=20, add_noise=True) # src/dima/ddpmx.py from __future__ import annotations from typing import Any, Optional, Dict import os import json import numpy as np 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, serialization as flax_ser import optax # --------------------------------------------------------------------- # Helpers (as in ddpmx.py) # --------------------------------------------------------------------- def _sigma_to_alpha_sigma_t(sigma: jnp.ndarray) -> tuple[jnp.ndarray, jnp.ndarray]: """ EDM-style sigma parameterization: alpha_t = 1 / sqrt(1 + sigma^2) sigma_t = sigma * alpha_t so that x = alpha_t * x0 + sigma_t * eps """ alpha_t = 1.0 / jnp.sqrt(1.0 + sigma**2) sigma_t = sigma * alpha_t return alpha_t, sigma_t def _make_lu_sigma_schedule(sigma_start: float, sigma_end: float, num_steps: int) -> np.ndarray: """ "Lu" schedule uniform in lambda = -log(sigma). """ sigma_start = float(max(sigma_start, 1e-12)) sigma_end = float(max(sigma_end, 1e-12)) lam_start = -np.log(sigma_start) lam_end = -np.log(sigma_end) lambdas = np.linspace(lam_start, lam_end, int(num_steps), dtype=np.float32) sigmas = np.exp(-lambdas).astype(np.float32) return sigmas def cosine_schedule(T: int, s: float = 0.008): """ Nichol & Dhariwal cosine schedule. Returns alpha, beta, alpha_bar with shape (T,). """ steps = jnp.arange(T + 1, dtype=jnp.float32) f = jnp.cos(((steps / T + s) / (1.0 + s)) * jnp.pi / 2.0) ** 2 alpha_bar_all = f / f[0] alpha_bar = alpha_bar_all[1:] # (T,) alpha = alpha_bar / jnp.concatenate([jnp.array([1.0], dtype=jnp.float32), alpha_bar[:-1]]) beta = 1.0 - alpha return alpha, beta, alpha_bar def sinusoidal_embedding(t_idx: jnp.ndarray, dim: int) -> jnp.ndarray: """ t_idx: (B,1) int32 or float32 returns: (B,dim) """ if t_idx.ndim != 2 or t_idx.shape[1] != 1: raise ValueError("t_idx must have shape (B,1)") t = t_idx.astype(jnp.float32) half = dim // 2 denom = float(max(half - 1, 1)) freqs = jnp.exp(-jnp.log(10_000.0) * jnp.arange(half, dtype=jnp.float32) / denom) args = t * freqs emb = jnp.concatenate([jnp.sin(args), jnp.cos(args)], axis=-1) if dim % 2 == 1: emb = jnp.pad(emb, ((0, 0), (0, 1))) return emb class EpsMLP(nn.Module): """Simple MLP epsilon-predictor for DDPM in R^D.""" hidden: int t_dim: int data_dim: int @nn.compact def __call__(self, x: jnp.ndarray, t_idx: jnp.ndarray) -> jnp.ndarray: t_emb = sinusoidal_embedding(t_idx, self.t_dim) t_h = nn.Dense(self.hidden)(t_emb) t_h = nn.gelu(t_h) h = nn.Dense(self.hidden)(x) h = nn.gelu(h + t_h) t_h2 = nn.Dense(self.hidden)(t_h) h = nn.Dense(self.hidden)(h) h = nn.gelu(h + t_h2) out = nn.Dense(self.data_dim)(h) return out @struct.dataclass class TrainStateEMA(train_state.TrainState): """Flax TrainState extended with EMA params.""" ema_params: Any = struct.field(pytree_node=True) def apply_gradients(self, *, grads, ema_decay: float): updates, new_opt_state = self.tx.update(grads, self.opt_state, self.params) new_params = optax.apply_updates(self.params, updates) new_ema = optax.incremental_update(new_params, self.ema_params, step_size=1.0 - ema_decay) return self.replace( step=self.step + 1, params=new_params, opt_state=new_opt_state, ema_params=new_ema, ) # --------------------------------------------------------------------- # DDPMX with HF upload/download integrated # --------------------------------------------------------------------- class DDPM: """ DDPM (+ fast DPM-Solver++(2M) sampler utilities) for D-dimensional latents. Added persistence utilities: - save_local / load_local - upload_to_huggingface / download_from_huggingface Serialization is done via flax.serialization.to_state_dict / from_state_dict to avoid msgpack failures with non-serializable Python objects (e.g., tuples). """ def __init__( self, Z_iX: jnp.ndarray, *, T: int = 100, hidden_dim: int = 128, t_embed_dim: int = 64, learning_rate: float = 1e-3, n_iter: int = 20_000, ema_decay: float = 0.999, beta_max: float = 0.02, batch_size: Optional[int] = None, key: jax.Array = random.PRNGKey(0), verbose_every: int = 0, eps: float = 1e-5, ): Z_iX = jnp.asarray(Z_iX, dtype=jnp.float32) if Z_iX.ndim != 2: raise ValueError("Z_iX must be 2D (N,D).") self.D = int(Z_iX.shape[1]) self.T = int(T) # store config for checkpointing self.hidden_dim = int(hidden_dim) self.t_embed_dim = int(t_embed_dim) self.learning_rate = float(learning_rate) self.ema_decay = float(ema_decay) self.beta_max = float(beta_max) self.batch_size = batch_size self.verbose_every = int(verbose_every) self.eps = float(eps) self.key = key # VP schedule (cosine + clip) alpha, beta, alpha_bar = cosine_schedule(self.T) beta = jnp.minimum(beta, self.beta_max) alpha = 1.0 - beta alpha_bar = jnp.cumprod(alpha) self.alpha_s = alpha.astype(jnp.float32) self.beta_s = beta.astype(jnp.float32) self.alpha_bar_s = alpha_bar.astype(jnp.float32) # Precompute EDM-style "sigma_in" for DPM++ schedule interpolation # sigma_in = sqrt((1 - a_bar) / a_bar) self.sigma_in_train = jnp.sqrt( jnp.clip( (1.0 - self.alpha_bar_s) / jnp.clip(self.alpha_bar_s, self.eps, 1.0), self.eps, 1e12, ) ).astype(jnp.float32) # model + optimizer + EMA state self.model = EpsMLP(hidden=self.hidden_dim, t_dim=self.t_embed_dim, data_dim=self.D) params = self.model.init( self.key, jnp.zeros((1, self.D), dtype=jnp.float32), jnp.zeros((1, 1), dtype=jnp.int32), )["params"] tx = optax.adam(self.learning_rate) self.state = TrainStateEMA.create( apply_fn=self.model.apply, params=params, tx=tx, ema_params=params, ) if int(n_iter) > 0: self._train(Z_iX, int(n_iter)) # ------------------------- # Training # ------------------------- @staticmethod def _loss(params, apply_fn, x_t, t_idx, eps_true): eps_pred = apply_fn({"params": params}, x_t, t_idx) return jnp.mean((eps_pred - eps_true) ** 2) @staticmethod @jax.jit def _train_step( state: TrainStateEMA, x0_batch: jnp.ndarray, key: jax.Array, alpha_bar_s: jnp.ndarray, ema_decay: float, eps: float, ): B = x0_batch.shape[0] key, k_eps, k_t = random.split(key, 3) eps_noise = random.normal(k_eps, shape=x0_batch.shape) t_idx = random.randint(k_t, shape=(B, 1), minval=0, maxval=alpha_bar_s.shape[0]) a_bar_t = jnp.take(alpha_bar_s, t_idx.squeeze(-1))[:, None] a_bar_t = jnp.clip(a_bar_t, eps, 1.0) x_t = jnp.sqrt(a_bar_t) * x0_batch + jnp.sqrt(1.0 - a_bar_t) * eps_noise def loss_fn(p): return DDPM._loss(p, state.apply_fn, x_t, t_idx, eps_noise) loss, grads = jax.value_and_grad(loss_fn)(state.params) new_state = state.apply_gradients(grads=grads, ema_decay=ema_decay) return new_state, loss, key def _train(self, Z_iX: jnp.ndarray, n_iter: int): N = int(Z_iX.shape[0]) bs = N if (self.batch_size is None) else min(int(self.batch_size), N) for it in range(n_iter): if bs >= N: batch = Z_iX else: self.key, k_perm = random.split(self.key) idx = random.permutation(k_perm, N)[:bs] batch = Z_iX[idx] self.state, loss, self.key = self._train_step( self.state, batch, self.key, self.alpha_bar_s, self.ema_decay, self.eps, ) if self.verbose_every and (it % self.verbose_every == 0 or it == n_iter - 1): print(f"iter {it:6d} loss {float(loss):.6f}", end="\r") if self.verbose_every: print("\ntraining complete.") # ------------------------- # Standard DDPM refine/sample # ------------------------- @staticmethod def _posterior_variance(alpha_s, beta_s, alpha_bar_s, t): a_bar_t = alpha_bar_s[t] a_bar_prev = jnp.where(t > 0, alpha_bar_s[t - 1], jnp.array(1.0, dtype=alpha_bar_s.dtype)) return ((1.0 - a_bar_prev) / (1.0 - a_bar_t)) * beta_s[t] @staticmethod def _make_sampler_step(params_ema, apply_fn, alpha_s, beta_s, alpha_bar_s, eps: float): @jax.jit def step(carry, _): key, t, x = carry key, k = random.split(key) alpha_t = jnp.clip(alpha_s[t], eps, 1.0) a_bar_t = jnp.clip(alpha_bar_s[t], eps, 1.0) sqrt_alpha = jnp.sqrt(alpha_t) sqrt_one_minus = jnp.sqrt(jnp.clip(1.0 - a_bar_t, 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) x0_hat = (x - sqrt_one_minus * eps_pred) / jnp.sqrt(a_bar_t) a_bar_prev = jnp.where(t > 0, alpha_bar_s[t - 1], jnp.array(1.0, dtype=alpha_bar_s.dtype)) denom = jnp.clip(1.0 - a_bar_t, eps, 1.0) coef1 = jnp.sqrt(jnp.clip(a_bar_prev, eps, 1.0)) * beta_s[t] / denom coef2 = sqrt_alpha * (1.0 - a_bar_prev) / denom mean = coef1 * x0_hat + coef2 * x beta_tilde = DDPM._posterior_variance(alpha_s, beta_s, alpha_bar_s, t) sigma = jnp.sqrt(jnp.clip(beta_tilde, 0.0, 1.0)) z = random.normal(k, x.shape) z = jnp.where(t == 0, 0.0, z) x_prev = mean + sigma * z return (key, t - 1, x_prev), x_prev return step def refine_latents( self, z0: jnp.ndarray, t_start: int = 10, key: Optional[jax.Array] = None, add_noise: bool = True, ) -> jnp.ndarray: 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) if key is None: self.key, key = random.split(self.key) else: self.key, _ = random.split(key) 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) if add_noise: z_t = jnp.sqrt(a_bar_t) * z0 + jnp.sqrt(1.0 - a_bar_t) * eps_noise else: z_t = z0 step = self._make_sampler_step( self.state.ema_params, self.state.apply_fn, self.alpha_s, self.beta_s, self.alpha_bar_s, self.eps, ) (final_key, _, _), trace = jax.lax.scan( step, (key, t_start, z_t), xs=None, length=t_start + 1, ) 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, ) -> jnp.ndarray: return self.refine_latents(z0, t_start=t_start, key=key, add_noise=add_noise) def reverse_from_T(self, x_T: jnp.ndarray) -> 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}).") step = self._make_sampler_step( self.state.ema_params, self.state.apply_fn, self.alpha_s, self.beta_s, self.alpha_bar_s, self.eps, ) self.key, k0 = random.split(self.key) (_, _, _), trace = jax.lax.scan( step, (k0, self.T - 1, x_T), xs=None, length=self.T, ) return trace[-1] def sample(self, N: int = 10_000) -> jnp.ndarray: self.key, k = random.split(self.key) noise = random.normal(k, (int(N), self.D)).astype(jnp.float32) return self.reverse_from_T(noise) # ------------------------- # DPM-Solver++(2M) schedule + sampler # ------------------------- def _make_dpmpp_schedule(self, *, num_steps: int, t_start: int) -> tuple[jnp.ndarray, jnp.ndarray]: """ Returns: sigmas_in: (K+1,) float32 decreasing, last one is 0 t_cont: (K,) float32 continuous "time" indices for model calls """ t_start = int(t_start) if not (0 <= t_start < self.T): raise ValueError(f"t_start must be in [0, {self.T-1}]") if int(num_steps) < 1: raise ValueError("num_steps must be >= 1") sigma_start = float(self.sigma_in_train[t_start]) sigma_end = float(self.sigma_in_train[0]) sigmas_k = _make_lu_sigma_schedule(sigma_start, sigma_end, int(num_steps)) sigmas = np.concatenate([sigmas_k, np.array([0.0], np.float32)], axis=0) sigma_train = np.array(self.sigma_in_train).astype(np.float32) # (T,) log_sig_train = np.log(np.maximum(sigma_train, 1e-12)) t_train = np.arange(self.T, dtype=np.float32) log_sig = np.log(np.maximum(sigmas[:-1], 1e-12)) t_cont = np.interp(log_sig, log_sig_train, t_train).astype(np.float32) return jnp.array(sigmas, dtype=jnp.float32), jnp.array(t_cont, dtype=jnp.float32) @staticmethod @jax.jit def _dpmpp_2m_midpoint_sample( params_ema: Any, apply_fn: Any, x_start: jnp.ndarray, # (B,D) sigmas_in: jnp.ndarray, # (K+1,) t_cont: jnp.ndarray, # (K,) eps: float, ) -> jnp.ndarray: """ DPM-Solver++ (2M, midpoint) sampler. """ sigma_s = sigmas_in[:-1] # (K,) sigma_t = sigmas_in[1:] # (K,) alpha_s, sigma_s_t = _sigma_to_alpha_sigma_t(sigma_s) alpha_t, sigma_t_t = _sigma_to_alpha_sigma_t(sigma_t) lambda_s = jnp.log(alpha_s) - jnp.log(sigma_s_t) lambda_t = jnp.log(alpha_t) - jnp.log(sigma_t_t) K = t_cont.shape[0] is_first = jnp.arange(K) == 0 is_last = jnp.arange(K) == (K - 1) def step(carry, inp): x, m_prev, lam_prev = carry (a_s, s_s, a_t, s_t, lam_s_i, lam_t_i, t_i, first_i, last_i) = inp B = x.shape[0] t_batch = jnp.full((B, 1), t_i, dtype=jnp.float32) eps_pred = apply_fn({"params": params_ema}, x, t_batch) a_s_b = jnp.clip(a_s, eps, 1.0) x0 = (x - s_s * eps_pred) / a_s_b h = lam_t_i - lam_s_i exp_neg_h = jnp.exp(-h) # 1st-order x_first = (s_t / s_s) * x - (a_t * (exp_neg_h - 1.0)) * x0 def do_second(_): h0 = lam_s_i - lam_prev r0 = h0 / jnp.clip(h, 1e-12) D1 = (x0 - m_prev) / jnp.clip(r0, 1e-12) x_second = (s_t / s_s) * x - (a_t * (exp_neg_h - 1.0)) * (x0 + 0.5 * D1) return x_second x_next = jax.lax.cond(first_i | last_i, lambda _: x_first, do_second, operand=None) return (x_next, x0, lam_s_i), x_next xs = ( alpha_s, sigma_s_t, alpha_t, sigma_t_t, lambda_s, lambda_t, t_cont, is_first, is_last ) x0_init = jnp.zeros_like(x_start) lam_init = jnp.array(0.0, dtype=jnp.float32) (x_final, _, _), _ = jax.lax.scan(step, (x_start, x0_init, lam_init), xs) return x_final def refine_latents_dpmpp( self, z0: jnp.ndarray, *, t_start: int = 10, num_steps: int = 20, key: Optional[jax.Array] = None, add_noise: bool = True, ) -> jnp.ndarray: 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}]") if key is None: self.key, key = random.split(self.key) else: self.key, _ = random.split(key) # forward-noise to t_start key, k_eps = random.split(key) eps_noise = random.normal(k_eps, z0.shape) a_bar = jnp.clip(self.alpha_bar_s[int(t_start)], self.eps, 1.0) if add_noise: x_start = jnp.sqrt(a_bar) * z0 + jnp.sqrt(1.0 - a_bar) * eps_noise else: x_start = z0 sigmas_in, t_cont = self._make_dpmpp_schedule(num_steps=int(num_steps), t_start=int(t_start)) x_final = self._dpmpp_2m_midpoint_sample( self.state.ema_params, self.state.apply_fn, x_start, sigmas_in, t_cont, self.eps, ) return x_final def reverse_from_T_dpmpp(self, x_T: jnp.ndarray, *, num_steps: int = 20) -> 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}).") sigmas_in, t_cont = self._make_dpmpp_schedule(num_steps=int(num_steps), t_start=self.T - 1) return self._dpmpp_2m_midpoint_sample( self.state.ema_params, self.state.apply_fn, x_T, sigmas_in, t_cont, self.eps, ) def sample_dpmpp(self, N: int = 10_000, *, num_steps: int = 20) -> jnp.ndarray: self.key, k = random.split(self.key) x_T = random.normal(k, (int(N), self.D)).astype(jnp.float32) return self.reverse_from_T_dpmpp(x_T, num_steps=int(num_steps)) # ----------------------------------------------------------------- # Persistence: state_dict / save_local / load_local # ----------------------------------------------------------------- def _config_dict(self) -> Dict[str, Any]: return { "class_name": "DDPMX", "T": int(self.T), "D": int(self.D), "hidden_dim": int(self.hidden_dim), "t_embed_dim": int(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), "key": np.array(self.key).tolist(), } def save_local(self, weights_file: str = "ddpmx_weights.msgpack", config_file: str = "ddpmx_config.json") -> None: """ Saves: - config_file: JSON with hyperparams + PRNG key - weights_file: msgpack with flax state_dict of TrainStateEMA """ cfg = self._config_dict() with open(config_file, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2, ensure_ascii=False) # Robust serialization (avoid msgpack tuple errors) state_sd = flax_ser.to_state_dict(self.state) blob = flax_ser.msgpack_serialize(state_sd) with open(weights_file, "wb") as f: f.write(blob) @classmethod def load_local( cls, weights_file: str, config_file: str, *, Z_iX: Optional[jnp.ndarray] = None, ) -> "DDPM": """ Reconstructs a DDPMX instance from local files. Z_iX is only used to provide shape (N,D) for initialization; training is skipped. If Z_iX is None, a dummy array of shape (1,D) is created. """ with open(config_file, "r", encoding="utf-8") as f: cfg = json.load(f) D = int(cfg["D"]) if Z_iX is None: Z_iX = jnp.zeros((1, D), dtype=jnp.float32) # Build a fresh instance with the same architecture, skip training obj = cls( Z_iX, T=int(cfg["T"]), hidden_dim=int(cfg["hidden_dim"]), t_embed_dim=int(cfg["t_embed_dim"]), learning_rate=float(cfg["learning_rate"]), n_iter=0, ema_decay=float(cfg["ema_decay"]), beta_max=float(cfg["beta_max"]), batch_size=cfg["batch_size"], key=random.PRNGKey(0), verbose_every=0, eps=float(cfg["eps"]), ) with open(weights_file, "rb") as f: state_sd = flax_ser.msgpack_restore(f.read()) obj.state = flax_ser.from_state_dict(obj.state, state_sd) key_list = cfg.get("key", None) if key_list is not None: obj.key = jnp.array(key_list, dtype=jnp.uint32) return obj # ----------------------------------------------------------------- # Hugging Face Hub: upload / download # ----------------------------------------------------------------- def upload_to_huggingface( self, repo_id: str, *, token: Optional[str] = None, weights_file: str = "ddpmx_weights.msgpack", config_file: str = "ddpmx_config.json", repo_type: str = "model", revision: Optional[str] = None, ) -> Dict[str, str]: """ Saves locally and uploads (weights_file, config_file) to Hugging Face Hub. """ try: from huggingface_hub import create_repo, upload_file except Exception as e: raise RuntimeError( "huggingface_hub not installed. Install it (e.g., `pip install huggingface_hub`)." ) from e self.save_local(weights_file=weights_file, config_file=config_file) create_repo(repo_id, token=token, repo_type=repo_type, exist_ok=True) w_name = os.path.basename(weights_file) c_name = os.path.basename(config_file) upload_file( path_or_fileobj=weights_file, path_in_repo=w_name, repo_id=repo_id, repo_type=repo_type, token=token, revision=revision, ) upload_file( path_or_fileobj=config_file, path_in_repo=c_name, repo_id=repo_id, repo_type=repo_type, token=token, revision=revision, ) return {"repo_id": repo_id, "weights": w_name, "config": c_name} @classmethod def download_from_huggingface( cls, repo_id: str, *, token: Optional[str] = None, weights_file: str = "ddpmx_weights.msgpack", config_file: str = "ddpmx_config.json", repo_type: str = "model", revision: Optional[str] = None, cache_dir: Optional[str] = None, Z_iX: Optional[jnp.ndarray] = None, ) -> "DDPM": """ Downloads (weights_file, config_file) from Hugging Face Hub and reconstructs the class. """ try: from huggingface_hub import hf_hub_download except Exception as e: raise RuntimeError( "huggingface_hub not installed. Install it (e.g., `pip install huggingface_hub`)." ) from e w_name = os.path.basename(weights_file) c_name = os.path.basename(config_file) w_path = hf_hub_download( repo_id=repo_id, filename=w_name, repo_type=repo_type, token=token, revision=revision, cache_dir=cache_dir, ) c_path = hf_hub_download( repo_id=repo_id, filename=c_name, repo_type=repo_type, token=token, revision=revision, cache_dir=cache_dir, ) return cls.load_local(w_path, c_path, Z_iX=Z_iX) __all__ = ["DDPM", "EpsMLP", "cosine_schedule", "sinusoidal_embedding"]