#!/usr/bin/env python3 """Report a plateau endpoint and gate strict solver convergence plus eight virials. Usage: check_plateau_physics.py RESULT_FOLDER Exit 0: solver_and_virial_milestone passes. Exit 3: it does not pass. Exit 2: required endpoint data is missing or malformed. Standard-library only. Local errors remain measurements; passing this gate is not full physical validation. """ import argparse import csv import json import math from pathlib import Path import sys THRESHOLD = 1.0e-5 QUADRATURES = (14, 18) VIRIALS = { "pressure_from_enthalpy_energy": "enthalpy_virial_error", "pressure_from_enthalpy_force": "enthalpy_force_virial_error", "pressure_from_density_energy": "virial_error", "pressure_from_density_force": "force_virial_error", } SIGN_METRICS = ( "minimum_density", "maximum_density", "negative_density_samples", "negative_density_mass_fraction", "negative_density_volume_fraction", "minimum_enthalpy", "maximum_enthalpy", "negative_enthalpy_samples", "negative_enthalpy_volume_fraction", ) CONSERVATION_METRICS = ( "mass", "mass_absolute_error", "angular_momentum_z", "angular_momentum_absolute_error", ) LOCAL_METRICS = ( "eos_density_mismatch_maximum", "eos_density_mismatch_maximum_scaled", "eos_density_mismatch_rms", "eos_density_mismatch_rms_scaled", "bernoulli_raw_rms", "bernoulli_raw_rms_scaled", "bernoulli_mean_subtracted_rms", "bernoulli_mean_subtracted_rms_scaled", "bernoulli_sampled_range", "hydrostatic_gradient_rms", ) PRESSURE_METRICS = ( "pressure_from_density_defined_everywhere", "pressure_integral", "pressure_from_enthalpy_defined_everywhere", "enthalpy_pressure_integral", ) REQUIRED_METRICS = ( *VIRIALS.values(), *SIGN_METRICS, *CONSERVATION_METRICS, *LOCAL_METRICS, *PRESSURE_METRICS, "complete_volume_measurement", ) def measurement(value): """Keep JSON standard-compliant without losing the reason for a null value.""" result = {"value": value if math.isfinite(value) else None, "finite": math.isfinite(value)} if not result["finite"]: result["nonfinite_value"] = "nan" if math.isnan(value) else ("+inf" if value > 0 else "-inf") return result def read_metadata(path): result = {} for line in path.read_text().splitlines(): if not line.strip(): continue if "=" not in line: raise ValueError(f"Malformed metadata line in {path}: {line!r}") key, value = line.split("=", 1) if not key or key in result: raise ValueError(f"Empty or duplicate metadata key in {path}: {key!r}") result[key] = value if result.get("converged") not in ("0", "1"): raise ValueError("metadata.txt must contain a completed converged=0 or converged=1 result") if "accepted_steps" not in result or int(result["accepted_steps"]) < 0: raise ValueError("metadata.txt must contain a nonnegative accepted_steps count") return result def read_physical(path): result = {q: {} for q in QUADRATURES} with path.open(newline="") as stream: reader = csv.DictReader(stream) if reader.fieldnames != ["quadrature", "metric", "value"]: raise ValueError("final_physical.csv requires the header quadrature,metric,value") for row in reader: if None in row or any(value is None for value in row.values()): raise ValueError("Malformed row in final_physical.csv") q = int(row["quadrature"]) if q not in result: continue key = row["metric"] if not key or key in result[q]: raise ValueError(f"Empty or duplicate physical metric at quadrature {q}: {key!r}") result[q][key] = float(row["value"]) for q, values in result.items(): missing = [key for key in REQUIRED_METRICS if key not in values] if missing: raise ValueError(f"Missing quadrature {q} endpoint metrics: {', '.join(missing)}") for key in ("complete_volume_measurement", "pressure_from_density_defined_everywhere", "pressure_from_enthalpy_defined_everywhere"): if values[key] not in (0.0, 1.0): raise ValueError(f"Invalid boolean metric at quadrature {q}: {key}") for key in VIRIALS.values(): if math.isfinite(values[key]) and values[key] < 0: raise ValueError(f"A virial error must be nonnegative: quadrature {q}, {key}") return result def pressure_report(values, kind): flag = values[f"pressure_from_{kind}_defined_everywhere"] == 1.0 integral = values["pressure_integral" if kind == "density" else "enthalpy_pressure_integral"] return { "defined_at_all_samples": flag, "integral": measurement(integral), "valid": flag and math.isfinite(integral), } def negative_samples_detected(values, kind): minimum = values[f"minimum_{kind}"] count = values[f"negative_{kind}_samples"] if not math.isfinite(minimum) or not math.isfinite(count): return None return minimum < 0.0 or count > 0.0 def assess(folder): meta = read_metadata(folder / "metadata.txt") physical = read_physical(folder / "final_physical.csv") per_quadrature = {} all_virials_pass = True for q, values in physical.items(): complete = values["complete_volume_measurement"] == 1.0 pressure = {kind: pressure_report(values, kind) for kind in ("enthalpy", "density")} virials = {} for name, key in VIRIALS.items(): value = values[key] kind = "density" if "density" in name else "enthalpy" check = measurement(value) if not complete or not pressure[kind]["valid"] or not math.isfinite(value): check["status"] = "undefined" check["reason"] = ( "incomplete_volume_measurement" if not complete else "pressure_not_defined_or_finite" if not pressure[kind]["valid"] else "nonfinite_virial" ) else: check["status"] = "pass" if value <= THRESHOLD else "fail" all_virials_pass = all_virials_pass and check["status"] == "pass" virials[name] = check per_quadrature[str(q)] = { "complete_volume_measurement": complete, "virials": virials, "pressure": pressure, "sampled_signs": { "negative_density_detected": negative_samples_detected(values, "density"), "negative_enthalpy_detected": negative_samples_detected(values, "enthalpy"), **{key: measurement(values[key]) for key in SIGN_METRICS}, }, "conservation": {key: measurement(values[key]) for key in CONSERVATION_METRICS}, "local_measurements_without_acceptance_tolerances": { key: measurement(values[key]) for key in LOCAL_METRICS }, } strict_converged = meta["converged"] == "1" passed = strict_converged and all_virials_pass report = { "schema_version": 1, "result_folder": str(folder.resolve()), "solver_and_virial_milestone": "pass" if passed else "fail", "strict_solver": { "converged": strict_converged, "accepted_steps": int(meta["accepted_steps"]), "failure": meta.get("failure"), "absolute_tolerance": meta.get("atol"), "relative_tolerance": meta.get("rtol"), }, "virial_threshold": THRESHOLD, "all_eight_virial_checks_pass": all_virials_pass, "quadratures": per_quadrature, "interpretation": [ "The milestone requires strict recorded solver convergence and all four defined, finite virial errors <= 1e-5 at both quadratures.", "The milestone is not full physical validation. Mass/J errors and local EOS/Bernoulli measurements have no acceptance tolerances here.", "Density and enthalpy signs are quadrature samples; sampled extrema and local errors are not certified pointwise bounds.", "An undefined pressure-from-density virial is kept separate from a finite virial that exceeds the threshold.", ], } return report, 0 if passed else 3 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("result_folder", type=Path) args = parser.parse_args() try: report, code = assess(args.result_folder) except (OSError, ValueError, csv.Error) as error: report, code = {"schema_version": 1, "input_error": str(error)}, 2 print(json.dumps(report, indent=2, allow_nan=False)) return code if __name__ == "__main__": sys.exit(main())