diff --git a/docs/validation/taylor_green.md b/docs/validation/taylor_green.md index 5309ebb..5454106 100644 --- a/docs/validation/taylor_green.md +++ b/docs/validation/taylor_green.md @@ -259,3 +259,24 @@ Run set: uniform N, uniform 2N, and refined depth-2 runs at root N. One flow, one phase and one rectangle, SRT only. The rectangle's vertical faces span 0.625 tangential periods and its horizontal faces 1.5, so this is not placement robustness. Two resolutions show a trend, not an order. + +### Receipt assessment and runner integrity + +- Fixed four-case N128 contract: + `python assess_shear_wave.py U128.json U256.json R128_constant.json R128_projected.json --output report.json --expected-revision `. + It reuses the shared provenance, build, workload and sample-time checks + from `assess_projected_pairs.py`, and adds shear metadata checks (flow, + mode, polarization, angle, phase, eps, limits, disabled early trace, and + pinned `shear_wave.py` hash). + - Missing, extra, duplicate or mixed-source records give `INVALID_SET`, + as does any failed shared check. + - Otherwise the result comes from `shear_wave.assess`, which separates + `INVALID_CONTROL` from `CONTROL_SCREEN_FAIL`. +- `run_case.verify_source(..., flow="shear_wave")` requires the imported + `shear_wave.py` to be the selected harness file with the pinned + `SHEAR_WAVE_SHA256`. The default TG flow does not import it. +- Sample and READY finiteness use `finite_row`, which accepts scalar and + list diagnostics. For TG's all-scalar rows it behaves the same as before. +- Merge note: native 92b8afe adds a `scaled_n256` receipt block. When the + two are merged, keep both blocks and make `--flow shear_wave` and + `scaled_n256` mutually exclusive. Not merged here. diff --git a/tests/validation/taylor_green/assess_shear_wave.py b/tests/validation/taylor_green/assess_shear_wave.py new file mode 100644 index 0000000..e5bf99a --- /dev/null +++ b/tests/validation/taylor_green/assess_shear_wave.py @@ -0,0 +1,81 @@ +"""Fixed N128 four-case shear-wave contract: U128, U256, constant R128, projected R128. + +Shared provenance, build, workload and sample-time checks are reused from +assess_projected_pairs (same cases, source pins and geometry); only the +Taylor-Green-specific checks are replaced by shear-wave metadata and physics. +""" + +import argparse +import json +from pathlib import Path + +import assess_projected_pairs as tg +import shear_wave as sw +from run_case import SHEAR_WAVE_SHA256 + +SHARED = ("complete_matrix", "no_extra_cases", "no_duplicate_cases", "finite_receipts", + "same_source_and_harness", "clean_a135252_numerical_source", + "same_recorded_runtime", "same_explicit_device", "same_early_sampling") +PER_CASE = ("complete_times", "scales", "actual_builds", "reference", "runtime_source", + "patch", "workload") + + +def metadata_ok(record, n): + return (record.get("flow") == "shear_wave" and record.get("mode") == list(sw.MODE) + and record.get("polarization") == sw.E.tolist() + and record.get("angle_deg") == sw.ANGLE_DEG and record.get("phase") == sw.PHI + and record.get("eps") == sw.eps(n) + and record.get("numerics") == tg.NUMERICS + and record.get("initialization") == sw.INITIALIZATION + and record.get("limits") == sw.LIMITS + and record.get("early_pressure") == dict(enabled=False, rows=[], + sampled_max_pressure_l2=None) + and (record.get("source") or {}).get("shear_wave_sha256") == SHEAR_WAVE_SHA256) + + +def assess(records, expected_revision=None): + 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 SHARED} + indexed = {} + for r in records: + indexed.setdefault(tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion")), r) + for key in tg.CASES: + _, n, depth, scheme = key + label = f"{n}/{depth}/{scheme}" + for name in PER_CASE: + checks[f"{label} {name}"] = base_checks.get(f"{label} {name}", False) + checks[f"{label} shear_metadata"] = key in indexed and metadata_ok(indexed[key], n) + contract_valid = all(checks.values()) + result = dict(checks=checks, contract_valid=contract_valid, + cases=[list(k) for k in tg.CASES], expected_revision=expected_revision, + scope="2D periodic Re100 oblique (2,1) shear wave; one phase and one " + "rectangle; no placement robustness or convergence order") + if not contract_valid: + result.update(status="INVALID_SET", passed=False) + return result + try: + physics = sw.assess(indexed[tg.CASES[0]], indexed[tg.CASES[1]], + [indexed[tg.CASES[2]], indexed[tg.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") + return result + + +if __name__ == "__main__": + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("files", type=Path, nargs=4) + p.add_argument("--output", type=Path, required=True) + p.add_argument("--expected-revision", required=True) + args = p.parse_args() + if args.output.exists(): + p.error("Output already exists; choose a new path to retain prior evidence") + result = assess([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) diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index df1efdd..d94eec6 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -18,6 +18,7 @@ SOURCE_SHA256 = "a5dbfd804befce868067d79a23e5fd47dce7f505b3f9b1bb2145fb57cb5f217 BUILD_CAPTURE_SHA256 = "3dbaa6547cb0a4cb1541054f3b64ad8b6e18cc70527ac6166e56b944e9bd464a" REFERENCE_HARNESS = "11901a7f77557a901e017a171eeae1758d584fa5" INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p" +SHEAR_WAVE_SHA256 = "49ab589d15d964d38ed90a49758e159c67a516191c7501c19fe2732e19742395" def flow_functions(flow="taylor_green"): @@ -30,6 +31,19 @@ def flow_functions(flow="taylor_green"): raise ValueError(f"unknown flow {flow!r}") +def finite_row(row): + """Every scalar or list-valued diagnostic is finite (TG rows are all scalars).""" + return all(np.isfinite(np.asarray(value, dtype=float)).all() for value in row.values()) + + +def sample(run, masks, step, n, viscosity, initial_mass, metrics, depth, label="sample"): + arrays = (run.download(),) if depth == 1 else run.download() + row = metrics(arrays, masks, step, n, viscosity, initial_mass) + if not finite_row(row): + raise FloatingPointError(f"nonfinite {label} diagnostics") + return row + + def sample_steps(n, early_pressure=False): """Root synchronization steps; early ladder matches the 707afe6 diagnostic.""" main = [round(tstar * n / U) for tstar in TIMES] @@ -74,7 +88,7 @@ def provenance(repo): ) -def verify_source(repo, revision): +def verify_source(repo, revision, flow="taylor_green"): """Exact clean harness commit, fixed production tree, and reviewed capture helper.""" import oracle from tests.validation.multigrid import check_projected_pairs as gate @@ -93,6 +107,13 @@ def verify_source(repo, revision): if result["source_tree_sha256"] != SOURCE_SHA256: raise RuntimeError("Wrong numerical source fingerprint") result["build_capture_sha256"] = BUILD_CAPTURE_SHA256 + if flow == "shear_wave": + import shear_wave + module = harness / "shear_wave.py" + if (Path(shear_wave.__file__).resolve() != module + or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_WAVE_SHA256): + raise RuntimeError("Wrong shear_wave module checkout or content") + result["shear_wave_sha256"] = SHEAR_WAVE_SHA256 return result @@ -175,7 +196,7 @@ def main(): 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) + source = verify_source(repo, args.expected_revision, args.flow) runtime = runtime_source(repo) if args.dry_run: print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True))) @@ -267,10 +288,8 @@ def main(): initial_mass = None for step in sorted(set(main_steps + early_steps)): run.step(step - run.step_count) - arrays = (run.download(),) if args.depth == 1 else run.download() - row = metrics(arrays, masks, step, args.n, cfg.viscosity, initial_mass) - if not all(np.isfinite(value) for value in row.values()): - raise FloatingPointError("nonfinite sample diagnostics") + row = sample(run, masks, step, args.n, cfg.viscosity, initial_mass, + metrics, args.depth) row["step"] = step if initial_mass is None: initial_mass = row["mass"] @@ -297,10 +316,8 @@ def main(): # Exercise real kernels/halos/transfers on every level before READY. # This is the first measured trajectory step, not discarded warmup. run.step(1) - probe = (run.download(),) if args.depth == 1 else run.download() - probe_metrics = metrics(probe, masks, 1, args.n, cfg.viscosity, initial_mass) - if not all(np.isfinite(value) for value in probe_metrics.values()): - raise FloatingPointError("nonfinite READY diagnostics") + sample(run, masks, 1, args.n, cfg.viscosity, initial_mass, + metrics, args.depth, "READY") report["ready_after_step"] = 1 save() print( diff --git a/tests/validation/taylor_green/test_shear_wave.py b/tests/validation/taylor_green/test_shear_wave.py index 53e6a20..99ad6fa 100644 --- a/tests/validation/taylor_green/test_shear_wave.py +++ b/tests/validation/taylor_green/test_shear_wave.py @@ -79,8 +79,21 @@ class ConventionTests(unittest.TestCase): dy = (f(x, y + h, t) - f(x, y - h, t)) / (2 * h) lap = (f(x + h, y, t) + f(x - h, y, t) + f(x, y + h, t) + f(x, y - h, t) - 4 * z) / h**2 np.testing.assert_allclose(dt, NU * lap, atol=1e-9) - np.testing.assert_allclose(z[0] * dx + z[1] * dy, 0.0, atol=1e-12) - np.testing.assert_allclose(dx[0] + dy[1], 0.0, atol=1e-12) + # Analytic identities: div u = A cos(theta) (E.k) and (u.grad)u = A sin(theta) (E.k) u/A. + k = sw.wavevector(N) + self.assertLess(abs(float(sw.E @ k)), 1e-17) + # Central differences do not cancel exactly: each derivative carries + # -(k_j h)^2/6 relative truncation, so div_h = -A cos(theta) h^2/6 sum_j E_j k_j^3 + # + O(h^4) (nonzero for an oblique k). Bound it with that leading term plus roundoff. + trunc = U * h * h / 6 * float(np.abs(sw.E * k**3).sum()) * 1.01 + # Argument rounding: eps*max|x|/h*U*|k| ~ 2.2e-16*21.6/1e-3*0.03*0.44 ~ 6e-14. + roundoff = 1e-13 + self.assertLess(np.max(np.abs(dx[0] + dy[1])), trunc + roundoff) + self.assertLess(np.max(np.abs(z[0] * dx + z[1] * dy)), U * trunc + roundoff) + # Halving h reduces the divergence truncation about fourfold. + hh = h / 2 + div2 = ((f(x + hh, y, t) - f(x - hh, y, t))[0] + (f(x, y + hh, t) - f(x, y - hh, t))[1]) / (2 * hh) + np.testing.assert_allclose(div2 * 4, dx[0] + dy[1], rtol=1e-2, atol=6 * roundoff) self.assertTrue(np.all(sw.exact(x, y, t, N, NU)[2] == 0)) @@ -256,6 +269,124 @@ class AssessTests(unittest.TestCase): self.assertEqual(sw.assess(base, fine, [refined])["status"], "FAIL") +class FakeRun: + def __init__(self, arrays): + self.arrays = arrays + + def download(self): + return self.arrays if len(self.arrays) > 1 else self.arrays[0] + + +class RunnerSampleTests(unittest.TestCase): + def test_refined_shear_sample_with_list_diagnostics(self): + masks = composite_masks(N) + arrays = [sw.initial(m.shape, level, N, NU) for level, m in enumerate(masks)] + row = run_case.sample(FakeRun(arrays), masks, 0, N, NU, None, sw.metrics, 2) + self.assertIsInstance(row["spurious_pressure_scaled_by_level"], list) + self.assertTrue(run_case.finite_row(row)) + bad = [a.copy() for a in arrays] + bad[1][0, masks[1]] = np.nan + with self.assertRaises(FloatingPointError): + run_case.sample(FakeRun(bad), masks, 0, N, NU, None, sw.metrics, 2, "READY") + self.assertFalse(run_case.finite_row(dict(a=1.0, b=[0.0, float("inf")]))) + + def test_taylor_green_sample_unchanged(self): + masks = composite_masks(N) + arrays = [oracle.initial(m.shape, level, N, NU) for level, m in enumerate(masks)] + row = run_case.sample(FakeRun(arrays), masks, 0, N, NU, None, oracle.metrics, 2) + self.assertEqual(row, oracle.metrics(arrays, masks, 0, N, NU, None)) + uniform = [oracle.initial((N, N), 0, N, NU)] + row = run_case.sample(FakeRun(uniform), [np.ones((N, N), bool)], 0, N, NU, None, + oracle.metrics, 1) + self.assertEqual(row["leaf_cells"], N * N) + + +def receipt(n, depth, explosion="constant"): + """Synthetic receipt satisfying every contract field of the fixed four-case set.""" + import assess_projected_pairs as tg + steps = run_case.sample_steps(n)[0] + rows = rows_for(n) + for row, step in zip(rows, steps): + row.update(step=step, tstar=step * U / n) + cells = [n * n] if depth == 1 else [8704, 30720] + hops = [16 * n * 2**level for level in range(depth)] + builds = [] + for level in range(depth): + size = n * 2**level + builds.append(dict(source_sha256=tg.BUILD_SOURCES[size], arch="sm_70", + flags=["--use_fast_math"], n_objects=0, ptx_sha256="a" * 64, + config=dict(nx=10, ny=10, nz=1, lattice_model="D2Q9", collision="SRT", + store_precision="FP32", streaming="double_buffer", + ddf_shifting=True, les_enabled=False, velocity=0.0, + viscosity=U * size / 100, rho=1.0, + omega=1 / (.5 + 3 * U * size / 100), + threads_per_block=256))) + checkout = "/x/CelerisLab" + return dict( + n=n, depth=depth, re_L=100, explosion=explosion, U=U, mach=U * math.sqrt(3), + viscosity=U * n / 100, tau_by_level=[0.5 + 3 * U * n / 100 * 2**l for l in range(depth)], + source=dict(revision="r", dirty=False, source_tree_sha256=run_case.SOURCE_SHA256, + harness_sha256={"x.py": "h"}, build_capture_sha256=run_case.BUILD_CAPTURE_SHA256, + shear_wave_sha256=run_case.SHEAR_WAVE_SHA256), + runtime_identity=dict(device=dict(name="V100", pci_bus_id="0", capability=[7, 0], + driver=1, runtime=[12, 0]), + nvcc_path="/nvcc", nvcc_version="v", python="3", numpy="1", + pycuda="2"), + runtime_source=dict(checkout=checkout, module=checkout + "/src/CelerisLab/__init__.py", + compiler=checkout + "/src/CelerisLab/cuda/compiler_v2.py", + kernel_root=checkout + "/src/CelerisLab/lbm/kernels"), + cuda_visible_devices="0", numerical_revision=run_case.NUMERICAL_REVISION, + reference_harness=run_case.REFERENCE_HARNESS, numerics=dict(tg.NUMERICS), + initialization=sw.INITIALIZATION, limits=dict(sw.LIMITS), + early_pressure=dict(enabled=False, rows=[], sampled_max_pressure_l2=None), + refinements=[] if depth == 1 else rectangles(n, depth), + workload=dict(active_cells_by_level=cells, level_hops=hops, root_intervals=16 * n, + active_collision_updates_by_level=[c * h for c, h in zip(cells, hops)], + active_collision_updates=sum(c * h for c, h in zip(cells, hops))), + actual_build_sidecars=builds, rows=rows, completed=True, + flow="shear_wave", mode=list(sw.MODE), polarization=sw.E.tolist(), + angle_deg=sw.ANGLE_DEG, phase=sw.PHI, eps=sw.eps(n)) + + +class ContractTests(unittest.TestCase): + def setUp(self): + import assess_shear_wave + self.assess = assess_shear_wave.assess + self.records = [receipt(128, 1), receipt(256, 1), receipt(128, 2), + receipt(128, 2, "projected_linear")] + + def test_complete_contract_passes(self): + result = self.assess(self.records, "r") + self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v]) + self.assertEqual(result["status"], "PASS") + + def test_reproduced_false_passes_are_rejected(self): + a = self.assess + self.assertEqual(a(self.records[:2], "r")["status"], "INVALID_SET") # empty refined + dup = [*self.records[:3], receipt(128, 2)] + self.assertEqual(a(dup, "r")["status"], "INVALID_SET") # duplicate constant-only + for mutate in ( + lambda r: r.pop("source"), + lambda r: r.update(flow="taylor_green"), + lambda r: r.update(mode=[1, 2]), + lambda r: r.update(re_L=1000), + lambda r: r.update(refinements=[[(0, 0, 8, 8)]]), + lambda r: r["actual_build_sidecars"][0].update(arch="sm_80"), + lambda r: r["rows"][0].update(step=1, tstar=U / 128), # wrong initial time + lambda r: r["source"].update(shear_wave_sha256="0" * 64), + ): + records = [dict(r, rows=[dict(x) for x in r["rows"]], + source=dict(r["source"]), + actual_build_sidecars=[dict(b) for b in r["actual_build_sidecars"]]) + for r in self.records] + mutate(records[2]) + self.assertEqual(a(records, "r")["status"], "INVALID_SET") + mixed = [dict(r) for r in self.records] + mixed[1] = dict(mixed[1], source=dict(mixed[1]["source"], revision="other")) + self.assertEqual(a(mixed, "r")["status"], "INVALID_SET") + self.assertEqual(a(self.records, "wrong-revision")["status"], "INVALID_SET") + + class DefaultPathTests(unittest.TestCase): def test_frozen_oracle_is_byte_identical(self): digest = hashlib.sha256((HERE / "oracle.py").read_bytes()).hexdigest()