resonance-engine/analysis/capture_and_predict.py

#!/usr/bin/env python3
# capture_and_predict.py
# PASSIVE PREDICT-TEST on the live 5561 coarse stream.
# Read-only: subscribes to the observer daemon, sends NOTHING.
# Captures N frames, then asks the ONE honest question:
#   Can ANY predictor beat persistence ("next = current") on the live field?
# Pre-committed, bootstrapped, no rescue. Writes results + a data file for visualization.
#
# Per KHRAGIXX_SOURCE_OF_TRUTH.md:
#  - short window (leak-safe)
#  - persistence + linear baselines, pre-committed
#  - bootstrap distribution, not single sample
#  - a clean null is a real result; do not rescue

import sys, io, json, time, struct
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
import numpy as np

try:
    import zmq
except ImportError:
    print("pip install pyzmq --break-system-packages"); sys.exit(1)

TILES, CH = 32, 6
NVALS = TILES * TILES * CH
HDR = 16
FRAME_BYTES = HDR + NVALS * 4
N_CAPTURE = 500          # ~50s at 100Hz. Short => leak-safe.
PORT = "tcp://localhost:5561"
CH_NAMES = ["rho", "ux", "uy", "sxx", "syy", "sxy"]
OUT_DATA = "predict_capture.npz"
OUT_REPORT = "predict_report.txt"

def capture(n):
    ctx = zmq.Context()
    s = ctx.socket(zmq.SUB)
    s.setsockopt(zmq.RCVTIMEO, 5000)
    s.setsockopt_string(zmq.SUBSCRIBE, "")
    s.connect(PORT)
    frames, cycles = [], []
    print(f"Capturing {n} frames from {PORT} (read-only)...", flush=True)
    t0 = time.time()
    while len(frames) < n:
        try:
            buf = s.recv()
        except zmq.Again:
            print(f"  timeout after {len(frames)} frames — is the daemon streaming 5561?"); break
        if len(buf) != FRAME_BYTES: continue
        if buf[:4] != b"KGCF": continue
        cyc = struct.unpack_from("<I", buf, 4)[0]
        vals = np.frombuffer(buf, dtype=np.float32, count=NVALS, offset=HDR).copy()
        frames.append(vals.reshape(TILES*TILES, CH))
        cycles.append(cyc)
        if len(frames) % 100 == 0:
            print(f"  {len(frames)}/{n}  cycle={cyc}", flush=True)
    s.close(); ctx.term()
    dt = time.time() - t0
    print(f"Captured {len(frames)} frames in {dt:.1f}s ({len(frames)/max(dt,1e-9):.1f} fps)", flush=True)
    return np.array(frames), np.array(cycles)

def zscore_per_channel(F):
    # F: (T, tiles, CH). Normalize each channel so no single channel dominates the error metric.
    mu = F.mean(axis=(0,1), keepdims=True)
    sd = F.std(axis=(0,1), keepdims=True); sd[sd < 1e-9] = 1.0
    return (F - mu) / sd

def frame_err(pred, truth):
    # RMSE over all tiles*channels of one frame-step, per step -> array of length T-1 (or T-k)
    d = pred - truth
    return np.sqrt(np.mean(d*d, axis=(1,2)))

