# app.py — Coherent_Compute_Engine (RFTSystems) # Live, measurable throughput + stability + energy proxy, with verification baselines + receipt download. # No estimates. No precomputed data. Same workload, same machine, same rules. import os import time import json import math import hashlib import platform from datetime import datetime, timezone import numpy as np import gradio as gr # Optional: numba acceleration try: import numba as nb NUMBA_OK = True except Exception: NUMBA_OK = False nb = None APP_TITLE = "Coherent Compute Engine" RESULTS_DIR = "receipts" # ----------------------------- # Definition: what an "item" is # ----------------------------- # One coherent state update of [Psi, E, L] per oscillator per step. # Items/sec = (N oscillators * steps) / elapsed_seconds # ----------------------------- # Core update: vectorised (NumPy) # ----------------------------- def np_step(Psi, E, L, scale=1.0): # Numerically tame, branchless-ish ops; stable for large N. phase = 0.997 * Psi + 0.003 * E drive = np.tanh(phase * scale) Psi_n = 0.999 * Psi + 0.001 * drive E_n = 0.995 * E + 0.004 * Psi_n L_n = 0.998 * L + 0.001 * (Psi_n * E_n) return Psi_n, E_n, L_n # ----------------------------- # Baseline: tiny Python loop (safety-capped) # ----------------------------- def pyloop_step(Psi, E, L, scale=1.0): # Scalar operations; intentionally slow baseline. phase = 0.997 * Psi + 0.003 * E drive = math.tanh(phase * scale) Psi_n = 0.999 * Psi + 0.001 * drive E_n = 0.995 * E + 0.004 * Psi_n L_n = 0.998 * L + 0.001 * (Psi_n * E_n) return Psi_n, E_n, L_n def run_python_loop_baseline(n, steps, seed=7, cap_seconds=1.2): """ Runs a scalar baseline on a small subset for a short capped duration. Reports throughput in *equivalent* items/sec for that baseline subset only. This is a "floor" baseline, not a competitor. """ rng = np.random.default_rng(seed) # Keep tiny so we don't lock the Space. n0 = min(n, 150_000) # safety subset Psi = rng.random(n0, dtype=np.float32) E = rng.random(n0, dtype=np.float32) L = rng.random(n0, dtype=np.float32) # Run until either steps done or time cap hit t0 = time.time() done = 0 for _ in range(int(steps)): # time cap guard if (time.time() - t0) > cap_seconds: break # scalar loop over subset for i in range(n0): Psi[i], E[i], L[i] = pyloop_step(float(Psi[i]), float(E[i]), float(L[i])) done += 1 elapsed = max(1e-9, time.time() - t0) items = done * n0 thr_Bps = (items / elapsed) / 1e9 return thr_Bps, elapsed, n0, done # ----------------------------- # Optional: Numba kernel # ----------------------------- if NUMBA_OK: @nb.njit(fastmath=True, parallel=True) def nb_run(Psi, E, L, steps): for _ in range(steps): # same math as numpy step, inside jit loop phase = 0.997 * Psi + 0.003 * E drive = np.tanh(phase) Psi = 0.999 * Psi + 0.001 * drive E = 0.995 * E + 0.004 * Psi L = 0.998 * L + 0.001 * (Psi * E) return Psi, E, L def compute_coherence(Psi_before, Psi_after): # Normalised dot product: magnitude is the point; can be signed. # We report |C| for "stability" to avoid phase sign confusion. v1 = Psi_before.astype(np.float64, copy=False) v2 = Psi_after.astype(np.float64, copy=False) num = float(np.dot(v1, v2)) + 1e-12 den = float(np.linalg.norm(v1) * np.linalg.norm(v2)) + 1e-12 return num / den def compute_energy(E): # Energy proxy: bounded mean in [0,1.5] return float(np.mean(np.clip(E, 0.0, 1.5))) def human_bps(x_bps): # x_bps is in billions/sec (B/s) if x_bps >= 1.0: return f"{x_bps:.3f} B/s" return f"{x_bps:.3f} B/s" def get_cpu_string(): # best-effort try: return platform.processor() or platform.uname().processor or "" except Exception: return "" def sha256_bytes(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def make_receipt(payload: dict): os.makedirs(RESULTS_DIR, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ") fname = f"receipt_{ts}.json" path = os.path.join(RESULTS_DIR, fname) # Canonical JSON for stable hashing canon = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") h = sha256_bytes(canon) payload_out = dict(payload) payload_out["receipt_sha256"] = h with open(path, "w", encoding="utf-8") as f: json.dump(payload_out, f, indent=2) return path, h, payload_out def run_engine(n_oscillators: int, steps: int, include_baselines: bool): # Hard safety rails for HF Spaces stability (still "full throttle" within reason) n = int(max(50_000, min(int(n_oscillators), 25_000_000))) steps = int(max(10, min(int(steps), 2_000))) rng = np.random.default_rng(7) Psi = rng.random(n, dtype=np.float32) E = rng.random(n, dtype=np.float32) L = rng.random(n, dtype=np.float32) # Snapshot for coherence metric (small sample) sample = min(n, 250_000) Psi0 = Psi[:sample].copy() # Choose engine: prefer numba if available, else numpy engine = "numpy" t0 = time.time() if NUMBA_OK: engine = "numba" # warm-up compile on a tiny slice if first run # (keeps first-run penalty from ruining the metric) try: _Psi_w = Psi[:50_000].copy() _E_w = E[:50_000].copy() _L_w = L[:50_000].copy() nb_run(_Psi_w, _E_w, _L_w, 2) except Exception: engine = "numpy" if engine == "numba": Psi, E, L = nb_run(Psi, E, L, steps) else: for _ in range(steps): Psi, E, L = np_step(Psi, E, L) elapsed = max(1e-9, time.time() - t0) # Metrics items = n * steps thr_Bps = (items / elapsed) / 1e9 coh = compute_coherence(Psi0, Psi[:sample]) coh_abs = abs(coh) meanE = compute_energy(E[:sample]) # Optional baselines (measured live) base_numpy = None base_py = None speedup_vs_py = None speedup_vs_numpy = None if include_baselines: # Baseline A: NumPy (forced) on same n/steps t1 = time.time() PsiA = rng.random(n, dtype=np.float32) EA = rng.random(n, dtype=np.float32) LA = rng.random(n, dtype=np.float32) for _ in range(steps): PsiA, EA, LA = np_step(PsiA, EA, LA) elA = max(1e-9, time.time() - t1) base_numpy = (n * steps / elA) / 1e9 # Baseline B: Python loop (subset, capped) base_py, py_elapsed, py_n, py_steps_done = run_python_loop_baseline(n=n, steps=steps, seed=7) # Speedups (honest: can be < 1.0) if base_py and base_py > 0: speedup_vs_py = thr_Bps / base_py if base_numpy and base_numpy > 0: speedup_vs_numpy = thr_Bps / base_numpy # Receipt payload payload = { "app": APP_TITLE, "timestamp_utc": datetime.now(timezone.utc).isoformat(), "definition_of_item": "One coherent update of [Psi,E,L] per oscillator per step", "n_oscillators": n, "steps": steps, "engine": engine, "elapsed_seconds": elapsed, "throughput_Bps": thr_Bps, "coherence_C": coh, "coherence_abs": coh_abs, "mean_energy_proxy": meanE, "cpu": get_cpu_string(), "cores_available": os.cpu_count() or 1, "baselines_enabled": bool(include_baselines), "baseline_numpy_Bps": base_numpy, "baseline_python_loop_Bps": base_py, "speedup_vs_python_loop_x": speedup_vs_py, "speedup_vs_numpy_x": speedup_vs_numpy, "notes": [ "All values measured live on the Space runtime machine.", "Baselines are measured on the same machine with the same workload settings.", "Python loop baseline is safety-capped and uses a subset to keep the Space responsive.", ], } receipt_path, receipt_sha, payload_out = make_receipt(payload) # UI-friendly output (minimal, factual) result = { "Throughput (B/s)": f"{thr_Bps:.3f}", "Coherence (|C|)": f"{coh_abs:.5f}", "Mean Energy": f"{meanE:.5f}", "Elapsed Time (s)": f"{elapsed:.2f}", "Oscillators": f"{n:,}", "Steps": f"{steps}", "Engine": engine, "CPU Cores Available": int(os.cpu_count() or 1), } if include_baselines: result["Baseline (Vectorised NumPy) (B/s)"] = f"{base_numpy:.3f}" if base_numpy is not None else "n/a" result["Baseline (Python loop, capped) (B/s)"] = f"{base_py:.3f}" if base_py is not None else "n/a" if speedup_vs_py is not None: result["Speedup vs Python loop (x)"] = f"{speedup_vs_py:.1f}" if speedup_vs_numpy is not None: result["Speedup vs NumPy (x)"] = f"{speedup_vs_numpy:.2f}" # add one explicit honesty line result["Note"] = "Speedups can be <1.0 depending on runtime/Numba warmup/CPU features. That is expected and is reported as-is." result["Receipt SHA-256 (in file)"] = "written in receipt" return json.dumps(result, indent=2), receipt_path # ----------------------------- # UI # ----------------------------- INTRO_MD = """ ### What this is - **No precomputed results** - **No GPUs required** - Measures **real throughput**, **stability**, and **energy behaviour** on the machine running this Space. ### What an “item” is - One coherent state update of **[Ψ, E, L]** per oscillator per step. Everything you see below is computed **right now**, on this machine. """ NOTES_MD = """ **Notes** - This runs on the Hugging Face Space runtime machine. Your browser just displays the UI. - If the Space is under load, throughput will vary — that variance is real and is part of the measurement. - Baselines are not “competitions”. They are **verification anchors** measured live on the same machine. """ with gr.Blocks(theme=gr.themes.Soft(), title=APP_TITLE) as demo: gr.Markdown(f"# {APP_TITLE}") gr.Markdown(INTRO_MD) with gr.Row(): n_slider = gr.Slider( minimum=250_000, maximum=25_000_000, value=6_400_000, step=50_000, label="Number of Oscillators", ) steps_slider = gr.Slider( minimum=50, maximum=2000, value=650, step=10, label="Simulation Steps", ) include_baselines = gr.Checkbox( value=True, label="Show verification baselines (same machine)", info="Baselines are measured live too. Python loop is safety-capped." ) run_btn = gr.Button("Run Engine", variant="primary") with gr.Row(): out_json = gr.Code(label="Results", language="json") receipt_file = gr.File(label="Receipt (download)", file_count="single") gr.Markdown(NOTES_MD) run_btn.click( fn=run_engine, inputs=[n_slider, steps_slider, include_baselines], outputs=[out_json, receipt_file], ) if __name__ == "__main__": demo.queue(concurrency_count=1).launch()