230 lines
11 KiB
Python
230 lines
11 KiB
Python
"""Grok: v0.3 adaptations of TRON test-review additions (2026-10-13), test-only.
|
|
1) SRT NaN-mask equality on coarse and fine.
|
|
2) Corner packets: all 8 unique corner+direction keys; four child packets + 0.25 weights.
|
|
3) coalesce_paths reconstructed test-side (removed from reference); hop2_valid semantics.
|
|
4) Diagnostic field delta vs v0.2.
|
|
"""
|
|
import os, sys, unittest, hashlib, platform, resource, importlib.util
|
|
import numpy as np
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import test_cpu_reference as T
|
|
ref = T.ref
|
|
CX, CY, N, NF, LO, HI, H = T.CX, T.CY, T.N, T.NF, T.LO, T.HI, T.H
|
|
|
|
|
|
def cls_of(a, b):
|
|
if T.f_active(a, b):
|
|
return "A"
|
|
return "I" if T.f_inner(a, b) else "O"
|
|
|
|
|
|
ENC = {"A": ref.ACTIVE, "I": ref.INNER_GHOST, "O": ref.OUTER_GHOST}
|
|
|
|
|
|
def reconstruct_coalesce_paths(maps):
|
|
"""Independent path records formerly returned as diagnostics.coalesce_paths."""
|
|
rows = [tuple(int(v) for v in r) for r in maps.coalesce.tolist()]
|
|
cells = np.zeros((len(rows), 4, 2, 2), dtype=np.int64) # [r,k] -> (intermed yx, origin yx)
|
|
classes = np.zeros((len(rows), 4, 2), dtype=np.int64)
|
|
for r, (q, ky, kx) in enumerate(rows):
|
|
for k, (a, b) in enumerate(T.children(kx, ky)): # (x,y) children
|
|
ym, xm = b - CY[q], a - CX[q]
|
|
yo, xo = b - 2 * CY[q], a - 2 * CX[q]
|
|
cells[r, k, 0] = (ym, xm)
|
|
cells[r, k, 1] = (yo, xo)
|
|
classes[r, k, 0] = ENC[cls_of(xm, ym)]
|
|
classes[r, k, 1] = ENC[cls_of(xo, yo)]
|
|
return {"rows": rows, "cells_yx": cells, "classes": classes,
|
|
"valid": np.ones(len(rows), dtype=bool)}
|
|
|
|
|
|
class TestSRTNanMask(unittest.TestCase):
|
|
def test_srt_coarse_and_fine_nan_masks_equal(self):
|
|
spec = ref.make_spec(collision="srt")
|
|
maps = ref.build_maps(spec)
|
|
st = T.perturbed_state(spec, 3)
|
|
fc, ff = st.coarse_f.copy(), st.fine_f.copy()
|
|
worst = 0.0
|
|
for _ in range(3):
|
|
st, _ = ref.advance_coarse_once(spec, maps, st)
|
|
fc, ff = T.independent_step(fc, ff, srt=True)
|
|
np.testing.assert_array_equal(np.isnan(st.coarse_f), np.isnan(fc))
|
|
np.testing.assert_array_equal(np.isnan(st.fine_f), np.isnan(ff))
|
|
exp_fine_nan = np.array([[not T.f_active(a, b) for a in range(NF)] for b in range(NF)])
|
|
np.testing.assert_array_equal(np.isnan(st.fine_f), np.broadcast_to(exp_fine_nan, st.fine_f.shape))
|
|
self.assertTrue(np.isfinite(st.fine_f[:, ~exp_fine_nan]).all())
|
|
worst = max(worst, float(np.max(np.abs(np.nan_to_num(st.coarse_f - fc)))),
|
|
float(np.max(np.abs(np.nan_to_num(st.fine_f - ff)))))
|
|
print(f"\n[srt masks] coarse+fine NaN masks equal over 3 steps; max diff {worst:.3e}")
|
|
self.assertLessEqual(worst, 1e-14)
|
|
|
|
|
|
FLIPS = {"lower_left": (False, False), "lower_right": (True, False),
|
|
"upper_left": (False, True), "upper_right": (True, True)}
|
|
MX = (0, 2, 1, 3, 4, 8, 7, 6, 5)
|
|
MY = (0, 1, 2, 4, 3, 7, 8, 5, 6)
|
|
|
|
|
|
def mq(q, fx, fy):
|
|
if fx: q = MX[q]
|
|
if fy: q = MY[q]
|
|
return q
|
|
|
|
|
|
class TestCornerIndividualPackets(unittest.TestCase):
|
|
def test_eight_unique_keys_and_child_packets(self):
|
|
spec = ref.make_spec(collision="srt")
|
|
maps = ref.build_maps(spec)
|
|
keys = [link.key for link in maps.corner_links]
|
|
self.assertEqual(len(keys), 8)
|
|
self.assertEqual(len(set(keys)), 8)
|
|
# expected keys from independent CORNER_PASS geometry
|
|
expected = set()
|
|
for tag, (fx, fy) in FLIPS.items():
|
|
for q0 in (7, 8):
|
|
expected.add(f"{tag}_q{mq(q0, fx, fy)}")
|
|
self.assertEqual(set(keys), expected)
|
|
eps = 1e-3
|
|
exp_avg, g8 = T.TestSRTCornerOracle()._oracle(eps, True)
|
|
mid8 = 4 * exp_avg - 3 * g8
|
|
for name, (fx, fy) in FLIPS.items():
|
|
st = ref.make_state(spec)
|
|
fc = st.coarse_f.copy()
|
|
sx, sy = (N - 1 - 4 if fx else 4), (N - 1 - 3 if fy else 3)
|
|
for q, s in ((5, 1), (6, 1), (7, -1), (8, -1)):
|
|
fc[mq(q, fx, fy), sy, sx] += s * eps
|
|
out, d = ref.advance_coarse_once(spec, maps, ref.State(fc, st.fine_f, "arrival_f", 0))
|
|
self.assertEqual(len(d["corner_packets"]), 8)
|
|
self.assertEqual(set(d["corner_packets"]), set(keys))
|
|
key = f"{name}_q{mq(8, fx, fy)}"
|
|
cp = d["corner_packets"][key]
|
|
kx, ky = (N - 1 - 3 if fx else 3), (N - 1 - 4 if fy else 4)
|
|
q = mq(8, fx, fy)
|
|
self.assertEqual(cp["direction"], q, key)
|
|
self.assertEqual(tuple(cp["target_coarse_xy"]), (kx, ky), key)
|
|
kids_xy = T.children(kx, ky)
|
|
self.assertEqual({(y, x) for y, x in cp["children_fine_yx"]}, {(b, a) for a, b in kids_xy}, key)
|
|
ax, ay = (NF - 1 - 3 if fx else 3), (NF - 1 - 5 if fy else 5)
|
|
for (y, x), p in zip(cp["children_fine_yx"], cp["packets"]):
|
|
xm, ym = x - CX[q], y - CY[q]
|
|
expect = mid8 if (x, y) == (ax, ay) else g8
|
|
self.assertEqual(cls_of(xm, ym) == "A", (x, y) == (ax, ay), key)
|
|
self.assertAlmostEqual(p, expect, places=15, msg=f"{key} child {(x, y)}")
|
|
self.assertEqual(list(cp["weights"]), [0.25] * 4, key)
|
|
self.assertAlmostEqual(sum(w * p for w, p in zip(cp["weights"], cp["packets"])), cp["coalesced"], places=16)
|
|
self.assertAlmostEqual(cp["coalesced"], out.coarse_f[q, ky, kx], places=16)
|
|
self.assertAlmostEqual(cp["coalesced"], exp_avg, places=15)
|
|
print(f"\n[8 corner keys] {sorted(keys)}; packets 3xg8+1xmid8, weights 0.25, avg={exp_avg!r}")
|
|
|
|
|
|
class TestRetainedDiagnostics(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.spec = ref.make_spec(collision="srt")
|
|
cls.maps = ref.build_maps(cls.spec)
|
|
cls.out, cls.d = ref.advance_coarse_once(cls.spec, cls.maps, T.perturbed_state(cls.spec, 6))
|
|
|
|
def test_coalesce_paths_reconstructed_test_side(self):
|
|
self.assertNotIn("coalesce_paths", self.d)
|
|
cp = reconstruct_coalesce_paths(self.maps)
|
|
rows = cp["rows"]
|
|
slots = {(q, y, x) for y in range(N) for x in range(N) for q in range(9) if T.slot_kind(q, x, y)}
|
|
self.assertEqual(len(rows), len(set(rows)))
|
|
self.assertEqual(set(rows), slots)
|
|
self.assertEqual(cp["cells_yx"].shape, (len(rows), 4, 2, 2))
|
|
self.assertTrue(np.asarray(cp["valid"]).all())
|
|
hist = {}
|
|
for r, (q, ky, kx) in enumerate(rows):
|
|
kids = {(b, a) for a, b in T.children(kx, ky)}
|
|
seen = set()
|
|
for k in range(4):
|
|
(ym, xm), (yo, xo) = [tuple(int(v) for v in c) for c in cp["cells_yx"][r, k]]
|
|
y, x = ym + CY[q], xm + CX[q]
|
|
self.assertIn((y, x), kids)
|
|
seen.add((y, x))
|
|
self.assertEqual((yo, xo), (y - 2 * CY[q], x - 2 * CX[q]))
|
|
cm, co = cls_of(xm, ym), cls_of(xo, yo)
|
|
self.assertEqual(int(cp["classes"][r, k, 0]), ENC[cm])
|
|
self.assertEqual(int(cp["classes"][r, k, 1]), ENC[co])
|
|
self.assertNotEqual(cm, "O")
|
|
self.assertEqual(cls_of(x, y), "I")
|
|
kind = T.slot_kind(q, kx, ky)
|
|
hist[(kind, cm, co)] = hist.get((kind, cm, co), 0) + 1
|
|
self.assertEqual(seen, kids)
|
|
print(f"\n[coalesce_paths test-side] {len(rows)} rows x 4; hist={dict(sorted(hist.items()))}")
|
|
|
|
def test_hop2_mask_semantics(self):
|
|
self.assertNotIn("fine_valid_after", self.d)
|
|
v2 = np.asarray(self.d["hop2_valid"])
|
|
tgt = lambda a, b: cls_of(a, b) != "O"
|
|
inb = lambda a, b: 0 <= a < NF and 0 <= b < NF
|
|
exp1 = np.zeros((9, NF, NF), bool)
|
|
exp2 = np.zeros((9, NF, NF), bool)
|
|
for q in range(9):
|
|
for b in range(NF):
|
|
for a in range(NF):
|
|
sa, sb = a - CX[q], b - CY[q]
|
|
exp1[q, b, a] = tgt(a, b) and inb(sa, sb)
|
|
for b in range(NF):
|
|
for a in range(NF):
|
|
sa, sb = a - CX[q], b - CY[q]
|
|
exp2[q, b, a] = (tgt(a, b) and inb(sa, sb) and cls_of(sa, sb) != "O"
|
|
and exp1[q, sb, sa])
|
|
np.testing.assert_array_equal(v2, exp2)
|
|
act = np.array([[T.f_active(a, b) for a in range(NF)] for b in range(NF)])
|
|
inner = np.array([[cls_of(a, b) == "I" for a in range(NF)] for b in range(NF)])
|
|
self.assertTrue(v2[:, act].all())
|
|
self.assertFalse(v2[:, ~(act | inner)].any())
|
|
self.assertTrue(np.isnan(self.out.fine_f[:, ~act]).all())
|
|
n_valid_inner = int(v2[:, inner].sum())
|
|
self.assertGreater(n_valid_inner, 0)
|
|
reads = self.d["coalesce_reads"] > 0
|
|
self.assertTrue(v2[reads].all())
|
|
print(f"\n[hop2_valid] matches independent derivation; valid inner={n_valid_inner}; "
|
|
f"reads hop2-valid={int(reads.sum())}")
|
|
|
|
|
|
class TestV02V03FieldDelta(unittest.TestCase):
|
|
def test_diagnostic_field_delta(self):
|
|
"""Compare top-level diagnostic keys and budget keys between v0.2 and v0.3."""
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
def load(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = mod
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
v2 = load("cpu_ref_v02", os.path.join(here, "cpu_reference_v0.2.py"))
|
|
v3 = ref
|
|
s2, s3 = v2.make_spec(collision="identity"), v3.make_spec(collision="identity")
|
|
m2, m3 = v2.build_maps(s2), v3.build_maps(s3)
|
|
st2, st3 = v2.make_state(s2), v3.make_state(s3)
|
|
_, d2 = v2.advance_coarse_once(s2, m2, st2)
|
|
_, d3 = v3.advance_coarse_once(s3, m3, st3)
|
|
k2, k3 = set(d2), set(d3)
|
|
removed = sorted(k2 - k3)
|
|
added = sorted(k3 - k2)
|
|
renamed = {"explosion_count", "coalesce_paths", "fine_valid_after"}
|
|
self.assertTrue(set(removed) <= renamed | {"explosion_count"})
|
|
self.assertIn("hop2_valid", added)
|
|
self.assertNotIn("coalesce_paths", d3)
|
|
self.assertNotIn("explosion_count", d3)
|
|
self.assertNotIn("fine_valid_after", d3)
|
|
self.assertIn("closure_error", d2["budget"])
|
|
self.assertIn("bookkeeping_check", d3["budget"])
|
|
self.assertNotIn("closure_error", d3["budget"])
|
|
self.assertEqual(len(d2["corner_packets"]), 4) # v0.2 overwrite bug
|
|
self.assertEqual(len(d3["corner_packets"]), 8)
|
|
print(f"\n[v0.2→v0.3 fields] removed={removed} added={added}; "
|
|
f"budget rename closure_error→bookkeeping_check; corner_packets 4→8")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("reference:", T.REF_PATH, "API", ref.API_VERSION)
|
|
for p in (T.REF_PATH, T.__file__, os.path.abspath(__file__)):
|
|
print("sha256", hashlib.sha256(open(p, "rb").read()).hexdigest(), os.path.basename(p))
|
|
print("python", sys.version.split()[0], "numpy", np.__version__, platform.platform())
|
|
prog = unittest.main(exit=False, verbosity=2)
|
|
print("peak_rss_MiB", resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024)
|
|
sys.exit(0 if prog.result.wasSuccessful() else 1)
|