144 lines
8.6 KiB
Python
144 lines
8.6 KiB
Python
"""Audit the pressure/shape campaign's completed outputs without running solvers.
|
|
|
|
The sole argument is the campaign directory containing captures/ and focused/.
|
|
The script reads existing results and writes focused_audit_checks.json there.
|
|
Unfinished cases are reported as pending; this is an observation audit, not a
|
|
stellar convergence or physical acceptance gate.
|
|
"""
|
|
import argparse
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("campaign_root", type=Path)
|
|
ROOT = parser.parse_args().campaign_root.resolve()
|
|
FOCUSED = ROOT / "focused"
|
|
|
|
|
|
def rows(path):
|
|
with path.open() as source:
|
|
result = list(csv.DictReader(source))
|
|
if any(None in r or any(v is None for v in r.values()) for r in result):
|
|
raise ValueError(f"Malformed CSV row: {path}")
|
|
return result
|
|
|
|
|
|
def metrics(path):
|
|
data = rows(path)
|
|
if len({r["metric"] for r in data}) != len(data):
|
|
raise ValueError(f"Duplicate metric: {path}")
|
|
return {r["metric"]: float(r["value"]) for r in data}
|
|
|
|
|
|
def sha(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def clean(value):
|
|
if isinstance(value, dict):
|
|
return {k: clean(v) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [clean(v) for v in value]
|
|
if isinstance(value, float) and not math.isfinite(value):
|
|
return None
|
|
return value
|
|
|
|
|
|
provenance = json.loads((FOCUSED / "provenance.json").read_text())
|
|
result = {
|
|
"source_commit": provenance["commit"],
|
|
"binary_hash_matches": sha(FOCUSED / "plateau_diagnostic") == provenance["binary_sha256"],
|
|
"completed": [],
|
|
"pending": [],
|
|
}
|
|
for case in json.loads((FOCUSED / "cases.json").read_text()):
|
|
case_id = case["id"]
|
|
manifest = FOCUSED / f"{case_id}_run.json"
|
|
if not manifest.exists():
|
|
result["pending"].append(case_id)
|
|
continue
|
|
run = json.loads(manifest.read_text())
|
|
if "exit_code" not in run:
|
|
result["pending"].append(case_id)
|
|
continue
|
|
folder = FOCUSED / case_id
|
|
data = {
|
|
"id": case_id,
|
|
"exit_code": run["exit_code"],
|
|
"seconds": run["seconds"],
|
|
"input_hash_count": len(run["input_state_hashes"]),
|
|
"input_hash_mismatches": [str(p) for p, h in run["input_state_hashes"].items() if sha(Path(p)) != h],
|
|
"output_hash_count": len(run["output_hashes"]),
|
|
"output_hash_mismatches": [p for p, h in run["output_hashes"].items() if sha(folder / p) != h],
|
|
"unhashed_data_files": sorted(str(p.relative_to(folder)) for p in folder.rglob("*") if p.is_file() and p.suffix in (".txt", ".csv", ".json") and str(p.relative_to(folder)) not in run["output_hashes"]),
|
|
"csv_rows": {p.name: len(rows(p)) for p in sorted(folder.glob("*.csv"))},
|
|
"frozen": metrics(folder / "frozen_input_checks.csv"),
|
|
}
|
|
args = case["arguments"]
|
|
direction = Path(args[args.index("--probe-direction") + 1])
|
|
physical = Path(args[args.index("--probe-state") + 1])
|
|
data["direction_copy_byte_identical"] = direction.read_bytes() == (folder / "probe_normalized_correction.txt").read_bytes()
|
|
data["physical_base_copy_byte_identical"] = physical.read_bytes() == (folder / "frozen_physical_base.txt").read_bytes()
|
|
if "iteration_" in case_id:
|
|
iteration = int(case_id.rsplit("_", 1)[1])
|
|
capture_root = ROOT / "captures" / ("n1p5_J0_capture" if case_id.startswith("n1p5") else "n1_J0p10_capture")
|
|
capture_manifest = json.loads((capture_root.parent / f"{capture_root.name}_run.json").read_text())
|
|
data["auxiliary_capture_input_hashes_match"] = all(sha(physical.parent / name) == capture_manifest["output_hashes"][str((physical.parent / name).relative_to(capture_root))] for name in ("normalized_base.txt", "normalized_residual.txt"))
|
|
old = next(r for r in rows(capture_root / "newton.csv") if int(r["iteration"]) == iteration)
|
|
data["historical_linear_defect_relative"] = float(old["linear_true"])
|
|
trial = next(r for r in rows(capture_root / "trials.csv") if int(r["iteration"]) == iteration and r["trial"] == "0")
|
|
data["original_first_trial"] = trial
|
|
replay = rows(folder / "replayed_candidate.csv")[0]
|
|
data["candidate_replay"] = replay
|
|
if trial["disposition"] == "4":
|
|
data["replayed_candidate_residual_exact"] = replay["admissible"] == "1" and replay["residual_norm"] == trial["F"]
|
|
data["replayed_candidate_residual_difference"] = float(replay["residual_norm"]) - float(trial["F"])
|
|
data["replayed_candidate_minimum_jacobian_exact"] = replay["minimum_jacobian"] == trial["min_jacobian"]
|
|
else:
|
|
previous = physical.parent
|
|
data["historical_linear_defect_relative"] = float(next(r for r in rows(previous / "frozen_linear.csv") if r["restart"] == "160")["true_relative"])
|
|
data["linear_defect_relative_difference"] = data["frozen"]["linear_defect_relative"] - data["historical_linear_defect_relative"]
|
|
layout = rows(folder / "layout.csv")
|
|
residual_blocks = {r["block"] for r in layout if r["kind"] == "residual"}
|
|
value_blocks = {r["block"] for r in layout if r["kind"] == "value"}
|
|
nr = sum(r["kind"] == "residual" for r in layout)
|
|
nv = sum(r["kind"] == "value" for r in layout)
|
|
if "--shape-curvature" in args:
|
|
data["shape_summary"] = metrics(folder / "shape_curvature_summary.csv")
|
|
expected = {"shape_curvature_directions.csv": 3 * nv, "shape_curvature_actions.csv": nr, "shape_curvature_preparations.csv": 22, "shape_curvature_restorations.csv": 4 * nr, "shape_curvature_remainders.csv": 18 * nr, "shape_curvature_mixed.csv": 6 * nr}
|
|
data["shape_row_count_mismatches"] = {f: [data["csv_rows"][f], count] for f, count in expected.items() if data["csv_rows"][f] != count}
|
|
restorations = rows(folder / "shape_curvature_restorations.csv")
|
|
data["shape_max_restoration_difference"] = max(float(r["restoration_difference_norm"]) for r in restorations)
|
|
mixed = rows(folder / "shape_curvature_mixed.csv")
|
|
alphas = {r["alpha"] for r in mixed}
|
|
remainders = rows(folder / "shape_curvature_remainders.csv")
|
|
data["shape_six_alphas"] = len(alphas) == 6
|
|
data["shape_mixed_keys_complete"] = {(r["alpha"], r["block"]) for r in mixed} == {(a, b) for a in alphas for b in residual_blocks}
|
|
data["shape_remainder_keys_complete"] = {(r["alpha"], r["direction"], r["block"]) for r in remainders} == {(a, d, b) for a in alphas for d in ("full", "shape", "fields") for b in residual_blocks}
|
|
data["shape_direction_keys_complete"] = {(r["direction"], r["value_block"]) for r in rows(folder / "shape_curvature_directions.csv")} == {(d, b) for d in ("full", "shape", "fields") for b in value_blocks}
|
|
data["shape_max_absolute_merit_identity_error"] = max(abs(float(r["merit_identity_error"])) for r in mixed)
|
|
data["shape_max_absolute_projection_decomposition_error"] = max(abs(float(r["projection_decomposition_error"])) for r in mixed)
|
|
data["shape_all_trials_admissible"] = all(r["all_trials_admissible"] == "1" for r in mixed)
|
|
data["shape_all_increments_resolved"] = all(r["all_state_increments_resolved"] == "1" for r in mixed)
|
|
data["shape_all_alpha_squared_valid"] = all(r["alpha_squared_valid"] == "1" for r in mixed)
|
|
if "--pressure-probe" in args:
|
|
data["pressure_summary"] = metrics(folder / "pressure_obstruction_summary.csv")
|
|
pressure = rows(folder / "pressure_obstruction_trials.csv")
|
|
blocks = rows(folder / "pressure_obstruction_blocks.csv")
|
|
data["pressure_admissible_trials"] = sum(r["admissible"] == "1" for r in pressure)
|
|
data["pressure_expected_block_rows"] = nr * data["pressure_admissible_trials"]
|
|
data["pressure_block_rows_complete"] = len(blocks) == data["pressure_expected_block_rows"]
|
|
data["pressure_trial_keys_unique"] = len({(r["direction"], r["alpha"]) for r in pressure}) == len(pressure)
|
|
data["pressure_block_keys_complete"] = {(r["direction"], r["alpha"], r["block"]) for r in blocks} == {(r["direction"], r["alpha"], b) for r in pressure if r["admissible"] == "1" for b in residual_blocks}
|
|
data["pressure_limiting_h_prediction_max_absolute_error"] = {
|
|
family: max(abs(float(r[f"{family}_limiting_h_prediction"]) - float(r[f"{family}_limiting_h_evaluated"])) for r in pressure)
|
|
for family in ("closure", "pressure")
|
|
}
|
|
result["completed"].append(data)
|
|
|
|
(ROOT / "focused_audit_checks.json").write_text(json.dumps(clean(result), indent=2) + "\n")
|
|
print(json.dumps({"completed": [r["id"] for r in result["completed"]], "pending": result["pending"]}, indent=2))
|