Skills › Scientific Agent Skills › clinical-decision-support › cohort_table_generator.py
cohort_table_generator.py
295 lines · skills/clinical-decision-support/scripts/cohort_table_generator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Generate disclosure-controlled tables from bounded aggregate cohort summaries."""
4from __future__ import annotations
6import argparse
7import csv
8import io
9import sys
10from typing import Any
12from _common import (
13 InputError,
14 IssueLog,
15 load_json_object,
16 nonnegative_int,
17 require_list,
18 require_nonempty_text,
19 write_text,
20)
22MAX_GROUPS = 12
23MAX_ROWS = 200
26def _format_count(count: int, denominator: int) -> str:
27 percentage = 100.0 * (count / denominator) if denominator else 0.0
28 return f"{count}/{denominator} ({percentage:.1f}%)"
31def _build_table(
32 document: dict[str, Any], minimum: int
33) -> tuple[IssueLog, list[str], list[list[str]], list[str]]:
34 log = IssueLog()
35 notes: list[str] = []
36 headers: list[str] = ["Characteristic"]
37 output_rows: list[list[str]] = []
39 metadata = document.get("metadata")
40 if not isinstance(metadata, dict):
41 raise InputError("metadata must be an object")
42 for field in ("table_id", "title", "purpose", "population", "data_cut_date"):
43 require_nonempty_text(metadata.get(field), f"metadata.{field}")
44 if not isinstance(metadata.get("data_level"), str) or metadata.get("data_level") not in {"aggregate", "synthetic"}:
45 log.errors.append("metadata.data_level must be aggregate or synthetic")
46 if metadata.get("raw_rows_supplied") is not False:
47 log.errors.append("metadata.raw_rows_supplied must be false")
48 if metadata.get("patient_care_use") is not False:
49 log.errors.append("metadata.patient_care_use must be false")
50 require_nonempty_text(
51 metadata.get("disclosure_policy"), "metadata.disclosure_policy"
52 )
53 require_nonempty_text(
54 metadata.get("human_review"), "metadata.human_review"
55 )
57 review = document.get("human_review")
58 if not isinstance(review, dict):
59 raise InputError("human_review must be an object")
60 if review.get("required") is not True:
61 log.errors.append("human_review.required must be true")
62 roles = require_list(review.get("roles"), "human_review.roles", maximum=20)
63 if not roles:
64 log.errors.append("At least one human-review role is required")
65 for index, role in enumerate(roles):
66 require_nonempty_text(role, f"human_review.roles[{index}]")
67 if review.get("completed") is not True:
68 log.warnings.append("Human review is not recorded as complete")
70 governance = document.get("governance")
71 if not isinstance(governance, dict):
72 raise InputError("governance must be an object")
73 for field in (
74 "version",
75 "owner",
76 "change_summary",
77 "auditability",
78 "monitoring",
79 ):
80 require_nonempty_text(governance.get(field), f"governance.{field}")
82 if not log.ok:
83 return log, headers, output_rows, notes
85 groups = require_list(document.get("groups"), "groups", maximum=MAX_GROUPS)
86 if not groups:
87 raise InputError("At least one group is required")
88 group_map: dict[str, dict[str, Any]] = {}
89 for index, group in enumerate(groups):
90 field = f"groups[{index}]"
91 if not isinstance(group, dict):
92 raise InputError(f"{field} must be an object")
93 identifier = require_nonempty_text(
94 group.get("id"), f"{field}.id", max_length=50
95 )
96 label = require_nonempty_text(
97 group.get("label"), f"{field}.label", max_length=120
98 )
99 n = nonnegative_int(group.get("n"), f"{field}.n")
100 if n == 0:
101 raise InputError(f"{field}.n must be positive")
102 if identifier in group_map:
103 raise InputError(f"Duplicate group id: {identifier}")
104 group_map[identifier] = {"label": label, "n": n}
105 headers.append(f"{label} (n={n if n >= minimum else 'SUPP'})")
107 rows = require_list(document.get("rows"), "rows", maximum=MAX_ROWS)
108 if not rows:
109 raise InputError("At least one aggregate table row is required")
110 for row_index, row in enumerate(rows):
111 field = f"rows[{row_index}]"
112 if not isinstance(row, dict):
113 raise InputError(f"{field} must be an object")
114 if "p_value" in row or "statistical_test" in row:
115 raise InputError(
116 f"{field} contains inferential fields; this helper creates descriptive tables only"
117 )
118 label = require_nonempty_text(row.get("label"), f"{field}.label", max_length=200)
119 level = row.get("level")
120 if level is not None:
121 label = f" {require_nonempty_text(level, f'{field}.level', max_length=120)}"
122 row_type = require_nonempty_text(
123 row.get("type"), f"{field}.type", max_length=30
124 )
125 if row_type not in {"categorical", "continuous", "header"}:
126 raise InputError(f"{field}.type is unsupported")
127 values = row.get("values")
128 if row_type == "header":
129 output_rows.append([label] + [""] * len(group_map))
130 continue
131 if not isinstance(values, dict):
132 raise InputError(f"{field}.values must be an object")
133 if set(values) != set(group_map):
134 raise InputError(f"{field}.values must contain exactly all group ids")
136 cells: dict[str, str] = {}
137 primary_suppressed: set[str] = set()
138 eligible_counts: dict[str, int] = {}
139 for group_id, group in group_map.items():
140 value = values[group_id]
141 cell_field = f"{field}.values.{group_id}"
142 if not isinstance(value, dict):
143 raise InputError(f"{cell_field} must be an object")
144 group_n = group["n"]
145 if row_type == "categorical":
146 count = nonnegative_int(value.get("count"), f"{cell_field}.count")
147 denominator = nonnegative_int(
148 value.get("denominator"), f"{cell_field}.denominator"
149 )
150 missing = nonnegative_int(
151 value.get("missing", group_n - denominator),
152 f"{cell_field}.missing",
153 )
154 if denominator == 0 or count > denominator:
155 raise InputError(f"{cell_field} has inconsistent counts")
156 if denominator + missing != group_n:
157 raise InputError(
158 f"{cell_field} denominator plus missing must equal group n"
159 )
160 if (
161 group_n < minimum
162 or 0 < count < minimum
163 or 0 < denominator - count < minimum
164 or 0 < missing < minimum
165 ):
166 cells[group_id] = "SUPP"
167 primary_suppressed.add(group_id)
168 else:
169 cells[group_id] = _format_count(count, denominator)
170 eligible_counts[group_id] = count
171 else:
172 summarized_n = nonnegative_int(value.get("n"), f"{cell_field}.n")
173 missing = nonnegative_int(
174 value.get("missing", group_n - summarized_n),
175 f"{cell_field}.missing",
176 )
177 if summarized_n + missing != group_n:
178 raise InputError(
179 f"{cell_field} n plus missing must equal group n"
180 )
181 summary = require_nonempty_text(
182 value.get("summary"), f"{cell_field}.summary", max_length=120
183 )
184 if group_n < minimum or summarized_n < minimum or 0 < missing < minimum:
185 cells[group_id] = "SUPP"
186 primary_suppressed.add(group_id)
187 else:
188 cells[group_id] = f"{summary}; n={summarized_n}"
190 if (
191 row_type == "categorical"
192 and len(primary_suppressed) == 1
193 and eligible_counts
194 ):
195 complementary = min(eligible_counts, key=eligible_counts.get)
196 cells[complementary] = "SUPP-C"
197 notes.append(
198 f"Complementary suppression applied to row {row_index + 1}."
199 )
200 output_rows.append([label] + [cells[group_id] for group_id in group_map])
202 notes.extend(
203 [
204 f"Cells use a minimum threshold of {minimum}. SUPP is primary suppression; SUPP-C is complementary suppression.",
205 "Thresholding is an operational disclosure control, not a HIPAA or privacy determination.",
206 "Descriptive aggregate table only; no patient-specific or clinical recommendation output.",
207 "Not for patient care or live clinical use.",
208 ]
209 )
210 if log.ok:
211 log.info.append("Disclosure-controlled aggregate table generated")
212 return log, headers, output_rows, notes
215def _to_markdown(
216 title: str, headers: list[str], rows: list[list[str]], notes: list[str]
217) -> str:
218 lines = [
219 f"# {title}",
220 "",
221 "**Research aggregate only — not for patient care or live clinical use.**",
222 "",
223 "| " + " | ".join(headers) + " |",
224 "|" + "|".join("---" for _ in headers) + "|",
225 ]
226 lines.extend("| " + " | ".join(row) + " |" for row in rows)
227 lines.extend(["", "## Disclosure notes"])
228 lines.extend(f"- {note}" for note in notes)
229 return "\n".join(lines) + "\n"
232def _to_csv(headers: list[str], rows: list[list[str]], notes: list[str]) -> str:
233 def safe_cell(value: str) -> str:
234 # Treat labels/summaries as text when opened by a spreadsheet program.
235 if value.lstrip().startswith(("=", "+", "-", "@")):
236 return "'" + value
237 return value
239 stream = io.StringIO()
240 writer = csv.writer(stream, lineterminator="\n")
241 writer.writerow([safe_cell(value) for value in headers])
242 writer.writerows([safe_cell(value) for value in row] for row in rows)
243 writer.writerow([])
244 writer.writerow(["Disclosure notes"])
245 for note in notes:
246 writer.writerow([note])
247 return stream.getvalue()
250def main() -> int:
251 parser = argparse.ArgumentParser(
252 description=(
253 "Generate a bounded descriptive cohort table from local aggregate JSON "
254 "with primary and complementary disclosure suppression."
255 )
256 )
257 parser.add_argument("input", help="Local aggregate JSON")
258 parser.add_argument("-o", "--output", help="Optional .md or .csv output")
259 parser.add_argument(
260 "--min-cell-size",
261 type=int,
262 default=11,
263 help="Operational suppression threshold (default: 11; not a legal standard)",
264 )
265 args = parser.parse_args()
266 if not 2 <= args.min_cell_size <= 1_000:
267 parser.error("--min-cell-size must be between 2 and 1000")
269 try:
270 document = load_json_object(args.input)
271 log, headers, rows, notes = _build_table(document, args.min_cell_size)
272 if log.errors:
273 for error in log.errors:
274 print(f"ERROR: {error}", file=sys.stderr)
275 return 1
276 title = require_nonempty_text(document["metadata"].get("title"), "metadata.title")
277 if args.output:
278 if args.output.lower().endswith(".csv"):
279 text = _to_csv(headers, rows, notes)
280 write_text(args.output, text, {".csv"})
281 elif args.output.lower().endswith(".md"):
282 text = _to_markdown(title, headers, rows, notes)
283 write_text(args.output, text, {".md"})
284 else:
285 raise InputError("Output must end in .md or .csv")
286 else:
287 print(_to_markdown(title, headers, rows, notes), end="")
288 except InputError as exc:
289 print(f"ERROR: {exc}", file=sys.stderr)
290 return 2
291 return 0
294if __name__ == "__main__":
295 sys.exit(main())