Skills › Scientific Agent Skills › clinical-decision-support › survival_plan_validator.py
survival_plan_validator.py
294 lines · skills/clinical-decision-support/scripts/survival_plan_validator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate an estimand-led survival-analysis plan without reading subject rows."""
4from __future__ import annotations
6import argparse
7import sys
8from typing import Any
10from _common import (
11 InputError,
12 IssueLog,
13 load_json_object,
14 print_report,
15 require_list,
16 require_nonempty_text,
17 source_ids,
18 validate_references,
19 write_json,
20)
22ALLOWED_METHODS = {
23 "kaplan_meier",
24 "cox_proportional_hazards",
25 "flexible_parametric",
26 "restricted_mean_survival_time",
27 "accelerated_failure_time",
28 "cumulative_incidence",
29 "cause_specific_hazard",
30 "fine_gray",
31 "multi_state",
32 "other_justified",
33}
34COMPETING_METHODS = {
35 "cumulative_incidence",
36 "cause_specific_hazard",
37 "fine_gray",
38 "multi_state",
39 "other_justified",
40}
43def validate_plan(document: dict[str, Any]) -> IssueLog:
44 log = IssueLog()
45 try:
46 require_nonempty_text(document.get("schema_version"), "schema_version")
47 known_sources = source_ids(document)
49 metadata = document.get("metadata")
50 if not isinstance(metadata, dict):
51 raise InputError("metadata must be an object")
52 for field in ("plan_id", "title", "version", "status", "purpose", "data_cut_date"):
53 require_nonempty_text(metadata.get(field), f"metadata.{field}")
54 if not isinstance(metadata.get("data_level"), str) or metadata.get("data_level") not in {"aggregate", "synthetic", "plan_only"}:
55 log.errors.append("metadata.data_level must be aggregate, synthetic, or plan_only")
56 if metadata.get("raw_rows_supplied") is not False:
57 log.errors.append("metadata.raw_rows_supplied must be false")
58 if metadata.get("patient_care_use") is not False:
59 log.errors.append("metadata.patient_care_use must be false")
60 if metadata.get("human_review_required") is not True:
61 log.errors.append("metadata.human_review_required must be true")
63 estimand = document.get("estimand")
64 if not isinstance(estimand, dict):
65 raise InputError("estimand must be an object")
66 for field in (
67 "population",
68 "condition_or_comparison",
69 "endpoint_variable",
70 "population_summary",
71 "time_horizon",
72 "rationale",
73 ):
74 require_nonempty_text(estimand.get(field), f"estimand.{field}")
75 intercurrent = require_list(
76 estimand.get("intercurrent_events"),
77 "estimand.intercurrent_events",
78 maximum=30,
79 )
80 if not intercurrent:
81 log.errors.append(
82 "estimand.intercurrent_events must document events or explicitly state none"
83 )
84 for index, item in enumerate(intercurrent):
85 if not isinstance(item, dict):
86 raise InputError(f"estimand.intercurrent_events[{index}] must be an object")
87 require_nonempty_text(
88 item.get("event"), f"estimand.intercurrent_events[{index}].event"
89 )
90 require_nonempty_text(
91 item.get("strategy"), f"estimand.intercurrent_events[{index}].strategy"
92 )
93 require_nonempty_text(
94 item.get("rationale"), f"estimand.intercurrent_events[{index}].rationale"
95 )
97 endpoint = document.get("endpoint")
98 if not isinstance(endpoint, dict):
99 raise InputError("endpoint must be an object")
100 for field in (
101 "time_zero",
102 "event_definition",
103 "time_scale",
104 "ascertainment",
105 "follow_up_end",
106 "delayed_entry",
107 "same_time_rules",
108 ):
109 require_nonempty_text(endpoint.get(field), f"endpoint.{field}")
110 censoring = require_list(
111 endpoint.get("censoring_rules"), "endpoint.censoring_rules", maximum=30
112 )
113 if not censoring:
114 log.errors.append("At least one censoring rule is required")
115 for index, rule in enumerate(censoring):
116 require_nonempty_text(rule, f"endpoint.censoring_rules[{index}]")
117 competing = require_list(
118 endpoint.get("competing_events"), "endpoint.competing_events", maximum=20
119 )
120 for index, event in enumerate(competing):
121 require_nonempty_text(event, f"endpoint.competing_events[{index}]")
123 analysis = document.get("analysis")
124 if not isinstance(analysis, dict):
125 raise InputError("analysis must be an object")
126 primary_method = require_nonempty_text(
127 analysis.get("primary_method"), "analysis.primary_method"
128 )
129 if primary_method not in ALLOWED_METHODS:
130 log.errors.append(f"Unsupported analysis.primary_method: {primary_method}")
131 for field in (
132 "effect_measure",
133 "analysis_population",
134 "covariate_strategy",
135 "missing_data",
136 "multiplicity",
137 "uncertainty",
138 "software_and_version",
139 "model_diagnostics",
140 ):
141 require_nonempty_text(analysis.get(field), f"analysis.{field}")
143 ph_assessment = require_nonempty_text(
144 analysis.get("proportional_hazards_assessment"),
145 "analysis.proportional_hazards_assessment",
146 )
147 non_ph_strategy = require_nonempty_text(
148 analysis.get("non_proportional_hazards_strategy"),
149 "analysis.non_proportional_hazards_strategy",
150 )
151 effect_measure = str(analysis.get("effect_measure", "")).lower()
152 if (
153 primary_method == "cox_proportional_hazards"
154 or "hazard ratio" in effect_measure
155 ):
156 if ph_assessment.lower() in {"none", "not applicable", "n/a"}:
157 log.errors.append(
158 "A proportional-hazards assessment is required for a hazard-ratio plan"
159 )
160 if non_ph_strategy.lower() in {"none", "not applicable", "n/a"}:
161 log.errors.append(
162 "Pre-specify an alternative if proportional hazards is not supported"
163 )
165 competing_method = require_nonempty_text(
166 analysis.get("competing_risk_method"),
167 "analysis.competing_risk_method",
168 )
169 if competing:
170 if competing_method not in COMPETING_METHODS:
171 log.errors.append(
172 "A competing-risk method is required when competing events are listed"
173 )
174 if (
175 primary_method == "kaplan_meier"
176 and "absolute" in effect_measure
177 ):
178 log.errors.append(
179 "Kaplan-Meier with competing events cannot be the sole absolute-incidence method"
180 )
181 elif competing_method not in {"not_applicable", "none"}:
182 log.warnings.append(
183 "A competing-risk method is named but no competing events are listed"
184 )
186 bias = document.get("bias_controls")
187 if not isinstance(bias, dict):
188 raise InputError("bias_controls must be an object")
189 for field in (
190 "informative_censoring",
191 "immortal_time",
192 "time_dependent_confounding",
193 "outcome_misclassification",
194 "informative_visits",
195 ):
196 require_nonempty_text(bias.get(field), f"bias_controls.{field}")
198 sensitivity = require_list(
199 document.get("sensitivity_analyses"),
200 "sensitivity_analyses",
201 maximum=30,
202 )
203 if len(sensitivity) < 2:
204 log.warnings.append("Fewer than two sensitivity analyses are specified")
205 for index, item in enumerate(sensitivity):
206 require_nonempty_text(item, f"sensitivity_analyses[{index}]")
208 subgroups = document.get("subgroups")
209 if not isinstance(subgroups, dict):
210 raise InputError("subgroups must be an object")
211 for field in (
212 "prespecification",
213 "interaction_testing",
214 "multiplicity",
215 "precision_and_disclosure",
216 ):
217 require_nonempty_text(subgroups.get(field), f"subgroups.{field}")
219 validation = document.get("validation")
220 if not isinstance(validation, dict):
221 raise InputError("validation must be an object")
222 for field in (
223 "internal_validation",
224 "external_validation",
225 "calibration",
226 "subgroup_performance",
227 ):
228 require_nonempty_text(validation.get(field), f"validation.{field}")
229 if "not performed" in str(validation.get("external_validation")).lower():
230 log.warnings.append("External validation is documented as not performed")
232 review = document.get("human_review")
233 if not isinstance(review, dict):
234 raise InputError("human_review must be an object")
235 roles = require_list(review.get("roles"), "human_review.roles", maximum=20)
236 if not roles:
237 log.errors.append("At least one human-review role is required")
238 for index, role in enumerate(roles):
239 require_nonempty_text(role, f"human_review.roles[{index}]")
240 require_nonempty_text(review.get("approval_boundary"), "human_review.approval_boundary")
241 if review.get("completed") is not True:
242 log.warnings.append("Human review is not recorded as complete")
244 governance = document.get("governance")
245 if not isinstance(governance, dict):
246 raise InputError("governance must be an object")
247 for field in (
248 "owner",
249 "change_control",
250 "auditability",
251 "monitoring",
252 "retirement",
253 ):
254 require_nonempty_text(governance.get(field), f"governance.{field}")
256 validate_references(document.get("source_ids"), known_sources, "source_ids")
257 except InputError as exc:
258 log.errors.append(str(exc))
260 if log.ok:
261 log.info.append("Survival plan contains the required estimand and analysis fields")
262 return log
265def main() -> int:
266 parser = argparse.ArgumentParser(
267 description=(
268 "Validate a local survival-analysis plan. No subject-level data are read "
269 "and no survival model is fitted."
270 )
271 )
272 parser.add_argument("input", help="Local plan JSON")
273 parser.add_argument("-o", "--output", help="Optional local JSON report")
274 parser.add_argument(
275 "--strict", action="store_true", help="Return failure when warnings are present"
276 )
277 args = parser.parse_args()
279 try:
280 document = load_json_object(args.input)
281 report = validate_plan(document).as_dict()
282 if args.output:
283 write_json(args.output, report)
284 print_report(report)
285 except InputError as exc:
286 print_report(IssueLog(errors=[str(exc)]).as_dict())
287 return 2
288 if report["status"] != "pass" or (args.strict and report["warnings"]):
289 return 1
290 return 0
293if __name__ == "__main__":
294 sys.exit(main())