Skills › Scientific Agent Skills › pydicom › extract_metadata.py
extract_metadata.py
339 lines · skills/pydicom/scripts/extract_metadata.py at 5f2bc03
1#!/usr/bin/env python3
2"""Emit a redacted, allowlisted technical DICOM metadata inventory."""
4from __future__ import annotations
6import argparse
7import sys
8from collections import Counter
9from typing import Any
11from _common import (
12 DEFAULT_MAX_FILES,
13 DEFAULT_MAX_INPUT_BYTES,
14 HARD_MAX_FILES,
15 HARD_MAX_INPUT_BYTES,
16 SCHEMA_VERSION,
17 TECHNICAL_KEYWORDS,
18 ToolError,
19 atomic_write,
20 bounded_int,
21 checked_input,
22 checked_output,
23 collect_local_files,
24 counter_dict,
25 emit_json,
26 fail_json,
27 frame_count,
28 json_bytes,
29 parse_size,
30 paths_overlap,
31 require_pydicom,
32 safe_dcmread,
33)
35TOOL = "extract_metadata"
38def _uid_description(value: Any) -> dict[str, str] | None:
39 if not value:
40 return None
41 pydicom = require_pydicom()
42 uid = pydicom.uid.UID(str(value))
43 return {"name": uid.name, "uid": str(uid)}
46def _status(value: Any) -> str:
47 text = str(value or "").strip().upper()
48 if not text:
49 return "ABSENT"
50 if text in {"YES", "NO"}:
51 return text
52 return "OTHER"
55def _lossy_status(value: Any) -> str:
56 # PS3.3 C.7.6.1.1.5 uses CS values 00/01, not YES/NO.
57 text = str(value or "").strip()
58 return text if text in {"00", "01"} else "ABSENT" if not text else "OTHER"
61def allowlisted_record(dataset: Any, *, file_id: str) -> dict[str, Any]:
62 """Return only non-identifying technical fields from a metadata-only read."""
64 pydicom = require_pydicom()
65 transfer = dataset.file_meta.get("TransferSyntaxUID")
66 transfer_info = _uid_description(transfer)
67 sop_info = _uid_description(dataset.get("SOPClassUID"))
68 frames, frame_warnings = frame_count(dataset)
69 record: dict[str, Any] = {
70 "bits_allocated": (
71 int(dataset.BitsAllocated) if "BitsAllocated" in dataset else None
72 ),
73 "bits_stored": int(dataset.BitsStored) if "BitsStored" in dataset else None,
74 "burned_in_annotation": _status(dataset.get("BurnedInAnnotation")),
75 "columns": int(dataset.Columns) if "Columns" in dataset else None,
76 "file_id": file_id,
77 "frames": frames,
78 "has_modality_transform": bool(
79 "ModalityLUTSequence" in dataset
80 or "RescaleSlope" in dataset
81 or "RescaleIntercept" in dataset
82 ),
83 "has_voi_transform": bool(
84 "VOILUTSequence" in dataset
85 or "WindowCenter" in dataset
86 or "WindowWidth" in dataset
87 ),
88 "high_bit": int(dataset.HighBit) if "HighBit" in dataset else None,
89 "lossy_image_compression": _lossy_status(dataset.get("LossyImageCompression")),
90 "modality": str(dataset.get("Modality", "UNSPECIFIED")),
91 "photometric_interpretation": str(
92 dataset.get("PhotometricInterpretation", "UNSPECIFIED")
93 ),
94 "pixel_representation": (
95 int(dataset.PixelRepresentation)
96 if "PixelRepresentation" in dataset
97 else None
98 ),
99 "recognizable_visual_features": _status(
100 dataset.get("RecognizableVisualFeatures")
101 ),
102 "rows": int(dataset.Rows) if "Rows" in dataset else None,
103 "samples_per_pixel": (
104 int(dataset.SamplesPerPixel) if "SamplesPerPixel" in dataset else None
105 ),
106 "sop_class": sop_info,
107 "transfer_syntax": transfer_info,
108 "warnings": frame_warnings,
109 }
110 if transfer_info is not None:
111 uid = pydicom.uid.UID(transfer_info["uid"])
112 record["transfer_syntax"]["compressed"] = (
113 bool(uid.is_compressed) if uid.is_transfer_syntax else None
114 )
115 if not uid.is_transfer_syntax:
116 record["warnings"].append("Transfer Syntax UID is not recognized.")
117 return record
120def aggregate_records(
121 records: list[dict[str, Any]], *, read_failures: int
122) -> dict[str, Any]:
123 modalities: Counter[str] = Counter()
124 sop_classes: Counter[str] = Counter()
125 transfer_syntaxes: Counter[str] = Counter()
126 photometric: Counter[str] = Counter()
127 burned_in: Counter[str] = Counter()
128 recognizable: Counter[str] = Counter()
129 compressed = 0
130 image_like = 0
131 total_frames = 0
132 for record in records:
133 modalities[record["modality"]] += 1
134 photometric[record["photometric_interpretation"]] += 1
135 burned_in[record["burned_in_annotation"]] += 1
136 recognizable[record["recognizable_visual_features"]] += 1
137 if record["sop_class"]:
138 sop_classes[record["sop_class"]["name"]] += 1
139 if record["transfer_syntax"]:
140 transfer_syntaxes[record["transfer_syntax"]["name"]] += 1
141 compressed += int(record["transfer_syntax"].get("compressed", False))
142 if record["rows"] is not None and record["columns"] is not None:
143 image_like += 1
144 total_frames += record["frames"]
145 return {
146 "burned_in_annotation": counter_dict(burned_in),
147 "compressed_files": compressed,
148 "image_like_files": image_like,
149 "modalities": counter_dict(modalities),
150 "photometric_interpretations": counter_dict(photometric),
151 "read_failures": read_failures,
152 "recognizable_visual_features": counter_dict(recognizable),
153 "sop_classes": counter_dict(sop_classes),
154 "successful_files": len(records),
155 "total_frames_declared": total_frames,
156 "transfer_syntaxes": counter_dict(transfer_syntaxes),
157 }
160def build_report(args: argparse.Namespace) -> dict[str, Any]:
161 max_bytes = parse_size(
162 args.max_input_bytes,
163 name="max_input_bytes",
164 maximum=HARD_MAX_INPUT_BYTES,
165 )
166 max_files = bounded_int(
167 args.max_files,
168 name="max_files",
169 minimum=1,
170 maximum=HARD_MAX_FILES,
171 )
172 source = checked_input(
173 args.input,
174 root=args.root,
175 kind="any",
176 max_bytes=max_bytes,
177 )
178 files = collect_local_files(
179 source,
180 max_files=max_files,
181 max_bytes=max_bytes,
182 recursive=args.recursive,
183 )
184 records: list[dict[str, Any]] = []
185 read_failures = 0
186 tags = list(TECHNICAL_KEYWORDS) + ["ModalityLUTSequence", "VOILUTSequence"]
187 for index, path in enumerate(files, start=1):
188 try:
189 dataset = safe_dcmread(
190 path,
191 stop_before_pixels=True,
192 force=args.force_dicom_read,
193 specific_tags=tags,
194 )
195 records.append(allowlisted_record(dataset, file_id=f"file-{index:06d}"))
196 except ToolError:
197 read_failures += 1
198 aggregate = aggregate_records(records, read_failures=read_failures)
199 report: dict[str, Any] = {
200 "aggregate": aggregate,
201 "allowlist_only": True,
202 "file_names_emitted": False,
203 "full_metadata_dump_supported": False,
204 "network_accessed": False,
205 "ok": bool(records),
206 "phi_values_emitted": False,
207 "schema_version": SCHEMA_VERSION,
208 "technical_inventory_only": True,
209 "tool": TOOL,
210 "warnings": [
211 "DICOM metadata and pixels may contain PHI.",
212 "This report omits patient, study, series, instance, date, and free-text identifiers.",
213 "Technical inventory is not DICOM conformance or diagnostic validation.",
214 ],
215 }
216 if args.per_file:
217 report["records"] = records
218 return report
221def render_text(report: dict[str, Any]) -> str:
222 aggregate = report["aggregate"]
223 lines = [
224 "Redacted DICOM technical inventory",
225 f"Successful files: {aggregate['successful_files']}",
226 f"Read failures: {aggregate['read_failures']}",
227 f"Image-like files: {aggregate['image_like_files']}",
228 f"Compressed files: {aggregate['compressed_files']}",
229 f"Declared frames: {aggregate['total_frames_declared']}",
230 "No file names, patient/study/series/instance identifiers, dates, or free text emitted.",
231 "Not a DICOM conformance or diagnostic report.",
232 ]
233 for heading, key in (
234 ("Modalities", "modalities"),
235 ("SOP classes", "sop_classes"),
236 ("Transfer syntaxes", "transfer_syntaxes"),
237 ):
238 lines.append(f"{heading}:")
239 values = aggregate[key]
240 if not values:
241 lines.append(" (none)")
242 for name, count in values.items():
243 lines.append(f" {name}: {count}")
244 return "\n".join(lines) + "\n"
247def build_parser() -> argparse.ArgumentParser:
248 parser = argparse.ArgumentParser(
249 description=(
250 "Create a redacted allowlisted technical DICOM inventory; never "
251 "prints a full metadata dump or known PHI fields."
252 ),
253 formatter_class=argparse.RawDescriptionHelpFormatter,
254 epilog="""
255Local authorized data only. DICOM metadata and pixels may contain PHI.
256The default is aggregate JSON with no file names or per-instance identifiers.
258Examples:
259 python extract_metadata.py image.dcm
260 python extract_metadata.py dicom-dir --recursive --output inventory.json
261 python extract_metadata.py image.dcm --per-file --format text
262""",
263 )
264 parser.add_argument("input", help="Local DICOM file or directory")
265 parser.add_argument("--root", default=".", help="Existing local I/O root")
266 parser.add_argument(
267 "--recursive", action="store_true", help="Recursively inspect a directory"
268 )
269 parser.add_argument(
270 "--per-file",
271 action="store_true",
272 help="Include allowlisted records keyed by synthetic file IDs",
273 )
274 parser.add_argument(
275 "--format", choices=("json", "text"), default="json", help="Output format"
276 )
277 parser.add_argument("--output", "-o", help="Private local report file")
278 parser.add_argument(
279 "--force-dicom-read",
280 action="store_true",
281 help="Parse datasets lacking a File Format header; does not validate them",
282 )
283 parser.add_argument(
284 "--max-files",
285 type=int,
286 default=DEFAULT_MAX_FILES,
287 help=f"Maximum files to inspect (default: {DEFAULT_MAX_FILES})",
288 )
289 parser.add_argument(
290 "--max-input-bytes",
291 default=str(DEFAULT_MAX_INPUT_BYTES),
292 help="Maximum bytes per input file (integer or B/KiB/MiB/GiB)",
293 )
294 parser.add_argument(
295 "--force", action="store_true", help="Overwrite an existing report only"
296 )
297 return parser
300def main(argv: list[str] | None = None) -> int:
301 args = build_parser().parse_args(argv)
302 try:
303 report = build_report(args)
304 if args.format == "json":
305 payload = json_bytes(report)
306 else:
307 payload = render_text(report).encode("utf-8")
308 if args.output:
309 destination = checked_output(args.output, root=args.root, force=args.force)
310 source = checked_input(
311 args.input,
312 root=args.root,
313 kind="any",
314 max_bytes=parse_size(
315 args.max_input_bytes,
316 name="max_input_bytes",
317 maximum=HARD_MAX_INPUT_BYTES,
318 ),
319 )
320 if source.is_file() and paths_overlap(source, destination):
321 raise ToolError("report must not overwrite its input")
322 atomic_write(destination, payload, force=args.force)
323 emit_json(
324 {
325 "network_accessed": False,
326 "ok": report["ok"],
327 "report_written": True,
328 "tool": TOOL,
329 }
330 )
331 else:
332 sys.stdout.buffer.write(payload)
333 return 0 if report["ok"] else 1
334 except Exception as exc: # noqa: BLE001 - sanitize unexpected library errors
335 return fail_json(TOOL, exc)
338if __name__ == "__main__":
339 sys.exit(main())