204 lines
8.6 KiB
Python
204 lines
8.6 KiB
Python
#!/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())
|