Skills › Scientific Agent Skills › pkpd-modeling › _common.py
_common.py
353 lines · skills/pkpd-modeling/scripts/_common.py at 5f2bc03
1"""Shared I/O, formatting, and reporting helpers for the pkpd-modeling scripts.
3Every script in this skill follows the same contract:
5* data goes to **stdout**, provenance and findings go to **stderr**, so
6 ``script.py ... > out.tsv`` keeps them apart;
7* ``--format table|tsv|json`` selects the stdout rendering, and ``json`` is
8 self-contained (it carries findings and notes too, because a machine reading
9 stdout should not have to also parse stderr);
10* exit status is ``0`` when nothing was flagged, ``1`` when at least one
11 finding was raised, and ``2`` for unusable input — so any script can gate a
12 pipeline.
14Nothing here does pharmacology. The numerics live in ``_models.py`` and in the
15individual scripts; this module only moves tables around.
16"""
18from __future__ import annotations
20import argparse
21import csv
22import json
23import math
24import sys
25from dataclasses import dataclass, field
26from pathlib import Path
27from typing import Any, Iterable, Mapping, Sequence
29EXIT_OK = 0
30EXIT_FINDINGS = 1
31EXIT_INPUT = 2
33_MISSING = {"", ".", "na", "n/a", "nan", "null", "none", "-"}
36class InputError(Exception):
37 """Unusable input. Callers map this to exit code 2."""
40# --------------------------------------------------------------- formatting
43def fmt(value: Any, digits: int = 6) -> str:
44 """Render a value for a fixed-width table without lying about precision."""
45 if value is None:
46 return "n/a"
47 if isinstance(value, bool):
48 return "yes" if value else "no"
49 if isinstance(value, str):
50 return value
51 if isinstance(value, (int,)):
52 return str(value)
53 try:
54 x = float(value)
55 except (TypeError, ValueError):
56 return str(value)
57 if math.isnan(x):
58 return "n/a"
59 if math.isinf(x):
60 return "inf" if x > 0 else "-inf"
61 if x == 0:
62 return "0"
63 return f"{x:.{digits}g}"
66def parse_float(text: Any, field_name: str = "value", allow_missing: bool = False):
67 """Parse a float, treating the usual missing-value spellings as ``None``."""
68 if text is None:
69 if allow_missing:
70 return None
71 raise InputError(f"{field_name}: missing")
72 if isinstance(text, (int, float)):
73 x = float(text)
74 if math.isnan(x) and allow_missing:
75 return None
76 if not math.isfinite(x):
77 raise InputError(f"{field_name}: must be finite")
78 return x
79 s = str(text).strip()
80 if s.lower() in _MISSING:
81 if allow_missing:
82 return None
83 raise InputError(f"{field_name}: missing")
84 try:
85 x = float(s)
86 if not math.isfinite(x):
87 raise InputError(f"{field_name}: must be finite")
88 return x
89 except ValueError as exc:
90 raise InputError(f"{field_name}: {s!r} is not a number") from exc
93def parse_positive(text: Any, field_name: str) -> float:
94 x = parse_float(text, field_name)
95 if x is None or x <= 0:
96 raise InputError(f"{field_name}: must be greater than zero, got {text!r}")
97 return x
100# ------------------------------------------------------------------- tables
103def _sniff_delimiter(sample: str, path: Path) -> str:
104 if path.suffix.lower() in {".tsv", ".tab"}:
105 return "\t"
106 if path.suffix.lower() == ".csv":
107 return ","
108 counts = {d: sample.count(d) for d in (",", "\t", ";")}
109 return max(counts, key=lambda d: counts[d]) if max(counts.values()) else ","
112def read_table(source: str | Path) -> list[dict[str, str]]:
113 """Read a delimited table into a list of row dicts.
115 Accepts ``-`` for stdin. Column names are lower-cased and stripped so a
116 file with ``Time`` and one with ``TIME`` behave the same; blank lines and
117 lines beginning with ``#`` are skipped, which lets a fixture carry a
118 provenance header.
119 """
120 if str(source) == "-":
121 text = sys.stdin.read()
122 path = Path("stdin.csv")
123 else:
124 path = Path(source)
125 if not path.is_file():
126 raise InputError(f"no such file: {path}")
127 text = path.read_text(encoding="utf-8-sig")
129 lines = [ln for ln in text.splitlines() if ln.strip() and not ln.lstrip().startswith("#")]
130 if not lines:
131 raise InputError(f"{path}: no data rows")
133 delimiter = _sniff_delimiter(lines[0], path)
134 reader = csv.DictReader(lines, delimiter=delimiter)
135 if not reader.fieldnames:
136 raise InputError(f"{path}: no header row")
138 fieldnames = [(name or "").strip().lower() for name in reader.fieldnames]
139 if len(set(fieldnames)) != len(fieldnames):
140 dupes = sorted({n for n in fieldnames if fieldnames.count(n) > 1})
141 raise InputError(f"{path}: duplicate column names: {', '.join(dupes)}")
143 rows: list[dict[str, str]] = []
144 for raw in reader:
145 row = {}
146 for key, value in zip(fieldnames, (raw.get(orig) for orig in reader.fieldnames)):
147 row[key] = (value or "").strip()
148 rows.append(row)
149 if not rows:
150 raise InputError(f"{path}: header only, no data rows")
151 return rows
154def require_columns(rows: Sequence[Mapping[str, Any]], required: Iterable[str], where: str) -> None:
155 present = set(rows[0].keys())
156 missing = [c for c in required if c not in present]
157 if missing:
158 raise InputError(
159 f"{where}: missing required column(s): {', '.join(missing)}. "
160 f"Found: {', '.join(sorted(present))}"
161 )
164def column(rows: Sequence[Mapping[str, Any]], name: str, allow_missing: bool = False) -> list:
165 return [parse_float(r.get(name), f"{name} (row {i + 1})", allow_missing) for i, r in enumerate(rows)]
168def group_by(rows: Sequence[Mapping[str, Any]], key: str) -> dict[str, list[Mapping[str, Any]]]:
169 """Stable grouping that preserves first-seen order of the keys."""
170 out: dict[str, list[Mapping[str, Any]]] = {}
171 for row in rows:
172 out.setdefault(str(row.get(key, "")).strip(), []).append(row)
173 return out
176# ------------------------------------------------------------------ reports
179@dataclass
180class Table:
181 title: str
182 rows: list[dict[str, Any]]
183 columns: list[str] | None = None
185 def headers(self) -> list[str]:
186 if self.columns:
187 return list(self.columns)
188 seen: list[str] = []
189 for row in self.rows:
190 for key in row:
191 if key not in seen:
192 seen.append(key)
193 return seen
196@dataclass
197class Report:
198 """Accumulates output, then renders it once at the end.
200 Findings are the reason a script exits non-zero. They are deliberately
201 plain sentences rather than codes: they are read by a person deciding
202 whether an analysis is defensible.
203 """
205 tables: list[Table] = field(default_factory=list)
206 findings: list[str] = field(default_factory=list)
207 notes: list[str] = field(default_factory=list)
208 scalars: dict[str, Any] = field(default_factory=dict)
210 def table(self, title: str, rows: Sequence[Mapping[str, Any]], columns: Sequence[str] | None = None) -> None:
211 self.tables.append(Table(title, [dict(r) for r in rows], list(columns) if columns else None))
213 def scalar(self, name: str, value: Any) -> None:
214 self.scalars[name] = value
216 def note(self, text: str) -> None:
217 self.notes.append(text)
219 def finding(self, text: str) -> None:
220 self.findings.append(text)
222 # -- rendering
224 def _render_table(self, table: Table) -> str:
225 headers = table.headers()
226 cells = [[fmt(row.get(h)) for h in headers] for row in table.rows]
227 widths = [len(h) for h in headers]
228 for line in cells:
229 for i, value in enumerate(line):
230 widths[i] = max(widths[i], len(value))
231 out = [""]
232 if table.title:
233 out.append(table.title)
234 out.append(" ".join(h.ljust(widths[i]) for i, h in enumerate(headers)).rstrip())
235 for line in cells:
236 out.append(" ".join(v.ljust(widths[i]) for i, v in enumerate(line)).rstrip())
237 return "\n".join(out)
239 def emit(self, fmt_name: str, stream=None, err=None) -> int:
240 stream = stream or sys.stdout
241 err = err or sys.stderr
243 if fmt_name == "json":
244 payload = {
245 "scalars": self.scalars,
246 "tables": [
247 {"title": t.title, "columns": t.headers(), "rows": t.rows} for t in self.tables
248 ],
249 "findings": self.findings,
250 "notes": self.notes,
251 }
252 print(json.dumps(_json_finite(payload), indent=2, allow_nan=False, default=_json_default), file=stream)
253 elif fmt_name == "tsv":
254 for key, value in self.scalars.items():
255 print(f"{key}\t{fmt(value)}", file=stream)
256 for table in self.tables:
257 headers = table.headers()
258 print("\t".join(headers), file=stream)
259 for row in table.rows:
260 print("\t".join(fmt(row.get(h)) for h in headers), file=stream)
261 else:
262 if self.scalars:
263 width = max(len(k) for k in self.scalars)
264 for key, value in self.scalars.items():
265 print(f"{key.ljust(width)} {fmt(value)}", file=stream)
266 for table in self.tables:
267 print(self._render_table(table), file=stream)
269 if fmt_name != "json":
270 for note in self.notes:
271 print(f"note: {note}", file=err)
272 for finding in self.findings:
273 print(f"finding: {finding}", file=err)
275 return EXIT_FINDINGS if self.findings else EXIT_OK
278def _json_finite(obj: Any) -> Any:
279 """Missing numerical diagnostics become JSON null, never nonstandard NaN/Infinity."""
280 if isinstance(obj, dict):
281 return {k: _json_finite(v) for k, v in obj.items()}
282 if isinstance(obj, (list, tuple)):
283 return [_json_finite(v) for v in obj]
284 if hasattr(obj, "tolist"):
285 return _json_finite(obj.tolist())
286 if isinstance(obj, float) and not math.isfinite(obj):
287 return None
288 return obj
291def validate_numeric_args(args: argparse.Namespace) -> None:
292 """Reject nonfinite CLI scalars before they enter a numerical model."""
293 for name, value in vars(args).items():
294 values = value if isinstance(value, (list, tuple)) else [value]
295 if any(isinstance(v, float) and not math.isfinite(v) for v in values):
296 raise InputError(f"--{name.replace('_', '-')}: must be finite")
299def _json_default(obj: Any) -> Any:
300 if hasattr(obj, "item"): # numpy scalar
301 return obj.item()
302 if hasattr(obj, "tolist"): # numpy array
303 return obj.tolist()
304 if isinstance(obj, float) and math.isnan(obj):
305 return None
306 raise TypeError(f"not JSON serialisable: {type(obj).__name__}")
309# --------------------------------------------------------------------- CLI
312def add_format_argument(parser: argparse.ArgumentParser) -> None:
313 parser.add_argument(
314 "--format",
315 choices=("table", "tsv", "json"),
316 default="table",
317 help="stdout rendering (default: table). json is self-contained.",
318 )
321def main_wrapper(func, argv: Sequence[str] | None = None) -> int:
322 """Run a script body, mapping InputError to exit code 2.
324 Keeps the ``if __name__`` block in every script down to one line and makes
325 the exit-code contract impossible to get wrong in one script and right in
326 another.
327 """
328 try:
329 return func(argv)
330 except InputError as exc:
331 print(f"error: {exc}", file=sys.stderr)
332 return EXIT_INPUT
333 except BrokenPipeError: # `| head`
334 return EXIT_OK
337__all__ = [
338 "EXIT_OK",
339 "EXIT_FINDINGS",
340 "EXIT_INPUT",
341 "InputError",
342 "Report",
343 "Table",
344 "add_format_argument",
345 "column",
346 "fmt",
347 "group_by",
348 "main_wrapper",
349 "parse_float",
350 "parse_positive",
351 "read_table",
352 "require_columns",
353]