Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| # app.py — SplatWorld on Hugging Face Spaces (Gradio, runs in a browser). | |
| # | |
| # The desktop explorer (explore_desktop.py) uses a live OpenCV window with | |
| # mouse-drag surfing. Spaces have no display, so this version puts the same | |
| # world behind sliders and buttons: | |
| # | |
| # Surf : radius = dive (core<->fire), morph sliders = drag, buttons = new face | |
| # Zoom : scrub one identity's dive, or render the full Shepard-zoom video | |
| # Gallery : sample a grid of faces | |
| # About : what this is and how it works | |
| # | |
| # Put splat_decoder.onnx (7.2 MB) next to this file. Without it the app still | |
| # boots and shows mock noise, so the Space never hard-fails on startup. | |
| # | |
| # python app.py # launch locally at http://127.0.0.1:7860 | |
| # | |
| # Loaded via onnxruntime (CPU); falls back to opencv's dnn, then to a mock. | |
| import os, math, tempfile, threading | |
| import numpy as np | |
| import gradio as gr | |
| from PIL import Image | |
| LATENT = 128 | |
| RSEED = 7 | |
| DISPLAY = 384 # on-screen render size (model is 96px, upscaled) | |
| R_ESC = 45.0 | |
| P_OCT = 3.0 | |
| MODEL = os.environ.get("SPLAT_MODEL", "splat_decoder.onnx") | |
| def smooth(t): | |
| t = float(np.clip(t, 0.0, 1.0)); return t * t * (3 - 2 * t) | |
| def shell_name(zn): | |
| if zn < 15: return "core (a face)" | |
| if zn < 35: return "ghost" | |
| if zn < 100: return "fire (the soup)" | |
| return "void" | |
| # ------------------------------------------------------------------ decoder | |
| class MockDecoder: | |
| def __init__(self, h=48): | |
| g = np.random.default_rng(99).standard_normal((3 * h * h, LATENT)) | |
| self.W = (g / math.sqrt(LATENT)).astype(np.float32); self.h = h | |
| def __call__(self, zs): | |
| y = np.tanh(zs.astype(np.float32) @ self.W.T) * 0.5 + 0.5 | |
| return y.reshape(len(zs), 3, self.h, self.h) | |
| class Decoder: | |
| """onnxruntime -> opencv dnn (classic engine, thread-local) -> mock. | |
| OpenCV 5's new dnn engine asserts 'buf.u == m.u' when a Net built on one | |
| thread is run on another (which Gradio always does). onnxruntime avoids | |
| this entirely; if only OpenCV is available we build one Net per worker | |
| thread and prefer the classic engine.""" | |
| def __init__(self, path=MODEL): | |
| self.backend = None; self.note = ""; self.path = path | |
| self._local = threading.local() | |
| self._engine = None | |
| if os.path.exists(path): | |
| try: | |
| import onnxruntime as ort | |
| self.sess = ort.InferenceSession( | |
| path, providers=["CPUExecutionProvider"]) | |
| self.iname = self.sess.get_inputs()[0].name | |
| self.oname = self.sess.get_outputs()[0].name | |
| self.backend = "onnxruntime" | |
| except Exception as e_ort: | |
| try: | |
| import cv2 as cv | |
| self._cv = cv | |
| # prefer the classic engine; fall back to whatever exists | |
| self._engine = getattr(cv.dnn, "ENGINE_CLASSIC", None) | |
| self._net() # validate it builds | |
| self.backend = "opencv-dnn" | |
| self.note = ("onnxruntime not installed — using OpenCV dnn. " | |
| "For best results run: pip install onnxruntime") | |
| except Exception as e_cv: | |
| self.note = f"onnx load failed: onnxruntime=({e_ort}); opencv=({e_cv})" | |
| if self.backend is None: | |
| self.mock = MockDecoder(); self.backend = "mock" | |
| if not self.note: | |
| self.note = (f"'{path}' not found — showing mock noise. " | |
| f"Add the model to see faces.") | |
| def _net(self): | |
| """A cv2 dnn Net local to the calling thread (dodges the cross-thread | |
| buffer assertion in OpenCV 5's new engine).""" | |
| n = getattr(self._local, "net", None) | |
| if n is None: | |
| cv = self._cv | |
| if self._engine is not None: | |
| try: | |
| n = cv.dnn.readNetFromONNX(self.path, engine=self._engine) | |
| except TypeError: # OpenCV 4: no engine kw | |
| n = cv.dnn.readNetFromONNX(self.path) | |
| else: | |
| n = cv.dnn.readNetFromONNX(self.path) | |
| self._local.net = n | |
| return n | |
| def __call__(self, zs): | |
| zs = np.ascontiguousarray(zs.astype(np.float32)) | |
| if self.backend == "onnxruntime": | |
| return self.sess.run([self.oname], {self.iname: zs})[0] | |
| if self.backend == "opencv-dnn": | |
| net = self._net() | |
| net.setInput(zs, "z_latent") | |
| return net.forward("rendered_image").copy() | |
| return self.mock(zs) | |
| DEC = Decoder() | |
| def to_rgb(out_chw, size=DISPLAY): | |
| """(3,H,W) float [0,1] -> upscaled HxWx3 uint8 RGB via PIL bicubic.""" | |
| im = np.transpose(np.clip(out_chw, 0, 1), (1, 2, 0)) | |
| im = (im * 255).astype(np.uint8) | |
| return np.asarray(Image.fromarray(im).resize((size, size), Image.BICUBIC)) | |
| def decode_one(z, size=DISPLAY): | |
| return to_rgb(DEC(z[None])[0], size) | |
| # ------------------------------------------------------------------- charts | |
| def direction_bank(seed=RSEED): | |
| g = np.random.default_rng(seed).standard_normal((LATENT, LATENT)) | |
| q, _ = np.linalg.qr(g) | |
| return q.T.astype(np.float32) | |
| BANK = direction_bank() | |
| def tangent_axes(d, e1, e2): | |
| """Three orthonormal directions in the tangent plane of unit vector d, | |
| so the morph sliders rotate the identity instead of rescaling |z|.""" | |
| a = e1 - (e1 @ d) * d; a /= (np.linalg.norm(a) + 1e-9) | |
| b = e2 - (e2 @ d) * d - (e2 @ a) * a; b /= (np.linalg.norm(b) + 1e-9) | |
| e3 = BANK[2] | |
| c = e3 - (e3 @ d) * d - (e3 @ a) * a - (e3 @ b) * b | |
| c /= (np.linalg.norm(c) + 1e-9) | |
| return np.stack([a, b, c]) | |
| # =================================================================== SURF === | |
| def new_dir(seed): | |
| rng = np.random.default_rng(int(seed) & 0x7fffffff) | |
| d = rng.standard_normal(LATENT); d /= np.linalg.norm(d) | |
| return d.astype(np.float32) | |
| def surf_render(state, radius, mx, my, mz): | |
| d = np.asarray(state["d"], np.float32) | |
| ax = tangent_axes(d, BANK[state["e"] % LATENT], BANK[(state["e"] + 1) % LATENT]) | |
| d2 = d + mx * ax[0] + my * ax[1] + mz * ax[2] | |
| d2 /= (np.linalg.norm(d2) + 1e-9) | |
| z = (d2 * float(radius)).astype(np.float32) | |
| zn = float(np.linalg.norm(z)) | |
| return decode_one(z), f"**|z| = {zn:5.1f}** · {shell_name(zn)}" | |
| def surf_new_face(state): | |
| state = dict(state); state["d"] = new_dir(np.random.randint(1 << 30)) | |
| img, info = surf_render(state, 8.0, 0.0, 0.0, 0.0) | |
| return state, img, info, 8.0, 0.0, 0.0, 0.0 | |
| def surf_reroll(state): | |
| state = dict(state); state["e"] = (state["e"] + 3) % LATENT | |
| return state, "_morph axes re-rolled — the sliders now push new directions_" | |
| # =================================================================== ZOOM === | |
| class Journey: | |
| def __init__(self, wps): self.wp = wps | |
| def ident(self, k): return self.wp[k % len(self.wp)] | |
| def z(self, k, phi): | |
| K, N = self.ident(k), self.ident(k + 1) | |
| nK, nN = np.linalg.norm(K), np.linalg.norm(N) | |
| dK, dN = K / nK, N / nN | |
| if phi < 0.35: | |
| t = smooth(phi / 0.35); r = R_ESC + (nK - R_ESC) * t | |
| return (dK * r).astype(np.float32) | |
| elif phi < 0.65: | |
| return K.astype(np.float32) | |
| else: | |
| t = smooth((phi - 0.65) / 0.35); r = nK + (R_ESC - nK) * t | |
| om = math.acos(float(np.clip(dK @ dN, -1, 1))) | |
| d = dK if om < 1e-5 else \ | |
| (math.sin((1-t)*om)*dK + math.sin(t*om)*dN) / math.sin(om) | |
| d /= np.linalg.norm(d) | |
| return (d * r).astype(np.float32) | |
| def zoom_waypoints(seed, n=48): | |
| rng = np.random.default_rng(int(seed) & 0x7fffffff) | |
| return [rng.standard_normal(LATENT).astype(np.float32) * 0.6 for _ in range(n)] | |
| def zoom_scrub(state, phi): | |
| jour = Journey(state["wps"]) | |
| z = jour.z(0, float(phi)) | |
| zn = float(np.linalg.norm(z)) | |
| stage = ("descending into the face" if phi < 0.35 else | |
| "dwelling inside the face" if phi < 0.65 else | |
| "rising back into the fire") | |
| return decode_one(z), f"phi {float(phi):.2f} · |z| {zn:5.1f} · {shell_name(zn)} · {stage}" | |
| def zoom_new_ids(state): | |
| state = dict(state); state["wps"] = zoom_waypoints(np.random.randint(1 << 30)) | |
| img, info = zoom_scrub(state, 0.0) | |
| return state, img, info, 0.0 | |
| def zoom_video(state, cycles, progress=gr.Progress()): | |
| """Render a short Shepard zoom to mp4: fall through several identities.""" | |
| import imageio | |
| jour = Journey(state["wps"]) | |
| fps, per = 24, 48 | |
| frames = [] | |
| total = int(cycles) * per | |
| for f in progress.tqdm(range(total), desc="rendering zoom"): | |
| u = f / per; k = int(u); phi = u - k | |
| frames.append(decode_one(jour.z(k, phi), size=256)) | |
| path = os.path.join(tempfile.mkdtemp(), "splat_zoom.mp4") | |
| imageio.mimwrite(path, frames, fps=fps, codec="libx264", | |
| quality=8, macro_block_size=None) | |
| return path | |
| # ================================================================ GALLERY === | |
| def gallery(n, radius, seed, progress=gr.Progress()): | |
| rng = np.random.default_rng(int(seed) & 0x7fffffff) | |
| n = int(n) | |
| zs = (rng.standard_normal((n, LATENT)) * float(radius)).astype(np.float32) | |
| imgs = [] | |
| for i in progress.tqdm(range(n), desc="sampling faces"): | |
| imgs.append(decode_one(zs[i], size=224)) | |
| return imgs | |
| # ================================================================== THEORY == | |
| ABOUT = """ | |
| ## SplatWorld — 202,599 faces in a 7 MB wave field | |
| This decoder doesn't store pixels. It maps a 128-D point **z** to **256 Gabor | |
| wave packets** — each one a little oriented wave with a position, a size, an | |
| orientation, a frequency, and a *complex* amplitude (a cosine weight and a sine | |
| weight). The picture is the **sum of all 256 packets interfering**. A face is a | |
| phase-locked standing wave. | |
| **Fire vs face.** Near the origin (`|z| < 15`) the packets phase-lock: peaks and | |
| troughs cancel everywhere except along an eyebrow or a cheekbone. Far out | |
| (`|z| > 35`) there's no training data, so the decoder stops orchestrating — the | |
| packets decorrelate into drifting "fire". Zoom rides the radius: dive from fire | |
| into a face and back, forever (a Shepard tone for the eye). | |
| **The space between faces.** Moving a feature from A to B is *transport*. In a | |
| fixed additive basis the only way is to fade one atom out while fading another | |
| in — mid-way both exist and their phases fight, and that fight *is* the fire. | |
| This is 1990s technology: **eigenfaces** interpolated faces linearly in 1991 and | |
| produced exactly these ghosts. The loophole: a complex atom can **translate by | |
| rotating its phase** (a Fourier shift) instead of crossfading — phase-transport | |
| leaves no ghost, and that's the direction this whole line of work points at. | |
| **Honest notes.** 96×96 is a VRAM limit, not a taste; hair and fine detail | |
| struggle and samples skew toward a mean face. The "standing wave" language is a | |
| faithful description of a Gabor renderer, not a claim of new physics. Trained on | |
| CelebA (non-commercial research use — check the dataset's terms). | |
| *Do not hype. Do not lie. Just show.* | |
| """ | |
| # =================================================================== BUILD == | |
| def build(): | |
| with gr.Blocks(title="SplatWorld") as demo: | |
| gr.Markdown("# 🌊 SplatWorld\n" | |
| "**202,599 CelebA faces compressed into a 7 MB wave-interference " | |
| "field.** Dive through it below.") | |
| if DEC.backend == "mock": | |
| gr.Markdown(f"> ⚠️ {DEC.note}") | |
| else: | |
| gr.Markdown(f"<sub>model backend: <code>{DEC.backend}</code></sub>") | |
| with gr.Tabs(): | |
| # ---- SURF ---- | |
| with gr.Tab("Surf"): | |
| s_state = gr.State(value={"d": new_dir(RSEED), "e": 0}) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| s_img = gr.Image(label="face", height=DISPLAY, width=DISPLAY) | |
| s_info = gr.Markdown() | |
| with gr.Column(scale=1): | |
| s_r = gr.Slider(1, 55, value=8, step=0.5, | |
| label="radius · dive core (a face) ↔ rise into the fire") | |
| gr.Markdown("**morph** — nudge the identity along three directions:") | |
| s_x = gr.Slider(-1.5, 1.5, value=0, step=0.02, label="morph A") | |
| s_y = gr.Slider(-1.5, 1.5, value=0, step=0.02, label="morph B") | |
| s_z = gr.Slider(-1.5, 1.5, value=0, step=0.02, label="morph C") | |
| with gr.Row(): | |
| s_new = gr.Button("🎲 new face") | |
| s_roll = gr.Button("↻ re-roll morph axes") | |
| s_roll_info = gr.Markdown() | |
| inp = [s_state, s_r, s_x, s_y, s_z] | |
| for ctrl in (s_r, s_x, s_y, s_z): | |
| ctrl.change(surf_render, inp, [s_img, s_info]) | |
| s_new.click(surf_new_face, s_state, | |
| [s_state, s_img, s_info, s_r, s_x, s_y, s_z]) | |
| s_roll.click(surf_reroll, s_state, [s_state, s_roll_info]) | |
| demo.load(surf_render, inp, [s_img, s_info]) | |
| # ---- ZOOM ---- | |
| with gr.Tab("Zoom"): | |
| z_state = gr.State(value={"wps": zoom_waypoints(RSEED)}) | |
| with gr.Row(): | |
| with gr.Column(): | |
| z_img = gr.Image(label="one identity's dive", | |
| height=DISPLAY, width=DISPLAY) | |
| z_info = gr.Markdown() | |
| with gr.Column(): | |
| z_phi = gr.Slider(0, 0.999, value=0, step=0.005, | |
| label="phi · scrub: fire → face → fire") | |
| z_new = gr.Button("🎲 new identities") | |
| gr.Markdown("**Full Shepard zoom** (falls through many faces, " | |
| "the scale reset hidden in the fire):") | |
| z_cycles = gr.Slider(2, 8, value=4, step=1, | |
| label="identities to fall through") | |
| z_go = gr.Button("🎬 render zoom video") | |
| z_vid = gr.Video(label="Shepard zoom") | |
| z_phi.change(zoom_scrub, [z_state, z_phi], [z_img, z_info]) | |
| z_new.click(zoom_new_ids, z_state, [z_state, z_img, z_info, z_phi]) | |
| z_go.click(zoom_video, [z_state, z_cycles], z_vid) | |
| demo.load(zoom_scrub, [z_state, z_phi], [z_img, z_info]) | |
| # ---- GALLERY ---- | |
| with gr.Tab("Gallery"): | |
| with gr.Row(): | |
| g_n = gr.Slider(4, 64, value=24, step=4, label="how many faces") | |
| g_r = gr.Slider(0.2, 3.0, value=0.6, step=0.1, label="radius (spread)") | |
| g_seed = gr.Number(value=7, label="seed", precision=0) | |
| g_go = gr.Button("🖼️ sample faces") | |
| g_out = gr.Gallery(label="samples", columns=6, height=560) | |
| g_go.click(gallery, [g_n, g_r, g_seed], g_out) | |
| # ---- ABOUT ---- | |
| with gr.Tab("About / Theory"): | |
| gr.Markdown(ABOUT) | |
| return demo | |
| if __name__ == "__main__": | |
| demo = build() | |
| try: | |
| demo.launch(theme=gr.themes.Soft()) # Gradio 6: theme lives here | |
| except TypeError: | |
| demo.launch() # older Gradio: theme was on Blocks | |