#!/usr/bin/env python3 """Compare two completed n=1 solves separated by one uniform h-refinement. Standard library only; never runs Newton or modifies input data. Exit 0 means the comparison is usable, not that physical verification passed; 3 indicates incompatible/incomplete comparison data, and 2 an execution/input error. """ import argparse import csv import hashlib import math from pathlib import Path import sys MODEL = "nonrotating_n1_fixed_mass_fixed_central_density_zero_surface_pressure" TEXT_KEYS = ("model", "normalization") CONSTANT_KEYS = ("G", "M", "R", "K", "rho_c") ORDER_KEYS = ("polynomial_increment", "density_order", "enthalpy_order", "potential_order", "gravity_flux_order", "displacement_order") TOLERANCE_KEYS = ("absolute_tolerance", "relative_tolerance", "linear_tolerance") ITERATION_KEYS = ("max_newton", "max_linear_iterations") # Optional rows were added after the original coarse solve. Their absence is # visible but does not erase the usable rows from that historical dataset. METRICS = ( ("density_relative_l2_error", "Interior density relative L2", True), ("enthalpy_relative_l2_error", "Interior enthalpy relative L2", True), ("potential_relative_l2_error", "Interior potential relative L2", True), ("gravity_gradient_relative_l2_error", "Interior gravity-gradient relative L2", True), ("pressure_relative_l2_error", "Interior pressure relative L2", True), ("surface_radius_relative_rms_error", "Surface radius RMS / R", True), ("volume_radius_relative_error", "Volume-equivalent radius relative error", True), ("virial_error", "Virial error, P(rho)", True), ("force_virial_error", "Force virial error, P(rho)", True), ("enthalpy_virial_error", "Virial error, P(h)", False), ("enthalpy_force_virial_error", "Force virial error, P(h)", False), ("gravity_energy_consistency", "Gravity-energy consistency error", True), ("mass_relative_error", "Mass relative error", True), ("binding_relative_error", "Binding-energy relative error", True), ("pressure_integral_relative_error", "Pressure-integral relative error", True), ("moment_of_inertia_relative_error", "Moment-of-inertia relative error", True), ("eos_enthalpy_scaled_rms", "Pointwise EOS RMS / central enthalpy", True), ("bernoulli_scaled_rms_variation", "Bernoulli RMS variation / GM/R", True), ("bernoulli_scaled_range", "Sampled Bernoulli range / GM/R", False), ("bernoulli_mean_scaled_error", "Bernoulli mean scaled error", False), ("normalized_bordered_residual", "Normalized bordered residual", True), ("normalized_unbordered_residual", "Normalized unbordered residual", True), ("normalized_central_border_action", "Normalized central-border action", False), ("quadrature_virial_absolute_change", "Virial quadrature-order change", True), ("quadrature_binding_relative_change", "Binding-energy quadrature-order change", True), ) EXTERIOR = ( ("potential_mean_error_scaled", "Finite-exterior maximum absolute shell-mean potential error"), ("potential_rms_error_scaled", "Finite-exterior maximum shell RMS potential error"), ("gravity_radial_mean_error_scaled", "Finite-exterior maximum absolute shell-mean radial-gravity error"), ("gravity_radial_rms_error_scaled", "Finite-exterior maximum shell RMS radial-gravity error"), ) def number(value): try: return float(value) except (ValueError, TypeError): return math.nan def read_metadata(path): result = {} for line in path.read_text().splitlines(): if "=" not in line: continue key, value = line.split("=", 1) if key in result: raise ValueError(f"Duplicate metadata key {key!r}: {path}") result[key] = value return result def read_metrics(path): result = {} with path.open(newline="") as stream: reader = csv.DictReader(stream) if reader.fieldnames != ["metric", "value"]: raise ValueError(f"Expected metric,value CSV schema: {path}") for row in reader: key = row["metric"] if not key or key in result or None in row: raise ValueError(f"Malformed/duplicate metric row: {path}") result[key] = number(row["value"]) return result def read_dataset(directory): directory = directory.resolve() return {"directory": directory, "metadata": read_metadata(directory / "metadata.txt"), "metrics": read_metrics(directory / "physical_metrics.csv")} def integer(value): try: # Unlike float->int, this rejects a nonintegral or nonfinite count. return int(value) except (ValueError, TypeError): return None def compatibility(coarse, fine, coarse_control=None, fine_control=None): checks = [] def check(name, okay, detail): checks.append((name, None if okay is None else bool(okay), detail)) def match(left, right, keys, numeric=False, integral=False, prefix="solve"): for key in keys: a, b = left.get(key), right.get(key) if integral: okay = integer(a) is not None and integer(a) == integer(b) elif numeric: okay = math.isfinite(number(a)) and number(a) == number(b) else: okay = a is not None and a == b check(f"{prefix}: matching {key}", okay, f"{a!r} / {b!r}") for name, data in (("coarse", coarse), ("fine", fine)): meta = data["metadata"] check(f"{name}: completed converged solve", meta.get("mode") == "solve" and meta.get("solver_converged") == "1" and meta.get("physical_screen_passed") in ("0", "1"), f"mode={meta.get('mode')}, converged={meta.get('solver_converged')}, " f"physical screen={meta.get('physical_screen_passed')} (not required to pass)") check(f"{name}: supported model and MPI ranks", meta.get("model") == MODEL and meta.get("mpi_ranks") == "1", "Requires this single-rank nonrotating n=1 benchmark.") check(f"{name}: positive finite constants", all(math.isfinite(number(meta.get(key))) and number(meta.get(key)) > 0 for key in CONSTANT_KEYS), "G, M, R, K, rho_c must be positive and finite.") check(f"{name}: zero rotation", data["metrics"].get("angular_velocity_norm") == 0.0, "Requires a saved angular_velocity_norm of exactly zero.") for metric, _, required in METRICS: if required: value = data["metrics"].get(metric) check(f"{name}: usable {metric}", value is not None and math.isfinite(value) and value >= 0, f"Saved value: {value!r}") a, b = coarse["metadata"], fine["metadata"] match(a, b, TEXT_KEYS) match(a, b, CONSTANT_KEYS + TOLERANCE_KEYS, numeric=True) match(a, b, ORDER_KEYS, integral=True) match(a, b, ITERATION_KEYS, integral=True) elements_a, elements_b = integer(a.get("elements")), integer(b.get("elements")) check("one uniform hexahedral level: 8x elements", elements_a is not None and elements_a > 0 and elements_b == 8 * elements_a, f"{elements_a} -> {elements_b}; element counts alone do not establish mesh ancestry.") quad_a, quad_b = coarse["metrics"].get("quadrature_order"), fine["metrics"].get("quadrature_order") check("matching diagnostic quadrature order", quad_a is not None and math.isfinite(quad_a) and quad_a == quad_b, f"{quad_a!r} / {quad_b!r}") for name, solve, control in (("coarse control", coarse, coarse_control), ("fine control", fine, fine_control)): if control is None: continue meta = control["metadata"] check(f"{name}: completed passing analytic control", meta.get("mode") == "analytic-mesh" and meta.get("physical_screen_passed") == "1" and meta.get("mpi_ranks") == "1", "Analytic-control status is read, not fabricated.") match(solve["metadata"], meta, TEXT_KEYS, prefix=name) match(solve["metadata"], meta, CONSTANT_KEYS, numeric=True, prefix=name) match(solve["metadata"], meta, ORDER_KEYS + ("elements",), integral=True, prefix=name) left = solve["metrics"].get("quadrature_order") right = control["metrics"].get("quadrature_order") check(f"{name}: matching diagnostic quadrature order", left is not None and math.isfinite(left) and left == right, f"{left!r} / {right!r}") solve_path = solve.get("directory") control_path = control.get("directory") solve_snapshot = solve_path / "input.smesh" if solve_path is not None else None control_snapshot = control_path / "input.smesh" if control_path is not None else None if solve_snapshot is not None and control_snapshot is not None and solve_snapshot.is_file() and control_snapshot.is_file(): solve_hash, control_hash = digest(solve_snapshot), digest(control_snapshot) check(f"{name}: identical input snapshot SHA-256", solve_hash == control_hash, f"{solve_hash} / {control_hash}") else: check(f"{name}: identical input snapshot SHA-256", None, "Unavailable: one or both snapshots absent; equal geometry is not established by the element-count check.") return checks def checks_satisfied(checks): # Optional unavailable provenance checks are not fabricated passes. return all(okay is not False for _, okay, _ in checks) def reduction(coarse, fine, eligible=True): if coarse is None or fine is None: return None, None, "missing" if not math.isfinite(coarse) or not math.isfinite(fine): return None, None, "nonfinite" if coarse < 0 or fine < 0: return None, None, "negative error magnitude" if not eligible: return None, None, "suppressed: compatibility checks failed" if fine == 0: return None, None, "both zero; no rate" if coarse == 0 else "fine zero; no finite rate" if coarse == 0: return 0.0, None, "coarse zero; no finite rate" ratio = coarse / fine rate = math.log2(coarse) - math.log2(fine) if ratio == 0: return ratio, rate, "ratio underflow; log-rate remains finite" return ratio, rate, "two-level observation" if math.isfinite(ratio) else "ratio overflow; log-rate remains finite" def exterior_diagnostics(data): path = data["directory"] / "radial_profiles.csv" if not path.exists(): return {}, "not available (radial_profiles.csv absent)" with path.open(newline="") as stream: rows = [row for row in csv.DictReader(stream) if number(row.get("r_over_R")) > 1.0] if not rows: return {}, "not available (no requested exterior shells)" signature = sorted({(str(row.get("mu_points")), str(row.get("phi_points"))) for row in rows}) description = (f"{len(rows)} shells, r/R={rows[0].get('r_over_R')}..{rows[-1].get('r_over_R')}, " f"angular grids={signature}; sampled-shell diagnostics only") values = {} for column, _ in EXTERIOR: samples = [number(row.get(column)) for row in rows] field = "potential" if column.startswith("potential_") else "gravity_radial" complete = all(number(row.get("located_weight_fraction")) >= 1.0 - 1e-12 and number(row.get(field + "_valid_weight_fraction")) >= 1.0 - 1e-12 for row in rows) values[column] = (max(abs(value) for value in samples) if complete and all(math.isfinite(value) for value in samples) else math.nan) return values, description def digest(path): if not path.is_file(): return "not available" value = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): value.update(chunk) return value.hexdigest() def display(value): if value is None: return "not available" if not math.isfinite(value): return str(value) return f"{value:.8g}" def markdown(value): return str(value).replace("|", "\\|").replace("\n", " ") def write_comparison(coarse, fine, output, coarse_control=None, fine_control=None): checks = compatibility(coarse, fine, coarse_control, fine_control) eligible = checks_satisfied(checks) rows = [] for metric, label, required in METRICS: a, b = coarse["metrics"].get(metric), fine["metrics"].get(metric) ratio, rate, status = reduction(a, b, eligible) rows.append({"metric": metric, "description": label, "category": "volume_surface_or_solver", "required_data": int(required), "coarse": a, "fine": b, "coarse_over_fine": ratio, "observed_log2_rate": rate, "status": status, "coarse_control": coarse_control["metrics"].get(metric) if coarse_control else None, "fine_control": fine_control["metrics"].get(metric) if fine_control else None}) exterior_a, sampling_a = exterior_diagnostics(coarse) exterior_b, sampling_b = exterior_diagnostics(fine) for metric, label in EXTERIOR: a, b = exterior_a.get(metric), exterior_b.get(metric) status = "diagnostic only; no gate/rate (angular/radial samples need independent convergence checks)" if a is None or b is None or not math.isfinite(a) or not math.isfinite(b): status = "diagnostic unavailable or incomplete; no gate/rate" rows.append({"metric": "sampled_exterior_max_" + metric, "description": label, "category": "sampled_exterior_diagnostic_only", "required_data": 0, "coarse": a, "fine": b, "coarse_over_fine": None, "observed_log2_rate": None, "status": status, "coarse_control": None, "fine_control": None}) # No overwrite, including output paths that alias an input directory. output.mkdir() with (output / "comparison.csv").open("x", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) lines = ["# Two-level polytrope refinement comparison", "", "Required comparison checks: " + ("satisfied." if eligible else "**FAILED; h-rates suppressed.**"), "", "This is an observed two-level error comparison, not an established asymptotic order or a physical-accuracy certificate. " "Exit 0 indicates usable comparison data, not a passing physical screen. " "Reported rates are log2(E_coarse/E_fine), conditional on one uniform level halving the logical cell scale. " "An 8x element ratio alone cannot prove mesh ancestry, stable source code, or identical geometry construction.", "", "Optional unavailable snapshot checks are marked unavailable, not passed; they do not establish identical control/solve geometry.", "", "Interior L2 errors use the saved full 3D stellar-volume diagnostics, not fitted spherical means. " "A negative rate means this error increased. Missing/nonfinite/negative errors and zero denominators never receive a fabricated rate. " "No existing physical screening budget is changed or reinterpreted.", "", "## Provenance", ""] for name, data in (("Coarse solve", coarse), ("Fine solve", fine), ("Coarse analytic control", coarse_control), ("Fine analytic control", fine_control)): if data is None: lines.append(f"- {name}: not supplied.") continue meta = data["metadata"] lines += [f"- {name}: `{markdown(data['directory'])}`; elements={markdown(meta.get('elements'))}; " f"saved physical_screen_passed={markdown(meta.get('physical_screen_passed'))}."] for filename in ("metadata.txt", "physical_metrics.csv", "input.smesh"): lines.append(f" - {filename} SHA-256: `{digest(data['directory'] / filename)}`") lines.append(f" - compiled={markdown(meta.get('compiled', 'not available'))}; compiler={markdown(meta.get('compiler', 'not available'))}.") lines += ["", "Input hashes identify these artifacts, not the production source/library version. " "Confirm unchanged physics/seed/normalization/mapping code separately. Geometry-aware STROID refinement can regenerate " "the curved mesh, rather than merely subdividing its old polynomial geometry.", "", "## Error comparison", "", "| Diagnostic | Coarse | Fine | E_coarse/E_fine | Observed log2 rate | Coarse control | Fine control | Status |", "|---|---:|---:|---:|---:|---:|---:|---|"] for row in rows: lines.append("| " + " | ".join(markdown(value) for value in ( row["description"], display(row["coarse"]), display(row["fine"]), display(row["coarse_over_fine"]), display(row["observed_log2_rate"]), display(row["coarse_control"]), display(row["fine_control"]), row["status"])) + " |") lines += ["", "Analytic-control field errors may be zero by construction; they are not FE best-approximation errors. " "Controls are shown without subtraction from solve errors. EOS projection floors, cancellation in integral errors, " "sampled extrema, and algebraic residual floors can produce rates unrelated to formal FE approximation order.", "", "## Finite-exterior sampling (diagnostic only)", "", f"- Coarse: {markdown(sampling_a)}.", f"- Fine: {markdown(sampling_b)}.", "", "Exterior rows are maxima across the saved requested shells with r/R > 1, not pointwise global maxima or volume L2 norms. " "Potential is scaled by GM/R and radial gravity by GM/R^2. No rate or pass gate is inferred from these samples; " "missing or incomplete shell coverage remains unavailable.", "", "## Compatibility checks", "", "| Check | Satisfied | Detail |", "|---|---|---|"] lines.extend(f"| {markdown(name)} | {'unavailable' if okay is None else ('yes' if okay else 'NO')} | {markdown(detail)} |" for name, okay, detail in checks) (output / "comparison.md").write_text("\n".join(lines) + "\n") return eligible def self_check(): """Synthetic-only checks; no project data, solver, or persistent outputs.""" import tempfile import unittest class ComparisonChecks(unittest.TestCase): @staticmethod def fixture(elements): meta = {key: "1" for key in CONSTANT_KEYS + ORDER_KEYS} meta.update({"model": MODEL, "normalization": "synthetic", "mode": "solve", "mpi_ranks": "1", "solver_converged": "1", "physical_screen_passed": "0", "elements": str(elements), "max_newton": "8", "max_linear_iterations": "80", "absolute_tolerance": "1e-8", "relative_tolerance": "1e-8", "linear_tolerance": ".03"}) metrics = {key: 1e-4 for key, _, _ in METRICS} metrics.update({"angular_velocity_norm": 0.0, "quadrature_order": 18.0}) return {"metadata": meta, "metrics": metrics} def test_rates_and_edge_cases(self): self.assertEqual(reduction(8.0, 1.0)[:2], (8.0, 3.0)) self.assertEqual(reduction(1.0, 4.0)[:2], (.25, -2.0)) for a, b in ((None, 1), (math.nan, 1), (1, math.inf), (-1, 1), (0, 0), (1, 0)): self.assertIsNone(reduction(a, b)[1]) self.assertEqual(reduction(0, 1)[:2], (0.0, None)) self.assertEqual(reduction(8, 1, False)[:2], (None, None)) def test_compatibility(self): a, b = self.fixture(19), self.fixture(152) self.assertTrue(checks_satisfied(compatibility(a, b))) for key, bad in (("mode", "replay"), ("solver_converged", "0"), ("elements", "151"), ("elements", "152.5"), ("G", "nan"), ("density_order", "2"), ("linear_tolerance", ".02"), ("max_newton", "9"), ("max_linear_iterations", "81")): broken = {"metadata": dict(b["metadata"], **{key: bad}), "metrics": b["metrics"]} self.assertFalse(checks_satisfied(compatibility(a, broken)), key) del b["metrics"]["density_relative_l2_error"] self.assertFalse(checks_satisfied(compatibility(a, b))) def test_control(self): a, b, control = self.fixture(19), self.fixture(152), self.fixture(19) control["metadata"].update(mode="analytic-mesh", physical_screen_passed="1") self.assertTrue(checks_satisfied(compatibility(a, b, control))) self.assertTrue(any(okay is None for _, okay, _ in compatibility(a, b, control))) control["metadata"]["elements"] = "152" self.assertFalse(checks_satisfied(compatibility(a, b, control))) def test_control_snapshot_hash(self): a, b, control = self.fixture(19), self.fixture(152), self.fixture(19) control["metadata"].update(mode="analytic-mesh", physical_screen_passed="1", max_newton="99") with tempfile.TemporaryDirectory(prefix="polytrope-snapshot-check-") as temporary: root = Path(temporary) for name, data in (("solve", a), ("control", control)): data["directory"] = root / name data["directory"].mkdir() (data["directory"] / "input.smesh").write_text("identical synthetic snapshot\n") checks = compatibility(a, b, control) self.assertTrue(checks_satisfied(checks)) self.assertTrue(any("snapshot SHA-256" in name and okay is True for name, okay, _ in checks)) (control["directory"] / "input.smesh").write_text("different geometry, same element count\n") checks = compatibility(a, b, control) self.assertFalse(checks_satisfied(checks)) self.assertTrue(any("snapshot SHA-256" in name and okay is False for name, okay, _ in checks)) def test_optional_metrics_and_exterior_coverage(self): a, b = self.fixture(19), self.fixture(152) del a["metrics"]["enthalpy_virial_error"] self.assertTrue(checks_satisfied(compatibility(a, b))) with tempfile.TemporaryDirectory(prefix="polytrope-exterior-check-") as temporary: directory = Path(temporary) a["directory"] = directory self.assertEqual(exterior_diagnostics(a)[0], {}) exterior = {"r_over_R": 1.001, "located_weight_fraction": 1, "potential_valid_weight_fraction": 1, "gravity_radial_valid_weight_fraction": 1, "mu_points": 6, "phi_points": 12, **{column: -.02 if "mean" in column else .03 for column, _ in EXTERIOR}} for coverage in (1, .5): exterior["potential_valid_weight_fraction"] = coverage with (directory / "radial_profiles.csv").open("w", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(exterior)) writer.writeheader() writer.writerow(exterior) values, _ = exterior_diagnostics(a) if coverage == 1: self.assertEqual(values["potential_mean_error_scaled"], .02) else: self.assertTrue(math.isnan(values["potential_mean_error_scaled"])) self.assertEqual(values["gravity_radial_rms_error_scaled"], .03) def test_round_trip_and_no_overwrite(self): with tempfile.TemporaryDirectory(prefix="polytrope-comparison-check-") as temporary: root = Path(temporary) data = [] for name, elements in (("coarse", 19), ("fine", 152)): fixture = self.fixture(elements) directory = root / name directory.mkdir() (directory / "metadata.txt").write_text("".join(f"{k}={v}\n" for k, v in fixture["metadata"].items())) with (directory / "physical_metrics.csv").open("w", newline="") as stream: writer = csv.writer(stream) writer.writerow(("metric", "value")) writer.writerows(fixture["metrics"].items()) data.append(read_dataset(directory)) output = root / "comparison" self.assertTrue(write_comparison(*data, output)) self.assertTrue((output / "comparison.csv").is_file()) self.assertIn("not an established asymptotic order", (output / "comparison.md").read_text()) with self.assertRaises(FileExistsError): write_comparison(*data, output) data[1]["metadata"]["solver_converged"] = "0" self.assertFalse(write_comparison(*data, root / "failed")) with (root / "failed" / "comparison.csv").open(newline="") as stream: self.assertTrue(all(not row["observed_log2_rate"] for row in csv.DictReader(stream))) suite = unittest.defaultTestLoader.loadTestsFromTestCase(ComparisonChecks) return unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("coarse", type=Path, nargs="?", help="Completed coarse solve directory (not a replay)") parser.add_argument("fine", type=Path, nargs="?", help="Completed fine solve directory (not a replay)") parser.add_argument("--coarse-control", type=Path) parser.add_argument("--fine-control", type=Path) parser.add_argument("--output", type=Path, help="Fresh output directory; never overwritten") parser.add_argument("--self-check", action="store_true", help="Run synthetic-only checks and exit") arguments = parser.parse_args() if arguments.self_check: if any((arguments.coarse, arguments.fine, arguments.output, arguments.coarse_control, arguments.fine_control)): parser.error("--self-check cannot be combined with dataset/output arguments") return 0 if self_check() else 3 if not all((arguments.coarse, arguments.fine, arguments.output)): parser.error("coarse, fine, and --output are required") try: eligible = write_comparison( read_dataset(arguments.coarse), read_dataset(arguments.fine), arguments.output, read_dataset(arguments.coarse_control) if arguments.coarse_control else None, read_dataset(arguments.fine_control) if arguments.fine_control else None) print(f"Wrote {arguments.output / 'comparison.md'}; comparison prerequisites satisfied={eligible} " "(not a physical verification pass)") return 0 if eligible else 3 except (OSError, ValueError, csv.Error) as error: print(f"Comparison error: {error}", file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())