Skills › Scientific Agent Skills › hypothesis-generation › check_operationalization.py
check_operationalization.py
237 lines · skills/hypothesis-generation/scripts/check_operationalization.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit a local operationalization and measurement checklist without scoring."""
4from __future__ import annotations
6import argparse
7from collections import Counter
8from typing import Any
10from _common import (
11 ValidationError,
12 error_exit,
13 issue,
14 read_json,
15 require_bool,
16 require_enum,
17 require_exact_keys,
18 require_identifier,
19 require_identifier_list,
20 require_list,
21 require_object,
22 require_text,
23 require_unique,
24 write_json_report,
25)
27ROLES = {
28 "intervention",
29 "exposure",
30 "outcome",
31 "mediator",
32 "confounder",
33 "selection",
34 "effect_modifier",
35 "negative_control",
36 "positive_control",
37 "other",
38}
39PLANNING_STATUSES = {"planned", "complete", "not_applicable", "unresolved"}
40REVIEW_STATUSES = {"pending", "complete", "specialist_required"}
41BOOLEAN_FIELDS = (
42 "construct_defined",
43 "operational_definition_recorded",
44 "population_scope_recorded",
45 "unit_or_categories_recorded",
46 "timing_recorded",
47 "instrument_or_method_recorded",
48 "validity_applicability_reviewed",
49 "reliability_or_repeatability_plan_recorded",
50 "missingness_plan_recorded",
51 "limitations_recorded",
52)
53STATUS_FIELDS = (
54 "calibration_or_quality_control_status",
55 "measurement_invariance_or_comparability_status",
56 "masking_status",
57 "threshold_or_cutpoint_status",
58)
61def load_checklist(payload: Any) -> dict[str, Any]:
62 root = require_object(payload, "checklist")
63 require_exact_keys(
64 root,
65 required={
66 "schema_version",
67 "checklist_id",
68 "record_id",
69 "human_reviewer",
70 "items",
71 },
72 context="checklist",
73 )
74 raw_items = require_list(root["items"], "checklist.items", minimum=1, maximum=200)
75 parsed_items: list[dict[str, Any]] = []
76 identifiers: list[str] = []
77 item_fields = {
78 "measurement_id",
79 "applicability",
80 "variable_role",
81 "validity_evidence_source_ids",
82 "human_review_status",
83 "note",
84 *BOOLEAN_FIELDS,
85 *STATUS_FIELDS,
86 }
87 for index, raw_item in enumerate(raw_items):
88 context = f"checklist.items[{index}]"
89 item = require_object(raw_item, context)
90 require_exact_keys(item, required=item_fields, context=context)
91 measurement_id = require_identifier(
92 item["measurement_id"], f"{context}.measurement_id"
93 )
94 identifiers.append(measurement_id)
95 parsed: dict[str, Any] = {
96 "measurement_id": measurement_id,
97 "applicability": require_enum(
98 item["applicability"],
99 {"applicable", "not_applicable"},
100 f"{context}.applicability",
101 ),
102 "variable_role": require_enum(
103 item["variable_role"], ROLES, f"{context}.variable_role"
104 ),
105 "validity_evidence_source_ids": require_identifier_list(
106 item["validity_evidence_source_ids"],
107 f"{context}.validity_evidence_source_ids",
108 maximum=100,
109 ),
110 "human_review_status": require_enum(
111 item["human_review_status"],
112 REVIEW_STATUSES,
113 f"{context}.human_review_status",
114 ),
115 "note": require_text(
116 item["note"], f"{context}.note", allow_empty=True, maximum=2_000
117 ),
118 }
119 for field in BOOLEAN_FIELDS:
120 parsed[field] = require_bool(item[field], f"{context}.{field}")
121 for field in STATUS_FIELDS:
122 parsed[field] = require_enum(
123 item[field], PLANNING_STATUSES, f"{context}.{field}"
124 )
125 parsed_items.append(parsed)
126 require_unique(identifiers, "checklist.items")
127 return {
128 "schema_version": require_enum(
129 root["schema_version"], {"2.0"}, "checklist.schema_version"
130 ),
131 "checklist_id": require_identifier(
132 root["checklist_id"], "checklist.checklist_id"
133 ),
134 "record_id": require_identifier(root["record_id"], "checklist.record_id"),
135 "human_reviewer": require_text(
136 root["human_reviewer"], "checklist.human_reviewer", minimum=3
137 ),
138 "items": parsed_items,
139 }
142def audit(checklist: dict[str, Any]) -> dict[str, Any]:
143 errors: list[dict[str, str]] = []
144 warnings: list[dict[str, str]] = []
145 gaps_by_measurement: dict[str, list[str]] = {}
146 role_counts: Counter[str] = Counter()
147 review_counts: Counter[str] = Counter()
149 for item in checklist["items"]:
150 measurement_id = item["measurement_id"]
151 role_counts[item["variable_role"]] += 1
152 review_counts[item["human_review_status"]] += 1
153 gaps: list[str] = []
155 if item["applicability"] == "not_applicable":
156 if not item["note"]:
157 errors.append(issue("NOT_APPLICABLE_RATIONALE_REQUIRED", measurement_id))
158 continue
160 for field in BOOLEAN_FIELDS:
161 if not item[field]:
162 gaps.append(field)
163 if not item["validity_evidence_source_ids"]:
164 errors.append(issue("VALIDITY_SOURCE_REQUIRED", measurement_id))
165 for field in STATUS_FIELDS:
166 status = item[field]
167 if status in {"planned", "unresolved"}:
168 gaps.append(field)
169 if status == "not_applicable" and not item["note"]:
170 errors.append(
171 issue("NOT_APPLICABLE_STATUS_NEEDS_RATIONALE", f"{measurement_id}:{field}")
172 )
173 if item["human_review_status"] != "complete":
174 gaps.append("human_review_status")
175 if not item["note"]:
176 warnings.append(issue("MEASUREMENT_NOTE_EMPTY", measurement_id))
177 if gaps:
178 gaps_by_measurement[measurement_id] = sorted(set(gaps))
180 if not role_counts["outcome"]:
181 warnings.append(issue("NO_OUTCOME_MEASUREMENT_DECLARED", "items"))
182 if not (role_counts["intervention"] or role_counts["exposure"]):
183 warnings.append(issue("NO_INTERVENTION_OR_EXPOSURE_DECLARED", "items"))
185 return {
186 "schema_version": "2.0",
187 "checklist_id": checklist["checklist_id"],
188 "record_id": checklist["record_id"],
189 "valid": not errors,
190 "status": (
191 "INVALID_CHECKLIST"
192 if errors
193 else "VALID_WITH_MEASUREMENT_GAPS"
194 if gaps_by_measurement
195 else "VALID_HUMAN_REVIEW_COMPLETE"
196 ),
197 "errors": errors,
198 "warnings": warnings,
199 "measurement_count": len(checklist["items"]),
200 "role_counts": dict(sorted(role_counts.items())),
201 "human_review_status_counts": dict(sorted(review_counts.items())),
202 "gap_fields_by_measurement_id": gaps_by_measurement,
203 "notice": (
204 "This checklist records declared measurement work. It does not test "
205 "an instrument, establish construct validity, certify comparability, "
206 "or score measurement quality. Qualified human review is required."
207 ),
208 }
211def build_parser() -> argparse.ArgumentParser:
212 parser = argparse.ArgumentParser(
213 description=(
214 "Audit a bounded local JSON operationalization checklist and report "
215 "measurement IDs and unresolved fields without scientific scoring."
216 )
217 )
218 parser.add_argument("checklist", help="Local operationalization checklist JSON")
219 parser.add_argument("-o", "--output", help="Optional local JSON report")
220 parser.add_argument(
221 "--force", action="store_true", help="Replace an existing output file"
222 )
223 return parser
226def main() -> int:
227 args = build_parser().parse_args()
228 try:
229 report = audit(load_checklist(read_json(args.checklist)))
230 write_json_report(report, args.output, force=args.force)
231 return 0 if report["valid"] else 1
232 except ValidationError as exc:
233 return error_exit(exc)
236if __name__ == "__main__":
237 raise SystemExit(main())