def main():
    F, C = capture(N_CAPTURE)
    if len(F) < 50:
        print("Too few frames; aborting."); sys.exit(1)

    # cycle sanity: should advance ~10/frame
    dcyc = np.diff(C)
    print(f"cycle step: median={np.median(dcyc):.0f} (expect ~10), min={dcyc.min()}, max={dcyc.max()}", flush=True)

    Fz = zscore_per_channel(F.astype(np.float64))   # (T, 1024, 6)
    T = len(Fz)

    # ---- PRE-COMMITTED PREDICTORS (declared before scoring) ----
    # 1) PERSISTENCE: pred[t] = frame[t-1]      (the baseline to beat)
    # 2) LINEAR:      pred[t] = 2*frame[t-1] - frame[t-2]  (constant-velocity extrapolation)
    # 3) MEAN:        pred[t] = running mean of all prior frames (dumb global)
    # Score = mean per-step RMSE over t = 2..T-1 (aligned window for all three).

    truth = Fz[2:]                       # frames 2..T-1
    persist = Fz[1:-1]                   # frame t-1
    linear  = 2*Fz[1:-1] - Fz[0:-2]      # 2*(t-1) - (t-2)
    # running mean predictor
    cummean = np.cumsum(Fz, axis=0) / np.arange(1, T+1)[:,None,None]
    meanpred = cummean[1:-1]             # mean of frames 0..t-1 approx (uses through t-1)

    e_persist = frame_err(persist, truth)
    e_linear  = frame_err(linear,  truth)
    e_mean    = frame_err(meanpred, truth)

    P = e_persist.mean(); L = e_linear.mean(); M = e_mean.mean()

    # ---- BOOTSTRAP: is linear reliably better (lower err) than persistence? ----
    # Pre-committed: linear "beats" persistence only if the 95% bootstrap CI of
    # (persist_err - linear_err) is entirely > 0 (i.e. linear strictly lower error),
    # AND the median improvement exceeds 2% of persistence error (not noise-width).
    rng = np.random.default_rng(42)
    diffs = e_persist - e_linear         # >0 means linear better
    n = len(diffs)
    boot = np.array([rng.choice(diffs, n, replace=True).mean() for _ in range(2000)])
    lo, hi = np.percentile(boot, [2.5, 97.5])
    improve_frac = (P - L) / P
    linear_beats = (lo > 0) and (improve_frac > 0.02)

    # How predictable is the field at all? (persistence error relative to frame-to-frame scale)
    # scale = typical magnitude of a z-scored frame's per-step change baseline
    field_change = frame_err(Fz[1:], Fz[:-1]).mean()   # avg step-to-step change

    lines = []
    def out(s): print(s, flush=True); lines.append(s)

    out("="*64)
    out("PASSIVE PREDICT-TEST — live 5561 coarse field")
    out("="*64)
    out(f"frames used: {T}  (cycles {C[0]} -> {C[-1]}, ~{(C[-1]-C[0])} cycles)")
    out(f"channels: {CH_NAMES}")
    out("")
    out("Per-step RMSE (z-scored field, lower = better predictor):")
    out(f"  PERSISTENCE (next=current):        {P:.4f}   <- baseline to beat")
    out(f"  LINEAR (const-velocity extrap):    {L:.4f}")
    out(f"  RUNNING MEAN (dumb global):        {M:.4f}")
    out("")
    out("PRE-COMMITTED TEST: does LINEAR reliably beat PERSISTENCE?")
    out(f"  improvement: {improve_frac*100:+.2f}%  (need > +2.00%)")
    out(f"  bootstrap 95% CI of (persist_err - linear_err): [{lo:+.4f}, {hi:+.4f}]  (need lo > 0)")
    out(f"  VERDICT: {'LINEAR BEATS PERSISTENCE — field carries short-horizon dynamics' if linear_beats else 'NULL — no predictor beats persistence at threshold'}")
    out("")
    # Interpretation guardrails (from the doc, so the next reader doesn't over-read)
    if linear_beats:
        out("READ: The field's next state is better predicted by its trajectory than by")
        out("assuming it stays still. That means the live fast field carries structured")
        out("short-horizon motion the 9 global scalars could not show. This is the FIRST")
        out("positive signal that the live organism has learnable dynamics. It is NOT yet")
        out("memory (that needs the poke/history test). It IS a green light to proceed.")
    else:
        out("READ: Nothing beats 'it stays the same' at the committed threshold. The live")
        out("coarse field, at this resolution and horizon, is either near-static frame to")
        out("frame or its motion is not linearly predictable. A clean null. Do not rescue.")
        out("Next: try shorter tiles (finer grid) or the stress channels alone before")
        out("concluding the substrate has no fast structure.")

    with open(OUT_REPORT, "w") as f:
        f.write("\n".join(lines) + "\n")

    # Save for visualization: raw frames (un-normalized), cycles, and per-step errors
    np.savez(OUT_DATA,
             frames=F.astype(np.float32), cycles=C,
             e_persist=e_persist, e_linear=e_linear,
             ch_names=np.array(CH_NAMES))
    print(f"\nWrote {OUT_REPORT} and {OUT_DATA}", flush=True)
    print("DONE.", flush=True)

if __name__ == "__main__":
    main()