app.py

# 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()