Skills › Scientific Agent Skills › pydicom › pixel_frame_planner.py
pixel_frame_planner.py
313 lines · skills/pydicom/scripts/pixel_frame_planner.py at 5f2bc03
1#!/usr/bin/env python3
2"""Plan bounded DICOM frame decoding from metadata without loading pixels."""
4from __future__ import annotations
6import argparse
7import re
8import sys
9from typing import Any
11from _common import (
12 DEFAULT_MAX_DECOMPRESSED_BYTES,
13 DEFAULT_MAX_FRAMES,
14 DEFAULT_MAX_INPUT_BYTES,
15 HARD_MAX_DECOMPRESSED_BYTES,
16 HARD_MAX_FRAMES,
17 HARD_MAX_INPUT_BYTES,
18 SCHEMA_VERSION,
19 ToolError,
20 atomic_write,
21 bounded_int,
22 checked_input,
23 checked_output,
24 emit_json,
25 fail_json,
26 json_bytes,
27 parse_size,
28 paths_overlap,
29 pixel_plan,
30 require_pydicom,
31 safe_dcmread,
32 valid_uid,
33)
35TOOL = "pixel_frame_planner"
36FRAME_TOKEN = re.compile(r"^(\d+)(?:-(\d+))?$")
37PIXEL_TAGS = [
38 "Modality",
39 "Rows",
40 "Columns",
41 "NumberOfFrames",
42 "SamplesPerPixel",
43 "PhotometricInterpretation",
44 "PlanarConfiguration",
45 "BitsAllocated",
46 "BitsStored",
47 "HighBit",
48 "PixelRepresentation",
49 "ModalityLUTSequence",
50 "RescaleIntercept",
51 "RescaleSlope",
52 "RescaleType",
53 "VOILUTSequence",
54 "WindowCenter",
55 "WindowWidth",
56 "VOILUTFunction",
57 "SharedFunctionalGroupsSequence",
58 "PerFrameFunctionalGroupsSequence",
59 "RealWorldValueMappingSequence",
60 "ICCProfile",
61 "BurnedInAnnotation",
62 "RecognizableVisualFeatures",
63]
66def parse_frame_selection(
67 specification: str, *, total_frames: int, max_selected: int
68) -> list[int]:
69 if specification.strip().casefold() == "all":
70 if total_frames > max_selected:
71 raise ToolError("'all' exceeds the selected-frame limit")
72 return list(range(total_frames))
73 selected: set[int] = set()
74 for token in specification.split(","):
75 token = token.strip()
76 match = FRAME_TOKEN.fullmatch(token)
77 if not match:
78 raise ToolError(
79 "frames must use comma-separated indices or inclusive ranges"
80 )
81 start = int(match.group(1))
82 stop = int(match.group(2) or start)
83 if stop < start:
84 raise ToolError("frame ranges must be ascending")
85 if stop >= total_frames:
86 raise ToolError("selected frame is outside the declared range")
87 if stop - start + 1 > max_selected:
88 raise ToolError("one frame range exceeds the selected-frame limit")
89 selected.update(range(start, stop + 1))
90 if len(selected) > max_selected:
91 raise ToolError("selected-frame limit exceeded")
92 if not selected:
93 raise ToolError("at least one frame must be selected")
94 return sorted(selected)
97def plan(args: argparse.Namespace) -> dict[str, Any]:
98 max_input_bytes = parse_size(
99 args.max_input_bytes,
100 name="max_input_bytes",
101 maximum=HARD_MAX_INPUT_BYTES,
102 )
103 max_decompressed_bytes = parse_size(
104 args.max_decompressed_bytes,
105 name="max_decompressed_bytes",
106 maximum=HARD_MAX_DECOMPRESSED_BYTES,
107 )
108 max_frames = bounded_int(
109 args.max_frames,
110 name="max_frames",
111 minimum=1,
112 maximum=HARD_MAX_FRAMES,
113 )
114 max_selected = bounded_int(
115 args.max_selected_frames,
116 name="max_selected_frames",
117 minimum=1,
118 maximum=100_000,
119 )
120 source = checked_input(
121 args.input,
122 root=args.root,
123 kind="file",
124 max_bytes=max_input_bytes,
125 )
126 dataset = safe_dcmread(
127 source,
128 stop_before_pixels=True,
129 force=False,
130 specific_tags=PIXEL_TAGS,
131 )
132 metadata_plan = pixel_plan(
133 dataset,
134 max_frames=max_frames,
135 max_decompressed_bytes=max_decompressed_bytes,
136 )
137 selected = parse_frame_selection(
138 args.frames,
139 total_frames=metadata_plan["frames"],
140 max_selected=max_selected,
141 )
142 selected_bytes = metadata_plan["bytes_per_frame"] * len(selected)
143 if selected_bytes > max_decompressed_bytes:
144 raise ToolError("selected decoded frames exceed the configured byte limit")
146 pydicom = require_pydicom()
147 transfer_value = str(dataset.file_meta.get("TransferSyntaxUID", ""))
148 if not valid_uid(transfer_value):
149 raise ToolError("Transfer Syntax UID is absent or invalid")
150 transfer = pydicom.uid.UID(transfer_value)
151 try:
152 transfer_compressed = bool(transfer.is_compressed)
153 except ValueError as exc:
154 raise ToolError("UID is not recognized as a transfer syntax") from exc
155 photometric = str(dataset.get("PhotometricInterpretation", ""))
156 samples = metadata_plan["samples_per_pixel"]
157 one_frame_shape = (
158 [metadata_plan["rows"], metadata_plan["columns"]]
159 if samples == 1
160 else [metadata_plan["rows"], metadata_plan["columns"], samples]
161 )
162 transforms: list[str] = []
163 if (
164 "ModalityLUTSequence" in dataset
165 or "RescaleSlope" in dataset
166 or "RescaleIntercept" in dataset
167 ):
168 transforms.append("optional modality LUT/rescale")
169 if "VOILUTSequence" in dataset or (
170 "WindowCenter" in dataset and "WindowWidth" in dataset
171 ):
172 transforms.append("optional VOI LUT/window after modality transform")
173 if photometric in {"YBR_FULL", "YBR_FULL_422"}:
174 transforms.append("default pydicom YCbCr-to-RGB conversion unless raw=True")
175 elif photometric.startswith("YBR"):
176 transforms.append("codec-specific color conversion requires validation")
177 if photometric == "PALETTE COLOR":
178 transforms.append("palette color LUT required for RGB rendering")
179 if photometric == "MONOCHROME1":
180 transforms.append("presentation inversion may be required")
181 if any(keyword in dataset for keyword in (
182 "SharedFunctionalGroupsSequence", "PerFrameFunctionalGroupsSequence",
183 "RealWorldValueMappingSequence",
184 )):
185 transforms.append("frame-specific functional groups/real-world mapping require separate review")
187 return {
188 "burned_in_annotation": str(
189 dataset.get("BurnedInAnnotation", "ABSENT")
190 ).upper(),
191 "decode_plan": {
192 "all_frames_shape": metadata_plan["shape"],
193 "bytes_per_frame_estimate": metadata_plan["bytes_per_frame"],
194 "frame_indices": selected,
195 "one_frame_shape": one_frame_shape,
196 "selected_decoded_bytes_estimate": selected_bytes,
197 "use_iter_pixels": len(selected) > 1,
198 "use_pixel_array_index": len(selected) == 1,
199 },
200 "metadata": {
201 "bits_allocated": metadata_plan["bits_allocated"],
202 "columns": metadata_plan["columns"],
203 "frames": metadata_plan["frames"],
204 "modality": str(dataset.get("Modality", "UNSPECIFIED")),
205 "photometric_interpretation": photometric,
206 "rows": metadata_plan["rows"],
207 "samples_per_pixel": samples,
208 "transfer_syntax": {
209 "compressed": transfer_compressed,
210 "name": transfer.name,
211 "uid": str(transfer),
212 },
213 },
214 "network_accessed": False,
215 "ok": True,
216 "pixel_data_loaded": False,
217 "recognizable_visual_features": str(
218 dataset.get("RecognizableVisualFeatures", "ABSENT")
219 ).upper(),
220 "schema_version": SCHEMA_VERSION,
221 "tool": TOOL,
222 "transforms_in_semantic_order": transforms,
223 "warnings": [
224 *metadata_plan["warnings"],
225 "Estimates use metadata and do not prove that the codestream matches it.",
226 "Pixel decoding can expose PHI and requires independent correctness checks.",
227 "Frame rendering is non-diagnostic unless validated in an appropriate system.",
228 ],
229 }
232def build_parser() -> argparse.ArgumentParser:
233 parser = argparse.ArgumentParser(
234 description=(
235 "Plan bounded DICOM pixel frame access, shapes, byte limits, and "
236 "display transforms from metadata without decoding pixels."
237 ),
238 formatter_class=argparse.RawDescriptionHelpFormatter,
239 epilog="""
240Examples:
241 python pixel_frame_planner.py image.dcm
242 python pixel_frame_planner.py multi.dcm --frames 0,2-4
243 python pixel_frame_planner.py multi.dcm --frames all --max-selected-frames 20
245Metadata can be inconsistent with compressed codestreams. This is a resource
246and semantics plan, not diagnostic or pixel-correctness validation.
247""",
248 )
249 parser.add_argument("input", help="Local DICOM input file")
250 parser.add_argument("--root", default=".", help="Existing local I/O root")
251 parser.add_argument(
252 "--frames",
253 default="0",
254 help="Comma-separated zero-based indices/ranges, or 'all' (default: 0)",
255 )
256 parser.add_argument("--output", "-o", help="Private JSON plan path")
257 parser.add_argument(
258 "--max-frames",
259 type=int,
260 default=DEFAULT_MAX_FRAMES,
261 help=f"Maximum declared frames (default: {DEFAULT_MAX_FRAMES})",
262 )
263 parser.add_argument(
264 "--max-selected-frames",
265 type=int,
266 default=100,
267 help="Maximum selected frames (default: 100)",
268 )
269 parser.add_argument(
270 "--max-input-bytes",
271 default=str(DEFAULT_MAX_INPUT_BYTES),
272 help="Maximum DICOM input bytes",
273 )
274 parser.add_argument(
275 "--max-decompressed-bytes",
276 default=str(DEFAULT_MAX_DECOMPRESSED_BYTES),
277 help="Maximum estimated decoded bytes",
278 )
279 parser.add_argument(
280 "--force", action="store_true", help="Overwrite an existing plan only"
281 )
282 return parser
285def main(argv: list[str] | None = None) -> int:
286 args = build_parser().parse_args(argv)
287 try:
288 report = plan(args)
289 if args.output:
290 destination = checked_output(args.output, root=args.root, force=args.force)
291 source = checked_input(
292 args.input,
293 root=args.root,
294 kind="file",
295 max_bytes=parse_size(
296 args.max_input_bytes,
297 name="max_input_bytes",
298 maximum=HARD_MAX_INPUT_BYTES,
299 ),
300 )
301 if paths_overlap(source, destination):
302 raise ToolError("plan must not overwrite its input")
303 atomic_write(destination, json_bytes(report), force=args.force)
304 emit_json({"ok": True, "plan_written": True, "tool": TOOL})
305 else:
306 emit_json(report)
307 return 0
308 except Exception as exc: # noqa: BLE001 - sanitize unexpected library errors
309 return fail_json(TOOL, exc)
312if __name__ == "__main__":
313 sys.exit(main())