#!/usr/bin/env python3 """ The scattering side, step 4 (October 15, 2026): the prospective replication of the exploratory pilot on fresh starts 147 to 170, as replication-plan-final.md fixes it. Its settings were fixed in replication_lock.json, sealed with the final plan by change-2026-10-15-10, before this code was written. First, the predictions made before the pilot are scored again, with the pilot's own labels and combined readings. Then Colson's three questions, with criteria written after the pilot; Question 2's is labelled a hypothesis set after the pilot. The pilot's sealed numbers are quoted in their own block and never pooled. A toy model; histories, not memory. scattering_side.py and scattering_pilot.py are imported unchanged: every model run, reading, label and combined reading is theirs. Nothing in their folders is written. A sandbox script kept with the project. Nothing here is served by the site. Usage (each command runs in the foreground with a hard limit and ends by itself; nothing is left running) timeout -k 10 285 python3 -B scattering_replication.py check # R0 to R5: replication_checks.txt/.json timeout -k 10 285 python3 -B scattering_replication.py base # the 24 fresh base runs (about 4 to 6 pieces) timeout -k 10 285 python3 -B scattering_replication.py moments # the 240 moments (about 12 to 18 pieces), then the report timeout -k 10 285 python3 -B scattering_replication.py after # R6 to R8: replication_after.txt/.json check refuses to write over its record. base, moments and after refuse to run unless the lock, the plans, the amendment and both codes are the sealed versions, every check passed with this same code in the same numerical environment, and replication_go.json, written on Colson's separate authorization and sealed, names this lock and the check record. """ import json import re import shutil import statistics import sys import tempfile import time import traceback from multiprocessing import Pool from pathlib import Path HERE = Path(__file__).resolve().parent sys.dont_write_bytecode = True # leave every folder as it is sys.path.insert(0, str(HERE)) import scattering_pilot as sp # noqa: E402 (it imports scattering_side first: one BLAS thread per worker) ss, mt, pc, np, ct = sp.ss, sp.mt, sp.pc, sp.np, sp.ct sha256, stamp, shown, write_json, plain = ss.sha256, ss.stamp, ss.shown, ss.write_json, ss.plain ROOT, ARMS, W = ss.ROOT, sp.ARMS, sp.W THIS = Path(__file__).resolve() REP_LOCK, REP_PLAN = HERE / "replication_lock.json", HERE / "replication-plan-final.md" DRAFT = HERE / "replication-plan-draft-1.md" SAFE = ROOT / "analysis" / "cancer-research" / "safekeeping" SEAL = SAFE / "change-2026-10-15-10.json" # sealed the lock and the final plan PILOT_RAW = SAFE / "raw-data-2026-10-15-8.sha256" # the pilot's 132 saved runs CHECKS_TXT, CHECKS_JSON = HERE / "replication_checks.txt", HERE / "replication_checks.json" GO = HERE / "replication_go.json" # written only on Colson's separate authorization LOG = HERE / "replication_runlog.json" RESULTS_TXT, RESULTS_JSON = HERE / "replication_results.txt", HERE / "replication_results.json" AFTER_TXT, AFTER_JSON = HERE / "replication_after.txt", HERE / "replication_after.json" PARTS = ss.WORK / "replication" # saved runs (kept out of the project's history) OWN_RECORDS = (REP_LOCK, CHECKS_JSON, GO, LOG, RESULTS_JSON, AFTER_JSON) REP_STARTS = tuple(range(147, 171)) REP_SEEDS = tuple(s - 1 for s in REP_STARTS) PILOT_SEEDS = sp.PILOT_SEEDS # the pilot's own list, restored after every call ALL_MOMENTS = len(REP_SEEDS) * mt.MOMENTS THEORY_RATES, BAND_WIDTH, RATE_BAND = (0.9601, 0.9641), 0.03, (0.9301, 0.9941) BAND_INSIDE, BAND_OUTSIDE = 2, 3 # bars: strictly inside; clearly outside RATE_BAR = dict(resamples=mt.RESAMPLES, seed=mt.RESAMPLE_SEED, sd_ddof=1) D_ANY_SIZE, D_ANY_INSIDE, D_B2_SIZE, D_B2_INSIDE = 0.1, 0.05, 0.005, 0.005 RISE_BARS, BELOW_BARS, PAIR_AT_LEAST = 3, 3, 2 PILOT_CHECK = dict(start=135, seed=134, moment=0, T=2285) # R2: a pilot moment, rerun from the pilot's base run REP_CHECK = dict(start=147, seed=146, moment=0, T=2235) # R7: a replication moment, rerun after the report MADE_UP_STREAM, AGREEMENT = 11, 1e-12 # R4: draws (20261015, 11, scenario) BUDGET, START_BY, WORKERS, HARD_LIMIT = sp.BUDGET, sp.START_BY, sp.WORKERS, sp.HARD_LIMIT NAN = float("nan") def rep_settings(): """The pilot's settings, with the fresh starts, their count of moments, the questions' settings and the checks' moments: replication_lock.json's 'settings' exactly.""" s = json.loads(json.dumps(plain(sp.PILOT_SETTINGS))) s["starts"] = dict(numbers=list(REP_STARTS), seeds=list(REP_SEEDS)) s["moments"]["all"] = ALL_MOMENTS s["questions"] = dict( theory_rates=dict(together=THEORY_RATES[0], opposite=THEORY_RATES[1]), band_width=BAND_WIDTH, band=list(RATE_BAND), band_bars=dict(inside=BAND_INSIDE, clearly_outside=BAND_OUTSIDE), rate_bar=dict(RATE_BAR), D_any=dict(size=D_ANY_SIZE, inside=D_ANY_INSIDE), D_B2=dict(size=D_B2_SIZE, inside=D_B2_INSIDE), rise_at_0_6_bars=RISE_BARS, below_bars=BELOW_BARS, pair_described_at_least=PAIR_AT_LEAST) s["replication_checks"] = dict(pilot_moment=dict(PILOT_CHECK), replication_moment=dict(REP_CHECK), made_up=dict(seed=ss.DRAW_SEED, stream=MADE_UP_STREAM), agreement=AGREEMENT) return s REP_SETTINGS = rep_settings() def rel(p): return mt.shown_path(Path(p)) # --------------------------------------------------------------------------------------------------------------------- # R0: the lock and the sealed files def versions(): v = sp.versions() return dict(replication_lock_sha256=sha256(REP_LOCK), replication_plan_sha256=sha256(REP_PLAN), replication_code_sha256=sha256(THIS), pilot_lock_sha256=v["pilot_lock_sha256"], plan_sha256=v["plan_sha256"], lock_sha256=v["lock_sha256"], amendment_sha256=v["amendment_sha256"], steps_1_and_2_code_sha256=v["steps_1_and_2_code_sha256"], pilot_code_sha256=v["code_sha256"]) def lock_problems(): """Everything the replication lock fixes that doesn't match now (nothing expected).""" lock = json.loads(REP_LOCK.read_text()) wrong = [] if json.loads(json.dumps(plain(REP_SETTINGS))) != lock["settings"]: wrong.append("the code's settings differ from the replication lock's") named = [("the final plan", REP_PLAN, lock["replication_plan"]), ("draft 1", DRAFT, lock["draft"]), ("plan-final.md", ss.PLAN, lock["plan"]), ("lock.json", ss.LOCK, lock["lock"]), ("the amendment", sp.AMENDMENT, lock["amendment"]), ("pilot_lock.json", sp.PILOT_LOCK, lock["pilot_lock"]), ("scattering_side.py", ss.THIS, lock["steps_1_and_2_code"]), ("scattering_pilot.py", sp.THIS, lock["pilot_code"])] named += [(f"the pilot's record {Path(e['file']).name}", ROOT / e["file"], e) for e in lock["pilot_records"].values()] for label, path, entry in named: if rel(path) != entry["file"] or not path.exists() or sha256(path) != entry["sha256"]: wrong.append(f"{label} is not the version the replication lock names") seal = json.loads(SEAL.read_text()) if SEAL.exists() else {} if seal.get("replication_lock", {}).get("sha256") != sha256(REP_LOCK): wrong.append("replication_lock.json is not the version change-2026-10-15-10 sealed") if seal.get("replication_plan", {}).get("sha256") != sha256(REP_PLAN): wrong.append("the final plan is not the version change-2026-10-15-10 sealed") wrong += [f"the pilot lock: {w}" for w in sp.pilot_problems()] mine, pilot = lock["settings"], json.loads(sp.PILOT_LOCK.read_text())["settings"] differ = sorted(k for k in set(mine) | set(pilot) if mine.get(k) != pilot.get(k)) if differ != ["moments", "questions", "replication_checks", "starts"]: wrong.append(f"the lock's settings differ from pilot_lock.json's in {differ}, not only in the starts, the count" " of moments and the two additions") else: others = [{k: v for k, v in d["moments"].items() if k != "all"} for d in (mine, pilot)] if others[0] != others[1] or mine["moments"]["all"] != ALL_MOMENTS: wrong.append("the lock's moments differ from pilot_lock.json's beyond their count") if lock["moments"] != {str(s + 1): mt.moment_times(s) for s in REP_SEEDS}: wrong.append("the fresh starts' moments are not the lock's") for c in (PILOT_CHECK, REP_CHECK): if mt.moment_times(c["seed"])[c["moment"]] != c["T"]: wrong.append(f"start {c['start']}'s moment {c['moment'] + 1} is not at T = {c['T']}") if not (RISE_BARS == BELOW_BARS == BAND_OUTSIDE == sp.CLEAR and BAND_INSIDE == sp.SIDE and RATE_BAR["resamples"] == mt.RESAMPLES and RATE_BAR["seed"] == mt.RESAMPLE_SEED): wrong.append("the questions' bars are not the plan's") return wrong def environment_problems(): """The numerical environment must match the pilot's checks and run log.""" now, wrong = pc.machine(), [] for label, path in (("the pilot's checks", sp.CHECKS_JSON), ("the pilot's run log", sp.LOG)): then = json.loads(path.read_text()).get("machine") if not then or not pc.same_environment(then, now): wrong.append(f"the numerical environment differs from {label}'") return wrong def locked(): wrong = lock_problems() if wrong: sys.exit("Stopping: " + "; ".join(wrong) + ".") def go_seal(): """The dated change record that sealed replication_go.json as it is now ('' if none).""" if not GO.exists(): return "" key, want = rel(GO), sha256(GO) for path in sorted(SAFE.glob("change-*.json")): try: if json.loads(path.read_text()).get("added", {}).get(key) == want: return path.name except ValueError: continue return "" def require_checks(): """The run goes ahead only after every check passed with this same code, lock and plans, on the same machine.""" if not CHECKS_JSON.exists(): sys.exit("No replication check record: run check first.") record = json.loads(CHECKS_JSON.read_text()) if not record.get("all_passed"): sys.exit("A replication check failed: nothing runs on a fresh start (replication_lock.json). Report the" " failure.") changed = [key for key, value in versions().items() if record.get(key) != value] if changed: sys.exit(f"Changed since the checks passed ({', '.join(changed)}): stopping.") if not pc.same_environment(record["machine"], pc.machine()): sys.exit("The numerical environment differs from the checks': stopping.") return record def require_go(): """Colson's separate authorization for the run: replication_go.json, naming this lock and this check record, and sealed by a dated change record.""" if not GO.exists(): sys.exit("The run needs Colson's separate authorization (replication_go.json). Nothing was run.") go = json.loads(GO.read_text()) if go.get("replication_lock_sha256") != sha256(REP_LOCK) or go.get("replication_checks_sha256") != sha256(CHECKS_JSON): sys.exit("replication_go.json doesn't name this lock and this check record. Nothing was run.") if not go_seal(): sys.exit("replication_go.json isn't sealed by a dated change record yet. Nothing was run.") return go # --------------------------------------------------------------------------------------------------------------------- # R5: the fresh starts appear in no earlier record START_KEYS = {"start", "starts", "start_numbers", "fresh_starts", "numbers"} SEED_KEYS = {"seed", "seeds"} TASK = re.compile(r"(?:base|moment)_(\d{3})(?:_\d{2})?(?:\.|$)") # a task or its saved file, by start number def task_start(text): m = TASK.match(str(text).rsplit("/", 1)[-1]) return int(m.group(1)) if m else None def start_moments(value): """A start's moments as the records keep them under its number: its ten moment times, or an entry holding T.""" if isinstance(value, list): return len(value) == mt.MOMENTS and all(isinstance(t, int) and not isinstance(t, bool) for t in value) return isinstance(value, dict) and "T" in value def mentions(value, key, found, aside): """Every mention of a fresh start in one json value: start or seed numbers under their keys, start numbers as keys of a 'moments' map holding a start's moment times, and task names. A 'moments' map keyed by other numbers (a lag, say) whose entries aren't moment times goes to aside, to be named.""" if isinstance(value, dict): for k, v in value.items(): if key == "moments" and str(k).isdigit() and int(k) in REP_STARTS: if start_moments(v): found.append(f"moments for start {k}") else: aside.append(f"'moments' keyed by {k}, its entry not a start's moment times") if task_start(k) in REP_STARTS: found.append(f"the task {k}") mentions(v, k, found, aside) elif isinstance(value, list): for v in value: mentions(v, key, found, aside) elif isinstance(value, bool) or value is None: return elif isinstance(value, int): if key in START_KEYS and value in REP_STARTS: found.append(f"'{key}': {value}") if key in SEED_KEYS and value in REP_SEEDS: found.append(f"'{key}': {value}") elif isinstance(value, str) and task_start(value) in REP_STARTS: found.append(f"the task {value}") def fresh_outputs(): """Anything the run has produced (nothing before Colson's authorization).""" found = [p for p in sorted(PARTS.rglob("*")) if p.is_file()] if PARTS.exists() else [] return [rel(p) for p in found + [q for q in (LOG, RESULTS_TXT, RESULTS_JSON, AFTER_TXT, AFTER_JSON) if q.exists()]] def r5(): own = {p.resolve() for p in OWN_RECORDS} records = sorted(p for p in (ROOT / "analysis").rglob("*.json") if p.is_file() and p.resolve() not in own) hits, unreadable, set_aside = {}, [], {} for path in records: try: data = json.loads(path.read_text()) except (ValueError, UnicodeDecodeError): unreadable.append(rel(path)) continue found, aside = [], [] mentions(data, None, found, aside) if found: hits[rel(path)] = found if aside: set_aside[rel(path)] = aside artifacts = ROOT / "generated-artifacts" files = [p for p in artifacts.rglob("*") if p.is_file() and not p.is_relative_to(PARTS)] if artifacts.exists() else [] named = sorted(rel(p) for p in files if task_start(p.name) in REP_STARTS) outputs = fresh_outputs() folder = PARTS.exists() ok = not hits and not unreadable and not named and ((not outputs and not folder) or GO.exists()) details = [f"json records read in analysis/ (the replication's own left out): {len(records)}; naming a fresh start" f" (147 to 170, seeds 146 to 169): {len(hits)}; unreadable: {len(unreadable)}", f"records with a 'moments' map keyed by numbers in 147 to 170 whose entries aren't a start's moment times" f" (a list of 10 or an entry holding T), so not start numbers; set aside and named here: {len(set_aside)}"] details += [f"set aside: {path}: {', '.join(aside[:3])}{' ...' if len(aside) > 3 else ''}" for path, aside in set_aside.items()] details += [f"files in generated-artifacts/ outside the replication's folder: {len(files)}; named for a fresh start:" f" {len(named)}", f"the replication's folder exists: {folder}; files from the run (saved runs, run log, report, after-report" f" record): {len(outputs)}; replication_go.json present: {GO.exists()}"] details += [f"{path}: {', '.join(found[:5])}" for path, found in hits.items()] details += [f"unreadable: {path}" for path in unreadable] + [f"named: {path}" for path in named[:10]] details += [f"from the run: {path}" for path in outputs[:10]] return ok, details # --------------------------------------------------------------------------------------------------------------------- # the new readings and the three questions (replication_lock.json "meaning") def analysis(numbers, seeds): """scattering_pilot.analyse, unchanged, for these starts: its start list (whose only use is its count check) is set for the call and the pilot's restored after, whatever happens.""" kept = sp.PILOT_SEEDS sp.PILOT_SEEDS = tuple(seeds) try: return sp.analyse(numbers) finally: sp.PILOT_SEEDS = kept def below(s): """More than 3 bars below zero, strictly; never with an undefined bar.""" return bool(np.isfinite(s["bar"]) and s["mean"] < -BELOW_BARS * s["bar"]) def rate_and_bar(per_start): """The median |Omega*| over every direction holding a rhythm, all starts pooled, and its resampled bar: the SD (ddof 1) of the pooled medians over the 400 draws of the starts that medium_test.summary makes (a start drawn twice counts twice). Undefined if a draw holds no direction.""" per_start = [np.asarray(x, float).ravel() for x in per_start] pooled = np.concatenate(per_start) if per_start else np.zeros(0) rate = float(np.median(pooled)) if len(pooled) else NAN n = len(per_start) if n < 2: return rate, NAN draws = np.random.default_rng(RATE_BAR["seed"]).integers(0, n, (RATE_BAR["resamples"], n)) medians = [] for row in draws: drawn = np.concatenate([per_start[i] for i in row]) if not len(drawn): return rate, NAN medians.append(float(np.median(drawn))) return rate, float(np.std(medians, ddof=RATE_BAR["sd_ddof"])) def band_reading(rate, bar): """'inside': the rate +- 2 bars strictly inside 0.9301 to 0.9941; 'clearly outside': more than 3 bars beyond either end, strictly; otherwise 'neither'. An undefined rate or bar is neither.""" if not (np.isfinite(rate) and np.isfinite(bar)): return "neither" if rate - BAND_INSIDE * bar > RATE_BAND[0] and rate + BAND_INSIDE * bar < RATE_BAND[1]: return "inside" if rate + BAND_OUTSIDE * bar < RATE_BAND[0] or rate - BAND_OUTSIDE * bar > RATE_BAND[1]: return "clearly outside" return "neither" BAND_WORDS = {"inside": "inside the band", "clearly outside": "clearly outside the band", "neither": "neither inside nor clearly outside the band"} def verdict_of(passes, fails): if passes and fails: return "INCONCLUSIVE" return "PASS" if passes else "FAIL" if fails else "INCONCLUSIVE" def question_1(combined, e_pair, e_any, band): passes = combined == 1 and band == "inside" fails = [text for text, hit in (("E-pair at 0.4 contradicts", e_pair == "contradicts"), ("E-any at 0.4 contradicts", e_any == "contradicts"), ("the rate is clearly outside the band", band == "clearly outside")) if hit] verdict = verdict_of(passes, fails) if passes and fails: why = ("PASS and FAIL conditions both hold: combined reading 1 with the rate inside the band, and " + "; ".join(fails)) elif passes: why = "combined reading 1 at 0.4, and the rate inside the band" elif fails: why = "; ".join(fails) else: why = f"combined reading {combined} at 0.4; E-pair {e_pair}; E-any {e_any}; the rate {BAND_WORDS[band]}" return dict(verdict=verdict, passes=passes, fails=fails, why=why) def question_2(b2, e_any, steady_inside): rise = sp.above(b2) passes = rise and e_any == "as expected" and steady_inside fails = [text for text, hit in ( ("B2 at 0.6 inside ±0.005 without being more than 3 bars above zero", sp.inside(b2, sp.B2_MARGIN) and not rise), ("B2 at 0.6 more than 3 bars below zero", below(b2)), ("E-any at 0.6 contradicts", e_any == "contradicts")) if hit] verdict = verdict_of(passes, fails) if passes: why = "B2 at 0.6 more than 3 bars above zero, E-any as expected and E-steady inside ±0.05" elif fails: why = "; ".join(fails) else: why = (f"B2 at 0.6 {'more than 3 bars above zero' if rise else 'neither more than 3 bars above zero nor inside ±0.005'};" f" E-any {e_any}; E-steady {'inside' if steady_inside else 'not inside'} ±0.05") return dict(verdict=verdict, passes=passes, fails=fails, why=why, rise=rise) def question_3(d_any, d_b2): la, lb = sp.label_up(d_any, D_ANY_SIZE, D_ANY_INSIDE), sp.label_up(d_b2, D_B2_SIZE, D_B2_INSIDE) passes = la == lb == "as expected" fails = [text for text, hit in (("D-any and D-B2 both inside their margins", la == lb == "contradicts"), ("D-any more than 3 bars below zero", below(d_any)), ("D-B2 more than 3 bars below zero", below(d_b2))) if hit] verdict = verdict_of(passes, fails) if passes: why = "D-any and D-B2 both as expected" elif fails: why = "; ".join(fails) elif la == "as expected": why = f"separated by the rhythm but not by B2 (D-B2 {lb})" elif lb == "as expected": why = f"separated by B2 but not by the rhythm (D-any {la})" else: why = f"neither resolved (D-any {la}, D-B2 {lb})" return dict(verdict=verdict, passes=passes, fails=fails, why=why, labels=dict(D_any=la, D_B2=lb)) def questions(numbers, res, seeds=REP_SEEDS): """The paired readings, the rate's bar and band, the three questions' verdicts and what they describe, from the same moment numbers and the pilot's analysis of them.""" starts = [s + 1 for s in seeds] by = {s: sorted((n for n in numbers if n["start"] == s), key=lambda n: n["moment"]) for s in starts} def flag(a, key): # (start, moment, group) return np.array([[n["arms"][a][key] for n in by[s]] for s in starts], bool) outer = {a: np.array([[n["arms"][a]["outer_R"].mean() for n in by[s]] for s in starts], float) for a in ("f06", "f04")} per_start = {k: (flag("f04", k).astype(float) - flag("f06", k).astype(float)).mean(axis=(1, 2)) for k in ("pair", "any", "steady")} per_start["B2"] = mt.by_start(outer["f04"] - outer["f06"]) D = {k: mt.summary(v) for k, v in per_start.items()} omegas = [np.concatenate([np.abs(n["arms"]["f04"]["rule"]["omega"][n["arms"]["f04"]["rule"]["rhythm"]]) for n in by[s]]) for s in starts] rate, bar = rate_and_bar(omegas) band = band_reading(rate, bar) L06, L04 = res["levels"]["f06"], res["levels"]["f04"] q1 = question_1(L04["combined"], L04["labels"]["E_pair"], L04["labels"]["E_any"], band) q2 = question_2(L06["B2"], L06["labels"]["E_any"], L06["steady_inside"]) q3 = question_3(D["any"], D["B2"]) pairs = flag("f04", "pair").sum(axis=(1, 2)) # each start's group-moments holding a pair at 0.4 any04, any06 = flag("f04", "any"), flag("f06", "any") described = dict( pair_starts_at_least_once=int((pairs >= 1).sum()), pair_starts_at_least_2=int((pairs >= PAIR_AT_LEAST).sum()), D_any_starts_at_least_size=int((per_start["any"] >= D_ANY_SIZE).sum()), D_B2_starts_at_least_size=int((per_start["B2"] >= D_B2_SIZE).sum()), both_starts_at_least_size=int(((per_start["any"] >= D_ANY_SIZE) & (per_start["B2"] >= D_B2_SIZE)).sum()), group_moments=int(any04.size), rhythm_at_0_4_not_0_6=int((any04 & ~any06).sum()), rhythm_at_0_6_not_0_4=int((any06 & ~any04).sum()), B2_above_zero=dict(f06=int(L06["B2"]["above"]), f04=int(L04["B2"]["above"])), starts=len(starts)) return dict(D=D, per_start={k: v.tolist() for k, v in per_start.items()}, rate=rate, rate_bar=bar, rate_directions=int(sum(len(o) for o in omegas)), band=band, q1=q1, q2=q2, q3=q3, described=described) # --------------------------------------------------------------------------------------------------------------------- # the report def unplain(s): return dict(s, mean=NAN if s.get("mean") is None else s["mean"], bar=NAN if s.get("bar") is None else s["bar"]) def pilot_quoted(): """The pilot's labelled readings, from pilot_results.json as sealed: quoted, never pooled or recomputed.""" saved = json.loads(sp.RESULTS_JSON.read_text())["levels"] b, bp = saved["f06"], saved["f04"] return dict(file=rel(sp.RESULTS_JSON), sha256=sha256(sp.RESULTS_JSON), f06=dict(B1=unplain(b["B1"]), B2=unplain(b["B2"]), E_any=unplain(b["E"]["any"]), labels=b["labels"], notes=b["notes"], combined=b["combined"], combined_name=b["combined_name"]), f04=dict(B2=unplain(bp["B2"]), E_pair=unplain(bp["E"]["pair"]), E_any=unplain(bp["E"]["any"]), labels=bp["labels"], notes=bp["notes"], rate=NAN if bp["rate"] is None else bp["rate"], rhythm_directions=bp["rhythm_directions"], combined=bp["combined"], combined_name=bp["combined_name"])) def report_lines(res, q, pilot, notes=None): b, bp, d = res["levels"]["f06"], res["levels"]["f04"], q["described"] pm, show, note = sp.pm, sp.show_label, sp.with_note q1, q2, q3 = q["q1"], q["q2"], q["q3"] lines = [ f"Scattering-side replication (step 4, prospective), {shown(stamp())}. Fresh starts 147 to 170 (seeds 146 to" " 169), 10 moments each by the medium pilot's rule: 240 moments, every start counting equally. At each moment" " four arms run from the base run's state at T, both copies, 200 units changed and 500 restored: the untouched" " rerun, the raised pull (K x 1.005), arm F (c_f = 0.6) and arm F' (c_f = 0.4), as in the pilot. A toy model;" " histories, not memory.", "Bars: starts are the units; the larger of the resampled bar (400 resamples, seed 20261016) and SD/sqrt(24);" " a share of averages (B1, the share left after R) takes the resampled bar. 'Inside': the reading ± 2 bars" " within ± the margin. 'Clear': more than 3 bars from zero.", "Sections 1 and 2 score again the predictions made before the pilot (plan-final.md), with the pilot's own labels" " and combined readings; section 3 describes. Sections 4 to 6 are Colson's three questions, with criteria" " written after the pilot. Section 7 quotes the pilot's sealed numbers, never pooled with these.", ""] lines += sp.report_lines(res)[3:] # the pilot's own text for its sections 1 to 3 rate_text = (f"{q['rate']:.4f} ± {sp.num(q['rate_bar'], 4)} (the bar resampled only)" if q["rate_directions"] else "undefined: no direction holds a rhythm") lines += [ "", "4. Question 1. Does the counter-rotating edge rhythm reliably emerge at 0.4, with a turning rate consistent with" " the analytical prediction? (criteria written after the pilot)", f" Section 2's combined reading at 0.4: {bp['combined']}. E-pair: {bp['labels']['E_pair']}. E-any:" f" {bp['labels']['E_any']}.", f" The median |Omega*| over the {q['rate_directions']} directions holding a rhythm in arm F': {rate_text}. The" " band runs from 0.9301 to 0.9941, 0.03 beyond the theory's 0.9601 (A and B together) and 0.9641 (opposite):" f" the rate is {BAND_WORDS[q['band']]}.", " PASS needs combined reading 1 and the rate ± 2 bars strictly inside the band. FAIL: E-pair contradicts," " E-any contradicts, or the rate more than 3 bars beyond either end of the band.", f" Verdict: {q1['verdict']} ({q1['why']}).", f" Described: {d['pair_starts_at_least_once']} of {d['starts']} starts hold a pair in arm F' at least once," f" and {d['pair_starts_at_least_2']} in at least {PAIR_AT_LEAST} of their 20 group-moments.", "", "5. Question 2 (a hypothesis set after the pilot). Does the 0.6 arm reproducibly produce a positive B2 change" " without establishing a separate rhythm?", " This criterion was written after the pilot, from the pilot alone. Before the pilot, the plan predicted no" " meaningful change in B2 at 0.6 (margin ±0.005); section 1 scores that prediction.", f" B2 at 0.6: {pm(b['B2'], 4)}; {d['B2_above_zero']['f06']} of {d['starts']} starts above zero;" f" {'more' if q2['rise'] else 'not more'} than 3 bars above zero.", f" By the margin set before the pilot (±0.005), section 1's label: {show(b['labels']['B2'], 'null')}" f"{note(b['notes']['B2'])}.", f" The pilot's historical reading (12 starts, exploratory; section 7): {pm(pilot['f06']['B2'], 4)}," f" {show(pilot['f06']['labels']['B2'], 'null')}{note(pilot['f06']['notes']['B2'])}, against the null predicted" " before the pilot.", f" E-any at 0.6: {show(b['labels']['E_any'], 'e06')}. E-steady at 0.6: {pm(b['E']['steady'])}," f" {'inside' if b['steady_inside'] else 'not inside'} ±0.05.", " PASS needs B2 more than 3 bars above zero, E-any as expected and E-steady inside ±0.05. FAIL: B2 inside" " ±0.005 without being more than 3 bars above zero, B2 more than 3 bars below zero, or E-any contradicts.", f" Verdict: {q2['verdict']} ({q2['why']}). The size and its bar above stand whatever the verdict.", "", "6. Question 3. Can we directly distinguish the 0.4 and 0.6 outcomes using a prespecified paired comparison?" " (criteria written after the pilot)", f" D-any, the share of group-moments holding a rhythm, arm F' minus arm F at the same moments (size 0.10, inside" f" ±0.05): {pm(q['D']['any'])}: {q3['labels']['D_any']}.", f" D-B2, the faster half's own coherence at R, arm F' minus arm F (size 0.005, inside ±0.005):" f" {pm(q['D']['B2'], 4)}: {q3['labels']['D_B2']}.", " PASS needs both as expected (more than 3 bars above zero and at least the size). FAIL: both inside their" " margins, or either more than 3 bars below zero.", f" Verdict: {q3['verdict']} ({q3['why']}).", f" Described: D-pair {pm(q['D']['pair'])}; D-steady {pm(q['D']['steady'])}. Starts with D-any of at least 0.10" f" on their own: {d['D_any_starts_at_least_size']} of {d['starts']}; with D-B2 of at least 0.005:" f" {d['D_B2_starts_at_least_size']}; both: {d['both_starts_at_least_size']}. Group-moments holding a rhythm in" f" arm F' but not in arm F: {d['rhythm_at_0_4_not_0_6']} of {d['group_moments']}; the reverse:" f" {d['rhythm_at_0_6_not_0_4']}.", f" B2 above zero at 0.4: {d['B2_above_zero']['f04']} of {d['starts']} starts (at 0.6: see section 5).", "", "7. The pilot's sealed readings, for reference only (fresh starts 135 to 146, exploratory; quoted from" " pilot_results.json as change-2026-10-15-8 sealed it; never pooled with these, never recomputed)", f" At 0.6: B1 {pm(pilot['f06']['B1'])}: {show(pilot['f06']['labels']['B1'], 'null')}" f"{note(pilot['f06']['notes']['B1'])}; B2 {pm(pilot['f06']['B2'], 4)}:" f" {show(pilot['f06']['labels']['B2'], 'null')}{note(pilot['f06']['notes']['B2'])}; E-any" f" {pm(pilot['f06']['E_any'])}: {show(pilot['f06']['labels']['E_any'], 'e06')}; combined reading" f" {pilot['f06']['combined']}. {pilot['f06']['combined_name']}.", f" At 0.4: B2 {pm(pilot['f04']['B2'], 4)}: {show(pilot['f04']['labels']['B2'], 'up')}; E-pair" f" {pm(pilot['f04']['E_pair'])}: {pilot['f04']['labels']['E_pair']}; E-any {pm(pilot['f04']['E_any'])}:" f" {pilot['f04']['labels']['E_any']}; the median |Omega*| over {pilot['f04']['rhythm_directions']} directions" f" {sp.num(pilot['f04']['rate'], 3)}; combined reading {pilot['f04']['combined']}." f" {pilot['f04']['combined_name']}."] if notes: lines += ["", "8. Notes"] + [f" {line}" for line in notes["lines"]] return lines def run_notes(numbers, log): n = len(numbers) same = sum(x["untouched_same_as_base"] for x in numbers) state = sum(x["state_R_same"] for x in numbers) restored = sum(x["restored"] for x in numbers) readings = sum(x["readings"] for x in numbers) largest = max(x["base_largest"] for x in numbers) saved = sum(p.stat().st_size for p in PARTS.glob("*.npz")) seconds = sum(p.get("seconds", 0) for p in log["pieces"]) keys = list(versions()) ran_under = sorted({tuple(p.get(k) for k in keys) for p in log["pieces"]}, key=str) same_versions = ran_under == [tuple(versions().values())] lock_unchanged = not pc.lock_changes() lines = [ f"The untouched rerun against the base run: coherence and gap growth identical to the last digit at {same} of" f" {n} moments (largest coherence difference {largest:.1e}); its first copy's state at R identical at {state} of" f" {n}.", f"Every arm on its scheduled pull and rates, and back on the original ones after R: {restored} of {n} moments." f" Every reading present and finite, the sums adding up, the rule running: {readings} of {n}.", "The pilot's starts aren't in sections 1 to 6: its sealed numbers are quoted in section 7 only.", f"Saved: {saved / 1e6:.1f} MB in generated-artifacts/scattering-side/replication/ (24 base runs and 240" " moments).", f"Compute: {len(log['pieces'])} pieces, {seconds / 60:.1f} minutes of foreground time ({WORKERS} workers at a" f" time). Machine: {log.get('machine', {}).get('cpu', 'unrecorded')}.", f"Every piece ran under this lock, the final plan, both codes, the pilot lock, plan, lock and amendment:" f" {'yes' if same_versions else 'NO'}.", f"The cause test's locked files unchanged: {lock_unchanged}.", f"Checks: replication_checks.json SHA-256 {sha256(CHECKS_JSON)}. Go: replication_go.json SHA-256 {sha256(GO)}" f" (sealed by {go_seal() or 'no record'}).", "Moments: " + "; ".join(f"start {s + 1}: T = " + ", ".join(str(t) for t in mt.moment_times(s)) for s in REP_SEEDS) + "."] return dict(lines=lines, untouched_same_as_base=same, state_R_same=state, restored=restored, readings=readings, largest_difference=largest, saved_bytes=saved, seconds=seconds, pieces=len(log["pieces"]), pieces_ran_under=[dict(zip(keys, v)) for v in ran_under], same_versions=same_versions, cause_test_lock_unchanged=lock_unchanged, machine=log.get("machine")) # --------------------------------------------------------------------------------------------------------------------- # R1 and R2: the earlier checks again, and a pilot moment bit for bit (each R2-style rerun in a worker) def same_files(a, b): """Bit for bit: the same array names, and for each array the same type, shape and bytes.""" with np.load(a) as x, np.load(b) as y: names = sorted(x.files) == sorted(y.files) both = sorted(set(x.files) & set(y.files)) differ = [k for k in both if not (x[k].dtype == y[k].dtype and x[k].shape == y[k].shape and x[k].tobytes() == y[k].tobytes())] return dict(names=names, arrays=len(both), differ=differ, same=names and not differ) def rerun_task(which): """One moment rerun by scattering_pilot.moment_task into a temporary folder (deleted after), compared bit for bit with its saved file: 'pilot' (R2, from the pilot's saved base run) or 'replication' (R7).""" clock = time.time() c, folder = (PILOT_CHECK, sp.PARTS) if which == "pilot" else (REP_CHECK, PARTS) tmp = Path(tempfile.mkdtemp(prefix=f"scattering-replication-{which}-")) try: name = f"moment_{c['start']:03d}_{c['moment'] + 1:02d}" part, problem = sp.moment_task(dict(name=name, seed=c["seed"], moment=c["moment"], parts=str(tmp), base=str(folder / f"base_{c['start']:03d}.npz"))) moment_seconds = time.time() - clock out = dict(name=name, problem=problem, moment_seconds=moment_seconds, error="", **same_files(part, folder / f"{name}.npz")) except Exception: # noqa: BLE001 (reported in the record, never hidden) out = dict(error=traceback.format_exc(limit=6)) finally: shutil.rmtree(tmp, ignore_errors=True) out["seconds"] = time.time() - clock return out def pilot_check_task(which): """In a worker: the pilot's P2 ('p2') or the R2 rerun ('pilot').""" return sp.moment_check_task(None) if which == "p2" else rerun_task(which) def p2_passed(p2): """P2's pass condition, as scattering_pilot.check states it.""" return bool(not p2["error"] and all(p2["untouched"].values()) and all(p2["raised"].values()) and p2["state_R_same"] and all(p2["restored"].values()) and all(p2["readings"].values()) and all(p2["S_read_same"].values()) and p2["numbers"] and p2["keys"] and not p2["problem"] and p2["T"] == p2["saved_T"] == sp.CHECK_T and p2["saved_size"] == ss.SIZE) def raw_fingerprints(): out = {} for line in PILOT_RAW.read_text().splitlines(): if line.strip(): digest, path = line.split(" ", 1) out[path] = digest return out def step1_checks(): """Step 1's 11 code checks: scattering_side.py's own check, its record written to a temporary folder (step 1's checks.txt and checks.json, and the pilot's records, stay as they are).""" tmp = Path(tempfile.mkdtemp(prefix="scattering-replication-step1-")) try: ss.CHECKS_TXT, ss.CHECKS_JSON = tmp / "checks.txt", tmp / "checks.json" ss.check() return json.loads(ss.CHECKS_JSON.read_text()), ss.CHECKS_TXT.read_text() finally: ss.CHECKS_TXT, ss.CHECKS_JSON = HERE / "checks.txt", HERE / "checks.json" shutil.rmtree(tmp, ignore_errors=True) # --------------------------------------------------------------------------------------------------------------------- # R3: the replication's analysis path reproduces the pilot's sealed analysis def leaves(x, path=""): """Every number, label and flag in a json value, with its place.""" if isinstance(x, dict): return [leaf for k in sorted(x) for leaf in leaves(x[k], f"{path}/{k}")] if isinstance(x, list): return [leaf for i, v in enumerate(x) for leaf in leaves(v, f"{path}/{i}")] return [(path, x)] def r3(): raw = raw_fingerprints() names = [f"moment_{s + 1:03d}_{m + 1:02d}" for s in PILOT_SEEDS for m in range(mt.MOMENTS)] matched = [name for name in names if raw.get(rel(sp.PARTS / f"{name}.npz")) == sha256(sp.PARTS / f"{name}.npz")] details = [f"the pilot's saved moments matching raw-data-2026-10-15-8.sha256: {len(matched)} of {len(names)}"] if len(matched) != len(names): return False, details numbers = [] for name in names: with np.load(sp.PARTS / f"{name}.npz") as p: numbers.append(sp.moment_numbers(p)) res = analysis(numbers, PILOT_SEEDS) saved = json.loads(sp.RESULTS_JSON.read_text()) ok = sp.PILOT_SEEDS == PILOT_SEEDS for key in ("levels", "descriptions"): mine, sealed = leaves(json.loads(json.dumps(plain(res[key])))), leaves(saved[key]) places_same = [p for p, _ in mine] == [p for p, _ in sealed] differ = sum(a != b for (_, a), (_, b) in zip(mine, sealed)) if places_same else len(mine) ok = ok and places_same and differ == 0 details.append(f"'{key}': {len(mine)} numbers, labels and flags in the same places as pilot_results.json's" f" {len(sealed)}: {places_same}; differing: {differ}") details.append(f"the pilot's start list as it was after the call: {sp.PILOT_SEEDS == PILOT_SEEDS}. Only the" " analysis the pilot had was run; no value is shown") return ok, details # --------------------------------------------------------------------------------------------------------------------- # R4: the new readings and the three questions on made-up numbers (fixtures fixed with this code; no model is run) BAND_CASES = ( # (rate, bar, reading) (0.962, 0.001, "inside"), (0.95, 0.0, "inside"), (0.9301, 0.0, "neither"), (0.9941, 0.0, "neither"), (0.932101, 0.001, "inside"), (0.992099, 0.001, "inside"), (0.962, 0.02, "neither"), (0.92, 0.003, "clearly outside"), (1.0, 0.001, "clearly outside"), (1.05, 0.001, "clearly outside"), (0.929, 0.0005, "neither"), (0.962, NAN, "neither"), (NAN, NAN, "neither")) BELOW_CASES = ((-0.004, 0.001, True), (-1.5, 0.5, False), (-1.6, 0.5, True), (0.0, 0.0, False), (-0.1, 0.0, True), (-1.0, NAN, False), (0.004, 0.001, False)) EXACT_BAR, EXACT_EDGE = 0.0009765625, 0.0029296875 # 2**-10 and 3 x 2**-10, exact in binary Q1_CASES = ( # (combined at 0.4, E-pair, E-any, band) and the verdict ((1, "as expected", "as expected", "inside"), "PASS"), ((1, "as expected", "contradicts", "inside"), "INCONCLUSIVE"), ((1, "as expected", "as expected", "neither"), "INCONCLUSIVE"), ((1, "as expected", "as expected", "clearly outside"), "FAIL"), ((2, "as expected", "as expected", "inside"), "INCONCLUSIVE"), ((3, "contradicts", "as expected", "inside"), "FAIL"), ((5, "contradicts", "contradicts", "neither"), "FAIL"), ((6, "unresolved", "unresolved", "inside"), "INCONCLUSIVE"), ((6, "unresolved", "contradicts", "neither"), "FAIL"), ((1, "as expected", "contradicts", "clearly outside"), "FAIL")) Q2_CASES = ( # ((B2 at 0.6, its bar), E-any at 0.6, E-steady inside) and the verdict (((0.006, 0.001), "as expected", True), "PASS"), (((0.003, EXACT_BAR), "as expected", True), "PASS"), (((EXACT_EDGE, EXACT_BAR), "as expected", True), "FAIL"), (((0.0, 0.001), "as expected", True), "FAIL"), (((-0.001, 0.001), "as expected", True), "FAIL"), (((-0.004, 0.001), "as expected", True), "FAIL"), (((0.006, 0.001), "contradicts", True), "FAIL"), (((0.006, 0.001), "unresolved", True), "INCONCLUSIVE"), (((0.006, 0.001), "as expected", False), "INCONCLUSIVE"), (((0.004, 0.002), "as expected", True), "INCONCLUSIVE"), (((0.006, NAN), "as expected", True), "INCONCLUSIVE")) Q3_CASES = ( # ((D-any, its bar), (D-B2, its bar)) and the verdict (((0.8, 0.02), (0.044, 0.001)), "PASS"), (((0.1, 0.02), (0.005, 0.001)), "PASS"), (((0.09, 0.02), (0.044, 0.001)), "INCONCLUSIVE"), (((0.8, 0.02), (0.0, 0.001)), "INCONCLUSIVE"), (((0.0, 0.0), (0.0, 0.001)), "FAIL"), (((0.8, 0.02), (-0.006, 0.001)), "FAIL"), (((-0.2, 0.02), (0.044, 0.001)), "FAIL"), (((0.03, 0.03), (0.003, 0.002)), "INCONCLUSIVE"), (((0.8, NAN), (0.044, 0.001)), "INCONCLUSIVE")) PAIR_04 = dict(rhythm="pair", share=0.8, rate=0.962, b2=0.05) R4_SCENARIOS = ( # name, untouched's, arm F's and arm F''s make-up; built for: combined (0.6, 0.4); verdicts (1, 2, 3) ("as the pilot showed", {}, dict(b2=0.006), PAIR_04, (2, 1), ("PASS", "PASS", "PASS")), ("the null at 0.6", {}, {}, PAIR_04, (1, 1), ("PASS", "FAIL", "PASS")), ("no rhythm at 0.4", {}, dict(b2=0.006), {}, (2, 5), ("FAIL", "PASS", "FAIL")), ("a rate clearly outside the band, within 0.90 to 1.10", {}, dict(b2=0.006), dict(PAIR_04, rate=1.05), (2, 1), ("FAIL", "PASS", "PASS")), ("PASS and FAIL meeting in Question 1", dict(rhythm="forward", share=1.0, rate=0.962), dict(b2=0.006), dict(PAIR_04, share=1.0), (4, 1), ("INCONCLUSIVE", "INCONCLUSIVE", "PASS")), ("B2 at 0.6 neither clear nor inside", {}, dict(b2=0.003, b2_noise=0.065), PAIR_04, (4, 1), ("PASS", "INCONCLUSIVE", "PASS")), ("edge synchronization at 0.6", {}, dict(rhythm="pair", share=0.3, rate=0.962, b2=0.006), PAIR_04, (3, 1), ("PASS", "FAIL", "PASS")), ("separated by the rhythm but not by B2", {}, dict(b2=0.05), PAIR_04, (2, 1), ("PASS", "PASS", "INCONCLUSIVE")), ("a rhythm at one start only: the rate's bar undefined", {}, dict(b2=0.006), dict(PAIR_04, only=1), (2, 6), ("INCONCLUSIVE", "PASS", "INCONCLUSIVE")), ) def made_up_numbers(index): """One made-up replication: 24 starts x 10 moments of the numbers the report reads, built for known readings (as scattering_pilot.made_up_numbers builds its pilots; 'only': the make-up holds at the first starts only).""" _, untouched, f06, f04, _, _ = R4_SCENARIOS[index] rng = np.random.default_rng((ss.DRAW_SEED, MADE_UP_STREAM, index)) specs = dict(untouched=untouched, raised=dict(gain=0.02), f06=f06, f04=f04) numbers = [] for i, start in enumerate(REP_STARTS): for m in range(mt.MOMENTS): C0, outer0 = 0.30 + 0.01 * rng.standard_normal(), 0.02 + 0.002 * rng.standard_normal(2) arms = {} for a in ARMS: spec, arm, changed = specs[a], sp.blank_arm(), a != "untouched" here = i < spec.get("only", len(REP_STARTS)) b2 = spec.get("b2", 0.0) if here else 0.0 arm["C_R"] = C0 + (spec.get("gain", 0.0) + 0.002 * rng.standard_normal() if changed else 0.0) arm["outer_R"] = outer0 + (b2 + spec.get("b2_noise", 0.0005) * rng.standard_normal(2) if changed else 0.0) arm["inner_R"] = np.full(2, 0.5) arm["R_R"] = np.full(2, arm["C_R"]) arm["contrib_R"] = arm["R_R"][:, None] * np.array([0.8, 0.15, 0.05]) arm["C_blocks"] = np.full(sp.UNITS // sp.C_BLOCK, arm["C_R"]) arm["C_later"] = C0 + 0.001 * rng.standard_normal() arm["growth_later"] = 0.02 + 0.001 * rng.standard_normal() arm["outer_blocks"] = np.full(W // sp.RULE_BLOCK, arm["outer_R"].mean()) if spec.get("rhythm") and here: for g in range(2): if rng.random() < spec["share"]: for d in ((0, 1) if spec["rhythm"] == "pair" else (0,)): om = (1.0 if d == 0 else -1.0) * (spec["rate"] + 0.005 * rng.standard_normal()) arm["rule"]["rhythm"][g, d] = arm["rule"]["steady"][g, d] = True arm["rule"]["omega"][g, d], arm["rule"]["S"][g, d], arm["rule"]["L"][g, d] = om, 0.1, 3.0 arm["rule"]["count"][g, d], arm["rule"]["natural_most"][g, d] = 100.0, 33.0 arm["S_blocks"][g, d] = np.linspace(0.02, 0.1, sp.BLOCKS) arm["turning_blocks"][g, d] = abs(om) arm["both"] = arm["rule"]["rhythm"].all(0) arm["in_step"] = arm["both"].copy() arm["rate_gap"] = np.where(arm["both"], 0.01, np.nan) arm["turn_gap"] = np.where(arm["both"], 0.1, np.nan) arms[a] = sp.with_flags(arm) numbers.append(dict(start=start, moment=m + 1, T=0, untouched_same_as_base=True, base_largest=0.0, state_R_same=True, restored=True, readings=True, delivered=np.full(2, 0.6), arms=arms)) return numbers def made_up_direct(numbers): """B1, B2, the E-readings, the paired readings, the rate and its bar, computed straight from made-up numbers with plain Python (statistics.median and statistics.stdev), not through the replication's functions.""" def per(fn): return [statistics.fmean([fn(n) for n in numbers if n["start"] == s]) for s in REP_STARTS] def mean(values): return statistics.fmean(values) def share(a, key, n): rule = n["arms"][a]["rule"] flags = rule["steady"].any(-1) if key == "steady" else rule["rhythm"].all(-1) if key == "pair" \ else rule["rhythm"].any(-1) return float(flags.mean()) none, out = "untouched", {} for a in ("f06", "f04"): g = per(lambda n, a=a: n["arms"][a]["C_R"] - n["arms"][none]["C_R"]) r = per(lambda n: n["arms"]["raised"]["C_R"] - n["arms"][none]["C_R"]) out[f"{a} B1"] = mean(g) / mean(r) out[f"{a} B2"] = mean(per(lambda n, a=a: float(n["arms"][a]["outer_R"].mean()) - float(n["arms"][none]["outer_R"].mean()))) for key in ("pair", "any", "steady"): out[f"{a} E-{key}"] = mean(per(lambda n, a=a, key=key: share(a, key, n) - share(none, key, n))) for key in ("pair", "any", "steady"): out[f"D-{key}"] = mean(per(lambda n, key=key: share("f04", key, n) - share("f06", key, n))) out["D-B2"] = mean(per(lambda n: float(n["arms"]["f04"]["outer_R"].mean()) - float(n["arms"]["f06"]["outer_R"].mean()))) omegas = [[abs(float(x)) for n in numbers if n["start"] == s for x in n["arms"]["f04"]["rule"]["omega"][n["arms"]["f04"]["rule"]["rhythm"]]] for s in REP_STARTS] pooled = [x for o in omegas for x in o] out["rate"] = statistics.median(pooled) if pooled else NAN medians = [] for row in np.random.default_rng(mt.RESAMPLE_SEED).integers(0, len(REP_STARTS), (mt.RESAMPLES, len(REP_STARTS))): drawn = [x for i in row for x in omegas[int(i)]] if not drawn: medians = None break medians.append(statistics.median(drawn)) out["rate bar"] = statistics.stdev(medians) if medians else NAN return out def agree(a, b): return bool((np.isnan(a) and np.isnan(b)) or abs(a - b) <= AGREEMENT) def r4(): details, wrong = [], [] for rate, bar, want in BAND_CASES: if band_reading(rate, bar) != want: wrong.append(f"band, {rate} ± {bar}: {band_reading(rate, bar)}, not {want}") for m, bar, want in BELOW_CASES: if below(dict(mean=m, bar=bar)) != want: wrong.append(f"below, {m} ± {bar}: {not want}, not {want}") spread = np.random.default_rng((ss.DRAW_SEED, MADE_UP_STREAM, 100)) varied = [list(0.9 + 0.1 * spread.random(int(spread.integers(1, 6)))) for _ in REP_STARTS] rate_cases = (("every start at 0.96", [[0.96, 0.96]] * 24, 0.96, 0.0), ("one start of 24 holding directions", [[0.96]] + [[]] * 23, 0.96, NAN), ("no start holding a direction", [[]] * 24, NAN, NAN), ("varied directions, against a direct loop", varied, None, None)) for label, per_start, want_rate, want_bar in rate_cases: rate, bar = rate_and_bar(per_start) if want_rate is None: pooled = [x for o in per_start for x in o] want_rate = statistics.median(pooled) draws = np.random.default_rng(mt.RESAMPLE_SEED).integers(0, 24, (400, 24)) want_bar = statistics.stdev([statistics.median([x for i in row for x in per_start[int(i)]]) for row in draws]) if not (agree(rate, want_rate) and agree(bar, want_bar)): wrong.append(f"the rate and its bar, {label}: {rate} ± {bar}, not {want_rate} ± {want_bar}") for args, want in Q1_CASES: if question_1(*args)["verdict"] != want: wrong.append(f"Question 1 for {args}: {question_1(*args)['verdict']}, not {want}") meet = question_1(1, "as expected", "contradicts", "inside") if not (meet["passes"] and meet["fails"] and "both hold" in meet["why"]): wrong.append("Question 1: PASS and FAIL meeting is not named as both") for ((m, bar), e_any, steady), want in Q2_CASES: got = question_2(dict(mean=m, bar=bar), e_any, steady)["verdict"] if got != want: wrong.append(f"Question 2 for B2 {m} ± {bar}, E-any {e_any}, E-steady inside {steady}: {got}, not {want}") for ((ma, ba), (mb, bb)), want in Q3_CASES: got = question_3(dict(mean=ma, bar=ba), dict(mean=mb, bar=bb))["verdict"] if got != want: wrong.append(f"Question 3 for D-any {ma} ± {ba}, D-B2 {mb} ± {bb}: {got}, not {want}") cases = len(BAND_CASES) + len(BELOW_CASES) + len(rate_cases) + len(Q1_CASES) + 1 + len(Q2_CASES) + len(Q3_CASES) details.append(f"boundary cases: {cases - len(wrong)} of {cases} as fixed ({len(BAND_CASES)} for the band," f" {len(BELOW_CASES)} for 'below', {len(rate_cases)} for the rate and its bar, {len(Q1_CASES) + 1} for" f" Question 1 (its meeting named), {len(Q2_CASES)} for Question 2, {len(Q3_CASES)} for Question 3)") details += wrong ok = not wrong pilot = pilot_quoted() for index, (name, _, _, _, want_c, want_q) in enumerate(R4_SCENARIOS): numbers = made_up_numbers(index) res = analysis(numbers, REP_SEEDS) q = questions(numbers, res) got_c = (res["levels"]["f06"]["combined"], res["levels"]["f04"]["combined"]) got_q = (q["q1"]["verdict"], q["q2"]["verdict"], q["q3"]["verdict"]) direct = made_up_direct(numbers) found = {"rate": q["rate"], "rate bar": q["rate_bar"], "D-B2": q["D"]["B2"]["mean"]} for a in ("f06", "f04"): L = res["levels"][a] found.update({f"{a} B1": L["B1"]["mean"], f"{a} B2": L["B2"]["mean"]}) found.update({f"{a} E-{k}": L["E"][k]["mean"] for k in ("pair", "any", "steady")}) found.update({f"D-{k}": q["D"][k]["mean"] for k in ("pair", "any", "steady")}) values_ok = sorted(found) == sorted(direct) and all(agree(found[k], direct[k]) for k in direct) rate_same = agree(q["rate"], res["levels"]["f04"]["rate"]) counted = res["starts"] == list(REP_STARTS) and sp.PILOT_SEEDS == PILOT_SEEDS lines = report_lines(res, q, pilot) written = len(lines) > 40 and all(any(line.startswith(t) for line in lines) for t in ( "1. Comparison B ", "2. Comparison B' ", "3. Described", "4. Question 1.", "5. Question 2 (a hypothesis set after the pilot).", "6. Question 3.", "7. The pilot's sealed readings")) passed = got_c == want_c and got_q == want_q and values_ok and rate_same and counted and written ok = ok and passed details.append(f"{name}: built for combined readings {want_c[0]} at 0.6 and {want_c[1]} at 0.4 and verdicts" f" {', '.join(want_q)}; found {got_c[0]} and {got_c[1]}, {', '.join(got_q)}; B1, B2, the" f" E-readings, the paired readings, the rate and its bar as computed directly (within 1e-12):" f" {values_ok and rate_same}; 24 starts counted, the pilot's list restored: {counted}; report" f" written: {written}: {'as built' if passed else 'NOT as built'}") return ok, details # --------------------------------------------------------------------------------------------------------------------- # the checks before the run def check(): if CHECKS_JSON.exists() or CHECKS_TXT.exists(): sys.exit("The replication's check record exists already; it is never written over. Nothing was run.") clock_all = time.time() lines = [f"Replication checks (step 4, before any fresh start), {shown(stamp())}. Equality and mechanics only:" " nothing runs on a fresh start, and no new number about the pilot is made. Replication lock:" " replication_lock.json; plan: replication-plan-final.md.", ""] results = {} def note(number, title, passed, details): results[number] = dict(number=number, title=title, passed=bool(passed), details=details) wrong = lock_problems() + environment_problems() note("R0", "The sealed files, the replication lock and its settings match, in the pilot's numerical environment", not wrong, wrong or [ "the final plan, draft 1, plan-final.md, lock.json, the amendment, pilot_lock.json, both codes and the" " pilot's 7 records are the versions the replication lock names; the lock and the final plan are the" " versions change-2026-10-15-10 sealed; scattering_pilot.pilot_problems finds nothing", "the code's settings equal the lock's after a JSON round trip; the lock's settings equal pilot_lock.json's" " except the starts and the count of moments, with the questions' settings and the checks' moments added;" " the fresh starts' moments are the lock's", "the numerical environment matches the pilot's checks and run log"]) r5_ok, r5_details = r5() note("R5", "Fresh starts 147 to 170 appear in no earlier record, and nothing has run on them", r5_ok, r5_details) step1, info = None, {} if not wrong and r5_ok: step1, step1_text = step1_checks() raw = raw_fingerprints() pilot_files = [sp.PARTS / f"base_{PILOT_CHECK['start']:03d}.npz", sp.PARTS / f"moment_{PILOT_CHECK['start']:03d}_{PILOT_CHECK['moment'] + 1:02d}.npz"] files_match = all(raw.get(rel(p)) == sha256(p) for p in pilot_files) with Pool(2) as pool: waiting = {which: pool.apply_async(pilot_check_task, (which,)) for which in ("p2", "pilot")} p3, p3_details = sp.made_up_check() # meanwhile, in this process r3_ok, r3_details = r3() r4_ok, r4_details = r4() p2, r2 = waiting["p2"].get(), waiting["pilot"].get() passed = sum(r["passed"] for r in step1["results"]) p2_ok = p2_passed(p2) p2_text = (f"failed to run: {p2['error'].strip().splitlines()[-1]}" if p2["error"] else f"the untouched rerun identical to the base run {all(p2['untouched'].values())}; the raised pull" f" identical to the medium pilot's saved arrays {all(p2['raised'].values())}; the state at R" f" {p2['state_R_same']}; every arm on its schedule {all(p2['restored'].values())}; every reading" f" {all(p2['readings'].values())}; S over (150, 200] {all(p2['S_read_same'].values())}; the report's" f" numbers {p2['numbers']}; the saved arrays {p2['keys']}; T {p2['T']}; problems:" f" {p2['problem'] or 'none'}") note("R1", "Step 1's 11 code checks and the pilot's checks P2 and P3 pass again, writing nothing to their files", step1["all_passed"] and p2_ok and p3, [f"step 1: {passed} of {len(step1['results'])} passed (scattering_side.py's own check, its record written" " to a temporary folder since deleted; step 1's checks.txt and checks.json are unchanged). Its own text:"] + [f"| {line}" if line else "|" for line in step1_text.rstrip("\n").split("\n")] + [f"P2, the pilot's moment code on the medium pilot's start 103, moment 1 (T = 2,055), into a temporary" f" folder since deleted: {'PASSED' if p2_ok else 'FAILED'}: {p2_text}. Mechanics only: no value of" " arms F and F' was shown or kept", f"P3, the pilot's labels and combined readings on its made-up numbers: {'PASSED' if p3 else 'FAILED'}"] + [f" {line}" for line in p3_details]) r2_ok = bool(files_match and not r2["error"] and r2["same"] and not r2["problem"]) note("R2", "The pilot's start 135, moment 1 (T = 2,285), rerun from its saved base run into a temporary folder" " since deleted, equals its saved moment bit for bit", r2_ok, [f"the saved base run and moment match raw-data-2026-10-15-8.sha256: {files_match}", (f"failed to run: {r2['error'].strip().splitlines()[-1]}" if r2["error"] else f"the same array names: {r2['names']}; arrays compared (type, shape and bytes): {r2['arrays']}; differing:" f" {len(r2['differ'])}{' (' + ', '.join(r2['differ'][:5]) + ')' if r2['differ'] else ''}; mechanical" f" problems: {r2['problem'] or 'none'}")]) note("R3", "The replication's analysis path reproduces the pilot's sealed analysis exactly from its 120 saved" " moments (counts only)", r3_ok, r3_details) note("R4", "The new readings and the three questions' rules on made-up numbers, no model run", r4_ok, r4_details) if not r2["error"]: per_moment = r2["moment_seconds"] per_worker = int(START_BY // per_moment) + 1 pieces = -(-ALL_MOMENTS // (per_worker * WORKERS)) info = dict(moment_seconds_one_worker=round(per_moment, 1), moments_per_worker_per_piece=per_worker, projected_moment_pieces=pieces, projected_piece_seconds=round(per_worker * per_moment)) order = ["R0", "R1", "R2", "R3", "R4", "R5"] for number in order: r = results.get(number) if r is None: lines.append(f"{number}: NOT RUN (R0 or R5 failed)") continue lines.append(f"{number}. {r['title']}: {'PASSED' if r['passed'] else 'FAILED'}") lines += [f" {line}" for line in (r["details"] if isinstance(r["details"], list) else [r["details"]])] all_passed = all(number in results and results[number]["passed"] for number in order) if info: lines += ["", "Reported only (not a check):", f" One moment (four arms, 2 copies x 700 units, every reading) took {info['moment_seconds_one_worker']:.0f}" f" s on one worker beside another. At that pace each worker begins {info['moments_per_worker_per_piece']}" f" moments a piece (none after {START_BY:.0f} s), so the 240 moments take about" f" {info['projected_moment_pieces']} pieces, under the {HARD_LIMIT} s limit; with 7 side by side each" " moment may take longer, as in the pilot."] lines += ["", f"All checks passed: {'yes' if all_passed else 'NO'}", "Run under " + ", ".join(f"{k.replace('_sha256', '')} SHA-256 {v}" for k, v in versions().items()) + f"; {time.time() - clock_all:.0f} s in all."] record = dict(made=stamp(), label="check", step=4, **versions(), settings=REP_SETTINGS, results=[results[n] for n in order if n in results], step1_record=step1, information=info, all_passed=all_passed, machine=pc.machine(), seconds=round(time.time() - clock_all, 1)) write_json(CHECKS_JSON, record) CHECKS_TXT.write_text("\n".join(lines) + "\n") print("\n".join(lines)) # --------------------------------------------------------------------------------------------------------------------- # the run (each command only after the checks passed and Colson's separate authorization) def do_task(task, deadline): if Path(task["parts"]) != PARTS or task["seed"] not in REP_SEEDS: raise ValueError("the run takes fresh starts 147 to 170 only, into the replication's folder") PARTS.mkdir(parents=True, exist_ok=True) if task["kind"] == "base": return mt.base_task(task, deadline, PARTS) part, problem = sp.moment_task(task) return dict(status="done", sha256=sha256(part), error=problem) def job(args): return pc.guard(do_task, args) def base_tasks(): return [dict(name=f"base_{s + 1:03d}", kind="base", seed=s, keep=mt.keep_times(s), logs=True, parts=str(PARTS)) for s in REP_SEEDS] def moment_tasks(): return [dict(name=f"moment_{s + 1:03d}_{m + 1:02d}", kind="moment", seed=s, moment=m, parts=str(PARTS), base=str(PARTS / f"base_{s + 1:03d}.npz")) for s in REP_SEEDS for m in range(mt.MOMENTS)] def piece(tasks): PARTS.mkdir(parents=True, exist_ok=True) pc.run_piece(LOG, tasks, job, BUDGET, START_BY) log = json.loads(LOG.read_text()) log["pieces"][-1].update(**versions(), checks_sha256=sha256(CHECKS_JSON), go_sha256=sha256(GO), start_by=START_BY, budget=BUDGET) write_json(LOG, log) def base_command(): locked() require_checks() require_go() todo = mt.unfinished(base_tasks()) if not todo: print("All 24 base runs are done: run moments.") return piece(todo) left = mt.unfinished(base_tasks()) print("All 24 base runs are done: run moments." if not left else f"Not finished: {', '.join(t['name'] for t in left)}; their exact states are kept. Run base again.") def moments_command(): clock = time.time() locked() require_checks() require_go() if mt.unfinished(base_tasks()): sys.exit("The base runs aren't finished: run base first.") if RESULTS_JSON.exists(): sys.exit("The replication's report exists already: it is made once. Nothing was run.") todo = mt.unfinished(moment_tasks()) if todo: piece(todo) left = mt.unfinished(moment_tasks()) if left: print(f"{len(left)} of 240 moments unfinished: run moments again.") return if time.time() - clock > START_BY: print("All 240 moments are done. Run moments again to make the report (too little time is left in this" " command).") return report() def report(): numbers = [] for t in moment_tasks(): with np.load(Path(t["parts"]) / f"{t['name']}.npz") as p: numbers.append(sp.moment_numbers(p)) if len(numbers) != ALL_MOMENTS or sorted({n["start"] for n in numbers}) != list(REP_STARTS): sys.exit("The report reads fresh starts 147 to 170 only, with all 240 moments. Stopping.") res = analysis(numbers, REP_SEEDS) q = questions(numbers, res) pilot = pilot_quoted() notes = run_notes(numbers, json.loads(LOG.read_text())) lines = report_lines(res, q, pilot, notes) RESULTS_TXT.write_text("\n".join(lines) + "\n") write_json(RESULTS_JSON, dict(made=stamp(), label="replication", step=4, prospective=True, **versions(), checks_sha256=sha256(CHECKS_JSON), go_sha256=sha256(GO), moments={str(s + 1): mt.moment_times(s) for s in REP_SEEDS}, levels=res["levels"], descriptions=res["descriptions"], questions=q, question_2_label="a hypothesis set after the pilot", pilot_quoted=pilot, notes={k: v for k, v in notes.items() if k != "lines"})) print("\n".join(lines)) # --------------------------------------------------------------------------------------------------------------------- # after the report: R6 to R8 (read-only, except a temporary folder) def moment_readings(p): """R6: one saved moment's readings, straight from its arrays (p: a mapping of them), without scattering_pilot.moment_numbers: the two groups' coherence at R from z; the faster half's own coherence at R from the arm's own sums over it (saved every unit); each group's pair, any and steady, and |Omega*|, from the saved rule arrays.""" out = {} for a in ARMS: rhythm, steady = np.asarray(p[f"rule_{a}_rhythm"], bool), np.asarray(p[f"rule_{a}_steady"], bool) out[a] = dict(C=float(np.abs(p[f"z_{a}"][:, W]).mean()), outer=float((np.abs(p[f"parts_{a}"][W, :, 1]) / sp.OUTER_COUNT).mean()), pair=rhythm.all(-1), any=rhythm.any(-1), steady=steady.any(-1), omegas=[abs(float(x)) for x in np.asarray(p[f"rule_{a}_omega"])[rhythm]]) return out def recomputed(readings): """R6: the labelled numbers from readings[start][moment] (each a moment_readings), without scattering_pilot.analyse: plain means per start, medium_test's uncertainty rules (summary, ratio_summary), and the median |Omega*| and its resampled bar by a direct loop (statistics.median and statistics.stdev).""" n_s = len(readings) def grid(a, key): # (start, moment) return np.array([[m[a][key] for m in start] for start in readings], float) def flags(a, key): # (start, moment, group) return np.array([[m[a][key] for m in start] for start in readings], bool) none, out = "untouched", {} for a in ("f06", "f04"): gain = (grid(a, "C") - grid(none, "C")).mean(axis=1) raised = (grid("raised", "C") - grid(none, "C")).mean(axis=1) E = {k: mt.summary(flags(a, k).mean(axis=(1, 2)) - flags(none, k).mean(axis=(1, 2))) for k in ("pair", "any", "steady")} out[a] = dict(B1=mt.ratio_summary(gain, raised), B2=mt.summary((grid(a, "outer") - grid(none, "outer")).mean(axis=1)), E=E) D = {k: mt.summary(flags("f04", k).mean(axis=(1, 2)) - flags("f06", k).mean(axis=(1, 2))) for k in ("pair", "any", "steady")} D["B2"] = mt.summary((grid("f04", "outer") - grid("f06", "outer")).mean(axis=1)) omegas = [[x for m in start for x in m["f04"]["omegas"]] for start in readings] pooled = [x for o in omegas for x in o] rate = statistics.median(pooled) if pooled else NAN medians = [] for row in np.random.default_rng(mt.RESAMPLE_SEED).integers(0, n_s, (mt.RESAMPLES, n_s)): drawn = [x for i in row for x in omegas[int(i)]] if not drawn: medians = None break medians.append(statistics.median(drawn)) return dict(levels=out, D=D, rate=rate, rate_bar=statistics.stdev(medians) if medians else NAN, rate_directions=len(pooled)) def recompute(seeds=REP_SEEDS, folder=PARTS): """R6: every labelled number, straight from the replication's saved moments.""" readings = [] for s in seeds: start = [] for m in range(mt.MOMENTS): with np.load(folder / f"moment_{s + 1:03d}_{m + 1:02d}.npz") as p: start.append(moment_readings(p)) readings.append(start) return recomputed(readings) def judged(x): """The same rules applied to recomputed numbers: the labels, combined readings, band and verdicts.""" b, bp = x["levels"]["f06"], x["levels"]["f04"] steady06, steady04 = sp.inside(b["E"]["steady"], sp.E_INSIDE), sp.inside(bp["E"]["steady"], sp.E_INSIDE) l06 = dict(B1=sp.label_null(b["B1"], sp.B1_MARGIN), B2=sp.label_null(b["B2"], sp.B2_MARGIN), E_any=sp.label_e06(b["E"]["any"])) l04 = dict(B2=sp.label_up(bp["B2"], sp.B2_MARGIN, sp.B2_MARGIN), E_pair=sp.label_up(bp["E"]["pair"], sp.E_EXPECTED, sp.E_INSIDE), E_any=sp.label_up(bp["E"]["any"], sp.E_EXPECTED, sp.E_INSIDE)) within = bool(np.isfinite(x["rate"]) and sp.RATE_RANGE[0] <= x["rate"] <= sp.RATE_RANGE[1]) c06 = sp.combined_06(l06["E_any"], steady06, l06["B2"]) c04 = sp.combined_04(l04["E_pair"], l04["E_any"], l04["B2"], steady04, within) band = band_reading(x["rate"], x["rate_bar"]) return dict(labels_06=l06, labels_04=l04, combined=(c06, c04), band=band, verdicts=(question_1(c04, l04["E_pair"], l04["E_any"], band)["verdict"], question_2(b["B2"], l06["E_any"], steady06)["verdict"], question_3(x["D"]["any"], x["D"]["B2"])["verdict"])) def r6(results, x=None): """R6: the recomputation (x: recompute()'s, unless given) agrees with the report: means and bars within 1e-12, and the same labels, combined readings and verdicts.""" x = recompute() if x is None else x levels, q = results["levels"], results["questions"] pairs = [] for a in ("f06", "f04"): mine, rep = x["levels"][a], levels[a] pairs += [(f"{a} {k}", mine[k], unplain(rep[k])) for k in ("B1", "B2")] pairs += [(f"{a} E-{k}", mine["E"][k], unplain(rep["E"][k])) for k in ("pair", "any", "steady")] pairs += [(f"D-{k}", x["D"][k], unplain(q["D"][k])) for k in ("pair", "any", "steady", "B2")] differ = [name for name, mine, rep in pairs if not (agree(mine["mean"], rep["mean"]) and agree(mine["bar"], rep["bar"]))] rate_rep = NAN if q["rate"] is None else q["rate"] bar_rep = NAN if q["rate_bar"] is None else q["rate_bar"] rate_ok = agree(x["rate"], rate_rep) and agree(x["rate_bar"], bar_rep) and x["rate_directions"] == q["rate_directions"] j = judged(x) rep_labels = (levels["f06"]["labels"], levels["f04"]["labels"]) labels_ok = (j["labels_06"] == rep_labels[0] and j["labels_04"] == rep_labels[1] and j["combined"] == (levels["f06"]["combined"], levels["f04"]["combined"]) and j["band"] == q["band"] and j["verdicts"] == (q["q1"]["verdict"], q["q2"]["verdict"], q["q3"]["verdict"])) ok = not differ and rate_ok and labels_ok return ok, [f"means and bars recomputed from the saved arrays: {len(pairs) - len(differ)} of {len(pairs)} agree" f" with the report within 1e-12{' (differing: ' + ', '.join(differ) + ')' if differ else ''}", f"the median |Omega*| over {x['rate_directions']} directions and its resampled bar agree: {rate_ok}", f"the same labels, combined readings ({j['combined'][0]} and {j['combined'][1]}), band ({j['band']}) and" f" verdicts ({', '.join(j['verdicts'])}) follow from the recomputed numbers: {labels_ok}"] def r8(results): """R8: what goes against the theory or fails, flagged for a read-only fault check of the arm and the code.""" levels, q = results["levels"], results["questions"] flags = [f"section 1, {name} at 0.6: contradicts" for name, label in levels["f06"]["labels"].items() if label == "contradicts"] flags += [f"section 2, {name} at 0.4: contradicts" for name, label in levels["f04"]["labels"].items() if label == "contradicts"] flags += [f"Question {i}: FAIL" for i, key in ((1, "q1"), (2, "q2"), (3, "q3")) if q[key]["verdict"] == "FAIL"] return flags def after(): locked() require_checks() require_go() if not RESULTS_JSON.exists(): sys.exit("No report yet: after runs once the report is made.") if AFTER_JSON.exists() or AFTER_TXT.exists(): sys.exit("The after-report record exists already; it is never written over. Nothing was run.") clock_all = time.time() results = json.loads(RESULTS_JSON.read_text()) with Pool(1) as pool: waiting = pool.apply_async(rerun_task, ("replication",)) r6_ok, r6_details = r6(results) # meanwhile, in this process r7 = waiting.get() r7_ok = bool(not r7["error"] and r7["same"] and not r7["problem"]) flags = r8(results) out = [dict(number="R6", title="An independent recomputation from the saved arrays agrees with the report", passed=r6_ok, details=r6_details), dict(number="R7", title="The replication's start 147, moment 1 (T = 2,235), rerun into a temporary folder" " since deleted, equals its saved moment bit for bit", passed=r7_ok, details=[f"failed to run: {r7['error'].strip().splitlines()[-1]}" if r7["error"] else f"the same array names: {r7['names']}; arrays compared (type, shape and bytes): {r7['arrays']};" f" differing: {len(r7['differ'])}; mechanical problems: {r7['problem'] or 'none'}"]), dict(number="R8", title="Anything against the theory, or any FAIL, flagged for a read-only fault check of the" " arm and the code; the report stays as made", passed=True, details=flags or ["nothing to flag"])] all_passed = r6_ok and r7_ok lines = [f"Replication after-report checks (step 4), {shown(stamp())}. Read-only, except a temporary folder. The" " report stays as made.", ""] for r in out: lines.append(f"{r['number']}. {r['title']}: {'PASSED' if r['passed'] else 'FAILED'}" if r["number"] != "R8" else f"{r['number']}. {r['title']}:") lines += [f" {line}" for line in r["details"]] lines += ["", f"R6 and R7 passed: {'yes' if all_passed else 'NO'}", "Run under " + ", ".join(f"{k.replace('_sha256', '')} SHA-256 {v}" for k, v in versions().items()) + f"; results SHA-256 {sha256(RESULTS_JSON)}; {time.time() - clock_all:.0f} s in all."] write_json(AFTER_JSON, dict(made=stamp(), label="after", step=4, **versions(), results_sha256=sha256(RESULTS_JSON), results=out, flagged=flags, all_passed=all_passed, machine=pc.machine(), seconds=round(time.time() - clock_all, 1))) AFTER_TXT.write_text("\n".join(lines) + "\n") print("\n".join(lines)) # --------------------------------------------------------------------------------------------------------------------- if __name__ == "__main__": commands = dict(check=check, base=base_command, moments=moments_command, after=after) if len(sys.argv) != 2 or sys.argv[1] not in commands: sys.exit(__doc__) commands[sys.argv[1]]()