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 <noreply@anthropic.com>
This commit is contained in:
Claude
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
9748ef91ef
commit
335624c27b
@@ -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
|
||||
`<output>.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 <sha>`.
|
||||
- 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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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")})
|
||||
|
||||
@@ -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")
|
||||
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user