"""Explicit tiny CUDA explosion parity and constructor/one-interval smoke check. No installs or device selection defaults. Run only on the authorized GPU host. This is operator/runtime evidence, not Taylor-Green accuracy qualification. """ import argparse import hashlib import json import os from pathlib import Path import sys import numpy as np def check_case(ctx, shifted, projected): import pycuda.driver as cuda from CelerisLab.config import LBMConfig from CelerisLab.lbm.refinement_stepper import FixedPatchStepper from CelerisLab.lbm.refinement_tables import projected_explosion_tables from CelerisLab.lbm.refinement import ACTIVE from test_projected_explosion import reconstruct, W cfg = LBMConfig(nx=16, ny=16, viscosity=.1, velocity=0, ddf_shifting=shifted) cfg.validate() options = {"explosion": "projected_linear"} if projected else {} # Omitting explosion in the constant case tests the production default. with FixedPatchStepper(cfg, context=ctx, refinements=[[(4, 4, 12, 12)]], level_blocks=[(4, 4), (8, 4)], **options) as run: parent, child = run._levels pair = run._pairs[0] patch = run.hierarchy.pairs[0] coarse_cells = int(np.prod(parent.layout.local_shape)) fine_cells = int(np.prod(child.layout.local_shape)) assert coarse_cells != fine_cells assert pair.parent_cells == coarse_cells assert len(pair.interpolation) == (2 if projected else 0) donor = patch.ghost_coarse.astype(np.intp) # Random variation must exercise the gradient, not just a constant field. c = np.random.default_rng(11).normal(0, .001, (9, 16, 16)) if not shifted: c += W[:, None, None] c = c.astype(np.float32) legal = np.zeros(256, bool) legal[donor] = True c.reshape(9, -1)[:, ~legal] = np.nan coarse = np.full(np.prod(parent.layout.blocked_shape), np.nan, np.float32) for q in range(9): coarse[parent.layout.slot(parent.layout.owner.ravel()[legal], q)] = c.reshape(9, -1)[q, legal] fine = np.full(np.prod(child.layout.blocked_shape), np.nan, np.float32) cuda.memcpy_htod(parent.current, coarse) cuda.memcpy_htod(child.current, fine) calls = [] real_hop = run._hop try: # Execute the production explosion launch exactly; isolate it from # subsequent streaming/collision without replacing the kernel/tables. run._hop = lambda *args, **kwargs: calls.append(args) run._advance(1, parent.current, block=run._block, stream=run._stream) run._stream.synchronize() finally: run._hop = real_hop assert calls == [(1, child.current, child.spare), (1, child.spare, child.current)] cuda.memcpy_dtoh(fine, child.current) targets = child.layout.owner.ravel()[patch.ghost_fine] slots = np.array([child.layout.slot(targets, q) for q in range(9)]) actual = fine[slots] neighbors, offsets = projected_explosion_tables(patch, parent.layout) inverse = np.full(parent.layout.nblocks * coarse_cells, -1, dtype=int) inverse[parent.layout.owner.ravel()] = np.arange(256) constant = c.reshape(9, -1)[:, donor].astype(float) expected = reconstruct(c.astype(float), patch, inverse[neighbors], offsets)[0] if projected else constant assert np.isfinite(actual).all() if projected: assert np.max(np.abs(expected - constant)) > 1e-5 np.testing.assert_allclose(actual, expected, atol=6e-8, rtol=2e-7) untouched = np.ones(fine.size, bool) untouched[slots.ravel()] = False assert np.isnan(fine[untouched]).all(), "explosion wrote outside target slots" error = float(np.max(np.abs(actual - expected))) # Restore ordinary execution and exercise real constructor wiring, # stream-only ghost hops, restriction, halo copies and collision once. initial = [np.broadcast_to(W[:, None, None], (9, *k.shape)).copy() for k in run.hierarchy.kinds] run.initialize(*initial) run.step(1) for values, kind in zip(run.download(), run.hierarchy.kinds): active = values[:, kind == ACTIVE] assert np.isfinite(active).all() np.testing.assert_allclose(active, np.broadcast_to(W[:, None], active.shape), atol=3e-7, rtol=1e-6) assert run.step_count == 1 return dict(shifted=shifted, explosion="projected_linear" if projected else "constant_default", parent_stride=coarse_cells, child_stride=fine_cells, compared_slots=int(actual.size), max_abs_error=error, poisoned_nondonors=True, untouched_slots_preserved=True, equilibrium_step=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--run-gpu", action="store_true", help="Explicitly opt into GPU execution") args = parser.parse_args() selected = os.environ.get("CUDA_VISIBLE_DEVICES", "") if not args.run_gpu or not selected or "," in selected or selected == "-1": parser.error("--run-gpu and exactly one explicit physical CUDA_VISIBLE_DEVICES are required") if args.output.exists(): parser.error("output already exists; choose a new receipt path") root = args.repo.resolve() sys.path[:0] = [str(root / "src"), str(root / "tests/unit"), str(root / "tests/validation/taylor_green")] import CelerisLab if Path(CelerisLab.__file__).resolve() != root / "src/CelerisLab/__init__.py": raise RuntimeError("Wrong imported source checkout") from run_case import provenance report = dict(source=provenance(root), numpy_oracle_sha256=hashlib.sha256((root / "tests/unit/test_projected_explosion.py").read_bytes()).hexdigest(), cuda_visible_devices=selected, logical_device=0, completed=False, cases=[]) args.output.parent.mkdir(parents=True, exist_ok=True) def save(): args.output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") save() try: from CelerisLab.cuda.context import CudaContext with CudaContext(0) as ctx: report["device"] = ctx.device.name() for shifted in (False, True): for projected in (False, True): report["cases"].append(check_case(ctx, shifted, projected)) save() report["completed"] = True save() except Exception as exc: report["error"] = f"{type(exc).__name__}: {exc}" save() raise print(json.dumps(report, indent=2)) if __name__ == "__main__": main()