Skills › Scientific Agent Skills › clinical-reports › consistency_checker.py
consistency_checker.py
393 lines · skills/clinical-reports/scripts/consistency_checker.py at 5f2bc03
1#!/usr/bin/env python3
2"""Check structured dates, units, denominators, percentages, and totals."""
4from __future__ import annotations
6import argparse
7import math
8import sys
9from typing import Any
11sys.dont_write_bytecode = True
13from _common import ( # noqa: E402
14 ValidationError,
15 error_report,
16 load_json_object,
17 parse_iso_date,
18 require_bool,
19 require_data_class,
20 require_exact_keys,
21 require_identifier,
22 require_nonnegative_int,
23 require_string,
24 write_json_report,
25)
27TOOL = "consistency_checker"
28TOP_LEVEL_FIELDS = {
29 "schema_version",
30 "artifact_kind",
31 "manifest_status",
32 "safety_notice",
33 "data_classification",
34 "authorized_purpose",
35 "authorization_verified",
36 "provenance_manifest",
37 "dates",
38 "date_ranges",
39 "quantities",
40 "proportions",
41 "totals",
42}
45def _array(data: dict[str, Any], field: str, errors: list[str]) -> list[Any]:
46 value = data.get(field, [])
47 if not isinstance(value, list):
48 errors.append(f"{field} must be an array")
49 return []
50 if len(value) > 10_000:
51 errors.append(f"{field} may contain at most 10,000 items")
52 return value[:10_000]
53 return value
56def _finite_number(value: Any, field: str) -> float:
57 if isinstance(value, bool) or not isinstance(value, (int, float)):
58 raise ValidationError(f"{field} must be a number")
59 try:
60 number = float(value)
61 except OverflowError as exc:
62 raise ValidationError(f"{field} must be a finite representable number") from exc
63 if not math.isfinite(number):
64 raise ValidationError(f"{field} must be finite")
65 return number
68def _track_id(
69 value: Any,
70 field: str,
71 seen: set[str],
72 errors: list[str],
73) -> str | None:
74 try:
75 identifier = require_identifier(value, field)
76 except ValidationError as exc:
77 errors.append(str(exc))
78 return None
79 if identifier in seen:
80 errors.append(f"duplicate check id: {identifier}")
81 seen.add(identifier)
82 return identifier
85def validate_consistency(data: dict[str, Any]) -> dict[str, Any]:
86 """Return deterministic discrepancies without changing any values."""
87 errors: list[str] = []
88 discrepancies: list[str] = []
89 warnings: list[str] = []
90 seen_ids: set[str] = set()
91 checked = 0
93 try:
94 require_exact_keys(data, TOP_LEVEL_FIELDS, "manifest")
95 if data.get("schema_version") != "2.0":
96 raise ValidationError("schema_version must be 2.0")
97 if data.get("manifest_status") != "BLOCKED_INCOMPLETE":
98 raise ValidationError("manifest_status must remain BLOCKED_INCOMPLETE")
99 require_string(data.get("safety_notice"), "safety_notice", max_length=1000)
100 require_string(data.get("authorized_purpose"), "authorized_purpose")
101 if not require_bool(data.get("authorization_verified"), "authorization_verified"):
102 raise ValidationError("authorization_verified must be true")
103 require_string(data.get("provenance_manifest"), "provenance_manifest")
104 except ValidationError as exc:
105 errors.append(str(exc))
106 if data.get("artifact_kind") != "consistency_manifest":
107 errors.append("artifact_kind must be consistency_manifest")
108 try:
109 require_data_class(data.get("data_classification"))
110 except ValidationError as exc:
111 errors.append(str(exc))
113 for index, item in enumerate(_array(data, "dates", errors)):
114 try:
115 item = require_exact_keys(
116 item,
117 {"id", "date", "source_fact_id"},
118 f"dates[{index}]",
119 )
120 except ValidationError as exc:
121 errors.append(str(exc))
122 continue
123 check_id = _track_id(item.get("id"), f"dates[{index}].id", seen_ids, errors)
124 try:
125 parse_iso_date(item.get("date"), f"dates[{index}].date")
126 require_identifier(
127 item.get("source_fact_id"),
128 f"dates[{index}].source_fact_id",
129 )
130 checked += 1
131 except ValidationError as exc:
132 errors.append(str(exc))
133 if check_id is None:
134 continue
136 for index, item in enumerate(_array(data, "date_ranges", errors)):
137 try:
138 item = require_exact_keys(
139 item,
140 {"id", "start", "end", "source_fact_id"},
141 f"date_ranges[{index}]",
142 )
143 except ValidationError as exc:
144 errors.append(str(exc))
145 continue
146 check_id = _track_id(
147 item.get("id"),
148 f"date_ranges[{index}].id",
149 seen_ids,
150 errors,
151 )
152 try:
153 start = parse_iso_date(item.get("start"), f"date_ranges[{index}].start")
154 end = parse_iso_date(item.get("end"), f"date_ranges[{index}].end")
155 require_identifier(
156 item.get("source_fact_id"),
157 f"date_ranges[{index}].source_fact_id",
158 )
159 checked += 1
160 if start > end and check_id:
161 discrepancies.append(f"{check_id}: start date is after end date")
162 except ValidationError as exc:
163 errors.append(str(exc))
165 series_units: dict[str, str] = {}
166 for index, item in enumerate(_array(data, "quantities", errors)):
167 try:
168 item = require_exact_keys(
169 item,
170 {
171 "id",
172 "series_id",
173 "value",
174 "unit",
175 "expected_unit",
176 "source_fact_id",
177 },
178 f"quantities[{index}]",
179 )
180 except ValidationError as exc:
181 errors.append(str(exc))
182 continue
183 check_id = _track_id(
184 item.get("id"),
185 f"quantities[{index}].id",
186 seen_ids,
187 errors,
188 )
189 try:
190 series_id = require_identifier(
191 item.get("series_id"),
192 f"quantities[{index}].series_id",
193 )
194 _finite_number(item.get("value"), f"quantities[{index}].value")
195 unit = require_string(
196 item.get("unit"),
197 f"quantities[{index}].unit",
198 max_length=80,
199 )
200 require_identifier(
201 item.get("source_fact_id"),
202 f"quantities[{index}].source_fact_id",
203 )
204 checked += 1
205 prior = series_units.get(series_id)
206 if prior is not None and prior != unit and check_id:
207 discrepancies.append(
208 f"{check_id}: unit {unit!r} differs from prior series unit {prior!r}"
209 )
210 else:
211 series_units[series_id] = unit
212 expected = item.get("expected_unit")
213 if expected is not None:
214 expected_unit = require_string(
215 expected,
216 f"quantities[{index}].expected_unit",
217 max_length=80,
218 )
219 if expected_unit != unit and check_id:
220 discrepancies.append(
221 f"{check_id}: unit does not match expected_unit"
222 )
223 except ValidationError as exc:
224 errors.append(str(exc))
226 for index, item in enumerate(_array(data, "proportions", errors)):
227 try:
228 item = require_exact_keys(
229 item,
230 {
231 "id",
232 "numerator",
233 "denominator",
234 "reported_percent",
235 "tolerance_percentage_points",
236 "source_fact_id",
237 },
238 f"proportions[{index}]",
239 )
240 except ValidationError as exc:
241 errors.append(str(exc))
242 continue
243 check_id = _track_id(
244 item.get("id"),
245 f"proportions[{index}].id",
246 seen_ids,
247 errors,
248 )
249 try:
250 numerator = require_nonnegative_int(
251 item.get("numerator"),
252 f"proportions[{index}].numerator",
253 )
254 denominator = require_nonnegative_int(
255 item.get("denominator"),
256 f"proportions[{index}].denominator",
257 )
258 reported = _finite_number(
259 item.get("reported_percent"),
260 f"proportions[{index}].reported_percent",
261 )
262 tolerance = _finite_number(
263 item.get("tolerance_percentage_points", 0.05),
264 f"proportions[{index}].tolerance_percentage_points",
265 )
266 require_identifier(
267 item.get("source_fact_id"),
268 f"proportions[{index}].source_fact_id",
269 )
270 checked += 1
271 if denominator == 0:
272 errors.append(f"proportions[{index}].denominator must be greater than zero")
273 elif numerator > denominator and check_id:
274 discrepancies.append(f"{check_id}: numerator exceeds denominator")
275 elif not 0 <= reported <= 100 and check_id:
276 discrepancies.append(f"{check_id}: reported_percent is outside 0-100")
277 elif tolerance < 0:
278 errors.append(
279 f"proportions[{index}].tolerance_percentage_points must be non-negative"
280 )
281 else:
282 calculated = numerator / denominator * 100
283 if abs(calculated - reported) > tolerance and check_id:
284 discrepancies.append(
285 f"{check_id}: reported_percent differs from n/N beyond tolerance"
286 )
287 except ValidationError as exc:
288 errors.append(str(exc))
290 for index, item in enumerate(_array(data, "totals", errors)):
291 try:
292 item = require_exact_keys(
293 item,
294 {"id", "components", "reported_total", "source_fact_ids"},
295 f"totals[{index}]",
296 )
297 except ValidationError as exc:
298 errors.append(str(exc))
299 continue
300 check_id = _track_id(
301 item.get("id"),
302 f"totals[{index}].id",
303 seen_ids,
304 errors,
305 )
306 try:
307 components = item.get("components")
308 if not isinstance(components, list) or not components:
309 raise ValidationError(f"totals[{index}].components must be non-empty")
310 parsed_components = [
311 require_nonnegative_int(value, f"totals[{index}].components")
312 for value in components
313 ]
314 reported_total = require_nonnegative_int(
315 item.get("reported_total"),
316 f"totals[{index}].reported_total",
317 )
318 source_fact_ids = item.get("source_fact_ids")
319 if not isinstance(source_fact_ids, list) or not source_fact_ids:
320 raise ValidationError(
321 f"totals[{index}].source_fact_ids must be non-empty"
322 )
323 for fact_id in source_fact_ids:
324 require_identifier(fact_id, f"totals[{index}].source_fact_ids")
325 checked += 1
326 if sum(parsed_components) != reported_total and check_id:
327 discrepancies.append(f"{check_id}: components do not equal reported_total")
328 except ValidationError as exc:
329 errors.append(str(exc))
331 if checked == 0:
332 errors.append("manifest contains no valid checks")
333 if not discrepancies and not errors:
334 warnings.append("No discrepancy was found within the declared checks and tolerances.")
336 status = (
337 "BLOCKED_INVALID_SCHEMA"
338 if errors
339 else "DISCREPANCIES_REQUIRE_RESOLUTION"
340 if discrepancies
341 else "CONSISTENT_WITHIN_DECLARED_TOLERANCES_REVIEW_REQUIRED"
342 )
343 return {
344 "tool": TOOL,
345 "status": status,
346 "checks_completed": checked,
347 "errors": errors,
348 "discrepancies": discrepancies,
349 "warnings": warnings,
350 "limitations": [
351 "No source was chosen as authoritative and no value was changed.",
352 "Arithmetic and format checks do not validate clinical or statistical meaning.",
353 ],
354 "review_required": True,
355 "authorizes_clinical_use_or_submission": False,
356 }
359def build_parser() -> argparse.ArgumentParser:
360 parser = argparse.ArgumentParser(
361 description=(
362 "Check bounded structured dates, units, n/N percentages, and totals. "
363 "Does not infer, convert, reconcile, or alter values."
364 )
365 )
366 parser.add_argument("input_file", help="Local consistency manifest (.json)")
367 parser.add_argument("-o", "--output", help="Optional local JSON report")
368 parser.add_argument("--overwrite", action="store_true")
369 return parser
372def main(argv: list[str] | None = None) -> int:
373 args = build_parser().parse_args(argv)
374 try:
375 _, data = load_json_object(args.input_file)
376 report = validate_consistency(data)
377 except (OSError, ValidationError) as exc:
378 report = error_report(TOOL, exc)
379 try:
380 write_json_report(report, args.output, overwrite=args.overwrite)
381 except (OSError, ValidationError) as exc:
382 print(str(exc), file=sys.stderr)
383 return 2
384 return (
385 0
386 if report["status"]
387 == "CONSISTENT_WITHIN_DECLARED_TOLERANCES_REVIEW_REQUIRED"
388 else 1
389 )
392if __name__ == "__main__":
393 raise SystemExit(main())