Skills › Scientific Agent Skills › clinical-reports › format_adverse_events.py
format_adverse_events.py
459 lines · skills/clinical-reports/scripts/format_adverse_events.py at 5f2bc03
1#!/usr/bin/env python3
2"""Format bounded aggregate adverse-event counts into a review-only Markdown table."""
4from __future__ import annotations
6import argparse
7import csv
8import re
9import sys
10from dataclasses import dataclass
11from pathlib import Path
13sys.dont_write_bytecode = True
15from _common import ( # noqa: E402
16 MAX_CSV_BYTES,
17 MAX_CSV_ROWS,
18 ValidationError,
19 load_json_object,
20 local_input_path,
21 local_output_path,
22 require_bool,
23 require_data_class,
24 require_exact_keys,
25 require_string,
26)
28TOOL = "format_adverse_events"
29REQUIRED_COLUMNS = (
30 "analysis_set",
31 "treatment_group",
32 "meddra_version",
33 "system_organ_class",
34 "preferred_term",
35 "subjects_affected",
36 "event_count",
37 "denominator",
38)
39FORBIDDEN_HEADER_FRAGMENTS = {
40 "patient",
41 "subject_id",
42 "participant",
43 "case_id",
44 "initial",
45 "name",
46 "email",
47 "phone",
48 "address",
49 "narrative",
50 "verbatim",
51 "onset",
52 "date",
53 "mrn",
54}
55VERSION_RE = re.compile(r"^\d{1,2}\.\d$")
56METADATA_FIELDS = {
57 "schema_version",
58 "artifact_kind",
59 "draft_status",
60 "safety_notice",
61 "data_classification",
62 "authorized_purpose",
63 "authorization_verified",
64 "local_only_handling_confirmed",
65 "analysis_metadata",
66 "input_csv",
67 "required_columns",
68 "prohibited_content",
69 "provenance_manifest",
70 "review",
71}
72ANALYSIS_METADATA_FIELDS = {
73 "protocol_reference",
74 "sap_reference",
75 "data_cut_reference",
76 "analysis_set",
77 "counting_rule",
78 "threshold_rule",
79 "meddra_version",
80 "meddra_language",
81 "coding_source_reference",
82}
85@dataclass(frozen=True)
86class AggregateRow:
87 analysis_set: str
88 treatment_group: str
89 meddra_version: str
90 system_organ_class: str
91 preferred_term: str
92 subjects_affected: int
93 event_count: int
94 denominator: int
97def validate_aggregate_metadata(
98 data: dict[str, object],
99 *,
100 input_name: str,
101) -> dict[str, object]:
102 """Require an authorized aggregate-only sidecar manifest."""
103 require_exact_keys(data, METADATA_FIELDS, "metadata")
104 if data.get("schema_version") != "2.0":
105 raise ValidationError("metadata schema_version must be 2.0")
106 if data.get("artifact_kind") != "clinical_trial_safety_aggregate_draft":
107 raise ValidationError(
108 "metadata artifact_kind must be clinical_trial_safety_aggregate_draft"
109 )
110 if (
111 data.get("draft_status")
112 != "BLOCKED_INCOMPLETE_NOT_AN_INDIVIDUAL_CASE_SAFETY_REPORT"
113 ):
114 raise ValidationError("metadata must preserve the blocked non-ICSR status")
115 require_string(data.get("safety_notice"), "metadata.safety_notice", max_length=1000)
116 if require_data_class(data.get("data_classification")) != "aggregate":
117 raise ValidationError("AE formatter accepts aggregate data only")
118 require_string(data.get("authorized_purpose"), "metadata.authorized_purpose")
119 if not require_bool(
120 data.get("authorization_verified"),
121 "metadata.authorization_verified",
122 ):
123 raise ValidationError("metadata.authorization_verified must be true")
124 if not require_bool(
125 data.get("local_only_handling_confirmed"),
126 "metadata.local_only_handling_confirmed",
127 ):
128 raise ValidationError("metadata.local_only_handling_confirmed must be true")
129 require_string(data.get("provenance_manifest"), "metadata.provenance_manifest")
130 if data.get("input_csv") != input_name:
131 raise ValidationError("metadata.input_csv must equal the aggregate CSV filename")
132 if data.get("required_columns") != list(REQUIRED_COLUMNS):
133 raise ValidationError("metadata.required_columns does not match the formatter schema")
134 prohibited = data.get("prohibited_content")
135 if not isinstance(prohibited, list) or not prohibited:
136 raise ValidationError("metadata.prohibited_content must be non-empty")
138 analysis = require_exact_keys(
139 data.get("analysis_metadata"),
140 ANALYSIS_METADATA_FIELDS,
141 "metadata.analysis_metadata",
142 )
143 for field in ANALYSIS_METADATA_FIELDS:
144 require_string(
145 analysis.get(field),
146 f"metadata.analysis_metadata.{field}",
147 max_length=500,
148 )
149 if not VERSION_RE.fullmatch(str(analysis["meddra_version"])):
150 raise ValidationError("metadata MedDRA version must look like 29.0")
152 review = require_exact_keys(
153 data.get("review"),
154 {
155 "safety_coding_review",
156 "statistical_review",
157 "privacy_review",
158 "regulatory_review",
159 "submission_authorized",
160 },
161 "metadata.review",
162 )
163 if review.get("submission_authorized") is not False:
164 raise ValidationError("metadata.review.submission_authorized must remain false")
165 return data
168def load_aggregate_metadata(
169 raw_path: str,
170 *,
171 input_name: str,
172) -> dict[str, object]:
173 """Load a bounded aggregate metadata sidecar."""
174 _, data = load_json_object(raw_path)
175 return validate_aggregate_metadata(data, input_name=input_name)
178def _label(value: str | None, field: str) -> str:
179 if value is None:
180 raise ValidationError(f"{field} is required")
181 text = value.strip()
182 if (
183 not 1 <= len(text) <= 200
184 or not text[0].isalnum()
185 or any(char in "|`<>\r\n" or ord(char) < 32 for char in text)
186 ):
187 raise ValidationError(
188 f"{field} must be a bounded label without markup or line breaks"
189 )
190 return text
193def _count(value: str | None, field: str) -> int:
194 if value is None or not re.fullmatch(r"\d{1,9}", value.strip()):
195 raise ValidationError(f"{field} must be a non-negative integer")
196 return int(value)
199def load_aggregate_csv(raw_path: str) -> list[AggregateRow]:
200 """Load and validate aggregate-only AE rows."""
201 path = local_input_path(
202 raw_path,
203 suffixes={".csv"},
204 max_bytes=MAX_CSV_BYTES,
205 )
206 try:
207 path.read_text(encoding="utf-8-sig")
208 except UnicodeDecodeError as exc:
209 raise ValidationError("CSV input must be UTF-8") from exc
210 handle = path.open("r", encoding="utf-8-sig", newline="")
212 rows: list[AggregateRow] = []
213 with handle:
214 reader = csv.DictReader(handle)
215 if reader.fieldnames is None:
216 raise ValidationError("CSV header is required")
217 normalized_headers = [header.strip() for header in reader.fieldnames]
218 if len(normalized_headers) != len(set(normalized_headers)):
219 raise ValidationError("duplicate CSV columns are not allowed")
220 if normalized_headers != reader.fieldnames:
221 raise ValidationError("CSV column names must not contain surrounding whitespace")
222 lowered = {header.lower() for header in normalized_headers}
223 for header in lowered:
224 if any(fragment in header for fragment in FORBIDDEN_HEADER_FRAGMENTS):
225 raise ValidationError(f"row-level or sensitive column is prohibited: {header}")
226 if set(normalized_headers) != set(REQUIRED_COLUMNS):
227 raise ValidationError(
228 f"CSV columns must be exactly: {list(REQUIRED_COLUMNS)}"
229 )
231 for line_number, raw in enumerate(reader, start=2):
232 if len(rows) >= MAX_CSV_ROWS:
233 raise ValidationError(f"CSV exceeds {MAX_CSV_ROWS} data rows")
234 if None in raw:
235 raise ValidationError(f"row {line_number} contains extra columns")
236 row = AggregateRow(
237 analysis_set=_label(raw.get("analysis_set"), f"row {line_number} analysis_set"),
238 treatment_group=_label(
239 raw.get("treatment_group"),
240 f"row {line_number} treatment_group",
241 ),
242 meddra_version=_label(
243 raw.get("meddra_version"),
244 f"row {line_number} meddra_version",
245 ),
246 system_organ_class=_label(
247 raw.get("system_organ_class"),
248 f"row {line_number} system_organ_class",
249 ),
250 preferred_term=_label(
251 raw.get("preferred_term"),
252 f"row {line_number} preferred_term",
253 ),
254 subjects_affected=_count(
255 raw.get("subjects_affected"),
256 f"row {line_number} subjects_affected",
257 ),
258 event_count=_count(
259 raw.get("event_count"),
260 f"row {line_number} event_count",
261 ),
262 denominator=_count(
263 raw.get("denominator"),
264 f"row {line_number} denominator",
265 ),
266 )
267 if not VERSION_RE.fullmatch(row.meddra_version):
268 raise ValidationError(
269 f"row {line_number} MedDRA version must look like 29.0"
270 )
271 if row.denominator == 0:
272 raise ValidationError(f"row {line_number} denominator must be greater than zero")
273 if row.subjects_affected > row.denominator:
274 raise ValidationError(
275 f"row {line_number} subjects_affected exceeds denominator"
276 )
277 if row.event_count < row.subjects_affected:
278 raise ValidationError(
279 f"row {line_number} event_count is less than subjects_affected"
280 )
281 rows.append(row)
283 if not rows:
284 raise ValidationError("CSV must contain at least one aggregate data row")
286 group_metadata: dict[tuple[str, str], tuple[int, str]] = {}
287 row_keys: set[tuple[str, str, str, str]] = set()
288 for row in rows:
289 group = (row.analysis_set, row.treatment_group)
290 metadata = (row.denominator, row.meddra_version)
291 if group in group_metadata and group_metadata[group] != metadata:
292 raise ValidationError(
293 f"inconsistent denominator or MedDRA version for group {group}"
294 )
295 group_metadata[group] = metadata
296 row_key = (
297 row.analysis_set,
298 row.treatment_group,
299 row.system_organ_class,
300 row.preferred_term,
301 )
302 if row_key in row_keys:
303 raise ValidationError(f"duplicate aggregate row: {row_key}")
304 row_keys.add(row_key)
305 return rows
308def render_markdown(
309 rows: list[AggregateRow],
310 *,
311 metadata: dict[str, object],
312 expected_meddra_version: str | None = None,
313) -> str:
314 """Render rows without inferring missing cells or statistical comparisons."""
315 versions = sorted({row.meddra_version for row in rows})
316 if len(versions) != 1:
317 raise ValidationError("one table may contain exactly one MedDRA version")
318 analysis_metadata = metadata["analysis_metadata"]
319 if not isinstance(analysis_metadata, dict):
320 raise ValidationError("metadata.analysis_metadata must be an object")
321 metadata_version = str(analysis_metadata["meddra_version"])
322 if versions != [metadata_version]:
323 raise ValidationError(
324 "CSV MedDRA version does not match metadata.analysis_metadata.meddra_version"
325 )
326 if expected_meddra_version is not None:
327 if not VERSION_RE.fullmatch(expected_meddra_version):
328 raise ValidationError("--expected-meddra-version must look like 29.0")
329 if versions != [expected_meddra_version]:
330 raise ValidationError(
331 f"input MedDRA versions {versions} do not match expected "
332 f"{expected_meddra_version}"
333 )
335 analysis_sets = sorted({row.analysis_set for row in rows})
336 if len(analysis_sets) != 1:
337 raise ValidationError(
338 "one output table may contain exactly one analysis_set"
339 )
340 if analysis_sets[0] != analysis_metadata["analysis_set"]:
341 raise ValidationError(
342 "CSV analysis_set does not match metadata.analysis_metadata.analysis_set"
343 )
344 groups = sorted({row.treatment_group for row in rows})
345 denominators = {
346 group: next(row.denominator for row in rows if row.treatment_group == group)
347 for group in groups
348 }
349 lookup = {
350 (row.system_organ_class, row.preferred_term, row.treatment_group): row
351 for row in rows
352 }
353 terms = sorted({(row.system_organ_class, row.preferred_term) for row in rows})
355 headers = ["System organ class", "Preferred term"]
356 for group in groups:
357 headers.extend(
358 [
359 f"{group}: subjects n/N (%)",
360 f"{group}: events",
361 ]
362 )
363 lines = [
364 "# Draft aggregate adverse-event table",
365 "",
366 "> DRAFT — aggregate display only. Not an ICSR, reportability decision, "
367 "clinical interpretation, filing, or submission. Qualified safety and "
368 "statistical review is required.",
369 "",
370 f"- Analysis set: {analysis_sets[0]}",
371 f"- MedDRA version(s), as supplied and not terminology-validated: {', '.join(versions)}",
372 f"- MedDRA language, as supplied: {analysis_metadata['meddra_language']}",
373 f"- Counting rule, as supplied: {analysis_metadata['counting_rule']}",
374 f"- Threshold rule, as supplied: {analysis_metadata['threshold_rule']}",
375 "- Missing group/term cells are shown as an em dash and are not assumed to be zero.",
376 "- Subjects and events are distinct; event counts may exceed subject counts.",
377 "",
378 "| " + " | ".join(headers) + " |",
379 "| " + " | ".join(["---"] * len(headers)) + " |",
380 ]
381 for soc, term in terms:
382 cells = [soc, term]
383 for group in groups:
384 row = lookup.get((soc, term, group))
385 if row is None:
386 cells.extend(["—", "—"])
387 continue
388 percent = row.subjects_affected / row.denominator * 100
389 cells.extend(
390 [
391 f"{row.subjects_affected}/{row.denominator} ({percent:.1f}%)",
392 str(row.event_count),
393 ]
394 )
395 lines.append("| " + " | ".join(cells) + " |")
396 lines.extend(
397 [
398 "",
399 "Denominators by treatment group: "
400 + "; ".join(f"{group} N={denominators[group]}" for group in groups)
401 + ".",
402 "",
403 "No MedDRA semantic validation, deduplication, inference, causal assessment, "
404 "or between-group testing was performed.",
405 ]
406 )
407 return "\n".join(lines) + "\n"
410def build_parser() -> argparse.ArgumentParser:
411 parser = argparse.ArgumentParser(
412 description=(
413 "Format aggregate-only AE CSV rows as review-only Markdown. "
414 "Rejects row-level identifiers and performs no safety assessment."
415 )
416 )
417 parser.add_argument("input_file", help="Local aggregate CSV")
418 parser.add_argument(
419 "--metadata",
420 required=True,
421 help="Authorized aggregate safety metadata sidecar (.json)",
422 )
423 parser.add_argument("-o", "--output", help="Optional local Markdown output")
424 parser.add_argument("--expected-meddra-version")
425 parser.add_argument("--overwrite", action="store_true")
426 return parser
429def main(argv: list[str] | None = None) -> int:
430 args = build_parser().parse_args(argv)
431 try:
432 rows = load_aggregate_csv(args.input_file)
433 input_name = Path(args.input_file).name
434 metadata = load_aggregate_metadata(
435 args.metadata,
436 input_name=input_name,
437 )
438 rendered = render_markdown(
439 rows,
440 metadata=metadata,
441 expected_meddra_version=args.expected_meddra_version,
442 )
443 if args.output:
444 output = local_output_path(
445 args.output,
446 suffixes={".md"},
447 overwrite=args.overwrite,
448 )
449 output.write_text(rendered, encoding="utf-8")
450 else:
451 print(rendered, end="")
452 return 0
453 except (OSError, ValidationError, csv.Error) as exc:
454 print(f"{TOOL}: BLOCKED_INVALID_INPUT: {exc}", file=sys.stderr)
455 return 1
458if __name__ == "__main__":
459 raise SystemExit(main())