Skills › Scientific Agent Skills › pydicom › _common.py
_common.py
916 lines · skills/pydicom/scripts/_common.py at 5f2bc03
1"""Shared, dependency-light safety helpers for the pydicom skill CLIs."""
3from __future__ import annotations
5import hashlib
6import hmac
7import json
8import os
9import re
10import stat
11import sys
12import tempfile
13from collections import Counter
14from collections.abc import Callable, Iterable, Mapping
15from pathlib import Path, PurePath
16from typing import Any
18SCHEMA_VERSION = "1.1"
19PYDICOM_VERSION = "3.0.2"
20NUMPY_VERSION = "2.5.3"
21PILLOW_VERSION = "12.3.0"
23KIB = 1024
24MIB = 1024**2
25GIB = 1024**3
26DEFAULT_MAX_INPUT_BYTES = 512 * MIB
27HARD_MAX_INPUT_BYTES = 16 * GIB
28DEFAULT_MAX_OUTPUT_BYTES = 512 * MIB
29HARD_MAX_OUTPUT_BYTES = 16 * GIB
30DEFAULT_MAX_DECOMPRESSED_BYTES = 256 * MIB
31HARD_MAX_DECOMPRESSED_BYTES = 4 * GIB
32DEFAULT_MAX_FILES = 1_000
33HARD_MAX_FILES = 100_000
34DEFAULT_MAX_ELEMENTS = 200_000
35HARD_MAX_ELEMENTS = 2_000_000
36DEFAULT_MAX_FRAMES = 1_000
37HARD_MAX_FRAMES = 1_000_000
38MAX_JSON_BYTES = 8 * MIB
39MAX_REPORT_BYTES = 32 * MIB
40MAX_SEQUENCE_DEPTH = 64
42_URI_PREFIXES = (
43 "http:",
44 "https:",
45 "ftp:",
46 "file:",
47 "s3:",
48 "gs:",
49 "ssh:",
50 "data:",
51)
52_SIZE_PATTERN = re.compile(r"^\s*(\d+)\s*(B|KIB|MIB|GIB)?\s*$", re.IGNORECASE)
53_UID_PATTERN = re.compile(r"^[0-9]+(?:\.[0-9]+)*$")
54_SAFE_PLUGIN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")
56TEXT_VRS = frozenset(
57 {
58 "AE",
59 "AS",
60 "CS",
61 "DA",
62 "DT",
63 "LO",
64 "LT",
65 "PN",
66 "SH",
67 "ST",
68 "TM",
69 "UC",
70 "UI",
71 "UR",
72 "UT",
73 }
74)
76# These are structural or coding UIDs, not instance identifiers. They must not be
77# replaced merely because their VR is UI.
78STRUCTURAL_UID_KEYWORDS = frozenset(
79 {
80 "AffectedSOPClassUID",
81 "CodingSchemeUID",
82 "ContextGroupExtensionCreatorUID",
83 "ContextGroupLocalVersion",
84 "ImplementationClassUID",
85 "MediaStorageSOPClassUID",
86 "PrivateInformationCreatorUID",
87 "ReferencedSOPClassUID",
88 "RelatedGeneralSOPClassUID",
89 "RequestedSOPClassUID",
90 "SOPClassUID",
91 "TransferSyntaxUID",
92 }
93)
95UID_REMAP_KEYWORDS = frozenset(
96 {
97 "AcquisitionUID",
98 "ConcatenationUID",
99 "ContrastBolusAgentNumber",
100 "DeviceUID",
101 "DimensionOrganizationUID",
102 "DoseReferenceUID",
103 "FiducialUID",
104 "FrameOfReferenceUID",
105 "IrradiationEventUID",
106 "MediaStorageSOPInstanceUID",
107 "ObservationUID",
108 "ReferencedDoseReferenceUID",
109 "ReferencedFiducialsUID",
110 "ReferencedFrameOfReferenceUID",
111 "ReferencedGeneralPurposeScheduledProcedureStepTransactionUID",
112 "ReferencedObservationUID",
113 "ReferencedSOPInstanceUID",
114 "SeriesInstanceUID",
115 "SOPInstanceUID",
116 "SpecimenUID",
117 "StorageMediaFileSetUID",
118 "StudyInstanceUID",
119 "SynchronizationFrameOfReferenceUID",
120 "TemplateExtensionCreatorUID",
121 "TemplateExtensionOrganizationUID",
122 "TrackingUID",
123 "TransactionUID",
124 "TreatmentPositionGroupUID",
125 "TreatmentSessionUID",
126 "UID",
127 }
128)
130# A conservative starter profile. It is intentionally not represented as a
131# DICOM PS3.15 conformance implementation.
132DEFAULT_ACTIONS: dict[str, str] = {
133 "AccessionNumber": "empty",
134 "AcquisitionContextSequence": "remove",
135 "AcquisitionComments": "remove",
136 "AdmittingDiagnosesCodeSequence": "remove",
137 "AdmittingDiagnosesDescription": "remove",
138 "Allergies": "remove",
139 "ClinicalTrialProtocolID": "remove",
140 "ClinicalTrialProtocolName": "remove",
141 "ClinicalTrialSiteID": "remove",
142 "ClinicalTrialSiteName": "remove",
143 "ClinicalTrialSponsorName": "remove",
144 "ClinicalTrialSubjectID": "pseudonym",
145 "ClinicalTrialSubjectReadingID": "pseudonym",
146 "ContentSequence": "remove",
147 "CurrentPatientLocation": "remove",
148 "DataSetTrailingPadding": "remove",
149 "DeidentificationMethod": "remove",
150 "DeidentificationMethodCodeSequence": "remove",
151 "DerivationDescription": "remove",
152 "DeviceSerialNumber": "remove",
153 "DigitalSignaturesSequence": "remove",
154 "DocumentTitle": "remove",
155 "EncryptedAttributesSequence": "remove",
156 "EncapsulatedDocument": "remove",
157 "EncapsulatedDocumentLength": "remove",
158 "EthnicGroup": "remove",
159 "FillerOrderNumberImagingServiceRequest": "empty",
160 "GraphicAnnotationSequence": "remove",
161 "HL7InstanceIdentifier": "remove",
162 "IconImageSequence": "remove",
163 "ImageComments": "remove",
164 "InstitutionAddress": "remove",
165 "InstitutionCodeSequence": "remove",
166 "InstitutionName": "remove",
167 "InstitutionalDepartmentName": "remove",
168 "InsurancePlanIdentification": "remove",
169 "IssuerOfAccessionNumberSequence": "remove",
170 "IssuerOfAdmissionID": "remove",
171 "IssuerOfPatientID": "remove",
172 "IssuerOfPatientIDQualifiersSequence": "remove",
173 "MACParametersSequence": "remove",
174 "MedicalAlerts": "remove",
175 "MedicalRecordLocator": "remove",
176 "MilitaryRank": "remove",
177 "MIMETypeOfEncapsulatedDocument": "remove",
178 "NameOfPhysiciansReadingStudy": "remove",
179 "Occupation": "remove",
180 "OperatorsIdentificationSequence": "remove",
181 "OperatorsName": "remove",
182 "OriginalAttributesSequence": "remove",
183 "OtherPatientIDs": "remove",
184 "OtherPatientIDsSequence": "remove",
185 "OtherPatientNames": "remove",
186 "PatientAddress": "remove",
187 "PatientAge": "remove",
188 "PatientBirthName": "remove",
189 "PatientComments": "remove",
190 "PatientID": "pseudonym",
191 "PatientInsurancePlanCodeSequence": "remove",
192 "PatientMotherBirthName": "remove",
193 "PatientName": "pseudonym",
194 "PatientReligiousPreference": "remove",
195 "PatientSex": "remove",
196 "PatientSize": "remove",
197 "PatientState": "remove",
198 "PatientTelephoneNumbers": "remove",
199 "PatientWeight": "remove",
200 "PerformedLocation": "remove",
201 "PerformedProcedureStepDescription": "remove",
202 "PerformedStationAETitle": "remove",
203 "PerformedStationGeographicLocationCodeSequence": "remove",
204 "PerformedStationName": "remove",
205 "PerformingPhysicianIdentificationSequence": "remove",
206 "PerformingPhysicianName": "remove",
207 "PersonAddress": "remove",
208 "PersonIdentificationCodeSequence": "remove",
209 "PersonName": "remove",
210 "PersonTelephoneNumbers": "remove",
211 "PhysiciansOfRecord": "remove",
212 "PhysiciansOfRecordIdentificationSequence": "remove",
213 "PhysiciansReadingStudyIdentificationSequence": "remove",
214 "PlacerOrderNumberImagingServiceRequest": "empty",
215 "ProtocolName": "remove",
216 "ReasonForImagingServiceRequest": "remove",
217 "ReasonForRequestedProcedure": "remove",
218 "ReferringPhysicianAddress": "remove",
219 "ReferringPhysicianIdentificationSequence": "remove",
220 "ReferringPhysicianName": "remove",
221 "ReferringPhysicianTelephoneNumbers": "remove",
222 "RegionOfResidence": "remove",
223 "RequestAttributesSequence": "remove",
224 "RequestedProcedureComments": "remove",
225 "RequestedProcedureDescription": "remove",
226 "RequestedProcedureID": "empty",
227 "RequestingPhysician": "remove",
228 "RequestingService": "remove",
229 "ResponsibleOrganization": "remove",
230 "ResponsiblePerson": "remove",
231 "ResponsiblePersonRole": "remove",
232 "ScheduledPerformingPhysicianIdentificationSequence": "remove",
233 "ScheduledPerformingPhysicianName": "remove",
234 "ScheduledProcedureStepDescription": "remove",
235 "ScheduledProcedureStepID": "empty",
236 "SeriesDescription": "remove",
237 "ServiceEpisodeDescription": "remove",
238 "ServiceEpisodeID": "remove",
239 "SmokingStatus": "remove",
240 "SpecialNeeds": "remove",
241 "SpecimenAccessionNumber": "remove",
242 "SpecimenIdentifier": "remove",
243 "StationName": "remove",
244 "StudyDescription": "remove",
245 "StudyID": "empty",
246 "TextComments": "remove",
247 "TextString": "remove",
248 "TextValue": "remove",
249 "UnformattedTextValue": "remove",
250 "AudioComments": "remove",
251 "AudioSampleData": "remove",
252}
254SENSITIVE_KEYWORDS = frozenset(
255 set(DEFAULT_ACTIONS)
256 | {
257 "PatientBirthDate",
258 "PatientBirthTime",
259 "PatientIdentityRemoved",
260 "StudyDate",
261 "StudyTime",
262 "SeriesDate",
263 "SeriesTime",
264 "AcquisitionDate",
265 "AcquisitionTime",
266 "AcquisitionDateTime",
267 "ContentDate",
268 "ContentTime",
269 "InstanceCreationDate",
270 "InstanceCreationTime",
271 *UID_REMAP_KEYWORDS,
272 }
273)
275TECHNICAL_KEYWORDS = (
276 "SOPClassUID",
277 "Modality",
278 "Rows",
279 "Columns",
280 "NumberOfFrames",
281 "SamplesPerPixel",
282 "PhotometricInterpretation",
283 "PlanarConfiguration",
284 "BitsAllocated",
285 "BitsStored",
286 "HighBit",
287 "PixelRepresentation",
288 "PixelSpacing",
289 "RescaleSlope",
290 "RescaleIntercept",
291 "RescaleType",
292 "WindowCenter",
293 "WindowWidth",
294 "VOILUTFunction",
295 "BurnedInAnnotation",
296 "RecognizableVisualFeatures",
297 "LossyImageCompression",
298)
301class ToolError(ValueError):
302 """Expected validation, dependency, or bounded local-I/O failure."""
305def bounded_int(value: Any, *, name: str, minimum: int, maximum: int) -> int:
306 if isinstance(value, bool):
307 raise ToolError(f"{name} must be an integer")
308 try:
309 result = int(value)
310 except (TypeError, ValueError, OverflowError) as exc:
311 raise ToolError(f"{name} must be an integer") from exc
312 if not minimum <= result <= maximum:
313 raise ToolError(f"{name} must be between {minimum} and {maximum}")
314 return result
317def parse_size(
318 value: str | int,
319 *,
320 name: str,
321 minimum: int = 1,
322 maximum: int = HARD_MAX_INPUT_BYTES,
323) -> int:
324 if isinstance(value, int):
325 result = value
326 elif isinstance(value, str):
327 match = _SIZE_PATTERN.fullmatch(value)
328 if not match:
329 raise ToolError(f"{name} must be an integer byte count or B/KiB/MiB/GiB")
330 result = int(match.group(1))
331 unit = (match.group(2) or "B").upper()
332 result *= {"B": 1, "KIB": KIB, "MIB": MIB, "GIB": GIB}[unit]
333 else:
334 raise ToolError(f"{name} must be a byte-size string")
335 return bounded_int(result, name=name, minimum=minimum, maximum=maximum)
338def require_text(
339 value: Any, *, name: str, minimum: int = 1, maximum: int = 1_000
340) -> str:
341 if not isinstance(value, str) or not minimum <= len(value) <= maximum:
342 raise ToolError(f"{name} must be a string of length {minimum}..{maximum}")
343 if any(ord(character) < 32 for character in value):
344 raise ToolError(f"{name} must not contain control characters")
345 return value
348def safe_plugin_name(value: str) -> str:
349 if not _SAFE_PLUGIN.fullmatch(value):
350 raise ToolError("decoding plugin name is invalid")
351 return value
354def _reject_path_text(value: str) -> None:
355 stripped = value.strip()
356 lowered = stripped.casefold()
357 if not stripped or "\x00" in value or stripped.startswith("~"):
358 raise ToolError("path must be nonempty, local, and contain no NUL")
359 if "://" in lowered or lowered.startswith(_URI_PREFIXES):
360 raise ToolError("only local filesystem paths are accepted")
361 if ".." in PurePath(stripped).parts:
362 raise ToolError("parent traversal is not accepted")
365def _absolute(path: Path) -> Path:
366 return Path(os.path.abspath(os.fspath(path)))
369def _reject_symlink_components(path: Path) -> None:
370 absolute = _absolute(path)
371 current = Path(absolute.anchor)
372 for part in absolute.parts[1:]:
373 current /= part
374 try:
375 mode = current.lstat().st_mode
376 except FileNotFoundError:
377 continue
378 except OSError as exc:
379 raise ToolError("a path component could not be inspected") from exc
380 if stat.S_ISLNK(mode):
381 raise ToolError("symlink path components are not accepted")
384def checked_root(value: str | os.PathLike[str]) -> Path:
385 raw = os.fspath(value)
386 _reject_path_text(raw)
387 candidate = _absolute(Path(raw))
388 _reject_symlink_components(candidate)
389 try:
390 resolved = candidate.resolve(strict=True)
391 info = resolved.stat()
392 except OSError as exc:
393 raise ToolError("root directory is not accessible") from exc
394 if not stat.S_ISDIR(info.st_mode):
395 raise ToolError("root must be a directory")
396 return resolved
399def _within_root(path: Path, root: Path) -> None:
400 try:
401 path.relative_to(root)
402 except ValueError as exc:
403 raise ToolError("path escapes the declared root") from exc
406def checked_input(
407 value: str | os.PathLike[str],
408 *,
409 root: str | os.PathLike[str] = ".",
410 kind: str = "file",
411 max_bytes: int = DEFAULT_MAX_INPUT_BYTES,
412) -> Path:
413 raw = os.fspath(value)
414 _reject_path_text(raw)
415 root_path = checked_root(root)
416 candidate = Path(raw)
417 if not candidate.is_absolute():
418 candidate = root_path / candidate
419 candidate = _absolute(candidate)
420 _reject_symlink_components(candidate)
421 try:
422 resolved = candidate.resolve(strict=True)
423 info = resolved.stat()
424 except OSError as exc:
425 raise ToolError("input path is not accessible") from exc
426 _within_root(resolved, root_path)
427 if kind == "file" and not stat.S_ISREG(info.st_mode):
428 raise ToolError("input must be a regular file")
429 if kind == "dir" and not stat.S_ISDIR(info.st_mode):
430 raise ToolError("input must be a directory")
431 if kind == "any" and not (stat.S_ISREG(info.st_mode) or stat.S_ISDIR(info.st_mode)):
432 raise ToolError("input must be a regular file or directory")
433 if stat.S_ISREG(info.st_mode):
434 if info.st_nlink != 1:
435 raise ToolError("multiply linked input files are not accepted")
436 if not 0 <= info.st_size <= max_bytes:
437 raise ToolError("input file exceeds the configured byte limit")
438 return resolved
441def checked_output(
442 value: str | os.PathLike[str],
443 *,
444 root: str | os.PathLike[str] = ".",
445 force: bool = False,
446) -> Path:
447 raw = os.fspath(value)
448 _reject_path_text(raw)
449 root_path = checked_root(root)
450 candidate = Path(raw)
451 if not candidate.is_absolute():
452 candidate = root_path / candidate
453 candidate = _absolute(candidate)
454 _reject_symlink_components(candidate)
455 parent = candidate.parent
456 try:
457 parent = parent.resolve(strict=True)
458 except OSError as exc:
459 raise ToolError("output parent directory must already exist") from exc
460 _within_root(parent, root_path)
461 destination = parent / candidate.name
462 if destination.exists():
463 if destination.is_symlink() or not destination.is_file():
464 raise ToolError("existing output is not a regular file")
465 if not force:
466 raise ToolError("refusing to overwrite existing output")
467 return destination
470def checked_output_directory(
471 value: str | os.PathLike[str],
472 *,
473 root: str | os.PathLike[str] = ".",
474) -> Path:
475 raw = os.fspath(value)
476 _reject_path_text(raw)
477 root_path = checked_root(root)
478 candidate = Path(raw)
479 if not candidate.is_absolute():
480 candidate = root_path / candidate
481 candidate = _absolute(candidate)
482 _reject_symlink_components(candidate)
483 parent = candidate.parent
484 try:
485 parent = parent.resolve(strict=True)
486 except OSError as exc:
487 raise ToolError("output directory parent must already exist") from exc
488 _within_root(parent, root_path)
489 destination = parent / candidate.name
490 if destination.exists():
491 if destination.is_symlink() or not destination.is_dir():
492 raise ToolError("existing destination is not a regular directory")
493 else:
494 try:
495 destination.mkdir(mode=0o700)
496 except OSError as exc:
497 raise ToolError("output directory could not be created") from exc
498 return destination.resolve(strict=True)
501def paths_overlap(first: Path, second: Path) -> bool:
502 first = first.resolve(strict=False)
503 second = second.resolve(strict=False)
504 return first == second or first in second.parents or second in first.parents
507def collect_local_files(
508 source: Path,
509 *,
510 max_files: int,
511 max_bytes: int,
512 recursive: bool = True,
513) -> list[Path]:
514 bounded_int(max_files, name="max_files", minimum=1, maximum=HARD_MAX_FILES)
515 if source.is_file():
516 return [source]
517 files: list[Path] = []
518 stack = [source]
519 while stack:
520 directory = stack.pop()
521 try:
522 entries = sorted(
523 os.scandir(directory), key=lambda entry: entry.name.casefold()
524 )
525 except OSError as exc:
526 raise ToolError("input directory cannot be scanned") from exc
527 for entry in entries:
528 try:
529 if entry.is_symlink():
530 raise ToolError("symlinks in input directories are rejected")
531 if entry.is_dir(follow_symlinks=False):
532 if recursive:
533 stack.append(Path(entry.path))
534 continue
535 if not entry.is_file(follow_symlinks=False):
536 continue
537 path = Path(entry.path).resolve(strict=True)
538 info = path.stat()
539 if info.st_nlink != 1:
540 raise ToolError("multiply linked input files are rejected")
541 if info.st_size > max_bytes:
542 raise ToolError("an input file exceeds the configured byte limit")
543 files.append(path)
544 if len(files) > max_files:
545 raise ToolError("input file-count limit exceeded")
546 except OSError as exc:
547 raise ToolError(
548 "an input directory entry could not be inspected"
549 ) from exc
550 return sorted(files, key=lambda path: path.as_posix().casefold())
553def _reject_json_constant(value: str) -> None:
554 raise ToolError(f"non-finite JSON number is not accepted: {value}")
557def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
558 result: dict[str, Any] = {}
559 for key, value in pairs:
560 if key in result:
561 raise ToolError(f"duplicate JSON key is not accepted: {key}")
562 result[key] = value
563 return result
566def load_json(path: Path, *, max_bytes: int = MAX_JSON_BYTES) -> Any:
567 if path.stat().st_size > max_bytes:
568 raise ToolError("JSON input exceeds the configured byte limit")
569 try:
570 with path.open("r", encoding="utf-8") as handle:
571 return json.load(
572 handle,
573 object_pairs_hook=_unique_object,
574 parse_constant=_reject_json_constant,
575 )
576 except ToolError:
577 raise
578 except (OSError, UnicodeError, json.JSONDecodeError, RecursionError) as exc:
579 raise ToolError("input is not valid bounded UTF-8 JSON") from exc
582def json_bytes(document: Any) -> bytes:
583 try:
584 payload = (
585 json.dumps(
586 document,
587 allow_nan=False,
588 ensure_ascii=False,
589 indent=2,
590 sort_keys=True,
591 )
592 + "\n"
593 ).encode("utf-8")
594 except (TypeError, ValueError) as exc:
595 raise ToolError("report cannot be serialized as strict JSON") from exc
596 if len(payload) > MAX_REPORT_BYTES:
597 raise ToolError("report exceeds the hard report-size limit")
598 return payload
601def emit_json(document: Any) -> None:
602 sys.stdout.buffer.write(json_bytes(document))
605def fail_json(tool: str, exc: Exception) -> int:
606 message = (
607 str(exc)[:500]
608 if isinstance(exc, ToolError)
609 else f"operation failed ({type(exc).__name__})"
610 )
611 emit_json(
612 {
613 "error": type(exc).__name__,
614 "message": message,
615 "ok": False,
616 "tool": tool,
617 }
618 )
619 return 2
622def atomic_write(
623 destination: Path,
624 payload: bytes,
625 *,
626 force: bool = False,
627 max_bytes: int = MAX_REPORT_BYTES,
628) -> None:
629 if len(payload) > max_bytes:
630 raise ToolError("generated output exceeds the configured byte limit")
631 descriptor, temporary_name = tempfile.mkstemp(
632 prefix=f".{destination.name}.", suffix=".tmp", dir=destination.parent
633 )
634 temporary = Path(temporary_name)
635 try:
636 with os.fdopen(descriptor, "wb") as handle:
637 handle.write(payload)
638 handle.flush()
639 os.fsync(handle.fileno())
640 os.chmod(temporary, 0o600)
641 if destination.exists() and not force:
642 raise ToolError("refusing to overwrite existing output")
643 os.replace(temporary, destination)
644 finally:
645 temporary.unlink(missing_ok=True)
648def atomic_generated_file(
649 destination: Path,
650 *,
651 writer: Callable[[Path], None],
652 validator: Callable[[Path], None] | None = None,
653 force: bool = False,
654 max_bytes: int = HARD_MAX_OUTPUT_BYTES,
655) -> None:
656 descriptor, temporary_name = tempfile.mkstemp(
657 prefix=f".{destination.name}.",
658 suffix=destination.suffix or ".tmp",
659 dir=destination.parent,
660 )
661 os.close(descriptor)
662 temporary = Path(temporary_name)
663 try:
664 os.chmod(temporary, 0o600)
665 writer(temporary)
666 info = temporary.stat()
667 if not stat.S_ISREG(info.st_mode) or info.st_size > max_bytes:
668 raise ToolError("generated output exceeds the configured byte limit")
669 if validator is not None:
670 validator(temporary)
671 with temporary.open("rb") as handle:
672 os.fsync(handle.fileno())
673 if destination.exists() and not force:
674 raise ToolError("refusing to overwrite existing output")
675 os.replace(temporary, destination)
676 except ToolError:
677 raise
678 except Exception as exc:
679 raise ToolError("generated output could not be safely written") from exc
680 finally:
681 temporary.unlink(missing_ok=True)
684def require_pydicom() -> Any:
685 try:
686 import pydicom
687 except ImportError as exc:
688 raise ToolError(
689 f"pydicom is required; install with: uv pip install pydicom=={PYDICOM_VERSION}"
690 ) from exc
691 if pydicom.__version__ != PYDICOM_VERSION:
692 raise ToolError(
693 f"this skill is verified with pydicom=={PYDICOM_VERSION}; "
694 f"found {pydicom.__version__}"
695 )
696 return pydicom
699def require_pixel_stack() -> tuple[Any, Any, Any]:
700 pydicom = require_pydicom()
701 try:
702 import numpy
703 from PIL import Image
704 except ImportError as exc:
705 raise ToolError(
706 "pixel conversion requires pinned NumPy and Pillow; install with: "
707 f"uv pip install pydicom=={PYDICOM_VERSION} "
708 f"numpy=={NUMPY_VERSION} Pillow=={PILLOW_VERSION}"
709 ) from exc
710 return pydicom, numpy, Image
713def safe_dcmread(
714 path: Path,
715 *,
716 stop_before_pixels: bool,
717 force: bool = False,
718 defer_size: str | int | None = None,
719 specific_tags: Iterable[str | int] | None = None,
720) -> Any:
721 pydicom = require_pydicom()
722 try:
723 return pydicom.dcmread(
724 path,
725 defer_size=defer_size,
726 stop_before_pixels=stop_before_pixels,
727 force=force,
728 specific_tags=list(specific_tags) if specific_tags is not None else None,
729 )
730 except Exception as exc:
731 raise ToolError("DICOM input could not be parsed") from exc
734def element_tag(element: Any) -> str:
735 return f"({int(element.tag.group):04X},{int(element.tag.element):04X})"
738def valid_uid(value: Any) -> bool:
739 if not isinstance(value, str) or not 1 <= len(value) <= 64:
740 return False
741 if not _UID_PATTERN.fullmatch(value):
742 return False
743 return all(
744 component == "0" or not component.startswith("0")
745 for component in value.split(".")
746 )
749def derive_uid(original: str, *, key: bytes, scope: str) -> str:
750 if not valid_uid(original):
751 raise ToolError("an instance UID selected for remapping is invalid")
752 digest = hmac.new(
753 key,
754 b"pydicom-skill-uid-v1\0"
755 + scope.encode("utf-8")
756 + b"\0"
757 + original.encode("ascii"),
758 hashlib.sha256,
759 ).digest()[:16]
760 number = int.from_bytes(digest, "big") or 1
761 result = f"2.25.{number}"
762 if not valid_uid(result):
763 raise ToolError("derived UID is invalid")
764 return result
767def derive_token(value: str, *, key: bytes, scope: str, length: int = 24) -> str:
768 digest = hmac.new(
769 key,
770 b"pydicom-skill-token-v1\0"
771 + scope.encode("utf-8")
772 + b"\0"
773 + value.encode("utf-8", errors="surrogatepass"),
774 hashlib.sha256,
775 ).hexdigest()
776 return digest[:length].upper()
779def sha256_text(value: str) -> str:
780 return hashlib.sha256(value.encode("utf-8")).hexdigest()
783def frame_count(dataset: Any) -> tuple[int, list[str]]:
784 warnings: list[str] = []
785 raw = dataset.get("NumberOfFrames", 1)
786 try:
787 count = int(raw)
788 except (TypeError, ValueError, OverflowError) as exc:
789 raise ToolError("NumberOfFrames is not an integer") from exc
790 if count == 0:
791 warnings.append("NumberOfFrames is zero; pydicom treats it as one")
792 count = 1
793 if count < 1:
794 raise ToolError("NumberOfFrames must be positive")
795 return count, warnings
798def pixel_plan(
799 dataset: Any,
800 *,
801 max_frames: int = HARD_MAX_FRAMES,
802 max_decompressed_bytes: int = HARD_MAX_DECOMPRESSED_BYTES,
803) -> dict[str, Any]:
804 rows = bounded_int(dataset.get("Rows"), name="Rows", minimum=1, maximum=1_000_000)
805 columns = bounded_int(
806 dataset.get("Columns"), name="Columns", minimum=1, maximum=1_000_000
807 )
808 samples = bounded_int(
809 dataset.get("SamplesPerPixel", 1),
810 name="SamplesPerPixel",
811 minimum=1,
812 maximum=256,
813 )
814 frames, warnings = frame_count(dataset)
815 bounded_int(frames, name="NumberOfFrames", minimum=1, maximum=max_frames)
816 if "DoubleFloatPixelData" in dataset:
817 bits_allocated = 64
818 pixel_element = "DoubleFloatPixelData"
819 elif "FloatPixelData" in dataset:
820 bits_allocated = 32
821 pixel_element = "FloatPixelData"
822 else:
823 bits_allocated = bounded_int(
824 dataset.get("BitsAllocated"),
825 name="BitsAllocated",
826 minimum=1,
827 maximum=64,
828 )
829 pixel_element = "PixelData"
830 pixels_per_frame = rows * columns * samples
831 packed_bytes_per_frame = (pixels_per_frame * bits_allocated + 7) // 8
832 decoded_item_bytes = (
833 1
834 if bits_allocated <= 8
835 else 2
836 if bits_allocated <= 16
837 else 4
838 if bits_allocated <= 32
839 else 8
840 )
841 bytes_per_frame = pixels_per_frame * decoded_item_bytes
842 total_bytes = bytes_per_frame * frames
843 if total_bytes > max_decompressed_bytes:
844 raise ToolError(
845 "estimated decompressed pixel data exceeds the configured limit"
846 )
847 if frames > 1:
848 shape = (
849 [frames, rows, columns]
850 if samples == 1
851 else [frames, rows, columns, samples]
852 )
853 else:
854 shape = [rows, columns] if samples == 1 else [rows, columns, samples]
855 return {
856 "bits_allocated": bits_allocated,
857 "bytes_per_frame": bytes_per_frame,
858 "columns": columns,
859 "estimated_decompressed_bytes": total_bytes,
860 "frames": frames,
861 "native_packed_bytes_per_frame": packed_bytes_per_frame,
862 "pixel_element": pixel_element,
863 "rows": rows,
864 "samples_per_pixel": samples,
865 "shape": shape,
866 "warnings": warnings,
867 }
870def counter_dict(counter: Counter[str]) -> dict[str, int]:
871 return {key: counter[key] for key in sorted(counter)}
874def validate_profile(document: Any) -> dict[str, Any]:
875 if not isinstance(document, Mapping):
876 raise ToolError("action profile must be a JSON object")
877 allowed = {"name", "version", "actions", "private_policy", "date_policy"}
878 unknown = sorted(set(document) - allowed)
879 if unknown:
880 raise ToolError(f"action profile has unsupported keys: {', '.join(unknown)}")
881 name = require_text(document.get("name"), name="profile name", maximum=128)
882 version = require_text(document.get("version"), name="profile version", maximum=32)
883 raw_actions = document.get("actions", {})
884 if not isinstance(raw_actions, Mapping) or len(raw_actions) > 2_000:
885 raise ToolError("profile actions must be an object with at most 2000 entries")
886 actions = dict(DEFAULT_ACTIONS)
887 valid_actions = {"empty", "keep", "pseudonym", "remove", "uid"}
888 for raw_key, raw_action in raw_actions.items():
889 key = require_text(raw_key, name="action key", maximum=128)
890 action = require_text(raw_action, name=f"action for {key}", maximum=16)
891 if action not in valid_actions:
892 raise ToolError(f"unsupported action for {key}")
893 actions[key] = action
894 private_policy = document.get("private_policy", "remove")
895 if private_policy not in {"keep", "reject", "remove"}:
896 raise ToolError("private_policy must be keep, reject, or remove")
897 date_policy = document.get("date_policy", "empty")
898 if date_policy not in {"empty", "keep", "shift"}:
899 raise ToolError("date_policy must be empty, keep, or shift")
900 return {
901 "actions": actions,
902 "date_policy": date_policy,
903 "name": name,
904 "private_policy": private_policy,
905 "version": version,
906 }
909def starter_profile() -> dict[str, Any]:
910 return {
911 "actions": dict(DEFAULT_ACTIONS),
912 "date_policy": "empty",
913 "name": "bounded-starter",
914 "private_policy": "remove",
915 "version": "1.1",
916 }