Skills › Scientific Agent Skills › scikit-survival › validate_survival_csv.py
validate_survival_csv.py
172 lines · skills/scikit-survival/scripts/validate_survival_csv.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a local survival CSV and optionally write a structured NumPy array."""
4from __future__ import annotations
6import argparse
7from typing import Any
9from _common import (
10 DEFAULT_SEED,
11 MAX_FEATURES,
12 CliError,
13 atomic_save_npy,
14 bounded_int,
15 emit_json,
16 parse_names,
17 read_csv,
18 structured_survival,
19 synthetic_survival_frame,
20)
23def validate_and_convert(
24 frame: Any,
25 *,
26 event_column: str,
27 time_column: str,
28 feature_columns: list[str] | None = None,
29) -> tuple[Any, dict[str, Any]]:
30 """Validate outcome/schema columns and build a two-field survival array."""
32 if event_column == time_column:
33 raise CliError("event and time columns must differ")
34 missing_outcomes = [
35 name for name in (event_column, time_column) if name not in frame.columns
36 ]
37 if missing_outcomes:
38 raise CliError(f"missing outcome columns: {', '.join(missing_outcomes)}")
40 if feature_columns is None:
41 features = [
42 str(name)
43 for name in frame.columns
44 if name not in {event_column, time_column}
45 ]
46 else:
47 features = list(feature_columns)
48 if len(features) > MAX_FEATURES:
49 raise CliError(f"at most {MAX_FEATURES} feature columns are allowed")
50 if event_column in features or time_column in features:
51 raise CliError("outcome columns must not be included as features")
52 missing_features = [name for name in features if name not in frame.columns]
53 if missing_features:
54 raise CliError(f"missing feature columns: {', '.join(missing_features)}")
56 outcome = structured_survival(
57 frame[event_column],
58 frame[time_column],
59 event_name=event_column,
60 time_name=time_column,
61 )
62 event = outcome[event_column]
63 time = outcome[time_column]
64 feature_dtypes = {name: str(frame[name].dtype) for name in features}
65 report = {
66 "censoring": {
67 "censored_count": int((~event).sum()),
68 "censored_fraction": float((~event).mean()),
69 "event_count": int(event.sum()),
70 "event_fraction": float(event.mean()),
71 "type": "right",
72 },
73 "columns": {
74 "event": event_column,
75 "features": features,
76 "feature_dtypes": feature_dtypes,
77 "time": time_column,
78 },
79 "rows": int(len(frame)),
80 "schema_version": "1.0",
81 "structured_dtype": [
82 [name, outcome.dtype.fields[name][0].str] for name in outcome.dtype.names
83 ],
84 "time": {
85 "maximum": float(time.max()),
86 "median": float(__import__("numpy").median(time)),
87 "minimum": float(time.min()),
88 "strictly_positive": True,
89 },
90 "validation": {
91 "binary_event": True,
92 "finite_time": True,
93 "outcome_excluded_from_features": True,
94 "valid": True,
95 },
96 }
97 return outcome, report
100def build_parser() -> argparse.ArgumentParser:
101 parser = argparse.ArgumentParser(
102 description=(
103 "Validate right-censored survival outcomes in a bounded local CSV. "
104 "With no --input, use deterministic synthetic data."
105 )
106 )
107 parser.add_argument(
108 "--input", help="Local .csv input; URLs and symlinks are rejected"
109 )
110 parser.add_argument("--event-column", default="event")
111 parser.add_argument("--time-column", default="time")
112 parser.add_argument(
113 "--feature-columns",
114 help="Optional explicit comma-separated features; defaults to other columns",
115 )
116 parser.add_argument(
117 "--synthetic-rows",
118 type=bounded_int(40, 20_000),
119 default=240,
120 help="Synthetic row count when --input is omitted (default: 240)",
121 )
122 parser.add_argument("--seed", type=bounded_int(0, 2**32 - 1), default=DEFAULT_SEED)
123 parser.add_argument(
124 "--structured-output",
125 help="Optional .npy output containing the structured survival array",
126 )
127 parser.add_argument("--output", help="Optional .json validation report")
128 parser.add_argument("--force", action="store_true", help="Replace explicit outputs")
129 return parser
132def main(argv: list[str] | None = None) -> int:
133 parser = build_parser()
134 args = parser.parse_args(argv)
135 try:
136 if args.input:
137 frame, source_name = read_csv(args.input)
138 source = {"kind": "local_csv", "name": source_name}
139 defaults: list[str] | None = None
140 else:
141 frame, numeric, categorical = synthetic_survival_frame(
142 rows=args.synthetic_rows, seed=args.seed
143 )
144 source = {
145 "kind": "synthetic",
146 "network_used": False,
147 "seed": args.seed,
148 }
149 defaults = numeric + categorical
151 requested_features = parse_names(args.feature_columns)
152 feature_columns = requested_features or defaults
153 outcome, report = validate_and_convert(
154 frame,
155 event_column=args.event_column,
156 time_column=args.time_column,
157 feature_columns=feature_columns,
158 )
159 report["source"] = source
160 report["structured_output"] = (
161 None if args.structured_output is None else args.structured_output
162 )
163 if args.structured_output:
164 atomic_save_npy(outcome, args.structured_output, force=args.force)
165 emit_json(report, output=args.output, force=args.force)
166 except CliError as exc:
167 parser.error(str(exc))
168 return 0
171if __name__ == "__main__":
172 raise SystemExit(main())