Skills › Scientific Agent Skills › treatment-plans › timeline_generator.py
timeline_generator.py
277 lines · skills/treatment-plans/scripts/timeline_generator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Create a schedule from explicitly supplied dates only."""
4from __future__ import annotations
6import argparse
7import sys
8from datetime import date
10from check_consistency import check_consistency
12from _common import (
13 NOTICE,
14 SCHEMA_VERSION,
15 Issue,
16 ValidationError,
17 atomic_write_json,
18 error_report,
19 load_package,
20 parse_iso_datetime,
21 print_report,
22 report_payload,
23 safe_output_file,
24 validate_package_structure,
25)
28def _date_bound(value: str | None, code: str) -> date | None:
29 if value is None:
30 return None
31 try:
32 parsed = date.fromisoformat(value)
33 if parsed.isoformat() != value:
34 raise ValueError
35 return parsed
36 except ValueError as exc:
37 raise ValidationError(code) from exc
40def build_schedule(
41 documents: dict[str, dict],
42 *,
43 from_date: str | None = None,
44 through_date: str | None = None,
45) -> tuple[dict, dict]:
46 issues = validate_package_structure(documents)
47 if issues:
48 return (
49 report_payload(
50 "explicit_date_schedule",
51 issues,
52 counts={"structural_issues": len(issues)},
53 ),
54 {},
55 )
57 consistency = check_consistency(documents)
58 if consistency["status"] == "fail":
59 return (
60 report_payload(
61 "explicit_date_schedule",
62 [Issue(item["code"], item["path"], item["level"])
63 for item in consistency["issues"]],
64 counts={"consistency_issues": len(consistency["issues"])},
65 ),
66 {},
67 )
69 lower = _date_bound(from_date, "FROM_DATE_INVALID")
70 upper = _date_bound(through_date, "THROUGH_DATE_INVALID")
71 if lower and upper and lower > upper:
72 raise ValidationError("DATE_FILTER_ORDER_INVALID")
74 events: list[dict] = []
76 def add_event(
77 value: str | None,
78 *,
79 event_type: str,
80 document_type: str,
81 record_index: int | None,
82 date_field: str,
83 ) -> None:
84 if value is None:
85 return
86 event_date = value[:10]
87 parsed = date.fromisoformat(event_date)
88 if lower and parsed < lower:
89 return
90 if upper and parsed > upper:
91 return
92 events.append(
93 {
94 "date": event_date,
95 "event_type": event_type,
96 "source_document_type": document_type,
97 "record_index": record_index,
98 "date_field": date_field,
99 }
100 )
102 interventions = documents[
103 "clinician_authored_intervention_record"
104 ]["interventions"]
105 for index, record in enumerate(interventions):
106 add_event(
107 record["start_date"],
108 event_type="intervention_start_as_supplied",
109 document_type="clinician_authored_intervention_record",
110 record_index=index,
111 date_field="start_date",
112 )
113 add_event(
114 record["end_date"],
115 event_type="intervention_end_as_supplied",
116 document_type="clinician_authored_intervention_record",
117 record_index=index,
118 date_field="end_date",
119 )
121 planning = documents["goals_monitoring_checkpoint_record"]
122 for index, record in enumerate(planning["goals"]):
123 add_event(
124 record["target_date"],
125 event_type="goal_target_date_as_supplied",
126 document_type="goals_monitoring_checkpoint_record",
127 record_index=index,
128 date_field="goals.target_date",
129 )
130 if record["target_date"] is None:
131 issues.append(
132 Issue(
133 "GOAL_HAS_NO_EXPLICIT_TARGET_DATE",
134 (
135 "goals_monitoring_checkpoint_record:"
136 f"$.goals[{index}].target_date"
137 ),
138 "warning",
139 )
140 )
141 for index, record in enumerate(planning["monitoring_items"]):
142 add_event(
143 record["next_due_date"],
144 event_type="monitoring_due_date_as_supplied",
145 document_type="goals_monitoring_checkpoint_record",
146 record_index=index,
147 date_field="monitoring_items.next_due_date",
148 )
149 if record["frequency_as_supplied"] and record["next_due_date"] is None:
150 issues.append(
151 Issue(
152 "FREQUENCY_TEXT_NOT_EXPANDED_WITHOUT_EXPLICIT_DATE",
153 (
154 "goals_monitoring_checkpoint_record:"
155 f"$.monitoring_items[{index}].next_due_date"
156 ),
157 "warning",
158 )
159 )
160 for index, record in enumerate(planning["checkpoints"]):
161 add_event(
162 record["checkpoint_date"],
163 event_type="checkpoint_date_as_supplied",
164 document_type="goals_monitoring_checkpoint_record",
165 record_index=index,
166 date_field="checkpoints.checkpoint_date",
167 )
169 transition = documents["transition_reconciliation_record"][
170 "transition"
171 ]
172 add_event(
173 transition["handoff_date"],
174 event_type="transition_handoff_date_as_supplied",
175 document_type="transition_reconciliation_record",
176 record_index=None,
177 date_field="transition.handoff_date",
178 )
180 intended = documents["intended_use_handoff_record"]
181 for field, event_type in (
182 ("sent_at", "documentation_handoff_sent_as_supplied"),
183 ):
184 timestamp = intended["handoff"][field]
185 if timestamp is not None and parse_iso_datetime(timestamp) is not None:
186 add_event(
187 timestamp,
188 event_type=event_type,
189 document_type="intended_use_handoff_record",
190 record_index=None,
191 date_field=f"handoff.{field}",
192 )
194 events.sort(
195 key=lambda item: (
196 item["date"],
197 item["event_type"],
198 item["source_document_type"],
199 -1 if item["record_index"] is None else item["record_index"],
200 )
201 )
202 if not events:
203 issues.append(Issue("NO_EXPLICIT_DATES_IN_FILTER", "$"))
205 schedule = {
206 "schema_version": SCHEMA_VERSION,
207 "document_type": "explicit_date_schedule",
208 "status": "DRAFT_NOT_FOR_CLINICAL_USE",
209 "notice": NOTICE,
210 "dates_inferred": False,
211 "recurrences_generated": False,
212 "clinical_intervals_generated": False,
213 "filters": {
214 "from_date": from_date,
215 "through_date": through_date,
216 },
217 "events": events,
218 }
219 issues.sort(key=lambda item: (item.path, item.code))
220 report = report_payload(
221 "explicit_date_schedule",
222 issues,
223 counts={"events": len(events)},
224 extra={
225 "dates_inferred": False,
226 "recurrences_generated": False,
227 "clinical_content_interpreted": False,
228 },
229 )
230 return report, schedule
233def build_parser() -> argparse.ArgumentParser:
234 parser = argparse.ArgumentParser(
235 description=(
236 "Write a local JSON schedule containing only dates explicitly "
237 "supplied by authorized clinicians. Frequencies are never expanded."
238 )
239 )
240 parser.add_argument("package", help="Complete local package directory")
241 parser.add_argument(
242 "--output",
243 required=True,
244 help="New local .json output; existing files are rejected",
245 )
246 parser.add_argument(
247 "--from-date",
248 help="Optional inclusive YYYY-MM-DD filter; no default is inferred",
249 )
250 parser.add_argument(
251 "--through-date",
252 help="Optional inclusive YYYY-MM-DD filter; no default is inferred",
253 )
254 return parser
257def main() -> int:
258 args = build_parser().parse_args()
259 try:
260 documents, _ = load_package(args.package)
261 report, schedule = build_schedule(
262 documents,
263 from_date=args.from_date,
264 through_date=args.through_date,
265 )
266 if report["status"] == "pass":
267 output = safe_output_file(args.output)
268 atomic_write_json(output, schedule)
269 report["output_filename"] = output.name
270 except ValidationError as exc:
271 report = error_report("explicit_date_schedule", exc)
272 print_report(report)
273 return 1 if report["status"] == "fail" else 0
276if __name__ == "__main__":
277 sys.exit(main())