Skills › Scientific Agent Skills › uncertainty-and-units › audit_units.py
audit_units.py
574 lines · skills/uncertainty-and-units/scripts/audit_units.py at 5f2bc03
1#!/usr/bin/env python3
2"""Scan Python source for the unit and uncertainty defects that stay silent.
4The heuristic rules identify code that warrants review, not a proof of error: a stripped unit, a rescaled covariance matrix, a destroyed
5correlation, a population standard deviation used as a standard uncertainty.
6The scan is static - it parses the file and never imports or runs it.
7"""
9# This module *is* the rule tables: it names every logarithmic unit and lists
10# every CODATA value it looks for, so scanning it flags its own data. The
11# directive below is the same mechanism any caller has, and suppressions are
12# counted in the report rather than hidden.
13# audit-units: ignore-file CONST001, UNIT004
15from __future__ import annotations
17import argparse
18import ast
19import math
20import re
21import sys
22from decimal import Decimal
23from typing import Any
25import _common
26from _common import CliError
29SEVERITY_ORDER = {"low": 1, "medium": 2, "high": 3}
30MAX_FILES = 50
32# `# audit-units: ignore [RULE, ...]` silences the rules on that line;
33# `# audit-units: ignore-file RULE[, ...]` silences them for the whole module.
34# Both forms are counted in the report, so a suppression stays visible.
35SUPPRESSION = re.compile(
36 r"#\s*audit-units:\s*ignore(?P<scope>-file)?(?P<rules>[\sA-Z0-9,]*)"
37)
39OFFSET_UNIT_TOKENS = {
40 "degC",
41 "degF",
42 "celsius",
43 "fahrenheit",
44 "degree_Celsius",
45 "degree_Fahrenheit",
46 "degreeC",
47 "degreeF",
48}
49LOGARITHMIC_UNIT_TOKENS = {
50 "dB",
51 "dBm",
52 "dBW",
53 "dBu",
54 "dBV",
55 "dBi",
56 "decibel",
57 "decibelmilliwatt",
58 "decibelwatt",
59}
60ELEMENTARY_FUNCTIONS = {
61 "acos",
62 "acosh",
63 "asin",
64 "asinh",
65 "atan",
66 "atan2",
67 "atanh",
68 "cos",
69 "cosh",
70 "degrees",
71 "erf",
72 "exp",
73 "expm1",
74 "fabs",
75 "hypot",
76 "log",
77 "log10",
78 "log1p",
79 "radians",
80 "sin",
81 "sinh",
82 "sqrt",
83 "tan",
84 "tanh",
85}
86UNIT_PRESERVING_CALLS = {"to", "to_base_units", "to_reduced_units", "m_as", "ito"}
88# CODATA 2022 recommended values, plus the constants the 2019 SI redefinition
89# fixed exactly. A literal within one part in a thousand of one of these is
90# almost always a constant somebody typed from memory. Every accessor below was
91# checked against scipy.constants 1.18.1, whose default data set is CODATA 2022.
92KNOWN_CONSTANTS: tuple[tuple[float, str, str], ...] = (
93 (299792458.0, "speed of light in vacuum", "value('speed of light in vacuum')"),
94 (6.62607015e-34, "Planck constant", "value('Planck constant')"),
95 (1.054571817e-34, "reduced Planck constant", "value('reduced Planck constant')"),
96 (1.602176634e-19, "elementary charge", "value('elementary charge')"),
97 (1.380649e-23, "Boltzmann constant", "value('Boltzmann constant')"),
98 (6.02214076e23, "Avogadro constant", "value('Avogadro constant')"),
99 (8.314462618, "molar gas constant", "value('molar gas constant')"),
100 (9.1093837139e-31, "electron mass", "value('electron mass')"),
101 (1.67262192595e-27, "proton mass", "value('proton mass')"),
102 (1.66053906892e-27, "atomic mass constant", "value('atomic mass constant')"),
103 (5.670374419e-8, "Stefan-Boltzmann constant", "value('Stefan-Boltzmann constant')"),
104 (
105 6.67430e-11,
106 "Newtonian constant of gravitation",
107 "value('Newtonian constant of gravitation')",
108 ),
109 (
110 9.80665,
111 "standard acceleration of gravity",
112 "value('standard acceleration of gravity')",
113 ),
114 (
115 8.8541878188e-12,
116 "vacuum electric permittivity",
117 "value('vacuum electric permittivity')",
118 ),
119 (
120 1.25663706127e-6,
121 "vacuum magnetic permeability",
122 "value('vacuum mag. permeability')",
123 ),
124 (5.29177210544e-11, "Bohr radius", "value('Bohr radius')"),
125 (7.2973525643e-3, "fine-structure constant", "value('fine-structure constant')"),
126 (10973731.568157, "Rydberg constant", "value('Rydberg constant')"),
127 (96485.33212, "Faraday constant", "value('Faraday constant')"),
128 (101325.0, "standard atmosphere", "value('standard atmosphere')"),
129 (273.15, "zero of the Celsius scale", "zero_Celsius"),
130)
133def significant_digits(value: float) -> int:
134 """Count the significant digits in a float's shortest representation."""
136 digits = list(Decimal(repr(abs(value))).as_tuple().digits)
137 while len(digits) > 1 and digits[-1] == 0:
138 digits.pop()
139 while len(digits) > 1 and digits[0] == 0:
140 digits.pop(0)
141 return len(digits)
144def parse_suppressions(source: str) -> tuple[dict[int, set[str]], set[str]]:
145 """Return per-line and whole-file rule suppressions.
147 A trailing directive applies to its own line. A directive alone on a line
148 applies to the next line, so a long statement does not have to carry the
149 comment. Naming no rule, or `ALL`, suppresses every rule and is recorded as
150 the `*` sentinel.
151 """
153 per_line: dict[int, set[str]] = {}
154 whole_file: set[str] = set()
155 for number, line in enumerate(source.splitlines(), start=1):
156 match = SUPPRESSION.search(line)
157 if match is None:
158 continue
159 rules = {
160 item.strip()
161 for item in match.group("rules").replace(" ", ",").split(",")
162 if item.strip()
163 }
164 if not rules or "ALL" in rules:
165 rules = {"*"}
166 if match.group("scope"):
167 whole_file |= rules
168 continue
169 target = number + 1 if line.lstrip().startswith("#") else number
170 existing = per_line.get(target)
171 if existing == {"*"} or rules == {"*"}:
172 per_line[target] = {"*"}
173 else:
174 per_line[target] = (existing or set()) | rules
175 return per_line, whole_file
178def _tokens(text: str) -> set[str]:
179 """Split a unit string into candidate unit tokens."""
181 current: list[str] = []
182 found: set[str] = set()
183 for character in text:
184 if character.isalnum() or character == "_":
185 current.append(character)
186 else:
187 if current:
188 found.add("".join(current))
189 current = []
190 if current:
191 found.add("".join(current))
192 return found
195class Auditor(ast.NodeVisitor):
196 """Collect unit and uncertainty findings from one parsed module."""
198 def __init__(self, source: str, filename: str) -> None:
199 self.filename = filename
200 self.findings: list[dict[str, Any]] = []
201 self.registry_sites: list[ast.AST] = []
202 self.imports_uncertainties = False
203 self.saw_delta_unit = False
204 self.saw_offset_autoconvert = "autoconvert_offset_to_baseunit" in source
205 self.line_suppressions, self.file_suppressions = parse_suppressions(source)
206 self.suppressed = 0
208 def report(
209 self,
210 node: ast.AST,
211 rule: str,
212 severity: str,
213 message: str,
214 remedy: str,
215 ) -> None:
216 self.findings.append(
217 {
218 "rule": rule,
219 "severity": severity,
220 "line": getattr(node, "lineno", 0),
221 "column": getattr(node, "col_offset", 0) + 1,
222 "message": message,
223 "remedy": remedy,
224 }
225 )
227 # --- imports -----------------------------------------------------------
229 def visit_Import(self, node: ast.Import) -> None:
230 for alias in node.names:
231 if alias.name.split(".", 1)[0] == "uncertainties":
232 self.imports_uncertainties = True
233 self.generic_visit(node)
235 def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
236 if node.module and node.module.split(".", 1)[0] == "uncertainties":
237 self.imports_uncertainties = True
238 self.generic_visit(node)
240 # --- string literals ---------------------------------------------------
242 def visit_Constant(self, node: ast.Constant) -> None:
243 if isinstance(node.value, str):
244 tokens = _tokens(node.value)
245 if any(token.startswith("delta_") for token in tokens):
246 self.saw_delta_unit = True
247 if tokens & OFFSET_UNIT_TOKENS:
248 self.report(
249 node,
250 "UNIT002",
251 "medium",
252 "offset temperature unit used; pint refuses to multiply or add "
253 "degC/degF quantities because the operation is ambiguous",
254 "express temperature differences in delta_degC, or build the "
255 "registry with UnitRegistry(autoconvert_offset_to_baseunit=True)",
256 )
257 if tokens & LOGARITHMIC_UNIT_TOKENS:
258 self.report(
259 node,
260 "UNIT004",
261 "medium",
262 "logarithmic unit used; adding two pint quantities in dB or dBm "
263 "multiplies the underlying linear quantities instead of summing "
264 "them, and reports the product in squared base units",
265 "convert to a linear unit before arithmetic, then convert back",
266 )
267 elif isinstance(node.value, (int, float)) and not isinstance(node.value, bool):
268 self._check_constant_literal(node, float(node.value))
269 self.generic_visit(node)
271 def _check_constant_literal(self, node: ast.Constant, value: float) -> None:
272 if value == 0 or not math.isfinite(value):
273 return
274 if significant_digits(value) < 3:
275 return
276 for reference, description, accessor in KNOWN_CONSTANTS:
277 if abs(value - reference) <= 1e-3 * abs(reference):
278 self.report(
279 node,
280 "CONST001",
281 "low",
282 f"hard-coded literal matches the {description}; the recommended "
283 "value may be measured or exact; confirm its source and uncertainty",
284 f"use scipy.constants.{accessor}, with scipy.constants.precision "
285 "for the relative standard uncertainty (0.0 when the SI fixes "
286 "the constant exactly)",
287 )
288 return
290 # --- attribute access --------------------------------------------------
292 def visit_Attribute(self, node: ast.Attribute) -> None:
293 if node.attr == "magnitude":
294 source = node.value
295 preserved = (
296 isinstance(source, ast.Call)
297 and isinstance(source.func, ast.Attribute)
298 and source.func.attr in UNIT_PRESERVING_CALLS
299 )
300 if not preserved:
301 self.report(
302 node,
303 "UNIT003",
304 "high",
305 ".magnitude strips the unit without stating which one, so the "
306 "number that comes out depends on whatever the quantity happened "
307 "to be carrying",
308 "call .to('unit').magnitude or .m_as('unit') so the scale is "
309 "fixed at the point of extraction",
310 )
311 self.generic_visit(node)
313 # --- calls -------------------------------------------------------------
315 def visit_Call(self, node: ast.Call) -> None:
316 name = self._call_name(node)
317 keywords = {keyword.arg for keyword in node.keywords if keyword.arg}
319 if name == "UnitRegistry":
320 self.registry_sites.append(node)
322 if name == "curve_fit" and "absolute_sigma" not in keywords:
323 self.report(
324 node,
325 "UNC001",
326 "high",
327 "curve_fit rescales the covariance matrix by the reduced chi-square "
328 "unless absolute_sigma=True, so parameter uncertainties silently "
329 "absorb the goodness of fit",
330 "pass absolute_sigma=True when sigma holds real standard "
331 "uncertainties; leave it False only for relative weights",
332 )
334 if name in {"std", "var", "nanstd", "nanvar"} and not ({"ddof", "correction"} & keywords):
335 if self._is_numpy_call(node):
336 self.report(
337 node,
338 "UNC002",
339 "medium",
340 f"numpy {name} defaults to ddof=0, which is the population "
341 "spread; a Type A standard uncertainty needs the sample "
342 "estimate",
343 "pass ddof=1, and divide by sqrt(n) as well when you want the "
344 "standard uncertainty of the mean rather than of one reading",
345 )
347 if name == "ufloat":
348 for argument in node.args + [kw.value for kw in node.keywords]:
349 if any(
350 isinstance(inner, ast.Attribute)
351 and inner.attr in {"nominal_value", "std_dev", "n", "s"}
352 for inner in ast.walk(argument)
353 ):
354 self.report(
355 node,
356 "UNC004",
357 "high",
358 "rebuilding a ufloat from another variable's nominal_value "
359 "and std_dev creates an independent variable, discarding "
360 "every correlation the original carried",
361 "pass the existing variable through, or rebuild the whole "
362 "set with correlated_values(values, covariance_matrix)",
363 )
364 break
366 if self.imports_uncertainties and name in ELEMENTARY_FUNCTIONS:
367 module = self._call_module(node)
368 if module in {"math", "np", "numpy"}:
369 self.report(
370 node,
371 "UNC003",
372 "medium",
373 f"{module}.{name} has no derivative rule for an uncertain value; "
374 "on a scalar it raises TypeError, and on an object array it "
375 "fails in the ufunc loop",
376 "use uncertainties.umath for scalars and "
377 "uncertainties.unumpy for arrays",
378 )
380 self.generic_visit(node)
382 @staticmethod
383 def _call_name(node: ast.Call) -> str | None:
384 if isinstance(node.func, ast.Name):
385 return node.func.id
386 if isinstance(node.func, ast.Attribute):
387 return node.func.attr
388 return None
390 @staticmethod
391 def _call_module(node: ast.Call) -> str | None:
392 if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name):
393 return node.func.value.id
394 return None
396 @staticmethod
397 def _is_numpy_call(node: ast.Call) -> bool:
398 return (
399 isinstance(node.func, ast.Attribute)
400 and isinstance(node.func.value, ast.Name)
401 and node.func.value.id in {"np", "numpy"}
402 )
404 def finish(self) -> list[dict[str, Any]]:
405 """Emit the findings that need whole-module context."""
407 if len(self.registry_sites) > 1:
408 for node in self.registry_sites[1:]:
409 self.report(
410 node,
411 "UNIT001",
412 "medium",
413 "a second UnitRegistry is constructed in this module; quantities "
414 "from different registries raise ValueError on any shared "
415 "operation",
416 "build one registry and share it, or call "
417 "pint.get_application_registry() everywhere",
418 )
419 if self.saw_offset_autoconvert or self.saw_delta_unit:
420 self.findings = [
421 finding for finding in self.findings if finding["rule"] != "UNIT002"
422 ]
423 kept = [finding for finding in self.findings if not self._suppressed(finding)]
424 self.suppressed = len(self.findings) - len(kept)
425 kept.sort(key=lambda item: (item["line"], item["column"], item["rule"]))
426 self.findings = kept
427 return kept
429 def _suppressed(self, finding: dict[str, Any]) -> bool:
430 """Report whether a directive comment silences this finding."""
432 if "*" in self.file_suppressions or finding["rule"] in self.file_suppressions:
433 return True
434 rules = self.line_suppressions.get(finding["line"])
435 if rules is None:
436 return False
437 return "*" in rules or finding["rule"] in rules
440def audit_source(source: str, filename: str) -> dict[str, Any]:
441 """Parse one module and return its findings and suppression count."""
443 try:
444 tree = ast.parse(source, filename=filename)
445 except SyntaxError as exc:
446 raise CliError(f"{filename}: cannot parse as Python: {exc}") from exc
447 auditor = Auditor(source, filename)
448 auditor.visit(tree)
449 findings = auditor.finish()
450 return {"findings": findings, "suppressed": auditor.suppressed}
453def render_markdown(document: dict[str, Any]) -> str:
454 """Render findings grouped by file."""
456 lines = ["# Unit and uncertainty audit", ""]
457 counts = document["counts"]
458 lines.append(
459 f"{counts['total']} findings across {document['files_scanned']} files "
460 f"({counts['high']} high, {counts['medium']} medium, {counts['low']} low), "
461 f"{counts['suppressed']} suppressed by directive comments."
462 )
463 for entry in document["results"]:
464 lines += ["", f"## {entry['file']}", ""]
465 if not entry["findings"]:
466 lines.append(f"No findings ({entry['suppressed']} suppressed).")
467 continue
468 lines.append("| Line | Rule | Severity | Finding |")
469 lines.append("| --- | --- | --- | --- |")
470 for finding in entry["findings"]:
471 lines.append(
472 f"| {finding['line']} | {finding['rule']} | {finding['severity']} | "
473 f"{finding['message']} |"
474 )
475 lines.append("")
476 for finding in entry["findings"]:
477 lines.append(
478 f"- **{finding['rule']}** (line {finding['line']}): "
479 f"{finding['remedy']}"
480 )
481 return "\n".join(lines) + "\n"
484def build_parser() -> argparse.ArgumentParser:
485 parser = argparse.ArgumentParser(
486 description=(
487 "Statically audit Python sources for stripped units, offset-temperature "
488 "arithmetic, broken uncertainty propagation, and hard-coded constants."
489 )
490 )
491 parser.add_argument(
492 "--input",
493 action="append",
494 default=[],
495 required=True,
496 help="Python file to audit; repeatable",
497 )
498 parser.add_argument(
499 "--fail-on",
500 choices=("none", "low", "medium", "high"),
501 default="high",
502 help="lowest severity that makes the command exit non-zero (default high)",
503 )
504 parser.add_argument("--format", choices=("json", "markdown"), default="json")
505 parser.add_argument("--output", help="write the report to this file")
506 parser.add_argument(
507 "--force", action="store_true", help="overwrite an existing output file"
508 )
509 return parser
512def run(arguments: argparse.Namespace) -> dict[str, Any]:
513 """Audit every requested file and summarize the findings."""
515 if len(arguments.input) > MAX_FILES:
516 raise CliError(f"at most {MAX_FILES} files can be audited in one run")
517 results = []
518 counts = {"total": 0, "low": 0, "medium": 0, "high": 0, "suppressed": 0}
519 for item in arguments.input:
520 source, path = _common.read_text_file(item, suffixes=(".py",))
521 audited = audit_source(source, path.name)
522 results.append(
523 {
524 "file": str(path),
525 "findings": audited["findings"],
526 "suppressed": audited["suppressed"],
527 }
528 )
529 counts["suppressed"] += audited["suppressed"]
530 for finding in audited["findings"]:
531 counts["total"] += 1
532 counts[finding["severity"]] += 1
533 return {
534 "files_scanned": len(results),
535 "counts": counts,
536 "results": results,
537 }
540def exit_code(document: dict[str, Any], fail_on: str) -> int:
541 """Return 1 when a finding meets the configured severity threshold."""
543 if fail_on == "none":
544 return 0
545 threshold = SEVERITY_ORDER[fail_on]
546 for entry in document["results"]:
547 for finding in entry["findings"]:
548 if SEVERITY_ORDER[finding["severity"]] >= threshold:
549 return 1
550 return 0
553def main(argv: list[str] | None = None) -> int:
554 arguments = build_parser().parse_args(argv)
555 try:
556 document = run(arguments)
557 if arguments.format == "markdown":
558 _common.emit_text(
559 render_markdown(document),
560 output=arguments.output,
561 force=arguments.force,
562 )
563 else:
564 _common.emit_json(
565 document, output=arguments.output, force=arguments.force
566 )
567 except CliError as error:
568 print(f"error: {error}", file=sys.stderr)
569 return 2
570 return exit_code(document, arguments.fail_on)
573if __name__ == "__main__":
574 raise SystemExit(main())