Skills › Scientific Agent Skills › pydicom › transfer_syntax_inspector.py
transfer_syntax_inspector.py
245 lines · skills/pydicom/scripts/transfer_syntax_inspector.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inspect installed pydicom transfer-syntax decoder/encoder capabilities."""
4from __future__ import annotations
6import argparse
7import importlib.metadata
8import sys
9from pathlib import Path
10from typing import Any
12from _common import (
13 DEFAULT_MAX_INPUT_BYTES,
14 HARD_MAX_INPUT_BYTES,
15 SCHEMA_VERSION,
16 ToolError,
17 atomic_write,
18 checked_input,
19 checked_output,
20 emit_json,
21 fail_json,
22 json_bytes,
23 parse_size,
24 paths_overlap,
25 require_pydicom,
26 safe_dcmread,
27 valid_uid,
28)
30TOOL = "transfer_syntax_inspector"
31KNOWN_TRANSFER_SYNTAXES = (
32 "1.2.840.10008.1.2",
33 "1.2.840.10008.1.2.1",
34 "1.2.840.10008.1.2.1.99",
35 "1.2.840.10008.1.2.2",
36 "1.2.840.10008.1.2.4.50",
37 "1.2.840.10008.1.2.4.51",
38 "1.2.840.10008.1.2.4.57",
39 "1.2.840.10008.1.2.4.70",
40 "1.2.840.10008.1.2.4.80",
41 "1.2.840.10008.1.2.4.81",
42 "1.2.840.10008.1.2.4.90",
43 "1.2.840.10008.1.2.4.91",
44 "1.2.840.10008.1.2.4.201",
45 "1.2.840.10008.1.2.4.202",
46 "1.2.840.10008.1.2.4.203",
47 "1.2.840.10008.1.2.5",
48)
49PACKAGES = (
50 "pydicom",
51 "numpy",
52 "Pillow",
53 "pylibjpeg",
54 "pylibjpeg-libjpeg",
55 "pylibjpeg-openjpeg",
56 "pylibjpeg-rle",
57 "pyjpegls",
58 "python-gdcm",
59)
62def _package_versions() -> dict[str, str | None]:
63 versions: dict[str, str | None] = {}
64 for package in PACKAGES:
65 try:
66 versions[package] = importlib.metadata.version(package)
67 except importlib.metadata.PackageNotFoundError:
68 versions[package] = None
69 return versions
72def _codec_capability(uid_value: str) -> dict[str, Any]:
73 pydicom = require_pydicom()
74 from pydicom.pixels import get_decoder, get_encoder
76 uid = pydicom.uid.UID(uid_value)
77 try:
78 compressed: bool | None = bool(uid.is_compressed)
79 implicit_vr: bool | None = bool(uid.is_implicit_VR)
80 little_endian: bool | None = bool(uid.is_little_endian)
81 except ValueError:
82 compressed = None
83 implicit_vr = None
84 little_endian = None
85 record: dict[str, Any] = {
86 "compressed": compressed,
87 "decoder": {
88 "available": False,
89 "available_plugins": [],
90 "implemented": False,
91 "missing_dependencies": [],
92 },
93 "encoder": {
94 "available": False,
95 "available_plugins": [],
96 "implemented": False,
97 "missing_dependencies": [],
98 },
99 "implicit_vr": implicit_vr,
100 "little_endian": little_endian,
101 "name": uid.name,
102 "retired": uid_value == "1.2.840.10008.1.2.2",
103 "uid": uid_value,
104 }
105 try:
106 decoder = get_decoder(uid)
107 record["decoder"] = {
108 "available": bool(decoder.is_available),
109 "available_plugins": list(decoder.available_plugins),
110 "implemented": True,
111 "missing_dependencies": list(decoder.missing_dependencies),
112 }
113 except (NotImplementedError, ValueError):
114 pass
115 try:
116 encoder = get_encoder(uid)
117 record["encoder"] = {
118 "available": bool(encoder.is_available),
119 "available_plugins": list(encoder.available_plugins),
120 "implemented": True,
121 "missing_dependencies": list(encoder.missing_dependencies),
122 }
123 except (NotImplementedError, ValueError):
124 pass
125 return record
128def inspect(args: argparse.Namespace) -> dict[str, Any]:
129 requested = list(args.uid or [])
130 selected_uid: str | None = None
131 source: Path | None = None
132 if args.input:
133 max_bytes = parse_size(
134 args.max_input_bytes,
135 name="max_input_bytes",
136 maximum=HARD_MAX_INPUT_BYTES,
137 )
138 source = checked_input(
139 args.input,
140 root=args.root,
141 kind="file",
142 max_bytes=max_bytes,
143 )
144 dataset = safe_dcmread(source, stop_before_pixels=True, force=False)
145 selected_uid = str(dataset.file_meta.get("TransferSyntaxUID", ""))
146 if not valid_uid(selected_uid):
147 raise ToolError("input Transfer Syntax UID is absent or invalid")
148 requested.append(selected_uid)
149 if not requested:
150 requested.extend(KNOWN_TRANSFER_SYNTAXES)
151 unique: list[str] = []
152 for value in requested:
153 if not valid_uid(value):
154 raise ToolError("each --uid must be a valid numeric DICOM UID")
155 if value not in unique:
156 unique.append(value)
157 if len(unique) > 256:
158 raise ToolError("at most 256 transfer syntaxes may be inspected")
159 records = [_codec_capability(value) for value in unique]
160 return {
161 "capability_scope": (
162 "Installed plugin discovery only; image-specific bit depth, color, "
163 "platform, and codestream constraints still apply."
164 ),
165 "input_selected_uid": selected_uid,
166 "network_accessed": False,
167 "ok": True,
168 "package_versions": _package_versions(),
169 "pixel_data_loaded": False,
170 "records": records,
171 "schema_version": SCHEMA_VERSION,
172 "tool": TOOL,
173 "warnings": [
174 "An available plugin is not proof that a particular image will decode correctly.",
175 "Verify decoded pixels independently before scientific or clinical use.",
176 "Compression can change SOP Instance UID and image metadata; review the pydicom API behavior.",
177 ],
178 }
181def build_parser() -> argparse.ArgumentParser:
182 parser = argparse.ArgumentParser(
183 description=(
184 "Inspect pydicom 3.0.2 decoder/encoder implementation and installed "
185 "plugin availability without loading pixel data."
186 ),
187 formatter_class=argparse.RawDescriptionHelpFormatter,
188 epilog="""
189Examples:
190 python transfer_syntax_inspector.py
191 python transfer_syntax_inspector.py --input image.dcm
192 python transfer_syntax_inspector.py --uid 1.2.840.10008.1.2.4.90
194Availability is a deployment preflight, not a pixel-correctness guarantee.
195""",
196 )
197 parser.add_argument("--input", help="Optional local DICOM file")
198 parser.add_argument(
199 "--uid",
200 action="append",
201 help="Transfer Syntax UID to inspect; repeat as needed",
202 )
203 parser.add_argument("--root", default=".", help="Existing local I/O root")
204 parser.add_argument("--output", "-o", help="Private JSON report path")
205 parser.add_argument(
206 "--max-input-bytes",
207 default=str(DEFAULT_MAX_INPUT_BYTES),
208 help="Maximum optional DICOM input bytes",
209 )
210 parser.add_argument(
211 "--force", action="store_true", help="Overwrite an existing report only"
212 )
213 return parser
216def main(argv: list[str] | None = None) -> int:
217 args = build_parser().parse_args(argv)
218 try:
219 report = inspect(args)
220 if args.output:
221 destination = checked_output(args.output, root=args.root, force=args.force)
222 if args.input:
223 source = checked_input(
224 args.input,
225 root=args.root,
226 kind="file",
227 max_bytes=parse_size(
228 args.max_input_bytes,
229 name="max_input_bytes",
230 maximum=HARD_MAX_INPUT_BYTES,
231 ),
232 )
233 if paths_overlap(source, destination):
234 raise ToolError("report must not overwrite its input")
235 atomic_write(destination, json_bytes(report), force=args.force)
236 emit_json({"ok": True, "report_written": True, "tool": TOOL})
237 else:
238 emit_json(report)
239 return 0
240 except Exception as exc: # noqa: BLE001 - sanitize unexpected library errors
241 return fail_json(TOOL, exc)
244if __name__ == "__main__":
245 sys.exit(main())