Skills › Scientific Agent Skills › pydicom › deidentification_audit.py
deidentification_audit.py
380 lines · skills/pydicom/scripts/deidentification_audit.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit DICOM metadata for bounded de-identification review signals."""
4from __future__ import annotations
6import argparse
7import sys
8from collections import Counter
9from typing import Any
11from _common import (
12 DEFAULT_MAX_ELEMENTS,
13 DEFAULT_MAX_FILES,
14 DEFAULT_MAX_INPUT_BYTES,
15 HARD_MAX_ELEMENTS,
16 HARD_MAX_FILES,
17 HARD_MAX_INPUT_BYTES,
18 MAX_SEQUENCE_DEPTH,
19 SCHEMA_VERSION,
20 SENSITIVE_KEYWORDS,
21 STRUCTURAL_UID_KEYWORDS,
22 TEXT_VRS,
23 UID_REMAP_KEYWORDS,
24 ToolError,
25 atomic_write,
26 bounded_int,
27 checked_input,
28 checked_output,
29 collect_local_files,
30 counter_dict,
31 element_tag,
32 emit_json,
33 fail_json,
34 json_bytes,
35 load_json,
36 parse_size,
37 paths_overlap,
38 safe_dcmread,
39 starter_profile,
40 validate_profile,
41)
43TOOL = "deidentification_audit"
44PIXEL_METADATA = {
45 "BitsAllocated",
46 "BitsStored",
47 "Columns",
48 "DoubleFloatPixelData",
49 "FloatPixelData",
50 "NumberOfFrames",
51 "PhotometricInterpretation",
52 "PixelData",
53 "Rows",
54 "SamplesPerPixel",
55}
56BULK_OR_CONTENT_KEYWORDS = {
57 "AcquisitionContextSequence",
58 "AudioSampleData",
59 "ContentSequence",
60 "EncapsulatedDocument",
61 "GraphicAnnotationSequence",
62 "IconImageSequence",
63 "SpectroscopyData",
64 "WaveformData",
65}
68def _has_value(element: Any) -> bool:
69 value = element.value
70 if value is None:
71 return False
72 if isinstance(value, (str, bytes)):
73 return bool(value)
74 try:
75 return len(value) > 0
76 except TypeError:
77 return True
80def audit_dataset(
81 dataset: Any,
82 *,
83 profile: dict[str, Any],
84 file_id: str,
85 max_elements: int,
86) -> dict[str, Any]:
87 actions = profile["actions"]
88 categories: Counter[str] = Counter()
89 residual_keywords: Counter[str] = Counter()
90 unclassified_text: Counter[str] = Counter()
91 unclassified_uids: Counter[str] = Counter()
92 stack: list[tuple[Any, int]] = [(dataset, 0)]
93 elements = 0
94 image_metadata_present = False
95 while stack:
96 current, depth = stack.pop()
97 if depth > MAX_SEQUENCE_DEPTH:
98 raise ToolError("sequence nesting exceeds the hard depth limit")
99 for element in current:
100 elements += 1
101 if elements > max_elements:
102 raise ToolError("data-element limit exceeded")
103 keyword = element.keyword or element_tag(element)
104 vr = str(element.VR)
105 has_value = _has_value(element)
107 if keyword in PIXEL_METADATA:
108 image_metadata_present = True
109 if keyword in BULK_OR_CONTENT_KEYWORDS:
110 categories["bulk_or_structured_content_elements"] += 1
111 if element.tag.is_private:
112 categories["private_elements"] += 1
113 if has_value:
114 categories["nonempty_private_elements"] += 1
115 group = int(element.tag.group)
116 if 0x5000 <= group <= 0x50FF or 0x6000 <= group <= 0x60FF:
117 categories["curve_or_overlay_elements"] += 1
119 action = actions.get(keyword, actions.get(element_tag(element)))
120 if action in {"remove", "empty"} and has_value:
121 residual_keywords[keyword] += 1
122 categories["profile_action_residuals"] += 1
123 elif action == "pseudonym" and has_value:
124 categories["pseudonym_values_requiring_provenance_review"] += 1
125 elif action == "keep" and has_value:
126 categories["explicit_profile_retention"] += 1
128 if vr == "SQ":
129 for item in reversed(list(element.value)):
130 stack.append((item, depth + 1))
131 continue
132 if not has_value:
133 continue
134 if vr in {"DA", "DT", "TM"}:
135 categories["nonempty_date_or_time_elements"] += 1
136 if vr == "UI":
137 if keyword in STRUCTURAL_UID_KEYWORDS:
138 categories["structural_uid_elements"] += 1
139 elif keyword in UID_REMAP_KEYWORDS or keyword.endswith(
140 ("InstanceUID", "FrameOfReferenceUID")
141 ):
142 categories["instance_uid_elements"] += 1
143 else:
144 unclassified_uids[keyword] += 1
145 if keyword in SENSITIVE_KEYWORDS:
146 categories["known_sensitive_elements_present"] += 1
147 elif vr in TEXT_VRS and keyword not in {"SpecificCharacterSet"}:
148 unclassified_text[keyword] += 1
150 burned_in = str(dataset.get("BurnedInAnnotation", "")).strip().upper()
151 recognizable = str(dataset.get("RecognizableVisualFeatures", "")).strip().upper()
152 identity_removed = str(dataset.get("PatientIdentityRemoved", "")).strip().upper()
153 pixel_review_required = image_metadata_present
154 high_risk_findings = (
155 categories["nonempty_private_elements"]
156 + categories["profile_action_residuals"]
157 + categories["curve_or_overlay_elements"]
158 + categories["bulk_or_structured_content_elements"]
159 + len(unclassified_text)
160 + len(unclassified_uids)
161 )
162 if image_metadata_present and burned_in != "NO":
163 high_risk_findings += 1
164 return {
165 "burned_in_annotation": burned_in or "ABSENT",
166 "categories": counter_dict(categories),
167 "elements_examined": elements,
168 "file_id": file_id,
169 "high_risk_findings": high_risk_findings,
170 "patient_identity_removed_claim": identity_removed or "ABSENT",
171 "pixel_review_required": pixel_review_required,
172 "profile_action_residuals_by_keyword": counter_dict(residual_keywords),
173 "recognizable_visual_features": recognizable or "ABSENT",
174 "review_passed": high_risk_findings == 0 and not pixel_review_required,
175 "unclassified_text_by_keyword": counter_dict(unclassified_text),
176 "unclassified_uids_by_keyword": counter_dict(unclassified_uids),
177 }
180def build_report(args: argparse.Namespace) -> dict[str, Any]:
181 max_input_bytes = parse_size(
182 args.max_input_bytes,
183 name="max_input_bytes",
184 maximum=HARD_MAX_INPUT_BYTES,
185 )
186 max_files = bounded_int(
187 args.max_files,
188 name="max_files",
189 minimum=1,
190 maximum=HARD_MAX_FILES,
191 )
192 max_elements = bounded_int(
193 args.max_elements,
194 name="max_elements",
195 minimum=1,
196 maximum=HARD_MAX_ELEMENTS,
197 )
198 source = checked_input(
199 args.input,
200 root=args.root,
201 kind="any",
202 max_bytes=max_input_bytes,
203 )
204 files = collect_local_files(
205 source,
206 max_files=max_files,
207 max_bytes=max_input_bytes,
208 recursive=args.recursive,
209 )
210 profile = starter_profile()
211 profile_source = "built-in"
212 if args.profile_json:
213 profile_path = checked_input(
214 args.profile_json,
215 root=args.root,
216 kind="file",
217 max_bytes=1 * 1024 * 1024,
218 )
219 profile = validate_profile(load_json(profile_path))
220 profile_source = "local-json"
221 records: list[dict[str, Any]] = []
222 parse_failures = 0
223 for index, path in enumerate(files, start=1):
224 try:
225 dataset = safe_dcmread(
226 path,
227 stop_before_pixels=True,
228 force=args.force_dicom_read,
229 )
230 records.append(
231 audit_dataset(
232 dataset,
233 profile=profile,
234 file_id=f"file-{index:06d}",
235 max_elements=max_elements,
236 )
237 )
238 except ToolError:
239 parse_failures += 1
240 records.append(
241 {
242 "file_id": f"file-{index:06d}",
243 "high_risk_findings": 1,
244 "parse_failed": True,
245 "pixel_review_required": True,
246 "review_passed": False,
247 }
248 )
249 category_totals: Counter[str] = Counter()
250 for record in records:
251 category_totals.update(record.get("categories", {}))
252 review_failures = sum(not record["review_passed"] for record in records)
253 return {
254 "aggregate": {
255 "categories": counter_dict(category_totals),
256 "files_examined": len(records),
257 "parse_failures": parse_failures,
258 "pixel_review_files": sum(
259 bool(record.get("pixel_review_required")) for record in records
260 ),
261 "review_failure_files": review_failures,
262 },
263 "file_names_emitted": False,
264 "metadata_only": True,
265 "network_accessed": False,
266 "not_a_compliance_determination": True,
267 "ok": bool(records),
268 "profile": {
269 "name": profile["name"],
270 "source": profile_source,
271 "version": profile["version"],
272 },
273 "records": records,
274 "schema_version": SCHEMA_VERSION,
275 "tool": TOOL,
276 "warnings": [
277 "Attribute auditing cannot establish that an Information Object is de-identified.",
278 "Patient Identity Removed=YES is not treated as proof.",
279 "Pixels, recognizable visual features, graphics, overlays, structured content, private data, and external context require expert review.",
280 "Profile and regulatory applicability are purpose, recipient, jurisdiction, and risk specific.",
281 ],
282 }
285def build_parser() -> argparse.ArgumentParser:
286 parser = argparse.ArgumentParser(
287 description=(
288 "Audit bounded DICOM metadata for residual identifier categories; "
289 "never claims de-identification or regulatory compliance."
290 ),
291 formatter_class=argparse.RawDescriptionHelpFormatter,
292 epilog="""
293Local authorized data only. This metadata-only audit never decompresses pixels.
294It is one input to a context-specific expert de-identification review, not a
295replacement for DICOM PS3.15 profile selection or re-identification risk analysis.
297Examples:
298 python deidentification_audit.py candidate.dcm
299 python deidentification_audit.py export-dir --recursive --output audit.json
300 python deidentification_audit.py candidate.dcm --profile-json site-profile.json
301""",
302 )
303 parser.add_argument("input", help="Local DICOM file or directory")
304 parser.add_argument("--root", default=".", help="Existing local I/O root")
305 parser.add_argument(
306 "--recursive", action="store_true", help="Recursively inspect directories"
307 )
308 parser.add_argument("--profile-json", help="Local bounded action-profile JSON")
309 parser.add_argument("--output", "-o", help="Private JSON report path")
310 parser.add_argument(
311 "--force-dicom-read",
312 action="store_true",
313 help="Parse missing File Format headers; does not validate them",
314 )
315 parser.add_argument(
316 "--fail-on-findings",
317 action="store_true",
318 help="Exit 1 when any file requires review",
319 )
320 parser.add_argument(
321 "--max-files",
322 type=int,
323 default=DEFAULT_MAX_FILES,
324 help=f"Maximum files (default: {DEFAULT_MAX_FILES})",
325 )
326 parser.add_argument(
327 "--max-elements",
328 type=int,
329 default=DEFAULT_MAX_ELEMENTS,
330 help=f"Maximum elements per file (default: {DEFAULT_MAX_ELEMENTS})",
331 )
332 parser.add_argument(
333 "--max-input-bytes",
334 default=str(DEFAULT_MAX_INPUT_BYTES),
335 help="Maximum bytes per input file",
336 )
337 parser.add_argument(
338 "--force", action="store_true", help="Overwrite an existing report only"
339 )
340 return parser
343def main(argv: list[str] | None = None) -> int:
344 args = build_parser().parse_args(argv)
345 try:
346 report = build_report(args)
347 if args.output:
348 destination = checked_output(args.output, root=args.root, force=args.force)
349 source = checked_input(
350 args.input,
351 root=args.root,
352 kind="any",
353 max_bytes=parse_size(
354 args.max_input_bytes,
355 name="max_input_bytes",
356 maximum=HARD_MAX_INPUT_BYTES,
357 ),
358 )
359 if source.is_file() and paths_overlap(source, destination):
360 raise ToolError("report must not overwrite its input")
361 atomic_write(destination, json_bytes(report), force=args.force)
362 emit_json(
363 {
364 "ok": report["ok"],
365 "report_written": True,
366 "tool": TOOL,
367 }
368 )
369 else:
370 emit_json(report)
371 findings = report["aggregate"]["review_failure_files"]
372 if args.fail_on_findings and findings:
373 return 1
374 return 0 if report["ok"] else 1
375 except Exception as exc: # noqa: BLE001 - sanitize unexpected library errors
376 return fail_json(TOOL, exc)
379if __name__ == "__main__":
380 sys.exit(main())