diff --git a/experiments/plot_frozen_linear_rates.py b/experiments/plot_frozen_linear_rates.py new file mode 100644 index 0000000..b795157 --- /dev/null +++ b/experiments/plot_frozen_linear_rates.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +"""Plot frozen FGMRES histories only after a completed full artifact audit. + + python plot_frozen_linear_rates.py linear_audit.json [followup_audit.json] \ + --output NEW_FIGURE_DIRECTORY + +Inputs are full summarize_coupled_convergence.py audits of benchmark campaigns, +not the supplemental CSV-only comparison. Small summary/history CSVs are checked +against the audit hashes; large saved vectors and FEM actions are not reread. +Rendering must be scheduled outside timed numerical campaigns. +""" +from __future__ import annotations + +import argparse +import csv +import hashlib +import io +import json +import math +from pathlib import Path +import sys + + +ROLES = ("baseline", "candidate") +VARIANT_LABELS = { + "gravity-schur": "gravity Schur", + "surface-border-schur": "surface/scalar Schur", + "gravity-schur-surface-border-schur": "gravity + surface/scalar Schur", + "stellar-ldu-gravity-schur": "stellar LDU + gravity Schur", +} + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def digest(raw): + return hashlib.sha256(raw).hexdigest() + + +def number(value, name, positive=False): + result = float(value) + require(math.isfinite(result) and (result > 0 if positive else result >= 0), + f"Invalid {name}: require finite {'positive' if positive else 'nonnegative'} value") + return result + + +def near(left, right, name): + require(math.isclose(left, right, rel_tol=3e-11, abs_tol=0.), f"Inconsistent {name}") + + +def flag(value, name): + require(value in ("0", "1"), f"Invalid {name} flag") + return value == "1" + + +def checked_csv(path, hashes, producer_hash, consumed): + path = path.resolve() + expected = hashes.get(str(path)) + require(isinstance(expected, str) and len(expected) == 64 and expected == producer_hash, + f"Missing or inconsistent audit/producer input hash: {path}") + raw = path.read_bytes() + require(digest(raw) == expected, f"Input hash differs from the passed audit: {path}") + consumed[str(path)] = expected + return list(csv.DictReader(io.StringIO(raw.decode()))) + + +def role_curve(row, samples): + role, variant = row["role"], row["variant"] + require(role in ROLES and bool(samples), "Missing linear history role") + rhs = number(row["rhs_norm"], "RHS norm", positive=True) + target = number(row["relative_tolerance"], "relative target", positive=True) + absolute_target = number(row["absolute_target"], "absolute target", positive=True) + near(absolute_target, target * rhs, "absolute/relative target") + true_norm = number(row["true_residual_norm"], "true endpoint defect", positive=True) + true_relative = number(row["true_relative"], "true relative endpoint defect", positive=True) + near(true_relative, true_norm / rhs, "true relative endpoint defect") + gate = flag(row["true_gate_pass"], "true gate") + require(gate == (true_norm <= absolute_target), "True gate contradicts the independently checked endpoint") + iterations, reported_iterations, maximum = (int(row[key]) for key in + ("iterations", "reported_iterations", "maximum_iterations")) + require(0 < iterations <= maximum and reported_iterations == iterations, + "Reported iteration count differs from the completed preconditioner application count") + x, values, final = [], [], [] + for index, sample in enumerate(samples): + require(sample["role"] == role and sample["variant"] == variant, "Linear history role/variant mismatch") + iteration = int(sample["iteration"]) + require(0 <= iteration <= iterations and (not x or iteration >= x[-1]), + "Linear history has invalid or unordered iteration numbers") + x.append(iteration) + # Exact zero is a valid numerical outcome, but needs a different plotting + # convention. Refuse it here rather than invent a small positive floor. + values.append(number(sample["reported_norm"], "reported log residual", positive=True) / rhs) + if flag(sample["final"], "final monitor"): + final.append(index) + require(set(x) == set(range(iterations + 1)), "Missing reported iteration samples") + require(final == [len(samples) - 1] and x[-1] == reported_iterations, + "Missing, repeated or misplaced final monitor sample") + near(values[0], number(row["reported_initial_norm"], "reported initial norm", positive=True) / rhs, + "initial monitor sample") + near(values[0], 1., "cold-start relative norm") + near(values[-1], number(row["reported_final_norm"], "reported final norm", positive=True) / rhs, + "final monitor sample") + # Repeated iteration numbers at restarts/final callbacks are meaningful. + # Keep their order and choose the flagged endpoint, never the best residual. + setup = math.fsum(number(row[key], key) for key in + ("construction_seconds", "normalization_binding_seconds", "krylov_setup_seconds")) + solve, check = (number(row[key], key) for key in ("krylov_seconds", "true_check_seconds")) + total = number(row["setup_solve_and_true_check_seconds"], "verified solve seconds", positive=True) + near(number(row["setup_plus_solve_seconds"], "setup/solve seconds"), setup + solve, "setup/solve timing sum") + near(total, setup + solve + check, "setup/solve/true-check timing sum") + return {"role": role, "variant": variant, "iteration": x, "reported_relative": values, + "final_reported_index": final[0], "completed_iterations": iterations, + "maximum_iterations": maximum, "restart": int(row["restart"]), + "rhs_norm": rhs, "relative_target": target, "absolute_target": absolute_target, + "endpoint_true_relative": true_relative, "true_gate_pass": gate, + "solver_recurrence_converged": flag(row["solver_converged"], "solver convergence"), + "setup_seconds": setup, "solve_seconds": solve, "true_check_seconds": check, + "setup_solve_true_check_seconds": total} + + +def load_cases(paths): + cases, sources, consumed, seen = [], [], {}, set() + for path in paths: + path = Path(path).resolve() + raw = path.read_bytes() + audit = json.loads(raw) + require(audit.get("complete") is True and audit.get("artifact_passed") is True, + f"Require a completed, artifact-passed full linear audit: {path}") + require(audit.get("auditor_sha256") and audit.get("benchmark_checker_sha256") + and audit.get("consumed_sha256"), "CSV-only summaries are not a full linear artifact audit") + sources.append({"path": str(path), "sha256": digest(raw), "auditor_sha256": audit["auditor_sha256"], + "benchmark_checker_sha256": audit["benchmark_checker_sha256"], + "scope": audit.get("scope"), "limitations": audit.get("limitations", [])}) + require(bool(audit["campaigns"]), "Empty audited campaign list") + for campaign in audit["campaigns"]: + require(campaign.get("complete") is True and campaign.get("artifact_passed") is True, + "Incomplete or failed linear campaign") + require(bool(campaign["cases"]), "Empty linear campaign") + for case in campaign["cases"]: + name = case["id"] + require(name not in seen, f"Duplicate case identifier across audits: {name}") + seen.add(name) + require(Path(name).name == name and name not in (".", ".."), "Invalid case identifier") + require(case.get("complete") is True and case.get("artifact_passed") is True + and case.get("kind") == "benchmark", f"Missing complete linear benchmark: {name}") + assessment, run = case["benchmark"], case["run"] + require(assessment.get("passed") is True and assessment["comparison"].get("comparison_valid") is True, + "Frozen endpoint artifact audit did not pass") + checks = {item["check"]: item["passed"] for item in assessment["checks"]} + required_checks = {role + suffix for role in ROLES for suffix in + ("_rhs_norm", "_defect_norm", "_relative_defect", "_honest_gate")} + require(required_checks <= checks.keys() and all(value is True for value in checks.values()), + "Missing or failed saved-vector endpoint checks") + require(run.get("finished_utc") and run.get("inputs_unchanged") is True + and run.get("exit_code") in (0, 3) and not run.get("timed_out"), "Unfinished linear run") + folder = Path(campaign["path"]) / name + tables = {stem: checked_csv(folder / (stem + ".csv"), audit["consumed_sha256"], + run["output_hashes"].get(stem + ".csv"), consumed) + for stem in ("preconditioner_summary", "preconditioner_history")} + summary, history = tables["preconditioner_summary"], tables["preconditioner_history"] + require(len(summary) == 2 and {row["role"] for row in summary} == set(ROLES), "Invalid paired summary") + require({row["role"] for row in history} == set(ROLES), "Missing or unknown history role") + curves = [] + for role in ROLES: + row = next(row for row in summary if row["role"] == role) + samples = [sample for sample in history if sample["role"] == role] + require(row == assessment["comparison"][role], "CSV summary differs from the audited endpoint record") + require(samples == case["linear_history"][role]["samples"], "CSV history differs from the audited history") + curves.append(role_curve(row, samples)) + baseline, candidate = curves + require(baseline["variant"] == "independent", "Unexpected baseline preconditioner") + for key in ("rhs_norm", "relative_target", "absolute_target", "restart", "maximum_iterations"): + require(baseline[key] == candidate[key], f"Unmatched baseline/candidate {key}") + outcome = ("both_pass" if all(c["true_gate_pass"] for c in curves) else "baseline_only_pass" + if baseline["true_gate_pass"] else "candidate_only_pass" if candidate["true_gate_pass"] else "neither_passes") + require(case["outcome"] == assessment["comparison"]["outcome"] == outcome, "Contradictory true-gate outcome") + cases.append({"id": name, "campaign": campaign["path"], "audit_path": str(path), + "producer_commit": campaign["producer_commit"], "mesh_sha256": campaign["mesh_sha256"], + "metadata": case["metadata"], "source_input_hashes": run["input_state_hashes"], + "benchmark_audit_scope": case["benchmark_audit_scope"], "outcome": outcome, "curves": curves}) + require(bool(cases), "No complete linear cases") + return cases, sources, consumed + + +def panel_shape(count): + require(count > 0, "Need at least one panel") + columns = min(3, count) + return math.ceil(count / columns), columns + + +def timing_label(curve): + status = "PASS" if curve["true_gate_pass"] else "FAIL" + return (f"{curve['role'][0].upper()} {status}; true {curve['endpoint_true_relative']:.2e}; " + f"{curve['setup_seconds']:.2f}+{curve['solve_seconds']:.2f}+{curve['true_check_seconds']:.2f}" + f"={curve['setup_solve_true_check_seconds']:.2f} s") + + +def render(cases): + import matplotlib + matplotlib.use("Agg") + import matplotlib.pyplot as plt + from matplotlib.lines import Line2D + + rows, columns = panel_shape(len(cases)) + height = 4.35 * rows + 1.2 + figure, axes = plt.subplots(rows, columns, squeeze=False, figsize=(max(10.5, 5.1 * columns), height)) + colors = ("#1769a8", "#c35b16") + for axis, case in zip(axes.flat, cases): + curves = case["curves"] + for curve, color, style in zip(curves, colors, ("-", "--")): + axis.plot(curve["iteration"], curve["reported_relative"], color=color, linestyle=style, linewidth=1.4) + final = curve["final_reported_index"] + axis.plot(curve["iteration"][final], curve["reported_relative"][final], marker="o", markersize=4, + markerfacecolor="white", markeredgecolor=color, linestyle="none", zorder=4) + if curve["true_gate_pass"]: + axis.scatter([curve["completed_iterations"]], [curve["endpoint_true_relative"]], marker="D", s=58, + facecolors="none", edgecolors=color, linewidths=1.5, zorder=5) + else: + axis.scatter([curve["completed_iterations"]], [curve["endpoint_true_relative"]], marker="x", s=58, + color=color, linewidths=1.7, zorder=5) + target = curves[0]["relative_target"] + axis.axhline(target, color="#555555", linestyle=":", linewidth=1.2) + axis.text(.97, .96, f"target {target:g}", ha="right", va="top", fontsize=8, transform=axis.transAxes, + bbox={"facecolor": "white", "edgecolor": "none", "alpha": .8}) + variant = curves[1]["variant"] + meta = case["metadata"] + axis.set_title(f"n={float(meta['n']):g}, J={float(meta['J']):g} | {VARIANT_LABELS.get(variant, variant)}\n{case['id']}", fontsize=9) + axis.set_yscale("log") + axis.set_xlim(0, max(c["completed_iterations"] for c in curves) * 1.04) + axis.set_ylabel("Reported relative residual /\nchecked endpoint true defect", fontsize=8) + axis.set_xlabel("FGMRES iteration\n" + "\n".join(timing_label(c) for c in curves), fontsize=8) + axis.tick_params(labelsize=8) + axis.grid(True, which="major", alpha=.22) + for axis in list(axes.flat)[len(cases):]: + axis.set_visible(False) + handles = [Line2D([], [], color=colors[0], label="Baseline (B): independent"), + Line2D([], [], color=colors[1], linestyle="--", label="Candidate (C)"), + Line2D([], [], color="#555555", linestyle=":", label="True-defect target"), + Line2D([], [], color="#222222", marker="o", markerfacecolor="white", linestyle="none", label="Final reported norm"), + Line2D([], [], color="#222222", marker="D", markerfacecolor="none", linestyle="none", label="Checked endpoint: PASS"), + Line2D([], [], color="#222222", marker="x", linestyle="none", label="Checked endpoint: FAIL")] + figure.suptitle("Frozen linear solves: reported FGMRES rates and checked endpoint defects", fontsize=14, y=1 - .10 / height) + figure.legend(handles=handles, loc="upper center", bbox_to_anchor=(.5, 1 - .42 / height), ncol=3, fontsize=9, frameon=False) + figure.text(.5, .12 / height, "Lines are reported recurrence norms / ||b||, not an independently checked true-residual history or a proven floor.\n" + "Endpoints were checked from saved true-defect vectors. PASS/FAIL is linear accuracy, not nonlinear equilibrium.\n" + "Times: setup + solve + true check = total seconds; no per-iteration time interpolation. Axes autoscale per panel.", + ha="center", va="bottom", fontsize=8) + figure.tight_layout(rect=(0, .72 / height, 1, 1 - 1. / height), h_pad=2.2, w_pad=1.5) + return figure, matplotlib.__version__ + + +def sidecar(cases, sources, consumed): + return {"schema_version": 1, "source_audits": sources, "rechecked_csv_sha256": consumed, + "plotter_sha256": digest(Path(__file__).read_bytes()), "cases": cases, + "scope": ["Requires completed, artifact-passed full linear campaign audits; failed linear gates remain failed.", + "Summary/history CSV hashes are rechecked against both the full audit and producer record.", + "Large vectors, executable and mesh hashes, and FEM actions are not reevaluated during plotting.", + "Reported recurrence norms are not an independently checked true-residual history or a proven residual floor.", + "True endpoint norms/gates come from the saved-vector checks in the full audit; no intermediate true defects are inferred.", + "Repeated monitor iteration numbers, including a distinct final callback, are preserved in recorded order.", + "No time history is inferred. Times include inverse construction/binding/Krylov setup, solve and final true check; frozen-state preparation and other diagnostics are excluded.", + "Iteration counts are not equal work units, and linear success does not establish nonlinear equilibrium.", + "Log plots refuse exact zero residuals instead of silently replacing them with a positive floor."]} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("audits", nargs="+", type=Path) + parser.add_argument("--output", required=True, type=Path, help="New figure directory outside the source checkout") + args = parser.parse_args() + require(not args.output.exists(), "Refusing to overwrite an existing figure directory") + require(not args.output.resolve().is_relative_to(Path(__file__).resolve().parents[1]), + "Write figure artifacts outside the source checkout") + cases, sources, consumed = load_cases(args.audits) + report = sidecar(cases, sources, consumed) + # Validate all inputs before starting Matplotlib or creating output, so an + # incomplete ablation cannot quietly turn into a smaller, biased comparison. + figure, version = render(cases) + args.output.mkdir(parents=True) + report.update(matplotlib_version=version, python_version=sys.version, figures=[]) + for extension in ("png", "pdf"): + path = args.output / f"frozen_linear_rates.{extension}" + figure.savefig(path, dpi=200) + report["figures"].append({"path": str(path.resolve()), "sha256": digest(path.read_bytes())}) + from matplotlib import pyplot as plt + plt.close(figure) + path = args.output / "frozen_linear_rates.json" + path.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") + print(path) + + +if __name__ == "__main__": + main() diff --git a/experiments/test_plot_frozen_linear_rates.py b/experiments/test_plot_frozen_linear_rates.py new file mode 100644 index 0000000..46ac748 --- /dev/null +++ b/experiments/test_plot_frozen_linear_rates.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +"""Tiny synthetic schema checks; opt in to rendering during serial validation. + + python -m unittest discover -s experiments -p test_plot_frozen_linear_rates.py -v + MEANFIELD_PLOT_RENDER_TESTS=1 python -m unittest discover -s experiments -p test_plot_frozen_linear_rates.py -v +""" +import copy +import csv +import hashlib +import json +import os +from pathlib import Path +import tempfile +import unittest + +import plot_frozen_linear_rates as plots + + +def role_data(role): + baseline = role == "baseline" + variant = "independent" if baseline else "gravity-schur" + values = [1., .2, .08, .12] if baseline else [1., .2, .04, .07] + true_relative = .11 if baseline else .06 + row = {"role": role, "variant": variant, "rhs_norm": "2", "relative_tolerance": ".1", "absolute_target": ".2", + "true_residual_norm": str(2 * true_relative), "true_relative": str(true_relative), + "true_gate_pass": "0" if baseline else "1", "solver_converged": "1", + "iterations": "2", "reported_iterations": "2", "maximum_iterations": "10", "restart": "5", + "reported_initial_norm": "2", "reported_final_norm": str(2 * values[-1]), + "construction_seconds": ".75", "normalization_binding_seconds": ".125", "krylov_setup_seconds": ".125", + "krylov_seconds": "2", "true_check_seconds": ".25", "setup_plus_solve_seconds": "3", + "setup_solve_and_true_check_seconds": "3.25"} + history = [{"role": role, "variant": variant, "iteration": str(i), "reported_norm": str(2 * value), + "final": "1" if index == 3 else "0"} for index, (i, value) in enumerate(zip((0, 1, 2, 2), values))] + return row, history + + +def write_fixture(directory, name="n2_late_gravity"): + campaign = directory / name / "campaign" + folder = campaign / name + folder.mkdir(parents=True) + summary, history = [], [] + comparison = {"comparison_valid": True, "outcome": "candidate_only_pass"} + audit_history, checks, hashes, outputs = {}, [], {}, {} + for role in plots.ROLES: + row, samples = role_data(role) + summary.append(row) + history.extend(samples) + comparison[role] = row + audit_history[role] = {"samples": samples} + checks.extend({"check": role + suffix, "passed": True} for suffix in + ("_rhs_norm", "_defect_norm", "_relative_defect", "_honest_gate")) + for stem, rows in (("preconditioner_summary", summary), ("preconditioner_history", history)): + path = folder / (stem + ".csv") + with path.open("w", newline="") as stream: + writer = csv.DictWriter(stream, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + outputs[path.name] = hashlib.sha256(path.read_bytes()).hexdigest() + hashes[str(path.resolve())] = outputs[path.name] + case = {"id": name, "complete": True, "artifact_passed": True, "kind": "benchmark", "outcome": "candidate_only_pass", + "metadata": {"n": "2", "J": "0", "density_order": "2"}, + "run": {"finished_utc": "2026-09-19T00:00:00Z", "inputs_unchanged": True, "exit_code": 0, + "output_hashes": outputs, "input_state_hashes": {"/synthetic/frozen.txt": "a" * 64}}, + "benchmark": {"passed": True, "checks": checks, "comparison": comparison}, + "linear_history": audit_history, "benchmark_audit_scope": {"scope": "synthetic schema; not a real FEM audit"}} + audit = {"complete": True, "artifact_passed": True, "auditor_sha256": "b" * 64, "benchmark_checker_sha256": "c" * 64, + "consumed_sha256": hashes, "scope": "synthetic schema control", "limitations": ["No real vectors audited"], + "campaigns": [{"path": str(campaign), "producer_commit": "d" * 40, "mesh_sha256": "e" * 64, + "complete": True, "artifact_passed": True, "cases": [case]}]} + path = directory / name / "audit.json" + path.write_text(json.dumps(audit)) + return path, audit, folder + + +class LinearPlotChecks(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.directory = Path(self.temporary.name).resolve() + self.path, self.audit, self.folder = write_fixture(self.directory) + + def tearDown(self): + self.temporary.cleanup() + + def save(self): + self.path.write_text(json.dumps(self.audit)) + + def test_final_callback_is_not_the_history_minimum(self): + row, history = role_data("baseline") + curve = plots.role_curve(row, history) + self.assertEqual(curve["iteration"], [0, 1, 2, 2]) + self.assertEqual(curve["reported_relative"], [1., .2, .08, .12]) + self.assertEqual(curve["final_reported_index"], 3) + self.assertEqual(curve["completed_iterations"], 2) + self.assertNotEqual(curve["reported_relative"][-1], min(curve["reported_relative"])) + self.assertEqual(curve["endpoint_true_relative"], .11) + + def test_true_gate_not_recurrence_flag_controls_failure(self): + cases, sources, hashes = plots.load_cases([self.path]) + baseline, candidate = cases[0]["curves"] + self.assertTrue(baseline["solver_recurrence_converged"]) + self.assertFalse(baseline["true_gate_pass"]) + self.assertTrue(candidate["true_gate_pass"]) + self.assertEqual(cases[0]["outcome"], "candidate_only_pass") + self.assertIn("B FAIL", plots.timing_label(baseline)) + self.assertEqual(baseline["setup_seconds"], 1.) + self.assertEqual(baseline["setup_solve_true_check_seconds"], 3.25) + self.assertEqual(len(hashes), 2) + self.assertEqual(sources[0]["sha256"], hashlib.sha256(self.path.read_bytes()).hexdigest()) + json.dumps(plots.sidecar(cases, sources, hashes), allow_nan=False) + + def test_exact_target_passes_but_contradictory_gate_fails(self): + row, history = role_data("candidate") + row.update(true_residual_norm=".2", true_relative=".1", true_gate_pass="1") + self.assertTrue(plots.role_curve(row, history)["true_gate_pass"]) + row["true_gate_pass"] = "0" + with self.assertRaisesRegex(ValueError, "True gate"): + plots.role_curve(row, history) + + def test_hash_change_after_audit_is_refused(self): + path = self.folder / "preconditioner_history.csv" + path.write_text(path.read_text().replace("0.16", "0.17")) + with self.assertRaisesRegex(ValueError, "Input hash differs"): + plots.load_cases([self.path]) + + def test_missing_input_hash_or_embedded_endpoint_mismatch_is_refused(self): + key = next(iter(self.audit["consumed_sha256"])) + self.audit["consumed_sha256"][key] = "f" * 64 + self.save() + with self.assertRaisesRegex(ValueError, "input hash"): + plots.load_cases([self.path]) + self.audit["consumed_sha256"][key] = hashlib.sha256(Path(key).read_bytes()).hexdigest() + self.audit["campaigns"][0]["cases"][0]["benchmark"]["comparison"]["candidate"]["true_relative"] = ".09" + self.save() + with self.assertRaisesRegex(ValueError, "audited endpoint record"): + plots.load_cases([self.path]) + + def test_full_audit_required_and_pending_or_failed_cases_refused(self): + for change in ("complete", "artifact_passed"): + original = copy.deepcopy(self.audit) + self.audit[change] = False + self.save() + with self.assertRaisesRegex(ValueError, "full linear audit"): + plots.load_cases([self.path]) + self.audit = original + self.audit["campaigns"][0]["cases"][0]["kind"] = "execution_failure" + self.save() + with self.assertRaisesRegex(ValueError, "complete linear benchmark"): + plots.load_cases([self.path]) + self.audit.pop("benchmark_checker_sha256") + self.save() + with self.assertRaisesRegex(ValueError, "not a full"): + plots.load_cases([self.path]) + + def test_saved_vector_checks_and_outcome_are_required(self): + case = self.audit["campaigns"][0]["cases"][0] + removed = case["benchmark"]["checks"].pop() + self.save() + with self.assertRaisesRegex(ValueError, "saved-vector"): + plots.load_cases([self.path]) + case["benchmark"]["checks"].append(removed) + case["outcome"] = "both_pass" + self.save() + with self.assertRaisesRegex(ValueError, "Contradictory true-gate"): + plots.load_cases([self.path]) + + def test_final_flags_missing_samples_and_mismatched_final_norm_are_refused(self): + row, history = role_data("candidate") + for modified, message in ((history[:-1], "final monitor"), + ([history[0], *history[2:]], "Missing reported")): + with self.subTest(message=message), self.assertRaisesRegex(ValueError, message): + plots.role_curve(row, modified) + history[0]["final"] = "1" + with self.assertRaisesRegex(ValueError, "final monitor"): + plots.role_curve(row, history) + history[0]["final"] = "0" + row["reported_final_norm"] = ".08" + with self.assertRaisesRegex(ValueError, "final monitor sample"): + plots.role_curve(row, history) + + def test_zero_log_values_are_not_floored_and_time_sum_is_checked(self): + row, history = role_data("candidate") + history[1]["reported_norm"] = "0" + with self.assertRaisesRegex(ValueError, "reported log residual"): + plots.role_curve(row, history) + row, history = role_data("candidate") + row["true_residual_norm"] = "0" + with self.assertRaisesRegex(ValueError, "true endpoint defect"): + plots.role_curve(row, history) + row, history = role_data("candidate") + row["setup_solve_and_true_check_seconds"] = "9" + with self.assertRaisesRegex(ValueError, "timing sum"): + plots.role_curve(row, history) + + def test_multiple_audits_dynamic_panels_and_no_trial_requirement(self): + followup, _, _ = write_fixture(self.directory, "n3_late_gravity") + cases, sources, _ = plots.load_cases([self.path, followup]) + self.assertEqual(len(cases), 2) + self.assertEqual(len(sources), 2) + self.assertEqual(plots.panel_shape(9), (3, 3)) + self.assertEqual(plots.panel_shape(10), (4, 3)) + self.assertEqual(plots.panel_shape(1), (1, 1)) + with self.assertRaisesRegex(ValueError, "Duplicate case"): + plots.load_cases([self.path, self.path]) + + @unittest.skipUnless(os.environ.get("MEANFIELD_PLOT_RENDER_TESTS") == "1", "Rendering reserved for serial validation interval") + def test_render_ten_cases_has_distinct_endpoints_and_hides_unused_panels(self): + cases, _, _ = plots.load_cases([self.path]) + cases = [dict(copy.deepcopy(cases[0]), id=f"synthetic_{index}") for index in range(10)] + figure, _ = plots.render(cases) + self.assertEqual(len(figure.axes), 12) + self.assertEqual(sum(axis.get_visible() for axis in figure.axes), 10) + axis = figure.axes[0] + self.assertEqual(axis.get_yscale(), "log") + self.assertEqual(list(axis.lines[0].get_xdata()), [0, 1, 2, 2]) + self.assertEqual(list(axis.lines[-1].get_ydata()), [.1, .1]) + self.assertEqual(list(axis.collections[0].get_offsets()[0]), [2., .11]) + self.assertEqual(list(axis.collections[1].get_offsets()[0]), [2., .06]) + self.assertIn("B FAIL", axis.get_xlabel()) + for extension in ("png", "pdf"): + path = self.directory / ("rates." + extension) + figure.savefig(path) + self.assertGreater(path.stat().st_size, 1000) + from matplotlib import pyplot as plt + plt.close(figure) + + +if __name__ == "__main__": + unittest.main()