Skills › Scientific Agent Skills › pydicom › dicom_inventory.py
dicom_inventory.py
405 lines · skills/pydicom/scripts/dicom_inventory.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded metadata-only DICOM technical inventory and structural checks."""
4from __future__ import annotations
6import argparse
7import sys
8from collections import Counter
9from typing import Any
11from _common import (
12 DEFAULT_MAX_DECOMPRESSED_BYTES,
13 DEFAULT_MAX_FILES,
14 DEFAULT_MAX_FRAMES,
15 DEFAULT_MAX_INPUT_BYTES,
16 HARD_MAX_DECOMPRESSED_BYTES,
17 HARD_MAX_FILES,
18 HARD_MAX_FRAMES,
19 HARD_MAX_INPUT_BYTES,
20 SCHEMA_VERSION,
21 TECHNICAL_KEYWORDS,
22 ToolError,
23 atomic_write,
24 bounded_int,
25 checked_input,
26 checked_output,
27 collect_local_files,
28 counter_dict,
29 emit_json,
30 fail_json,
31 json_bytes,
32 parse_size,
33 paths_overlap,
34 pixel_plan,
35 require_pydicom,
36 safe_dcmread,
37 valid_uid,
38)
40TOOL = "dicom_inventory"
41INVENTORY_TAGS = list(TECHNICAL_KEYWORDS) + [
42 "SOPInstanceUID",
43 "PatientIdentityRemoved",
44]
47def _issue(code: str, message: str, severity: str = "error") -> dict[str, str]:
48 return {"code": code, "message": message, "severity": severity}
51def inspect_dataset(
52 dataset: Any,
53 *,
54 file_id: str,
55 file_size: int,
56 max_frames: int,
57 max_decompressed_bytes: int,
58 forced_read: bool,
59) -> dict[str, Any]:
60 pydicom = require_pydicom()
61 issues: list[dict[str, str]] = []
62 file_meta = dataset.file_meta
63 try:
64 pydicom.dataset.validate_file_meta(file_meta, enforce_standard=True)
65 except Exception: # noqa: BLE001 - any validation failure becomes a finding
66 issues.append(
67 _issue(
68 "file_meta_invalid",
69 "Required DICOM File Meta Information is missing, empty, or invalid.",
70 )
71 )
73 transfer_value = str(file_meta.get("TransferSyntaxUID", ""))
74 transfer = None
75 if not valid_uid(transfer_value):
76 issues.append(
77 _issue(
78 "transfer_syntax_invalid", "Transfer Syntax UID is absent or invalid."
79 )
80 )
81 else:
82 transfer = pydicom.uid.UID(transfer_value)
83 if not transfer.is_transfer_syntax:
84 issues.append(_issue("transfer_syntax_unknown", "UID is not a recognized transfer syntax."))
85 transfer = None
87 sop_class = str(dataset.get("SOPClassUID", ""))
88 sop_instance = str(dataset.get("SOPInstanceUID", ""))
89 if not valid_uid(sop_class):
90 issues.append(
91 _issue("sop_class_invalid", "SOP Class UID is absent or invalid.")
92 )
93 if not valid_uid(sop_instance):
94 issues.append(
95 _issue("sop_instance_invalid", "SOP Instance UID is absent or invalid.")
96 )
97 media_class = str(file_meta.get("MediaStorageSOPClassUID", ""))
98 media_instance = str(file_meta.get("MediaStorageSOPInstanceUID", ""))
99 if sop_class and media_class and sop_class != media_class:
100 issues.append(
101 _issue(
102 "media_sop_class_mismatch",
103 "Media Storage SOP Class UID does not match the dataset SOP Class UID.",
104 )
105 )
106 if sop_instance and media_instance and sop_instance != media_instance:
107 issues.append(
108 _issue(
109 "media_sop_instance_mismatch",
110 "Media Storage SOP Instance UID does not match the dataset SOP Instance UID.",
111 )
112 )
114 image_plan: dict[str, Any] | None = None
115 if "Rows" in dataset or "Columns" in dataset:
116 try:
117 image_plan = pixel_plan(
118 dataset,
119 max_frames=max_frames,
120 max_decompressed_bytes=max_decompressed_bytes,
121 )
122 bits_allocated = image_plan["bits_allocated"]
123 bits_stored = int(dataset.get("BitsStored", bits_allocated))
124 high_bit = int(dataset.get("HighBit", bits_stored - 1))
125 if bits_stored < 1 or bits_stored > bits_allocated:
126 issues.append(
127 _issue(
128 "bits_stored_invalid",
129 "Bits Stored must be positive and no greater than Bits Allocated.",
130 )
131 )
132 if high_bit != bits_stored - 1:
133 issues.append(
134 _issue(
135 "high_bit_unexpected",
136 "High Bit does not equal Bits Stored minus one.",
137 "warning",
138 )
139 )
140 representation = dataset.get("PixelRepresentation")
141 if representation is None:
142 issues.append(
143 _issue(
144 "pixel_representation_absent",
145 "Pixel Representation is absent; review whether integer or float pixel data is used.",
146 "warning",
147 )
148 )
149 elif int(representation) not in {0, 1}:
150 issues.append(
151 _issue(
152 "pixel_representation_invalid",
153 "Pixel Representation must be 0 or 1 for integer Pixel Data.",
154 )
155 )
156 samples = image_plan["samples_per_pixel"]
157 if samples > 1 and "PlanarConfiguration" not in dataset:
158 issues.append(
159 _issue(
160 "planar_configuration_missing",
161 "Planar Configuration is required for multi-sample native pixel data.",
162 "warning",
163 )
164 )
165 if not dataset.get("PhotometricInterpretation"):
166 issues.append(
167 _issue(
168 "photometric_interpretation_missing",
169 "Photometric Interpretation is absent.",
170 )
171 )
172 except ToolError as exc:
173 issues.append(_issue("pixel_metadata_invalid", str(exc)))
175 if forced_read:
176 issues.append(
177 _issue(
178 "forced_read",
179 "Dataset was parsed without requiring a DICOM File Format header.",
180 "warning",
181 )
182 )
183 errors = sum(issue["severity"] == "error" for issue in issues)
184 warnings = sum(issue["severity"] == "warning" for issue in issues)
185 return {
186 "errors": errors,
187 "file_id": file_id,
188 "file_size_bytes": file_size,
189 "forced_read": forced_read,
190 "image": image_plan,
191 "issues": issues,
192 "modality": str(dataset.get("Modality", "UNSPECIFIED")),
193 "ok": errors == 0,
194 "patient_identity_removed_claim": str(
195 dataset.get("PatientIdentityRemoved", "ABSENT")
196 ).upper(),
197 "photometric_interpretation": str(
198 dataset.get("PhotometricInterpretation", "UNSPECIFIED")
199 ),
200 "sop_class": {
201 "name": pydicom.uid.UID(sop_class).name if valid_uid(sop_class) else None,
202 "uid": sop_class or None,
203 },
204 "transfer_syntax": {
205 "compressed": bool(transfer.is_compressed) if transfer else None,
206 "name": transfer.name if transfer else None,
207 "uid": transfer_value or None,
208 },
209 "warnings": warnings,
210 }
213def inventory(args: argparse.Namespace) -> dict[str, Any]:
214 max_input_bytes = parse_size(
215 args.max_input_bytes,
216 name="max_input_bytes",
217 maximum=HARD_MAX_INPUT_BYTES,
218 )
219 max_decompressed_bytes = parse_size(
220 args.max_decompressed_bytes,
221 name="max_decompressed_bytes",
222 maximum=HARD_MAX_DECOMPRESSED_BYTES,
223 )
224 max_files = bounded_int(
225 args.max_files,
226 name="max_files",
227 minimum=1,
228 maximum=HARD_MAX_FILES,
229 )
230 max_frames = bounded_int(
231 args.max_frames,
232 name="max_frames",
233 minimum=1,
234 maximum=HARD_MAX_FRAMES,
235 )
236 source = checked_input(
237 args.input,
238 root=args.root,
239 kind="any",
240 max_bytes=max_input_bytes,
241 )
242 files = collect_local_files(
243 source,
244 max_files=max_files,
245 max_bytes=max_input_bytes,
246 recursive=args.recursive,
247 )
248 records: list[dict[str, Any]] = []
249 parse_failures = 0
250 for index, path in enumerate(files, start=1):
251 file_id = f"file-{index:06d}"
252 try:
253 dataset = safe_dcmread(
254 path,
255 stop_before_pixels=True,
256 force=args.force_dicom_read,
257 specific_tags=INVENTORY_TAGS,
258 )
259 records.append(
260 inspect_dataset(
261 dataset,
262 file_id=file_id,
263 file_size=path.stat().st_size,
264 max_frames=max_frames,
265 max_decompressed_bytes=max_decompressed_bytes,
266 forced_read=args.force_dicom_read,
267 )
268 )
269 except ToolError:
270 parse_failures += 1
271 records.append(
272 {
273 "errors": 1,
274 "file_id": file_id,
275 "file_size_bytes": path.stat().st_size,
276 "issues": [
277 _issue(
278 "parse_failed",
279 "File could not be parsed as bounded DICOM metadata.",
280 )
281 ],
282 "ok": False,
283 "warnings": 0,
284 }
285 )
286 issue_codes: Counter[str] = Counter()
287 for record in records:
288 for issue in record["issues"]:
289 issue_codes[issue["code"]] += 1
290 error_files = sum(not record["ok"] for record in records)
291 return {
292 "aggregate": {
293 "error_files": error_files,
294 "files_examined": len(records),
295 "issue_codes": counter_dict(issue_codes),
296 "parse_failures": parse_failures,
297 "warning_files": sum(bool(record["warnings"]) for record in records),
298 },
299 "file_names_emitted": False,
300 "metadata_only": True,
301 "network_accessed": False,
302 "ok": error_files == 0 and bool(records),
303 "records": records,
304 "schema_version": SCHEMA_VERSION,
305 "technical_checks_only": True,
306 "tool": TOOL,
307 "warnings": [
308 "No pixel data was loaded or decompressed.",
309 "This is not complete IOD validation, clinical validation, or a diagnostic claim.",
310 "Patient Identity Removed=YES, if present, is reported but not trusted as proof.",
311 ],
312 }
315def build_parser() -> argparse.ArgumentParser:
316 parser = argparse.ArgumentParser(
317 description=(
318 "Perform bounded metadata-only DICOM File Format and image-pixel "
319 "technical checks without printing PHI values."
320 ),
321 formatter_class=argparse.RawDescriptionHelpFormatter,
322 epilog="""
323Local authorized data only. Metadata and pixels may contain PHI. This tool does
324not load pixels, validate every IOD rule, establish de-identification, or make
325diagnostic claims.
327Examples:
328 python dicom_inventory.py image.dcm
329 python dicom_inventory.py dicom-dir --recursive --output inventory.json
330""",
331 )
332 parser.add_argument("input", help="Local DICOM file or directory")
333 parser.add_argument("--root", default=".", help="Existing local I/O root")
334 parser.add_argument(
335 "--recursive", action="store_true", help="Recursively inspect directories"
336 )
337 parser.add_argument("--output", "-o", help="Private JSON report path")
338 parser.add_argument(
339 "--force-dicom-read",
340 action="store_true",
341 help="Parse missing File Format headers; records a warning",
342 )
343 parser.add_argument(
344 "--max-files",
345 type=int,
346 default=DEFAULT_MAX_FILES,
347 help=f"Maximum files (default: {DEFAULT_MAX_FILES})",
348 )
349 parser.add_argument(
350 "--max-frames",
351 type=int,
352 default=DEFAULT_MAX_FRAMES,
353 help=f"Maximum declared frames (default: {DEFAULT_MAX_FRAMES})",
354 )
355 parser.add_argument(
356 "--max-input-bytes",
357 default=str(DEFAULT_MAX_INPUT_BYTES),
358 help="Maximum bytes per input file",
359 )
360 parser.add_argument(
361 "--max-decompressed-bytes",
362 default=str(DEFAULT_MAX_DECOMPRESSED_BYTES),
363 help="Maximum estimated full uncompressed pixel bytes",
364 )
365 parser.add_argument(
366 "--force", action="store_true", help="Overwrite an existing report only"
367 )
368 return parser
371def main(argv: list[str] | None = None) -> int:
372 args = build_parser().parse_args(argv)
373 try:
374 report = inventory(args)
375 if args.output:
376 destination = checked_output(args.output, root=args.root, force=args.force)
377 source = checked_input(
378 args.input,
379 root=args.root,
380 kind="any",
381 max_bytes=parse_size(
382 args.max_input_bytes,
383 name="max_input_bytes",
384 maximum=HARD_MAX_INPUT_BYTES,
385 ),
386 )
387 if source.is_file() and paths_overlap(source, destination):
388 raise ToolError("report must not overwrite its input")
389 atomic_write(destination, json_bytes(report), force=args.force)
390 emit_json(
391 {
392 "ok": report["ok"],
393 "report_written": True,
394 "tool": TOOL,
395 }
396 )
397 else:
398 emit_json(report)
399 return 0 if report["ok"] else 1
400 except Exception as exc: # noqa: BLE001 - sanitize unexpected library errors
401 return fail_json(TOOL, exc)
404if __name__ == "__main__":
405 sys.exit(main())