ClaudeandClaude Opus 5.5
3554cb2fde
- run_case: finite_row accepts list diagnostics; sample() shared by sample and READY paths; verify_source pins shear_wave.py path+SHA256 for --flow shear_wave (default TG flow unchanged, does not import it). - assess_shear_wave.py: fixed N128 four-case contract and 4-JSON CLI, reusing assess_projected_pairs provenance/build/workload/time checks plus shear metadata; missing/duplicate/mixed/extra records -> INVALID_SET. - tests: mocked refined runner sample (list diagnostics, NaN, TG unchanged), contract pass + reproduced false-pass mutations, divergence check uses analytic E.k=0 and a derived h^2 truncation + roundoff bound with h-halving scaling instead of a fixed 1e-12. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
82 lines
3.8 KiB
Python
82 lines
3.8 KiB
Python
"""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)
|