Skills › Scientific Agent Skills › pathml › validate_spatial_schema.py
validate_spatial_schema.py
431 lines · skills/pathml/scripts/validate_spatial_schema.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate bounded graph JSON and multiplex cell-table CSV schemas."""
4from __future__ import annotations
6import argparse
7import csv
8import math
9import re
10from collections import Counter
11from pathlib import Path
12from typing import Any
14from _common import (
15 CliError,
16 MAX_CSV_BYTES,
17 MAX_JSON_BYTES,
18 checked_input_file,
19 checked_root,
20 emit_json,
21 finite_float,
22 load_json_object,
23 parse_name_list,
24 run_cli,
25 validate_keys,
26)
29ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
30COORDINATE_UNITS = {"level_pixels", "level0_pixels", "um"}
33def _identifier(value: Any, *, context: str) -> str:
34 if not isinstance(value, str) or not ID_PATTERN.fullmatch(value):
35 raise CliError(
36 f"{context} must use 1-128 letters, digits, dots, underscores, "
37 "colons, or hyphens"
38 )
39 return value
42def validate_graph(args: argparse.Namespace) -> dict[str, Any]:
43 root = checked_root(args.root)
44 document = load_json_object(
45 args.input,
46 root=root,
47 max_bytes=args.max_input_bytes,
48 )
49 validate_keys(
50 document,
51 allowed={
52 "schema_version",
53 "slide_id",
54 "coordinate_unit",
55 "level",
56 "nodes",
57 "edges",
58 "metadata",
59 },
60 required={"schema_version", "slide_id", "coordinate_unit", "nodes", "edges"},
61 context="graph",
62 )
63 if document["schema_version"] != "1.0":
64 raise CliError("graph schema_version must be '1.0'")
65 _identifier(document["slide_id"], context="slide_id")
66 coordinate_unit = document["coordinate_unit"]
67 if not isinstance(coordinate_unit, str) or coordinate_unit not in COORDINATE_UNITS:
68 raise CliError(
69 "coordinate_unit must be one of: "
70 + ", ".join(sorted(COORDINATE_UNITS))
71 )
72 if "level" in document:
73 level = document["level"]
74 if isinstance(level, bool) or not isinstance(level, int) or level < 0:
75 raise CliError("level must be a nonnegative integer")
76 elif coordinate_unit == "level_pixels":
77 raise CliError("level is required for level_pixels coordinates")
78 nodes = document["nodes"]
79 edges = document["edges"]
80 if not isinstance(nodes, list) or not isinstance(edges, list):
81 raise CliError("nodes and edges must be arrays")
82 if not 1 <= len(nodes) <= args.max_nodes:
83 raise CliError(f"node count must be in [1, {args.max_nodes}]")
84 if len(edges) > args.max_edges:
85 raise CliError(f"edge count exceeds --max-edges={args.max_edges}")
87 node_ids: set[str] = set()
88 feature_length: int | None = None
89 features_present: bool | None = None
90 for index, node in enumerate(nodes):
91 if not isinstance(node, dict):
92 raise CliError(f"node {index} must be an object")
93 validate_keys(
94 node,
95 allowed={"id", "x", "y", "features", "label"},
96 required={"id", "x", "y"},
97 context=f"node {index}",
98 )
99 node_id = _identifier(node["id"], context=f"node {index} id")
100 if node_id in node_ids:
101 raise CliError(f"duplicate node id: {node_id!r}")
102 node_ids.add(node_id)
103 finite_float(node["x"], name=f"node {index} x", minimum=0, maximum=1e12)
104 finite_float(node["y"], name=f"node {index} y", minimum=0, maximum=1e12)
105 if features_present is None:
106 features_present = "features" in node
107 elif features_present != ("features" in node):
108 raise CliError("features must be present for all nodes or none")
109 if "features" in node:
110 features = node["features"]
111 if not isinstance(features, list):
112 raise CliError(f"node {index} features must be an array")
113 if len(features) > args.max_features:
114 raise CliError(
115 f"node {index} has more than {args.max_features} features"
116 )
117 if feature_length is None:
118 feature_length = len(features)
119 elif len(features) != feature_length:
120 raise CliError("all feature arrays must have the same length")
121 for feature_index, value in enumerate(features):
122 finite_float(
123 value,
124 name=f"node {index} feature {feature_index}",
125 minimum=-1e15,
126 maximum=1e15,
127 )
129 seen_edges: set[tuple[str, str]] = set()
130 self_loops = 0
131 for index, edge in enumerate(edges):
132 if not isinstance(edge, dict):
133 raise CliError(f"edge {index} must be an object")
134 validate_keys(
135 edge,
136 allowed={"source", "target", "weight", "edge_type"},
137 required={"source", "target"},
138 context=f"edge {index}",
139 )
140 source = _identifier(edge["source"], context=f"edge {index} source")
141 target = _identifier(edge["target"], context=f"edge {index} target")
142 if source not in node_ids or target not in node_ids:
143 raise CliError(f"edge {index} references an unknown node")
144 if source == target:
145 self_loops += 1
146 if not args.allow_self_loops:
147 raise CliError(f"edge {index} is a self-loop")
148 key = tuple(sorted((source, target))) if args.undirected else (source, target)
149 if key in seen_edges:
150 raise CliError(f"edge {index} duplicates an existing edge")
151 seen_edges.add(key)
152 if "weight" in edge:
153 finite_float(
154 edge["weight"],
155 name=f"edge {index} weight",
156 minimum=-1e15,
157 maximum=1e15,
158 )
160 return {
161 "command": "graph",
162 "valid": True,
163 "node_count": len(nodes),
164 "edge_count": len(edges),
165 "feature_count": feature_length or 0,
166 "coordinate_unit": coordinate_unit,
167 "directed_edge_keys": not args.undirected,
168 "self_loop_count": self_loops,
169 "checks": {
170 "strict_json": True,
171 "bounded": True,
172 "unique_node_ids": True,
173 "finite_coordinates": True,
174 "uniform_feature_shape": True,
175 "valid_edge_endpoints": True,
176 "unique_edges": True,
177 },
178 "path_redacted": True,
179 }
182def _parse_csv_float(
183 raw: str,
184 *,
185 name: str,
186 minimum: float | None = None,
187) -> float:
188 try:
189 value = float(raw)
190 except ValueError as exc:
191 raise CliError(f"{name} must be numeric") from exc
192 if not math.isfinite(value):
193 raise CliError(f"{name} must be finite")
194 if minimum is not None and value < minimum:
195 raise CliError(f"{name} must be at least {minimum}")
196 return value
199def validate_multiplex(args: argparse.Namespace) -> dict[str, Any]:
200 root = checked_root(args.root)
201 path = checked_input_file(
202 args.input,
203 root=root,
204 suffixes={".csv"},
205 max_bytes=args.max_input_bytes,
206 )
207 requested_markers = parse_name_list(
208 args.marker_columns, name="--marker-columns"
209 )
210 keys: set[tuple[str, str]] = set()
211 slide_units: dict[str, str] = {}
212 slide_levels: dict[str, int] = {}
213 patient_splits: dict[str, str] = {}
214 unit_counts: Counter[str] = Counter()
215 marker_missing: Counter[str] = Counter()
216 slide_ids: set[str] = set()
217 patient_ids: set[str] = set()
219 try:
220 with path.open("r", encoding="utf-8-sig", newline="") as handle:
221 reader = csv.DictReader(handle)
222 headers = reader.fieldnames
223 if headers is None:
224 raise CliError("multiplex CSV has no header")
225 headers = [header.strip() for header in headers]
226 if len(headers) != len(set(headers)):
227 raise CliError("multiplex CSV header contains duplicates")
228 required = {"cell_id", "slide_id", "x", "y", "coordinate_unit"}
229 missing = sorted(required - set(headers))
230 if missing:
231 raise CliError(
232 "multiplex CSV is missing required columns: "
233 + ", ".join(missing)
234 )
235 marker_columns = requested_markers or [
236 header for header in headers if header.startswith("marker:")
237 ]
238 if not marker_columns:
239 raise CliError(
240 "provide --marker-columns or use at least one marker:* column"
241 )
242 missing_markers = sorted(set(marker_columns) - set(headers))
243 if missing_markers:
244 raise CliError(
245 "requested marker columns are absent: "
246 + ", ".join(missing_markers)
247 )
249 row_count = 0
250 for row_count, raw_row in enumerate(reader, start=1):
251 if row_count > args.max_rows:
252 raise CliError(f"CSV exceeds --max-rows={args.max_rows}")
253 row_number = row_count + 1
254 if None in raw_row or any(value is None for value in raw_row.values()):
255 raise CliError(f"row {row_number}: CSV field count does not match header")
256 row = {
257 key.strip(): (value or "").strip()
258 for key, value in raw_row.items()
259 if key is not None
260 }
261 slide_id = _identifier(
262 row["slide_id"], context=f"row {row_number} slide_id"
263 )
264 cell_id = _identifier(
265 row["cell_id"], context=f"row {row_number} cell_id"
266 )
267 key = (slide_id, cell_id)
268 if key in keys:
269 raise CliError(
270 f"row {row_number}: duplicate (slide_id, cell_id)"
271 )
272 keys.add(key)
273 slide_ids.add(slide_id)
274 _parse_csv_float(
275 row["x"], name=f"row {row_number} x", minimum=0
276 )
277 _parse_csv_float(
278 row["y"], name=f"row {row_number} y", minimum=0
279 )
280 unit = row["coordinate_unit"]
281 if unit not in COORDINATE_UNITS:
282 raise CliError(
283 f"row {row_number}: coordinate_unit must be one of "
284 + ", ".join(sorted(COORDINATE_UNITS))
285 )
286 previous_unit = slide_units.setdefault(slide_id, unit)
287 if previous_unit != unit:
288 raise CliError(
289 f"row {row_number}: slide has inconsistent coordinate units"
290 )
291 unit_counts[unit] += 1
293 level_raw = row.get("level", "")
294 if (unit == "level_pixels" or "level" in headers) and not level_raw:
295 raise CliError(f"row {row_number}: level must be supplied")
296 if level_raw:
297 try:
298 level = int(level_raw)
299 except ValueError as exc:
300 raise CliError(
301 f"row {row_number}: level must be an integer"
302 ) from exc
303 if level < 0:
304 raise CliError(
305 f"row {row_number}: level must be nonnegative"
306 )
307 previous_level = slide_levels.setdefault(slide_id, level)
308 if previous_level != level:
309 raise CliError(
310 f"row {row_number}: slide has inconsistent levels"
311 )
313 patient_id = row.get("patient_id", "")
314 split = row.get("split", "")
315 if patient_id:
316 patient_id = _identifier(
317 patient_id, context=f"row {row_number} patient_id"
318 )
319 patient_ids.add(patient_id)
320 if split and patient_id:
321 previous_split = patient_splits.setdefault(patient_id, split)
322 if previous_split != split:
323 raise CliError(
324 f"row {row_number}: patient appears in multiple splits"
325 )
327 for marker in marker_columns:
328 raw_value = row.get(marker, "")
329 if raw_value == "":
330 marker_missing[marker] += 1
331 if not args.allow_missing_markers:
332 raise CliError(
333 f"row {row_number}: marker {marker!r} is missing"
334 )
335 continue
336 _parse_csv_float(
337 raw_value,
338 name=f"row {row_number} marker {marker}",
339 )
341 if row_count == 0:
342 raise CliError("multiplex CSV contains no rows")
343 except CliError:
344 raise
345 except (OSError, UnicodeError, csv.Error) as exc:
346 raise CliError(f"cannot parse multiplex CSV: {exc}") from exc
348 return {
349 "command": "multiplex",
350 "valid": True,
351 "row_count": len(keys),
352 "slide_count": len(slide_ids),
353 "patient_count": len(patient_ids),
354 "marker_columns": marker_columns,
355 "missing_marker_values": dict(sorted(marker_missing.items())),
356 "coordinate_unit_counts": dict(sorted(unit_counts.items())),
357 "checks": {
358 "bounded": True,
359 "unique_slide_cell_ids": True,
360 "finite_nonnegative_coordinates": True,
361 "per_slide_coordinate_units": True,
362 "per_slide_levels": True,
363 "finite_marker_values": True,
364 "patient_split_isolation_when_present": True,
365 },
366 "path_redacted": True,
367 }
370def build_parser() -> argparse.ArgumentParser:
371 parser = argparse.ArgumentParser(
372 description=(
373 "Validate bounded local graph JSON or multiplex cell CSV without "
374 "loading PathML, Torch, models, or checkpoints."
375 )
376 )
377 subparsers = parser.add_subparsers(dest="command", required=True)
379 graph = subparsers.add_parser("graph", help="validate graph JSON")
380 graph.add_argument("--input", required=True)
381 graph.add_argument("--root", default=".")
382 graph.add_argument("--max-input-bytes", type=int, default=MAX_JSON_BYTES)
383 graph.add_argument("--max-nodes", type=int, default=100_000)
384 graph.add_argument("--max-edges", type=int, default=1_000_000)
385 graph.add_argument("--max-features", type=int, default=4096)
386 graph.add_argument("--undirected", action="store_true")
387 graph.add_argument("--allow-self-loops", action="store_true")
388 graph.add_argument("--output")
389 graph.add_argument("--force", action="store_true")
391 multiplex = subparsers.add_parser(
392 "multiplex", help="validate a cell-by-marker CSV"
393 )
394 multiplex.add_argument("--input", required=True)
395 multiplex.add_argument("--root", default=".")
396 multiplex.add_argument("--max-input-bytes", type=int, default=MAX_CSV_BYTES)
397 multiplex.add_argument("--max-rows", type=int, default=100_000)
398 multiplex.add_argument("--marker-columns")
399 multiplex.add_argument("--allow-missing-markers", action="store_true")
400 multiplex.add_argument("--output")
401 multiplex.add_argument("--force", action="store_true")
402 return parser
405def main() -> None:
406 args = build_parser().parse_args()
407 if args.command == "graph":
408 if not 1 <= args.max_input_bytes <= MAX_JSON_BYTES:
409 raise CliError(
410 f"--max-input-bytes must be between 1 and {MAX_JSON_BYTES}"
411 )
412 if not 1 <= args.max_nodes <= 10_000_000:
413 raise CliError("--max-nodes must be between 1 and 10000000")
414 if not 0 <= args.max_edges <= 100_000_000:
415 raise CliError("--max-edges must be between 0 and 100000000")
416 if not 0 <= args.max_features <= 100_000:
417 raise CliError("--max-features must be between 0 and 100000")
418 report = validate_graph(args)
419 else:
420 if not 1 <= args.max_input_bytes <= MAX_CSV_BYTES:
421 raise CliError(
422 f"--max-input-bytes must be between 1 and {MAX_CSV_BYTES}"
423 )
424 if not 1 <= args.max_rows <= 1_000_000:
425 raise CliError("--max-rows must be between 1 and 1000000")
426 report = validate_multiplex(args)
427 emit_json(report, output=args.output, root=args.root, force=args.force)
430if __name__ == "__main__":
431 raise SystemExit(run_cli(main))