Skills › Scientific Agent Skills › scientific-writing › check_consistency.py
check_consistency.py
408 lines · skills/scientific-writing/scripts/check_consistency.py at 5f2bc03
1"""Check numeric and methods-results consistency in a bounded JSON registry."""
3from __future__ import annotations
5import argparse
6import math
7import re
8from typing import Any
10from _common import (
11 InputError,
12 Issue,
13 emit_report,
14 is_nonempty_string,
15 is_placeholder,
16 issue,
17 read_json,
18 require_list,
19 require_object,
20 run,
21)
23TOOL = "check_consistency"
24FACT_ID_RE = re.compile(r"^N[0-9]{3,8}$")
25METHOD_ID_RE = re.compile(r"^M[0-9]{3,8}$")
26RESULT_ID_RE = re.compile(r"^R[0-9]{3,8}$")
27OUTCOME_ID_RE = re.compile(r"^O[0-9]{3,8}$")
28EVIDENCE_ID_RE = re.compile(r"^E[0-9]{3,8}$")
29ANALYSIS_INTENT = {"confirmatory", "exploratory", "descriptive"}
30PROTOCOL_STATUS = {
31 "prespecified",
32 "amended_before_analysis",
33 "post_hoc",
34 "not_applicable",
35}
36ROOT_FIELDS = {"schema_version", "numeric_facts", "methods", "results"}
37FACT_FIELDS = {
38 "fact_id",
39 "concept",
40 "section",
41 "value",
42 "unit",
43 "numerator",
44 "denominator",
45 "sample_size",
46 "analysis_set",
47 "evidence_ids",
48}
49METHOD_FIELDS = {
50 "method_id",
51 "name",
52 "analysis_intent",
53 "protocol_status",
54 "outcome_ids",
55}
56RESULT_FIELDS = {
57 "result_id",
58 "method_id",
59 "outcome_id",
60 "analysis_intent",
61 "sample_size",
62 "evidence_ids",
63 "reported_sections",
64}
67def _is_number(value: Any) -> bool:
68 return (
69 isinstance(value, (int, float))
70 and not isinstance(value, bool)
71 and math.isfinite(value)
72 )
75def _check_fields(
76 value: dict[str, Any],
77 expected: set[str],
78 *,
79 location: str,
80 issues: list[Issue],
81) -> None:
82 for key in sorted(set(value) - expected):
83 issues.append(
84 issue("error", "UNKNOWN_SCHEMA_FIELD", location=location, item_id=key)
85 )
86 for key in sorted(expected - set(value)):
87 issues.append(
88 issue("error", "MISSING_SCHEMA_FIELD", location=location, item_id=key)
89 )
92def _validate_ids(
93 values: Any,
94 *,
95 pattern: re.Pattern[str],
96 location: str,
97 code: str,
98 issues: list[Issue],
99) -> list[str]:
100 ids = require_list(values, location)
101 valid: list[str] = []
102 for raw_value in ids:
103 if not isinstance(raw_value, str) or not pattern.fullmatch(raw_value):
104 issues.append(issue("error", code, location=location))
105 else:
106 valid.append(raw_value)
107 return valid
110def validate_numeric_facts(data: dict[str, Any]) -> tuple[list[Issue], int]:
111 issues: list[Issue] = []
112 facts = require_list(data.get("numeric_facts"), "numeric_facts")
113 seen_ids: set[str] = set()
114 by_concept: dict[
115 tuple[str, str],
116 tuple[float, str, str, int | None, float | None, float | None],
117 ] = {}
118 for index, raw_fact in enumerate(facts):
119 fact = require_object(raw_fact, f"numeric_facts[{index}]")
120 location = f"numeric_facts[{index}]"
121 _check_fields(fact, FACT_FIELDS, location=location, issues=issues)
122 fact_id = fact.get("fact_id")
123 if not isinstance(fact_id, str) or not FACT_ID_RE.fullmatch(fact_id):
124 issues.append(issue("error", "INVALID_FACT_ID", location=location))
125 fact_id = None
126 elif fact_id in seen_ids:
127 issues.append(issue("error", "DUPLICATE_FACT_ID", item_id=fact_id))
128 else:
129 seen_ids.add(fact_id)
131 concept = fact.get("concept")
132 section = fact.get("section")
133 unit = fact.get("unit")
134 analysis_set = fact.get("analysis_set")
135 for key, value in {
136 "concept": concept,
137 "section": section,
138 "unit": unit,
139 "analysis_set": analysis_set,
140 }.items():
141 if not is_nonempty_string(value) or is_placeholder(value):
142 issues.append(
143 issue("error", "MISSING_FACT_FIELD", location=key, item_id=fact_id)
144 )
146 value = fact.get("value")
147 if not _is_number(value):
148 issues.append(issue("error", "INVALID_FACT_VALUE", item_id=fact_id))
149 sample_size = fact.get("sample_size")
150 if sample_size is not None and (
151 not isinstance(sample_size, int)
152 or isinstance(sample_size, bool)
153 or sample_size <= 0
154 ):
155 issues.append(issue("error", "INVALID_FACT_SAMPLE_SIZE", item_id=fact_id))
157 numerator = fact.get("numerator")
158 denominator = fact.get("denominator")
159 if numerator is not None or denominator is not None:
160 if (
161 not _is_number(numerator)
162 or not _is_number(denominator)
163 or denominator <= 0
164 ):
165 issues.append(
166 issue("error", "INVALID_NUMERATOR_DENOMINATOR", item_id=fact_id)
167 )
168 elif numerator > denominator:
169 issues.append(
170 issue("error", "NUMERATOR_EXCEEDS_DENOMINATOR", item_id=fact_id)
171 )
172 elif unit == "percent" and _is_number(value):
173 expected = 100.0 * float(numerator) / float(denominator)
174 if not math.isclose(float(value), expected, rel_tol=0.0, abs_tol=0.05):
175 issues.append(
176 issue("error", "PERCENT_DENOMINATOR_MISMATCH", item_id=fact_id)
177 )
179 evidence_ids = _validate_ids(
180 fact.get("evidence_ids"),
181 pattern=EVIDENCE_ID_RE,
182 location=f"{location}.evidence_ids",
183 code="INVALID_FACT_EVIDENCE_ID",
184 issues=issues,
185 )
186 if not evidence_ids:
187 issues.append(issue("error", "FACT_WITHOUT_EVIDENCE", item_id=fact_id))
189 if (
190 is_nonempty_string(concept)
191 and is_nonempty_string(analysis_set)
192 and is_nonempty_string(unit)
193 and _is_number(value)
194 and is_nonempty_string(section)
195 ):
196 key = (str(concept), str(analysis_set))
197 prior = by_concept.get(key)
198 current = (
199 float(value),
200 str(unit),
201 str(section),
202 (
203 sample_size
204 if isinstance(sample_size, int)
205 and not isinstance(sample_size, bool)
206 and sample_size > 0
207 else None
208 ),
209 float(numerator) if _is_number(numerator) else None,
210 float(denominator) if _is_number(denominator) else None,
211 )
212 if prior is not None:
213 if prior[1] != current[1]:
214 issues.append(issue("error", "UNIT_MISMATCH", item_id=fact_id))
215 elif not math.isclose(prior[0], current[0], rel_tol=0.0, abs_tol=1e-12):
216 issues.append(
217 issue("error", "CROSS_SECTION_VALUE_MISMATCH", item_id=fact_id)
218 )
219 if prior[3] != current[3]:
220 issues.append(
221 issue(
222 "error",
223 "CROSS_SECTION_SAMPLE_SIZE_MISMATCH",
224 item_id=fact_id,
225 )
226 )
227 if prior[4:] != current[4:]:
228 issues.append(
229 issue(
230 "error",
231 "CROSS_SECTION_DENOMINATOR_MISMATCH",
232 item_id=fact_id,
233 )
234 )
235 else:
236 by_concept[key] = current
237 return issues, len(facts)
240def validate_methods_results(data: dict[str, Any]) -> tuple[list[Issue], int, int]:
241 issues: list[Issue] = []
242 methods: dict[str, dict[str, Any]] = {}
243 expected_outcomes: dict[str, set[str]] = {}
244 for index, raw_method in enumerate(require_list(data.get("methods"), "methods")):
245 method = require_object(raw_method, f"methods[{index}]")
246 _check_fields(
247 method,
248 METHOD_FIELDS,
249 location=f"methods[{index}]",
250 issues=issues,
251 )
252 method_id = method.get("method_id")
253 if not isinstance(method_id, str) or not METHOD_ID_RE.fullmatch(method_id):
254 issues.append(
255 issue("error", "INVALID_METHOD_ID", location=f"methods[{index}]")
256 )
257 continue
258 if method_id in methods:
259 issues.append(issue("error", "DUPLICATE_METHOD_ID", item_id=method_id))
260 continue
261 methods[method_id] = method
262 if not is_nonempty_string(method.get("name")) or is_placeholder(
263 method.get("name")
264 ):
265 issues.append(issue("error", "MISSING_METHOD_NAME", item_id=method_id))
266 if method.get("analysis_intent") not in ANALYSIS_INTENT:
267 issues.append(
268 issue("error", "INVALID_METHOD_ANALYSIS_INTENT", item_id=method_id)
269 )
270 if method.get("protocol_status") not in PROTOCOL_STATUS:
271 issues.append(issue("error", "INVALID_PROTOCOL_STATUS", item_id=method_id))
272 outcomes = set(
273 _validate_ids(
274 method.get("outcome_ids"),
275 pattern=OUTCOME_ID_RE,
276 location=f"methods[{index}].outcome_ids",
277 code="INVALID_METHOD_OUTCOME_ID",
278 issues=issues,
279 )
280 )
281 if not outcomes:
282 issues.append(issue("error", "METHOD_WITHOUT_OUTCOME", item_id=method_id))
283 expected_outcomes[method_id] = outcomes
285 results = require_list(data.get("results"), "results")
286 seen_result_ids: set[str] = set()
287 observed_outcomes: dict[str, set[str]] = {method_id: set() for method_id in methods}
288 for index, raw_result in enumerate(results):
289 result = require_object(raw_result, f"results[{index}]")
290 _check_fields(
291 result,
292 RESULT_FIELDS,
293 location=f"results[{index}]",
294 issues=issues,
295 )
296 result_id = result.get("result_id")
297 if not isinstance(result_id, str) or not RESULT_ID_RE.fullmatch(result_id):
298 issues.append(
299 issue("error", "INVALID_RESULT_ID", location=f"results[{index}]")
300 )
301 result_id = None
302 elif result_id in seen_result_ids:
303 issues.append(issue("error", "DUPLICATE_RESULT_ID", item_id=result_id))
304 else:
305 seen_result_ids.add(result_id)
307 method_id = result.get("method_id")
308 outcome_id = result.get("outcome_id")
309 if method_id not in methods:
310 issues.append(
311 issue("error", "RESULT_WITHOUT_DECLARED_METHOD", item_id=result_id)
312 )
313 else:
314 if not isinstance(outcome_id, str) or not OUTCOME_ID_RE.fullmatch(
315 outcome_id
316 ):
317 issues.append(
318 issue("error", "INVALID_RESULT_OUTCOME_ID", item_id=result_id)
319 )
320 else:
321 observed_outcomes[method_id].add(outcome_id)
322 if outcome_id not in expected_outcomes[method_id]:
323 issues.append(
324 issue("error", "UNDECLARED_RESULT_OUTCOME", item_id=result_id)
325 )
326 if result.get("analysis_intent") != methods[method_id].get(
327 "analysis_intent"
328 ):
329 issues.append(
330 issue("error", "ANALYSIS_INTENT_MISMATCH", item_id=result_id)
331 )
333 sample_size = result.get("sample_size")
334 if (
335 not isinstance(sample_size, int)
336 or isinstance(sample_size, bool)
337 or sample_size <= 0
338 ):
339 issues.append(
340 issue("error", "INVALID_RESULT_SAMPLE_SIZE", item_id=result_id)
341 )
342 evidence_ids = _validate_ids(
343 result.get("evidence_ids"),
344 pattern=EVIDENCE_ID_RE,
345 location=f"results[{index}].evidence_ids",
346 code="INVALID_RESULT_EVIDENCE_ID",
347 issues=issues,
348 )
349 if not evidence_ids:
350 issues.append(issue("error", "RESULT_WITHOUT_EVIDENCE", item_id=result_id))
351 sections = result.get("reported_sections")
352 if (
353 not isinstance(sections, list)
354 or not sections
355 or not all(is_nonempty_string(value) for value in sections)
356 ):
357 issues.append(
358 issue("error", "RESULT_WITHOUT_REPORTED_SECTION", item_id=result_id)
359 )
361 for method_id, outcomes in expected_outcomes.items():
362 for outcome_id in sorted(outcomes - observed_outcomes.get(method_id, set())):
363 issues.append(
364 issue(
365 "error",
366 "METHOD_OUTCOME_WITHOUT_RESULT",
367 location=method_id,
368 item_id=outcome_id,
369 )
370 )
371 return issues, len(methods), len(results)
374def build_parser() -> argparse.ArgumentParser:
375 parser = argparse.ArgumentParser(
376 description=(
377 "Check a local JSON registry for repeated numeric values, units, "
378 "denominators, sample sizes, and declared methods-results mappings."
379 )
380 )
381 parser.add_argument("registry", help="UTF-8 JSON consistency registry")
382 return parser
385def cli() -> int:
386 args = build_parser().parse_args()
387 data = require_object(read_json(args.registry), "consistency_registry")
388 root_issues: list[Issue] = []
389 _check_fields(data, ROOT_FIELDS, location="root", issues=root_issues)
390 if data.get("schema_version") != "1.0":
391 raise InputError("unsupported consistency registry version")
392 issues, fact_count = validate_numeric_facts(data)
393 issues.extend(root_issues)
394 method_issues, method_count, result_count = validate_methods_results(data)
395 issues.extend(method_issues)
396 return emit_report(
397 TOOL,
398 issues,
399 summary={
400 "numeric_facts": fact_count,
401 "methods": method_count,
402 "results": result_count,
403 },
404 )
407if __name__ == "__main__":
408 run(TOOL, cli)