From 335624c27b8fc9925e708fb61bde5eea5c04d2a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 02:14:23 +0000 Subject: [PATCH] test: fixed three-level/two-region nested shear preset (nested_shear_n128) Test-only opt-in. New shear_nested.py: fixed parent-frame rectangles (L0->L1 (16,16,60,112),(68,24,112,120); L1->L2 (44,44,108,212), (148,60,212,228)), exact leaves/hops/work/nu/tau, depth-3 + 4N-finest generalization of the shear assessment (same summaries/screens/budgets), bounded final macro snapshot writer/reader (packed ACTIVE leaves + rho/u, metadata, sha256, 8 MiB cap; no populations/restart/time series). run_case: --diagnostic nested_shear_n128 (U128, U512, constant/projected R128 depth 3), --final-snapshot only for projected depth 3, shear_nested.py path+hash pin. assess_shear_wave: assess_nested contract and CLI. shear_wave.py and oracle.py byte-identical; old guards/defaults unchanged. Co-Authored-By: Claude Opus 5.5 --- docs/validation/taylor_green.md | 43 ++++ .../taylor_green/assess_shear_wave.py | 111 +++++++++- tests/validation/taylor_green/run_case.py | 53 ++++- tests/validation/taylor_green/shear_nested.py | 174 +++++++++++++++ .../taylor_green/test_shear_wave.py | 205 ++++++++++++++++++ 5 files changed, 575 insertions(+), 11 deletions(-) create mode 100644 tests/validation/taylor_green/shear_nested.py diff --git a/docs/validation/taylor_green.md b/docs/validation/taylor_green.md index cb2281e..8fb7acd 100644 --- a/docs/validation/taylor_green.md +++ b/docs/validation/taylor_green.md @@ -342,3 +342,46 @@ Assessment: differed by one ULP: 0.012047856934923534 on Horsea vs 0.012047856934923532 elsewhere. Source and file hashes, case identity, the other metadata and every scientific threshold stay exact. + +### Nested shear preset (opt-in `nested_shear_n128`) + +A fixed three-level, two-region preset, defined in `shear_nested.py`. The +wave, phase, Re100/U=0.03, initialization, five sample times, eps(root N), +limits and floors are the same as before. + +- Refinements, given as parent-frame `(x0, y0, x1, y1)` rectangles on + level-wide, zero-origin unions: + - L0→L1: `(16,16,60,112)`, `(68,24,112,120)` + - L1→L2: `(44,44,108,212)`, `(148,60,212,228)` +- Leaves `[7936, 12288, 86016]`, hops `[2048, 4096, 8192]`, work + 771227648 per refined case; 3723493376 for all four cases. +- Level ν `[.0384, .0768, .1536]`, τ `[.6152, .7304, .9608]`. + +Cases (`run_case.py --flow shear_wave --diagnostic nested_shear_n128 --re 100`): +- `--n 128 --depth 1` +- `--n 512 --depth 1` — the finest control is 4N, not 2N +- `--n 128 --depth 3` +- `--n 128 --depth 3 --explosion projected_linear [--final-snapshot]` + +`--final-snapshot` is allowed only for the projected depth-3 case. It writes +`.final_macro.npz`, containing: +- ACTIVE leaves packed as `level`/`i`/`j`; +- `rho`, `ux`, `uy` from the FP64 physical populations at the final root + synchronization; +- embedded metadata: exact time, phase, mode and config. + +It contains no populations, restart state or time series. Its SHA-256 and +size go into the receipt. A file over 8 MiB is removed and reported, never +kept. All old paths stay scalar-only. + +Assessment: `assess_shear_wave.py U128.json U512.json R128c.json R128p.json +--diagnostic nested_shear_n128 --output report.json --expected-revision `. +- It reuses the case-agnostic provenance checks from the TG assessor, plus + exact nested scales, builds, patch, workload, metadata and the pinned + `shear_nested.py` hash. +- The physics is `shear_wave.assess` generalized only in its depth-3 / 4N + shape contract; the metrics are unchanged. +- `shear_wave.py` is byte-identical (81d5c8ee…), so historical receipt + identities and verdicts are untouched. +- This preset changes placement and refined volume as well as depth. It + gives no isolated cost of an extra level and no resolution trend. diff --git a/tests/validation/taylor_green/assess_shear_wave.py b/tests/validation/taylor_green/assess_shear_wave.py index 0793a9b..452f4a6 100644 --- a/tests/validation/taylor_green/assess_shear_wave.py +++ b/tests/validation/taylor_green/assess_shear_wave.py @@ -18,7 +18,9 @@ from pathlib import Path import assess_projected_pairs as tg import shear_wave as sw -from run_case import SCALED_DIAGNOSTIC, SCALED_SHEAR, SHEAR_WAVE_SHA256 +import shear_nested as nested +from run_case import (NESTED_SHEAR, SCALED_DIAGNOSTIC, SCALED_SHEAR, SHEAR_NESTED_SHA256, + SHEAR_WAVE_SHA256, sample_steps) # Audited de5b0cc N128 receipts (evidence commit cba6bca, shear-wave-de5b0cc zip). REFERENCE_REVISION = "de5b0cc080794d0ce44d37f845aa8e8dbf8c2016" @@ -114,6 +116,102 @@ def assess(records, expected_revision=None, *, scaled=False): return result +CASE_AGNOSTIC = ("finite_receipts", "same_source_and_harness", "clean_a135252_numerical_source", + "same_recorded_runtime", "same_explicit_device", "same_early_sampling") + + +def _nested_case_ok(r, n, depth): + """Per-case identity of the fixed nested preset (scales, builds, patch, work, pins).""" + runtime = r.get("runtime_source", {}) + package = Path(runtime.get("checkout", "")) / "src/CelerisLab" + cells = [n * n] if depth == 1 else nested.LEAVES + hops = [16 * n * 2**level for level in range(depth)] + work = r.get("workload", {}) + snapshot = r.get("final_snapshot") + expected_patch = [] if depth == 1 else nested.REFINEMENTS + return { + "complete_times": r.get("completed") is True and tg.at_steps( + r.get("rows", []), sample_steps(n)[0], n), + "scales": (r.get("U") == sw.U and tg.finite(r.get("viscosity")) + and abs(r["viscosity"] - sw.U * n / 100) < 1e-15 + and r.get("tau_by_level") == [0.5 + 3 * sw.U * n / 100 * 2**l + for l in range(depth)] + and r.get("mach") == sw.U * math.sqrt(3)), + "actual_builds": tg.valid_builds(r, n, depth), + "reference": (r.get("numerical_revision") == tg.NUMERICAL_REVISION + and r.get("reference_harness") == tg.REFERENCE_HARNESS), + "runtime_source": bool(runtime.get("checkout")) and all( + runtime.get(k) == str(package / tail) for k, tail in + (("module", "__init__.py"), ("compiler", "cuda/compiler_v2.py"), + ("kernel_root", "lbm/kernels"))), + "patch": json.dumps(r.get("refinements")) == json.dumps(expected_patch), + "workload": (work.get("active_cells_by_level") == cells and work.get("level_hops") == hops + and work.get("root_intervals") == 16 * n + and work.get("active_collision_updates_by_level") == [ + c * h for c, h in zip(cells, hops)] + and work.get("active_collision_updates") == sum( + c * h for c, h in zip(cells, hops))), + "shear_metadata": metadata_ok(r, n), + "nested_module_pin": (r.get("source") or {}).get("shear_nested_sha256") + == SHEAR_NESTED_SHA256, + # Optional, projected depth-3 only; bounded and hashed when present. + "snapshot": snapshot is None or ( + depth == nested.DEPTH and r.get("explosion") == "projected_linear" + and snapshot.get("kept") is True and isinstance(snapshot.get("sha256"), str) + and len(snapshot["sha256"]) == 64 + and 0 < snapshot.get("bytes", 0) <= nested.SNAPSHOT_MAX_BYTES + and snapshot.get("leaf_counts") == nested.LEAVES + and snapshot.get("step") == sample_steps(n)[0][-1]), + } + + +def assess_nested(records, expected_revision=None): + """Fixed nested_shear_n128 four-case contract (U128, U512, constant/projected R128 d3).""" + if not isinstance(records, list) or not all(isinstance(r, dict) for r in records): + raise ValueError("records must be a list of receipt objects") + base_checks = tg.assess(records, expected_revision)["checks"] + checks = {name: base_checks.get(name, False) for name in CASE_AGNOSTIC} + indexed, duplicates = {}, [] + for r in records: + key = tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion")) + (duplicates.append(key) if key in indexed else indexed.__setitem__(key, r)) + checks.update(complete_matrix=all(k in indexed for k in nested.CASES), + no_extra_cases=set(indexed) <= set(nested.CASES), + no_duplicate_cases=not duplicates, + shear_diagnostic_identity=all(r.get("diagnostic") == NESTED_SHEAR + for r in records)) + for key in nested.CASES: + _, n, depth, scheme = key + label = f"{n}/{depth}/{scheme}" + case = _nested_case_ok(indexed[key], n, depth) if key in indexed else {} + for name in ("complete_times", "scales", "actual_builds", "reference", "runtime_source", + "patch", "workload", "shear_metadata", "nested_module_pin", "snapshot"): + checks[f"{label} {name}"] = bool(case.get(name, False)) + contract_valid = all(checks.values()) + result = dict(checks=checks, contract_valid=contract_valid, diagnostic=NESTED_SHEAR, + cases=[list(k) for k in nested.CASES], expected_revision=expected_revision, + finest_control="uniform 4N (U512)", + scope="2D periodic Re100 oblique (2,1) shear wave; fixed three-level/two-region " + "placement; changes placement and refined volume as well as depth, so no " + "isolated-level cost or resolution trend") + if not contract_valid: + result.update(status="INVALID_SET", passed=False) + return result + try: + physics = nested.assess(indexed[nested.CASES[0]], indexed[nested.CASES[1]], + [indexed[nested.CASES[2]], indexed[nested.CASES[3]]]) + except (KeyError, ValueError) as error: + result.update(status="INVALID_SET", passed=False, reason=str(error)) + return result + result.update(physics=physics, status=physics["status"], passed=physics["status"] == "PASS") + try: + # Role "uniform_fine" here is the 4N finest control. + result["signed"] = signed_summary([indexed[k] for k in nested.CASES]) + except (KeyError, TypeError, ValueError) as error: + result["signed_unavailable"] = str(error) + return result + + def _signed(record): """Raw signed quantities of one valid record (None when the fit is unavailable).""" s = sw.summarize(record) @@ -183,11 +281,20 @@ if __name__ == "__main__": p.add_argument("files", type=Path, nargs=4) p.add_argument("--output", type=Path, required=True) p.add_argument("--expected-revision", required=True) - p.add_argument("--diagnostic", choices=[SCALED_SHEAR]) + p.add_argument("--diagnostic", choices=[SCALED_SHEAR, NESTED_SHEAR]) p.add_argument("--reference-files", type=Path, nargs=4) args = p.parse_args() if args.output.exists(): p.error("Output already exists; choose a new path to retain prior evidence") + if args.diagnostic == NESTED_SHEAR: + if args.reference_files: + p.error("nested_shear_n128 takes no reference files (no resolution trend)") + result = assess_nested([json.loads(path.read_text()) for path in args.files], + args.expected_revision) + text = json.dumps(result, indent=2, allow_nan=False) + "\n" + args.output.write_text(text) + print(text, end="") + raise SystemExit(0 if result["passed"] else 1) if bool(args.diagnostic) != bool(args.reference_files): p.error("--diagnostic scaled_shear_n256 requires --reference-files (N128 receipts)") result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision, diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index b63fbd8..5889d1b 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -21,6 +21,8 @@ INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p" SHEAR_WAVE_SHA256 = "81d5c8ee60864372887cddf9239580c82fb6750d956038f1e3ee9de61aa4e877" SCALED_DIAGNOSTIC = "scaled_n256" SCALED_SHEAR = "scaled_shear_n256" # mutually exclusive with scaled_n256 (one flow each) +NESTED_SHEAR = "nested_shear_n128" # fixed three-level/two-region shear preset +SHEAR_NESTED_SHA256 = "fa6cfe87fb4df8c6d732c637c885276833c81973b950c7724d734f8fc350bcb0" def flow_functions(flow="taylor_green"): @@ -109,7 +111,14 @@ def verify_source(repo, revision, flow="taylor_green"): if result["source_tree_sha256"] != SOURCE_SHA256: raise RuntimeError("Wrong numerical source fingerprint") result["build_capture_sha256"] = BUILD_CAPTURE_SHA256 - if flow == "shear_wave": + if flow == "shear_nested": + import shear_nested + module = harness / "shear_nested.py" + if (Path(shear_nested.__file__).resolve() != module + or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_NESTED_SHA256): + raise RuntimeError("Wrong shear_nested module checkout or content") + result["shear_nested_sha256"] = SHEAR_NESTED_SHA256 + if flow in ("shear_wave", "shear_nested"): import shear_wave module = harness / "shear_wave.py" if (Path(shear_wave.__file__).resolve() != module @@ -182,21 +191,35 @@ def main(): p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True) p.add_argument("--output", type=Path, required=True) p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant") - p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR], + p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR, NESTED_SHEAR], help="explicit four-case N256/fine512 same-placement diagnostic " - "(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave)") + "(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave) or the " + "fixed three-level/two-region shear preset (nested_shear_n128)") + p.add_argument("--final-snapshot", action="store_true", + help="nested_shear_n128 projected depth-3 only: final macro snapshot") p.add_argument("--early-pressure", action="store_true", help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64") p.add_argument("--flow", choices=["taylor_green", "shear_wave"], default="taylor_green") args = p.parse_args() - if args.flow == "shear_wave" and (args.re != 100 or args.depth == 3 or args.early_pressure): + nested = args.diagnostic == NESTED_SHEAR + if args.flow == "shear_wave" and (args.re != 100 or (args.depth == 3 and not nested) + or args.early_pressure): p.error("shear_wave requires --re 100, depth 1 or 2 and no --early-pressure") + if args.final_snapshot and not (nested and args.depth == 3 + and args.explosion == "projected_linear"): + p.error("--final-snapshot requires nested_shear_n128 projected depth 3") initial, metrics, limits, initialization = flow_functions(args.flow) if args.output.exists(): p.error("Output already exists; choose a new path to retain prior evidence") - if args.explosion != "constant" and args.depth != 2: + if args.explosion != "constant" and args.depth != 2 and not nested: p.error("projected_linear requires depth 2") - if args.diagnostic: + if nested: + import shear_nested + if args.flow != "shear_wave": + p.error(f"--diagnostic {NESTED_SHEAR} requires --flow shear_wave") + if (args.re, args.n, args.depth, args.explosion) not in shear_nested.CASES: + p.error("nested_shear_n128 requires one of its four Re100 cases") + elif args.diagnostic: cases = {(256, 1, "constant"), (512, 1, "constant"), (256, 2, "constant"), (256, 2, "projected_linear")} flow = "shear_wave" if args.diagnostic == SCALED_SHEAR else "taylor_green" @@ -209,13 +232,13 @@ def main(): if args.diagnostic == SCALED_SHEAR and (args.n, args.depth, args.explosion) not in cases: p.error("scaled_shear_n256 requires one of its four Re100 cases") elif args.n == 512: - p.error("N512 requires --diagnostic scaled_n256 or scaled_shear_n256") + p.error("N512 requires --diagnostic scaled_n256, scaled_shear_n256 or nested_shear_n128") selected = os.environ.get("CUDA_VISIBLE_DEVICES", "") if not args.dry_run and (not args.run_gpu or not selected or "," in selected or selected == "-1"): p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required") repo = args.repo.resolve() sys.path[:0] = [str(repo / "src"), str(repo)] - source = verify_source(repo, args.expected_revision, args.flow) + source = verify_source(repo, args.expected_revision, "shear_nested" if nested else args.flow) runtime = runtime_source(repo) if args.dry_run: print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True, @@ -229,7 +252,12 @@ def main(): from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition cfg = config(args.n, args.re) - refinements = [] if args.depth == 1 else rectangles(args.n, args.depth) + if args.depth == 1: + refinements = [] + elif nested: + refinements = [list(level) for level in shear_nested.REFINEMENTS] + else: + refinements = rectangles(args.n, args.depth) h = build_hierarchy(cfg, refinements) np.testing.assert_array_equal(leaf_partition(h), np.ones((args.n, args.n))) masks = [k == ACTIVE for k in h.kinds] @@ -317,6 +345,13 @@ def main(): initial_mass = row["mass"] if step in main_steps: report["rows"].append(row) + if args.final_snapshot and step == main_steps[-1]: + # Optional macro-only snapshot; scalar receipt stays authoritative. + report["final_snapshot"] = shear_nested.write_snapshot( + args.output.with_name(args.output.stem + ".final_macro.npz"), + run.download(), masks, + shear_nested.snapshot_meta(step, cfg.viscosity, args.explosion, + source.get("revision"))) if step in early_steps: trace = report["early_pressure"] trace["rows"].append({key: row[key] for key in ("step", "tstar", "pressure_l2")}) diff --git a/tests/validation/taylor_green/shear_nested.py b/tests/validation/taylor_green/shear_nested.py new file mode 100644 index 0000000..f1a9d09 --- /dev/null +++ b/tests/validation/taylor_green/shear_nested.py @@ -0,0 +1,174 @@ +"""Fixed three-level/two-region shear preset (opt-in nested_shear_n128). + +Four fresh cases: U128, U512 (finest control = 4N), constant R128 depth 3, +projected R128 depth 3. Same wave, phase, Re100/U=0.03, initialization, five +sample times, eps(root N), limits and floors as shear_wave. This changes +placement and refined volume as well as depth: no isolated-level cost or +resolution-trend label. shear_wave.py is reused unchanged (historical receipt +identities stay valid); only the depth/finest-control contract is generalized +here. +""" + +import hashlib +import json +from pathlib import Path + +import numpy as np + +import shear_wave as sw +from oracle import CX, CY, U, centers + +ROOT_N = 128 +DEPTH = 3 +FINEST_FACTOR = 2 ** (DEPTH - 1) # finest-spacing uniform control is 4N +# Parent-frame (x0, y0, x1, y1) rectangles; level-wide, zero-origin unions. +REFINEMENTS = [ + [(16, 16, 60, 112), (68, 24, 112, 120)], # L0 -> L1, root coordinates + [(44, 44, 108, 212), (148, 60, 212, 228)], # L1 -> L2, L1 coordinates +] +CASES = ((100, ROOT_N, 1, "constant"), (100, FINEST_FACTOR * ROOT_N, 1, "constant"), + (100, ROOT_N, DEPTH, "constant"), (100, ROOT_N, DEPTH, "projected_linear")) +ROOT_STEPS = 16 * ROOT_N +LEAVES = [7936, 12288, 86016] +HOPS = [ROOT_STEPS * 2**level for level in range(DEPTH)] # [2048, 4096, 8192] +WORK = sum(c * h for c, h in zip(LEAVES, HOPS)) # 771227648 per refined case +NU = [U * ROOT_N / sw.RE * 2**level for level in range(DEPTH)] # [.0384, .0768, .1536] +TAU = [0.5 + 3 * nu for nu in NU] # [.6152, .7304, .9608] +SNAPSHOT_MAX_BYTES = 8 * 1024 * 1024 +SNAPSHOT_KEYS = ("level", "i", "j", "rho", "ux", "uy", "meta") + + +def leaf_masks(): + """Independent numpy leaf masks of the fixed preset (root, L1, L2 level-wide grids).""" + masks, covered = [], None + for level in range(DEPTH): + size = ROOT_N * 2**level + inside = np.ones((size, size), bool) if level == 0 else covered + if level < DEPTH - 1: + child = np.zeros((size, size), bool) + for x0, y0, x1, y1 in REFINEMENTS[level]: + child[y0:y1, x0:x1] = True + masks.append(inside & ~child) + covered = np.kron(child, np.ones((2, 2), bool)) + else: + masks.append(inside) + return masks + + +def assess(base_control, finest_control, refined): + """shear_wave.assess generalized to depth 3 with a 4N finest control. + + Same summaries, screens, budgets and outcome semantics; only the + (depth, finest-control) shape contract differs. + """ + if base_control is None or finest_control is None or not refined: + return dict(status="INVALID_CONTROL", reason="missing control or refined run") + n = base_control["n"] + if (base_control["depth"], finest_control["depth"], finest_control["n"]) != ( + 1, 1, FINEST_FACTOR * n): + raise ValueError("controls must be uniform N and uniform 4N") + if any(r["depth"] != DEPTH or r["n"] != n for r in refined): + raise ValueError("refined runs must be depth 3 at root N") + if len({r.get("explosion") for r in refined}) != len(refined): + raise ValueError("duplicate refined explosion scheme") + sources = {repr(r.get("source")) for r in [base_control, finest_control, *refined]} + base, fine = sw._usable(base_control), sw._usable(finest_control) + if base is None or fine is None or len(sources) != 1: + return dict(status="INVALID_CONTROL", reason="incomplete/invalid control or mixed source") + controls_pass = base["own_screens_pass"] and fine["own_screens_pass"] + report = dict(base_control=base, finest_control=fine, refined={}, + controls_pass=controls_pass) + statuses = [] + for r in refined: + try: + s = sw.summarize(r) + except (KeyError, TypeError, ValueError) as error: + report["refined"][r["explosion"]] = dict(valid=False, reason=str(error)) + statuses.append("INVALID_RUN") + continue + s["valid"] = True + s["budget"] = { + key: (None if base[key] is None else max(sw.BASELINE_FACTOR * base[key], floor)) + for key, floor in sw.FLOORS.items() + } + s["budget_pass"] = all( + s[key] is not None and s["budget"][key] is not None and s[key] <= s["budget"][key] + for key in sw.FLOORS) + s["control_ratios"] = { + name: {key: (s[key] / c[key] if s[key] is not None and c[key] else None) + for key in (*sw.FLOORS, "spurious_pressure_rms")} + for name, c in (("root", base), ("finest", fine)) + } + s["controlled_pass"] = controls_pass and s["own_screens_pass"] and s["budget_pass"] + report["refined"][r["explosion"]] = s + statuses.append("PASS" if s["controlled_pass"] else "FAIL") + if "INVALID_RUN" in statuses: + report["status"] = "INVALID_RUN" + elif not controls_pass: + report["status"] = "CONTROL_SCREEN_FAIL" + else: + report["status"] = "PASS" if all(v == "PASS" for v in statuses) else "FAIL" + return report + + +def write_snapshot(path, arrays, masks, meta): + """Final macro snapshot: packed ACTIVE leaves (level, i, j), rho, ux, uy only. + + No populations, no restart state, no time series. Returns the receipt block; + a file over SNAPSHOT_MAX_BYTES is removed and reported, never kept. + """ + path = Path(path) + if path.exists(): + raise FileExistsError(path) + cols = {k: [] for k in ("level", "i", "j", "rho", "ux", "uy")} + counts = [] + for level, (g, mask) in enumerate(zip(arrays, masks)): + if g.shape != (9, *mask.shape): + raise ValueError("population/mask shape mismatch") + values = g[:, mask].astype(float) + rho = values.sum(axis=0) + j, i = np.nonzero(mask) # row = y index, column = x index + cols["level"].append(np.full(rho.size, level, np.uint8)) + cols["i"].append(i.astype(np.uint16)) + cols["j"].append(j.astype(np.uint16)) + cols["rho"].append(rho) + cols["ux"].append(CX @ values / rho) + cols["uy"].append(CY @ values / rho) + counts.append(int(rho.size)) + packed = {k: np.concatenate(v) for k, v in cols.items()} + if not all(np.isfinite(packed[k]).all() for k in ("rho", "ux", "uy")): + raise FloatingPointError("nonfinite snapshot macros") + meta = dict(meta, leaf_counts=counts, cell_center="h*(index+0.5), h=2**-level", + fields="rho, ux, uy from FP64 physical populations at ACTIVE leaves") + with path.open("xb") as handle: + np.savez_compressed(handle, meta=np.array(json.dumps(meta, sort_keys=True)), **packed) + size = path.stat().st_size + block = dict(file=path.name, bytes=size, leaf_counts=counts, max_bytes=SNAPSHOT_MAX_BYTES, + step=meta.get("step"), tstar=meta.get("tstar")) + if size > SNAPSHOT_MAX_BYTES: + path.unlink() + return dict(block, kept=False, sha256=None, reason="exceeds snapshot byte cap") + block.update(kept=True, sha256=hashlib.sha256(path.read_bytes()).hexdigest()) + return block + + +def read_snapshot(path): + """Validate a snapshot: exact keys, no populations, consistent packed arrays.""" + with np.load(path, allow_pickle=False) as data: + if tuple(sorted(data.files)) != tuple(sorted(SNAPSHOT_KEYS)): + raise ValueError("unexpected snapshot contents") + out = {k: data[k] for k in data.files} + meta = json.loads(str(out.pop("meta"))) + n = len(out["rho"]) + if any(len(v) != n for v in out.values()) or sum(meta["leaf_counts"]) != n: + raise ValueError("inconsistent snapshot arrays") + return meta, out + + +def snapshot_meta(step, viscosity, explosion, source_revision): + """Exact time/phase/config identity stored inside the snapshot.""" + return dict(step=step, tstar=step * U / ROOT_N, phase=sw.PHI, mode=list(sw.MODE), + polarization=sw.E.tolist(), amplitude=U, re_L=sw.RE, n=ROOT_N, depth=DEPTH, + explosion=explosion, viscosity=viscosity, tau_by_level=TAU, + refinements=REFINEMENTS, source_revision=source_revision, + origin="level-wide zero-origin grids", time="root synchronization, postcollision") diff --git a/tests/validation/taylor_green/test_shear_wave.py b/tests/validation/taylor_green/test_shear_wave.py index fe593ed..b739391 100644 --- a/tests/validation/taylor_green/test_shear_wave.py +++ b/tests/validation/taylor_green/test_shear_wave.py @@ -571,3 +571,208 @@ class DefaultPathTests(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class NestedPresetTests(unittest.TestCase): + """Fixed three-level/two-region shear preset (nested_shear_n128).""" + + def setUp(self): + import assess_shear_wave + import shear_nested + self.mod, self.nested = assess_shear_wave, shear_nested + + def test_exact_geometry_work_and_levels(self): + nested = self.nested + masks = nested.leaf_masks() + self.assertEqual([int(m.sum()) for m in masks], [7936, 12288, 86016]) + self.assertEqual(nested.LEAVES, [7936, 12288, 86016]) + self.assertEqual(nested.HOPS, [2048, 4096, 8192]) + self.assertEqual(nested.WORK, 771227648) + uniform = 128 * 128 * 2048 + 512 * 512 * 8192 + self.assertEqual(uniform + 2 * nested.WORK, 3723493376) + np.testing.assert_allclose(nested.NU, [.0384, .0768, .1536], rtol=0, atol=1e-15) + np.testing.assert_allclose(nested.TAU, [.6152, .7304, .9608], rtol=0, atol=1e-14) + self.assertEqual(sum(float(m.sum()) * 0.25**l for l, m in enumerate(masks)), 128.0**2) + # Production host hierarchy builds the same leaves from the same rectangles. + from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition + from CelerisLab.lbm.refinement import ACTIVE + h = build_hierarchy(run_case.config(128, 100), nested.REFINEMENTS) + np.testing.assert_array_equal(leaf_partition(h), np.ones((128, 128))) + for mine, kinds in zip(masks, h.kinds): + np.testing.assert_array_equal(mine, kinds == ACTIVE) + + def test_historical_shear_identity_is_preserved(self): + digest = hashlib.sha256((HERE / "shear_wave.py").read_bytes()).hexdigest() + self.assertEqual(digest, "81d5c8ee60864372887cddf9239580c82fb6750d956038f1e3ee9de61aa4e877") + self.assertEqual(run_case.SHEAR_WAVE_SHA256, digest) + nested_digest = hashlib.sha256((HERE / "shear_nested.py").read_bytes()).hexdigest() + self.assertEqual(run_case.SHEAR_NESTED_SHA256, nested_digest) + + def nested_set(self): + out = [] + for n, depth, scheme in ((128, 1, "constant"), (512, 1, "constant"), + (128, 3, "constant"), (128, 3, "projected_linear")): + r = receipt(n, depth, scheme, diagnostic=run_case.NESTED_SHEAR) + r["source"]["shear_nested_sha256"] = run_case.SHEAR_NESTED_SHA256 + if depth == 3: + cells, hops = self.nested.LEAVES, self.nested.HOPS + r["refinements"] = [list(level) for level in self.nested.REFINEMENTS] + r["workload"] = dict(active_cells_by_level=cells, level_hops=hops, + root_intervals=2048, + active_collision_updates_by_level=[c * h for c, h in zip(cells, hops)], + active_collision_updates=self.nested.WORK) + out.append(r) + return out + + def test_nested_contract_and_rejections(self): + a = self.mod.assess_nested + good = self.nested_set() + result = a(good, "r") + self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v]) + self.assertEqual(result["status"], "PASS") + self.assertIn("finest", result["physics"]["refined"]["constant"]["control_ratios"]) + wrong_finest = self.nested_set() + wrong_finest[1] = receipt(256, 1, diagnostic=run_case.NESTED_SHEAR) + wrong_finest[1]["source"]["shear_nested_sha256"] = run_case.SHEAR_NESTED_SHA256 + self.assertEqual(a(wrong_finest, "r")["status"], "INVALID_SET") + for mutate in ( + lambda s: s[2].update(refinements=[s[2]["refinements"][1], s[2]["refinements"][0]]), + lambda s: s[3].update(refinements=rectangles(128, 3)), + lambda s: s[2]["rows"][2].update(spurious_pressure_scaled_by_level=[0.0, 0.0]), + lambda s: s[0].update(diagnostic=run_case.SCALED_SHEAR), + lambda s: s[2]["source"].pop("shear_nested_sha256"), + lambda s: s[2].update(final_snapshot=dict(kept=True, sha256="a" * 64, bytes=10, + leaf_counts=self.nested.LEAVES, step=2048)), + lambda s: s[3].update(final_snapshot=dict(kept=True, sha256="a" * 64, + bytes=self.nested.SNAPSHOT_MAX_BYTES + 1, + leaf_counts=self.nested.LEAVES, step=2048)), + ): + records = self.nested_set() + mutate(records) + self.assertNotEqual(a(records, "r")["status"], "PASS") + ok = self.nested_set() + ok[3]["final_snapshot"] = dict(kept=True, sha256="a" * 64, bytes=1000, + leaf_counts=self.nested.LEAVES, step=2048) + self.assertEqual(a(ok, "r")["status"], "PASS") + with self.assertRaises(ValueError): + self.nested.assess(good[0], good[1], [good[2], dict(good[2])]) + + def test_nested_runner_options(self): + repo = Path(__file__).resolve().parents[3] + with tempfile.TemporaryDirectory() as tmp: + base = ["run_case.py", "--repo", str(repo), "--expected-revision", "f" * 40, + "--output", str(Path(tmp) / "case.json"), "--dry-run", "--re", "100"] + nest = ["--flow", "shear_wave", "--diagnostic", run_case.NESTED_SHEAR] + proj = ["--explosion", "projected_linear"] + for extra, valid in ( + (["--n", "128", "--depth", "1"] + nest, True), + (["--n", "512", "--depth", "1"] + nest, True), + (["--n", "128", "--depth", "3"] + nest, True), + (["--n", "128", "--depth", "3"] + proj + nest + ["--final-snapshot"], True), + (["--n", "256", "--depth", "1"] + nest, False), # finest control is 4N + (["--n", "128", "--depth", "2"] + nest, False), + (["--n", "128", "--depth", "3", "--flow", "shear_wave"], False), + (["--n", "128", "--depth", "3", "--diagnostic", run_case.NESTED_SHEAR], False), + (["--n", "128", "--depth", "3"] + nest + ["--final-snapshot"], False), + (["--n", "128", "--depth", "3"] + proj + ["--flow", "shear_wave"], False), + (["--n", "128", "--depth", "2", "--flow", "shear_wave", "--final-snapshot"], False)): + argv = base + extra + with self.subTest(argv=argv), patch.object(sys, "argv", argv), \ + patch.object(run_case, "verify_source", return_value={}) as verify, \ + patch.object(run_case, "runtime_source", return_value={}), \ + redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()): + if valid: + self.assertEqual(run_case.main(), 0) + self.assertEqual(verify.call_args.args[2], "shear_nested") + else: + with self.assertRaises(SystemExit): + run_case.main() + + def test_mocked_nested_execution_with_bounded_snapshot(self): + from types import SimpleNamespace + from contextlib import nullcontext + from CelerisLab.lbm.hierarchy import build_hierarchy + h = build_hierarchy(run_case.config(128, 100), self.nested.REFINEMENTS) + nu = U * 128 / sw.RE + arrays = [sw.initial(k.shape, level, 128, nu) for level, k in enumerate(h.kinds)] + + class FakeRun: + def __init__(self, *args, **kwargs): + self.step_count = 0 + self.kwargs = kwargs + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def initialize(self, *given): + assert len(given) == 3 + + def step(self, count): + assert count >= 0 + self.step_count += count + + def download(self): + return arrays + + modules = {"CelerisLab.cuda.context": SimpleNamespace(CudaContext=FakeRun), + "CelerisLab.lbm.block_stepper": SimpleNamespace(PeriodicBlockStepper=FakeRun), + "CelerisLab.lbm.refinement_stepper": SimpleNamespace(FixedPatchStepper=FakeRun)} + repo = Path(__file__).resolve().parents[3] + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "projected_nested.json" + argv = ["run_case.py", "--repo", str(repo), "--re", "100", "--n", "128", "--depth", "3", + "--explosion", "projected_linear", "--flow", "shear_wave", + "--diagnostic", run_case.NESTED_SHEAR, "--final-snapshot", + "--output", str(output), "--expected-revision", "f" * 40, "--run-gpu"] + with patch.dict(sys.modules, modules), patch.object(sys, "argv", argv), \ + patch.dict("os.environ", CUDA_VISIBLE_DEVICES="mock"), \ + patch.object(run_case, "verify_source", return_value=dict(revision="f" * 40)), \ + patch.object(run_case, "runtime_source", return_value={}), \ + patch.object(run_case, "runtime_identity", return_value={}), \ + patch.object(run_case, "build_capture", return_value=nullcontext()), \ + redirect_stdout(io.StringIO()): + run_case.main() + rec = json.loads(output.read_text()) + self.assertTrue(rec["completed"]) + self.assertEqual(rec["diagnostic"], run_case.NESTED_SHEAR) + self.assertEqual(rec["workload"]["active_cells_by_level"], self.nested.LEAVES) + self.assertEqual(rec["workload"]["active_collision_updates"], self.nested.WORK) + self.assertEqual(rec["tau_by_level"], self.nested.TAU) + self.assertTrue(all(len(r["spurious_pressure_scaled_by_level"]) == 3 for r in rec["rows"])) + snap = rec["final_snapshot"] + path = output.with_name(output.stem + ".final_macro.npz") + self.assertTrue(snap["kept"]) + self.assertEqual(snap["sha256"], hashlib.sha256(path.read_bytes()).hexdigest()) + self.assertLessEqual(snap["bytes"], self.nested.SNAPSHOT_MAX_BYTES) + meta, fields = self.nested.read_snapshot(path) + self.assertEqual((meta["step"], meta["phase"]), (2048, sw.PHI)) + self.assertEqual(meta["tstar"], 2048 * U / 128) + self.assertEqual(meta["leaf_counts"], self.nested.LEAVES) + self.assertEqual(sorted(fields), ["i", "j", "level", "rho", "ux", "uy"]) + np.testing.assert_allclose(fields["rho"], 1.0, atol=1e-12) + + def test_snapshot_contents_and_cap(self): + nested = self.nested + masks = nested.leaf_masks() + arrays = [sw.initial(m.shape, level, 128, 0.0384) for level, m in enumerate(masks)] + meta = nested.snapshot_meta(2048, 0.0384, "projected_linear", "r") + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "s.npz" + block = nested.write_snapshot(path, arrays, masks, meta) + self.assertTrue(block["kept"]) + got, fields = nested.read_snapshot(path) + self.assertEqual(got["refinements"], [[list(r) for r in level] for level in nested.REFINEMENTS]) + self.assertEqual(int(fields["level"].max()), 2) + with self.assertRaises(FileExistsError): + nested.write_snapshot(path, arrays, masks, meta) + bad = Path(tmp) / "bad.npz" + np.savez_compressed(bad, populations=np.zeros(3), **{k: np.zeros(1) for k in ("level", "i", "j", "rho", "ux", "uy")}, meta=np.array("{}")) + with self.assertRaises(ValueError): + nested.read_snapshot(bad) + with patch.object(nested, "SNAPSHOT_MAX_BYTES", 10): + capped = nested.write_snapshot(Path(tmp) / "c.npz", arrays, masks, meta) + self.assertFalse(capped["kept"]) + self.assertFalse((Path(tmp) / "c.npz").exists())