Skills › Scientific Agent Skills › qutip › result_audit.py
result_audit.py
402 lines · skills/qutip/scripts/result_audit.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit portable QuTiP skill JSON without loading Python objects."""
4from __future__ import annotations
6import argparse
7import math
8from collections.abc import Mapping
9from typing import Any
11from _common import (
12 MAX_SWEEP_RUNS,
13 MAX_TIME_POINTS,
14 QUTIP_VERSION,
15 CliError,
16 add_output_arguments,
17 emit_json,
18 finite_float,
19 load_json_object,
20 run_cli,
21)
24def _check(
25 checks: list[dict[str, Any]],
26 *,
27 name: str,
28 passed: bool,
29 detail: str,
30 severity: str = "error",
31) -> None:
32 checks.append(
33 {
34 "name": name,
35 "passed": bool(passed),
36 "severity": severity,
37 "detail": detail,
38 }
39 )
42def _finite_sequence(
43 value: Any,
44 *,
45 name: str,
46 maximum: int,
47) -> list[float]:
48 if not isinstance(value, list) or not 2 <= len(value) <= maximum:
49 raise CliError(f"{name} must contain from 2 through {maximum} values")
50 result: list[float] = []
51 for index, item in enumerate(value):
52 number = finite_float(item, name=f"{name}[{index}]")
53 result.append(number)
54 return result
57def _audit_simulation(
58 document: Mapping[str, Any],
59 *,
60 tolerance: float,
61) -> list[dict[str, Any]]:
62 checks: list[dict[str, Any]] = []
63 required = {
64 "qutip_version",
65 "configuration",
66 "model",
67 "times",
68 "expectations",
69 "analytic_reference",
70 "trajectory_statistics",
71 "checks",
72 "solver",
73 }
74 missing = sorted(required - set(document))
75 _check(
76 checks,
77 name="required_fields",
78 passed=not missing,
79 detail="all required fields present" if not missing else f"missing: {missing}",
80 )
81 if missing:
82 return checks
84 _check(
85 checks,
86 name="exact_qutip_version",
87 passed=document["qutip_version"] == QUTIP_VERSION,
88 detail=f"reported {document['qutip_version']!r}; expected {QUTIP_VERSION}",
89 )
91 try:
92 times = _finite_sequence(
93 document["times"],
94 name="times",
95 maximum=MAX_TIME_POINTS,
96 )
97 increasing = all(right > left for left, right in zip(times, times[1:]))
98 except CliError as exc:
99 times = []
100 increasing = False
101 time_detail = str(exc)
102 else:
103 time_detail = f"{len(times)} finite values"
104 _check(
105 checks,
106 name="bounded_monotonic_time_grid",
107 passed=bool(times) and increasing,
108 detail=time_detail if not times or increasing else "times are not increasing",
109 )
111 expectations = document["expectations"]
112 if not isinstance(expectations, Mapping) or "excited_population" not in expectations:
113 populations: list[float] = []
114 population_detail = "expectations.excited_population is missing"
115 else:
116 try:
117 populations = _finite_sequence(
118 expectations["excited_population"],
119 name="expectations.excited_population",
120 maximum=MAX_TIME_POINTS,
121 )
122 except CliError as exc:
123 populations = []
124 population_detail = str(exc)
125 else:
126 population_detail = f"{len(populations)} finite values"
127 lengths_match = bool(times) and bool(populations) and len(times) == len(populations)
128 _check(
129 checks,
130 name="population_length",
131 passed=lengths_match,
132 detail=population_detail,
133 )
134 if populations:
135 violation = max(
136 max(0.0, -min(populations)),
137 max(0.0, max(populations) - 1.0),
138 )
139 else:
140 violation = math.inf
141 _check(
142 checks,
143 name="population_bounds",
144 passed=violation <= tolerance,
145 detail=f"maximum [0,1] violation is {violation}",
146 )
148 analytic = document["analytic_reference"]
149 analytic_ok = True
150 analytic_detail = "not applicable"
151 if isinstance(analytic, Mapping) and analytic.get("applicable"):
152 try:
153 reference = _finite_sequence(
154 analytic["excited_population"],
155 name="analytic_reference.excited_population",
156 maximum=MAX_TIME_POINTS,
157 )
158 reported_error = finite_float(
159 analytic["max_abs_error"],
160 name="analytic_reference.max_abs_error",
161 minimum=0.0,
162 )
163 recomputed = max(
164 abs(observed - expected)
165 for observed, expected in zip(populations, reference)
166 )
167 analytic_ok = (
168 len(reference) == len(populations)
169 and abs(recomputed - reported_error)
170 <= max(1.0e-12, 10.0 * tolerance)
171 )
172 analytic_detail = (
173 f"reported error {reported_error}; recomputed {recomputed}"
174 )
175 except (CliError, KeyError, ValueError) as exc:
176 analytic_ok = False
177 analytic_detail = str(exc)
178 _check(
179 checks,
180 name="analytic_reference_consistency",
181 passed=analytic_ok,
182 detail=analytic_detail,
183 )
185 model = document["model"]
186 assumptions = model.get("assumptions") if isinstance(model, Mapping) else None
187 _check(
188 checks,
189 name="physical_assumptions_recorded",
190 passed=isinstance(assumptions, list) and len(assumptions) >= 3,
191 detail="model assumptions are present" if assumptions else "assumptions missing",
192 severity="warning",
193 )
195 solver = document["solver"]
196 stats = solver.get("stats") if isinstance(solver, Mapping) else None
197 _check(
198 checks,
199 name="solver_stats_recorded",
200 passed=isinstance(stats, Mapping) and bool(stats),
201 detail="solver stats present" if stats else "solver stats missing or empty",
202 severity="warning",
203 )
205 configuration = document["configuration"]
206 solver_name = (
207 configuration.get("solver") if isinstance(configuration, Mapping) else None
208 )
209 trajectory = document["trajectory_statistics"]
210 if solver_name == "mcsolve":
211 count = trajectory.get("trajectories_run") if isinstance(trajectory, Mapping) else None
212 seeds = trajectory.get("seeds") if isinstance(trajectory, Mapping) else None
213 trajectory_ok = (
214 isinstance(count, int)
215 and count > 0
216 and isinstance(seeds, list)
217 and len(seeds) == count
218 )
219 trajectory_detail = (
220 f"trajectories_run={count}, seed_records="
221 f"{len(seeds) if isinstance(seeds, list) else 'missing'}"
222 )
223 else:
224 trajectory_ok = True
225 trajectory_detail = "deterministic solver"
226 _check(
227 checks,
228 name="trajectory_seed_manifest",
229 passed=trajectory_ok,
230 detail=trajectory_detail,
231 )
232 return checks
235def _audit_convergence(
236 document: Mapping[str, Any],
237 *,
238 tolerance: float,
239) -> list[dict[str, Any]]:
240 checks: list[dict[str, Any]] = []
241 runs = document.get("runs")
242 comparisons = document.get("comparisons")
243 _check(
244 checks,
245 name="bounded_run_count",
246 passed=isinstance(runs, list) and 2 <= len(runs) <= MAX_SWEEP_RUNS,
247 detail=(
248 f"run count {len(runs)}"
249 if isinstance(runs, list)
250 else "runs is not an array"
251 ),
252 )
253 _check(
254 checks,
255 name="comparisons_present",
256 passed=isinstance(comparisons, list) and bool(comparisons),
257 detail=(
258 f"comparison count {len(comparisons)}"
259 if isinstance(comparisons, list)
260 else "comparisons is not an array"
261 ),
262 )
263 _check(
264 checks,
265 name="exact_qutip_version",
266 passed=document.get("qutip_version") == QUTIP_VERSION,
267 detail=f"reported {document.get('qutip_version')!r}",
268 )
269 finite = True
270 maximum_difference = 0.0
271 if isinstance(comparisons, list):
272 for index, comparison in enumerate(comparisons):
273 if not isinstance(comparison, Mapping):
274 finite = False
275 continue
276 candidates = [
277 value
278 for key, value in comparison.items()
279 if "difference" in key and isinstance(value, (int, float))
280 ]
281 for value in candidates:
282 try:
283 number = finite_float(
284 value,
285 name=f"comparisons[{index}] difference",
286 minimum=0.0,
287 )
288 except CliError:
289 finite = False
290 else:
291 maximum_difference = max(maximum_difference, number)
292 _check(
293 checks,
294 name="finite_convergence_differences",
295 passed=finite,
296 detail=f"largest recorded absolute difference is {maximum_difference}",
297 )
298 design = document.get("design")
299 varied = design.get("varied") if isinstance(design, Mapping) else None
300 _check(
301 checks,
302 name="sweep_design_recorded",
303 passed=isinstance(varied, list) and bool(varied),
304 detail=f"varied controls: {varied!r}",
305 severity="warning",
306 )
307 _check(
308 checks,
309 name="requested_tolerance_recorded",
310 passed=isinstance(document.get("acceptance"), (int, float))
311 and abs(float(document["acceptance"]) - tolerance) >= 0.0,
312 detail=f"report acceptance: {document.get('acceptance')!r}",
313 severity="warning",
314 )
315 return checks
318def audit_document(
319 document: Mapping[str, Any],
320 *,
321 tolerance: float,
322) -> dict[str, Any]:
323 """Audit one recognized portable report."""
325 if type(document.get("schema_version")) is not int or document["schema_version"] != 1:
326 raise CliError("schema_version must be the integer 1")
327 report_type = document.get("report_type")
328 if report_type == "qutip.two_level_simulation":
329 checks = _audit_simulation(document, tolerance=tolerance)
330 elif report_type == "qutip.convergence_sweep":
331 checks = _audit_convergence(document, tolerance=tolerance)
332 else:
333 raise CliError(
334 "unsupported report_type; expected qutip.two_level_simulation "
335 "or qutip.convergence_sweep"
336 )
338 failed = [
339 check
340 for check in checks
341 if not check["passed"] and check["severity"] == "error"
342 ]
343 warnings = [
344 check
345 for check in checks
346 if not check["passed"] and check["severity"] == "warning"
347 ]
348 status = "fail" if failed else ("pass_with_warnings" if warnings else "pass")
349 return {
350 "report_type": "qutip.result_audit",
351 "schema_version": 1,
352 "audited_report_type": report_type,
353 "tolerance": tolerance,
354 "status": status,
355 "interpretation": (
356 "Status describes report consistency, not physical validity or "
357 "convergence. Review analytic error, invariant audits, and the "
358 "original simulation or sweep acceptance status separately."
359 ),
360 "checks": checks,
361 "summary": {
362 "passed": sum(check["passed"] for check in checks),
363 "errors": len(failed),
364 "warnings": len(warnings),
365 },
366 "safety": {
367 "input_format": "strict JSON",
368 "python_object_deserialization": False,
369 "network": False,
370 },
371 }
374def build_parser() -> argparse.ArgumentParser:
375 parser = argparse.ArgumentParser(
376 description=(
377 "Audit bounded QuTiP skill JSON without importing QuTiP or "
378 "deserializing Python objects."
379 )
380 )
381 parser.add_argument("report", help="local strict-JSON report path")
382 parser.add_argument("--tolerance", type=float, default=1.0e-6)
383 add_output_arguments(parser)
384 return parser
387def main(argv: list[str] | None = None) -> int:
388 args = build_parser().parse_args(argv)
389 tolerance = finite_float(
390 args.tolerance,
391 name="tolerance",
392 minimum=1.0e-12,
393 maximum=0.1,
394 )
395 document = load_json_object(args.report)
396 report = audit_document(document, tolerance=tolerance)
397 emit_json(report, output=args.output, force=args.force)
398 return 0 if report["status"] != "fail" else 1
401if __name__ == "__main__":
402 raise SystemExit(run_cli(main))