125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
"""Frozen assessment of the complete 16-case matrix, independent of GPU launch."""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
|
|
from oracle import TIMES, LIMITS, U
|
|
|
|
# Numerical floors reflect finite-Mach/FP32 and equilibrium startup, not observed runs.
|
|
FLOORS = dict(velocity_l2=0.002, pressure_l2=0.02, energy_relative=0.004)
|
|
BASELINE_FACTOR = 2.0
|
|
REFINEMENT_FACTOR = 0.9
|
|
|
|
|
|
def assess(records):
|
|
indexed = {}
|
|
for r in records:
|
|
key = (r["re_L"], r["n"], r["depth"])
|
|
if key in indexed:
|
|
raise ValueError(f"duplicate case {key}")
|
|
indexed[key] = r
|
|
expected = {(re, n, 1) for re in (100, 1000) for n in (32, 64, 128, 256)}
|
|
expected |= {(re, n, d) for re in (100, 1000) for n in (32, 64) for d in (2, 3)}
|
|
missing = sorted(expected - indexed.keys())
|
|
checks = {
|
|
"complete_matrix": not missing,
|
|
"no_extra_cases": set(indexed) <= expected,
|
|
}
|
|
identities = {json.dumps(r["source"], sort_keys=True) for r in records}
|
|
checks["same_source_and_harness"] = len(identities) == 1
|
|
checks["clean_source"] = bool(records) and all(
|
|
not r["source"]["dirty"] for r in records
|
|
)
|
|
for key, r in indexed.items():
|
|
label = str(key)
|
|
rows = r["rows"]
|
|
checks[label + " complete"] = r["completed"] and len(rows) == len(TIMES)
|
|
checks[label + " scales"] = (
|
|
r["U"] == U and abs(r["viscosity"] - U * r["n"] / r["re_L"]) < 1e-15
|
|
)
|
|
checks[label + " times"] = len(rows) == len(TIMES) and all(
|
|
abs(v["tstar"] - t) < 1e-12 for v, t in zip(rows, TIMES)
|
|
)
|
|
checks[label + " absolute"] = bool(rows) and all(
|
|
0 <= row[k] <= limit for row in rows for k, limit in LIMITS.items()
|
|
)
|
|
comparisons = []
|
|
|
|
def worst(key, metric):
|
|
values = [r[metric] for r in indexed.get(key, {}).get("rows", [])[1:]]
|
|
return max(values) if values and all(math.isfinite(v) for v in values) else None
|
|
|
|
# Diagnostics survive failed screens and partial pilot matrices. Missing or
|
|
# nonfinite operands are explicit nulls; they cannot acquire a passing gate.
|
|
for re in (100, 1000):
|
|
for depth in (2, 3):
|
|
for n in (32, 64):
|
|
if (re, n, depth) not in indexed:
|
|
continue
|
|
for metric, floor in FLOORS.items():
|
|
amr = worst((re, n, depth), metric)
|
|
coarse = worst((re, n, 1), metric)
|
|
fine = worst((re, n * 2 ** (depth - 1), 1), metric)
|
|
ok = (
|
|
amr is not None
|
|
and coarse is not None
|
|
and amr <= max(BASELINE_FACTOR * coarse, floor)
|
|
)
|
|
comparisons.append(
|
|
dict(
|
|
re_L=re,
|
|
n=n,
|
|
depth=depth,
|
|
metric=metric,
|
|
amr=amr,
|
|
uniform_coarse=coarse,
|
|
uniform_finest=fine,
|
|
amr_to_finest=(
|
|
None
|
|
if amr is None or fine is None
|
|
else amr / max(fine, 1e-30)
|
|
),
|
|
passed=ok,
|
|
)
|
|
)
|
|
checks[f"baseline {re}/{n}/{depth}/{metric}"] = ok
|
|
for metric, floor in FLOORS.items():
|
|
low = worst((re, 32, depth), metric)
|
|
high = worst((re, 64, depth), metric)
|
|
checks[f"refinement {re}/{depth}/{metric}"] = (
|
|
low is not None
|
|
and high is not None
|
|
and high <= max(REFINEMENT_FACTOR * low, floor)
|
|
)
|
|
for metric, floor in FLOORS.items():
|
|
low = worst((re, 32, 1), metric)
|
|
high = worst((re, 256, 1), metric)
|
|
checks[f"uniform refinement {re}/{metric}"] = (
|
|
low is not None
|
|
and high is not None
|
|
and high <= max(REFINEMENT_FACTOR * low, floor)
|
|
)
|
|
return dict(
|
|
passed=all(checks.values()),
|
|
missing=missing,
|
|
checks=checks,
|
|
comparisons=comparisons,
|
|
floors=FLOORS,
|
|
baseline_factor=BASELINE_FACTOR,
|
|
refinement_factor=REFINEMENT_FACTOR,
|
|
scope="2D periodic smooth-decay AMR verification only; no turbulence, LES, wall or Jet certification",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("files", type=Path, nargs="+")
|
|
p.add_argument("--output", type=Path, required=True)
|
|
a = p.parse_args()
|
|
result = assess([json.loads(path.read_text()) for path in a.files])
|
|
a.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
|
|
print(json.dumps(result, indent=2))
|
|
raise SystemExit(0 if result["passed"] else 1)
|