Skills › Scientific Agent Skills › pathml › slide_manifest.py
slide_manifest.py
411 lines · skills/pathml/scripts/slide_manifest.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate local slide manifests and inspect allowlisted technical metadata."""
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_ROWS,
18 PINNED_INSTALL,
19 checked_input_file,
20 checked_root,
21 emit_json,
22 parse_name_list,
23 run_cli,
24 sha256_file,
25)
28SLIDE_SUFFIXES = {
29 ".svs",
30 ".tif",
31 ".tiff",
32 ".ome.tif",
33 ".ome.tiff",
34 ".bif",
35 ".ndpi",
36 ".vms",
37 ".vmu",
38 ".scn",
39 ".mrxs",
40 ".svslide",
41 ".qptiff",
42 ".dcm",
43 ".dicom",
44 ".czi",
45 ".vsi",
46 ".zvi",
47 ".h5",
48 ".h5path",
49}
50OPENSLIDE_SUFFIXES = {
51 ".svs",
52 ".tif",
53 ".tiff",
54 ".bif",
55 ".ndpi",
56 ".vms",
57 ".vmu",
58 ".scn",
59 ".mrxs",
60 ".svslide",
61}
62DICOM_SUFFIXES = {".dcm", ".dicom"}
63ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
66def _matched_suffix(path: Path) -> str:
67 lowered = path.name.lower()
68 matches = [suffix for suffix in SLIDE_SUFFIXES if lowered.endswith(suffix)]
69 if not matches:
70 raise CliError(f"unsupported slide suffix: {path.suffix.lower() or '<none>'}")
71 return max(matches, key=len)
74def _validate_identifier(value: str, *, field: str, row_number: int) -> str:
75 clean = value.strip()
76 if not ID_PATTERN.fullmatch(clean):
77 raise CliError(
78 f"row {row_number}: {field} must be a pseudonymous identifier "
79 "using 1-128 letters, digits, dots, underscores, or hyphens"
80 )
81 return clean
84def validate_manifest(args: argparse.Namespace) -> dict[str, Any]:
85 root = checked_root(args.root)
86 manifest = checked_input_file(
87 args.manifest,
88 root=root,
89 suffixes={".csv"},
90 max_bytes=args.max_manifest_bytes,
91 )
92 allowed_splits = set(parse_name_list(args.splits, name="--splits"))
93 max_slide_bytes = int(args.max_slide_gib * 1024**3)
95 slide_ids: set[str] = set()
96 resolved_paths: set[Path] = set()
97 patient_splits: dict[str, str] = {}
98 split_counts: Counter[str] = Counter()
99 suffix_counts: Counter[str] = Counter()
100 patient_ids: set[str] = set()
102 try:
103 with manifest.open("r", encoding="utf-8-sig", newline="") as handle:
104 reader = csv.DictReader(handle)
105 headers = reader.fieldnames
106 if headers is None:
107 raise CliError("manifest has no header")
108 normalized_headers = [header.strip() for header in headers]
109 if len(normalized_headers) != len(set(normalized_headers)):
110 raise CliError("manifest header contains duplicate column names")
111 required = {"slide_id", "patient_id", "path"}
112 missing = sorted(required - set(normalized_headers))
113 if missing:
114 raise CliError(
115 f"manifest is missing required columns: {', '.join(missing)}"
116 )
118 row_count = 0
119 for row_count, raw_row in enumerate(reader, start=1):
120 if row_count > args.max_rows:
121 raise CliError(f"manifest exceeds --max-rows={args.max_rows}")
122 row_number = row_count + 1
123 if None in raw_row or any(value is None for value in raw_row.values()):
124 raise CliError(f"row {row_number}: CSV field count does not match header")
125 row = {
126 key.strip(): (value or "").strip()
127 for key, value in raw_row.items()
128 if key is not None
129 }
130 slide_id = _validate_identifier(
131 row["slide_id"], field="slide_id", row_number=row_number
132 )
133 patient_id = _validate_identifier(
134 row["patient_id"], field="patient_id", row_number=row_number
135 )
136 if slide_id in slide_ids:
137 raise CliError(f"row {row_number}: duplicate slide_id")
138 slide_ids.add(slide_id)
139 patient_ids.add(patient_id)
141 path_value = row["path"]
142 if not path_value:
143 raise CliError(f"row {row_number}: path is empty")
144 slide_path = checked_input_file(
145 path_value,
146 root=root,
147 suffixes=SLIDE_SUFFIXES,
148 max_bytes=max_slide_bytes,
149 )
150 if slide_path in resolved_paths:
151 raise CliError(f"row {row_number}: duplicate resolved slide path")
152 resolved_paths.add(slide_path)
153 suffix_counts[_matched_suffix(slide_path)] += 1
155 split = row.get("split", "")
156 if "split" in normalized_headers and not split:
157 raise CliError(f"row {row_number}: split column must not contain blanks")
158 if split:
159 if allowed_splits and split not in allowed_splits:
160 raise CliError(
161 f"row {row_number}: split must be one of "
162 f"{', '.join(sorted(allowed_splits))}"
163 )
164 previous = patient_splits.setdefault(patient_id, split)
165 if previous != split:
166 raise CliError(
167 f"row {row_number}: patient_id appears in multiple splits: "
168 f"{previous!r} and {split!r}"
169 )
170 split_counts[split] += 1
172 if row_count == 0:
173 raise CliError("manifest contains no data rows")
174 except CliError:
175 raise
176 except (OSError, UnicodeError, csv.Error) as exc:
177 raise CliError(f"cannot parse manifest CSV: {exc}") from exc
179 return {
180 "command": "validate",
181 "valid": True,
182 "row_count": len(slide_ids),
183 "patient_count": len(patient_ids),
184 "split_counts": dict(sorted(split_counts.items())),
185 "suffix_counts": dict(sorted(suffix_counts.items())),
186 "checks": {
187 "local_regular_files": True,
188 "no_urls": True,
189 "no_symlinks": True,
190 "unique_slide_ids": True,
191 "unique_slide_paths": True,
192 "patient_split_isolation": bool(patient_splits),
193 },
194 "privacy_note": (
195 "Identifiers were validated syntactically only; confirm they are "
196 "pseudonyms and keep the linkage key outside the analysis workspace."
197 ),
198 }
201def _inspect_openslide(path: Path) -> dict[str, Any]:
202 try:
203 import openslide
204 except ModuleNotFoundError as exc:
205 raise CliError(
206 f"technical OpenSlide metadata requires PathML dependencies; {PINNED_INSTALL}"
207 ) from exc
209 slide = openslide.OpenSlide(str(path))
210 try:
211 properties = slide.properties
212 technical = {
213 "backend": "openslide",
214 "level_count": int(slide.level_count),
215 "level_dimensions_xy": [
216 [int(width), int(height)] for width, height in slide.level_dimensions
217 ],
218 "level_downsamples": [float(value) for value in slide.level_downsamples],
219 }
220 allowlist = {
221 "openslide.vendor": "vendor",
222 "openslide.objective-power": "objective_power",
223 "openslide.mpp-x": "mpp_x",
224 "openslide.mpp-y": "mpp_y",
225 "openslide.comment": None,
226 }
227 for source, destination in allowlist.items():
228 if destination is not None and source in properties:
229 value: Any = properties[source]
230 if destination in {"objective_power", "mpp_x", "mpp_y"}:
231 try:
232 value = float(value)
233 if not math.isfinite(value) or value <= 0:
234 value = None
235 except (TypeError, ValueError):
236 value = None
237 technical[destination] = value
238 return technical
239 finally:
240 slide.close()
243def _inspect_dicom(path: Path) -> dict[str, Any]:
244 try:
245 import pydicom
246 except ModuleNotFoundError as exc:
247 raise CliError(
248 f"technical DICOM metadata requires PathML dependencies; {PINNED_INSTALL}"
249 ) from exc
251 fields = (
252 "Rows",
253 "Columns",
254 "TotalPixelMatrixRows",
255 "TotalPixelMatrixColumns",
256 "NumberOfFrames",
257 "SamplesPerPixel",
258 "PhotometricInterpretation",
259 "BitsAllocated",
260 )
261 try:
262 dataset = pydicom.dcmread(
263 path,
264 stop_before_pixels=True,
265 specific_tags=list(fields),
266 force=False,
267 )
268 except Exception as exc:
269 raise CliError(f"cannot parse allowlisted DICOM metadata: {exc}") from exc
270 output: dict[str, Any] = {"backend": "dicom"}
271 for field in fields:
272 if hasattr(dataset, field):
273 value = getattr(dataset, field)
274 if field == "PhotometricInterpretation":
275 output[field] = str(value)
276 else:
277 try:
278 output[field] = int(value)
279 except (TypeError, ValueError):
280 output[field] = str(value)
281 return output
284def _inspect_raster(path: Path, max_pixels: int) -> dict[str, Any]:
285 try:
286 from PIL import Image
287 except ModuleNotFoundError as exc:
288 raise CliError(
289 f"technical raster metadata requires PathML/Pillow dependencies; {PINNED_INSTALL}"
290 ) from exc
292 Image.MAX_IMAGE_PIXELS = max_pixels
293 try:
294 with Image.open(path) as image:
295 width, height = image.size
296 if width * height > max_pixels:
297 raise CliError(
298 f"image has {width * height} pixels; limit is {max_pixels}"
299 )
300 return {
301 "backend": "pillow",
302 "width": int(width),
303 "height": int(height),
304 "mode": str(image.mode),
305 "frame_count": int(getattr(image, "n_frames", 1)),
306 "format": str(image.format or "unknown"),
307 }
308 except CliError:
309 raise
310 except Exception as exc:
311 raise CliError(f"cannot parse allowlisted raster metadata: {exc}") from exc
314def inspect_slide(args: argparse.Namespace) -> dict[str, Any]:
315 root = checked_root(args.root)
316 max_slide_bytes = int(args.max_slide_gib * 1024**3)
317 slide = checked_input_file(
318 args.slide,
319 root=root,
320 suffixes=SLIDE_SUFFIXES,
321 max_bytes=max_slide_bytes,
322 )
323 suffix = _matched_suffix(slide)
324 report: dict[str, Any] = {
325 "command": "inspect",
326 "local_regular_file": True,
327 "symlink": False,
328 "size_bytes": slide.stat().st_size,
329 "suffix": suffix,
330 "technical_metadata_included": False,
331 "path_redacted": True,
332 }
333 if args.sha256:
334 report["sha256"] = sha256_file(
335 slide, max_bytes=int(args.max_hash_gib * 1024**3)
336 )
337 if args.technical_metadata:
338 if suffix in DICOM_SUFFIXES:
339 technical = _inspect_dicom(slide)
340 elif suffix in OPENSLIDE_SUFFIXES:
341 technical = _inspect_openslide(slide)
342 else:
343 technical = _inspect_raster(slide, args.max_pixels)
344 report["technical_metadata"] = technical
345 report["technical_metadata_included"] = True
346 return report
349def build_parser() -> argparse.ArgumentParser:
350 parser = argparse.ArgumentParser(
351 description=(
352 "Validate local pseudonymous slide manifests or inspect only "
353 "allowlisted technical metadata. No network access."
354 )
355 )
356 subparsers = parser.add_subparsers(dest="command", required=True)
358 validate = subparsers.add_parser(
359 "validate", help="validate a bounded local CSV manifest"
360 )
361 validate.add_argument("--manifest", required=True)
362 validate.add_argument("--root", default=".")
363 validate.add_argument("--splits", default="train,validation,test")
364 validate.add_argument("--max-rows", type=int, default=100_000)
365 validate.add_argument(
366 "--max-manifest-bytes", type=int, default=MAX_CSV_BYTES
367 )
368 validate.add_argument("--max-slide-gib", type=float, default=1024.0)
369 validate.add_argument("--output")
370 validate.add_argument("--force", action="store_true")
372 inspect = subparsers.add_parser(
373 "inspect", help="inspect one local slide without emitting its path"
374 )
375 inspect.add_argument("--slide", required=True)
376 inspect.add_argument("--root", default=".")
377 inspect.add_argument("--max-slide-gib", type=float, default=1024.0)
378 inspect.add_argument("--technical-metadata", action="store_true")
379 inspect.add_argument("--max-pixels", type=int, default=16_000_000)
380 inspect.add_argument("--sha256", action="store_true")
381 inspect.add_argument("--max-hash-gib", type=float, default=64.0)
382 inspect.add_argument("--output")
383 inspect.add_argument("--force", action="store_true")
384 return parser
387def main() -> None:
388 args = build_parser().parse_args()
389 if args.command == "validate":
390 if not 1 <= args.max_rows <= MAX_ROWS:
391 raise CliError(f"--max-rows must be between 1 and {MAX_ROWS}")
392 if not 0 < args.max_manifest_bytes <= MAX_CSV_BYTES:
393 raise CliError(
394 f"--max-manifest-bytes must be between 1 and {MAX_CSV_BYTES}"
395 )
396 if not math.isfinite(args.max_slide_gib) or not 0 < args.max_slide_gib <= 4096:
397 raise CliError("--max-slide-gib must be finite and in (0, 4096]")
398 report = validate_manifest(args)
399 else:
400 if not math.isfinite(args.max_slide_gib) or not 0 < args.max_slide_gib <= 4096:
401 raise CliError("--max-slide-gib must be finite and in (0, 4096]")
402 if not math.isfinite(args.max_hash_gib) or not 0 < args.max_hash_gib <= 4096:
403 raise CliError("--max-hash-gib must be finite and in (0, 4096]")
404 if not 1 <= args.max_pixels <= 100_000_000:
405 raise CliError("--max-pixels must be between 1 and 100000000")
406 report = inspect_slide(args)
407 emit_json(report, output=args.output, root=args.root, force=args.force)
410if __name__ == "__main__":
411 raise SystemExit(run_cli(main))