#!/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(" 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()