Source code for monai.deploy.operators.dicom_utils

# Copyright 2022 MONAI Consortium
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import datetime
import logging
from pathlib import Path
from random import randint
from typing import Any, Optional

from monai.deploy.utils.importutil import optional_import

dcmread, _ = optional_import("pydicom", name="dcmread")
dcmwrite, _ = optional_import("pydicom.filewriter", name="dcmwrite")
generate_uid, _ = optional_import("pydicom.uid", name="generate_uid")
ImplicitVRLittleEndian, _ = optional_import("pydicom.uid", name="ImplicitVRLittleEndian")
Dataset_, _ = optional_import("pydicom.dataset", name="Dataset")
FileDataset, _ = optional_import("pydicom.dataset", name="FileDataset")
Sequence, _ = optional_import("pydicom.sequence", name="Sequence")

from monai.deploy.core.domain.dicom_series import DICOMSeries
from monai.deploy.utils.version import get_sdk_semver

# To address mypy complaint
Dataset: Any = Dataset_

__all__ = [

[docs]class ModelInfo(object): """Class encapsulating AI model information, according to IHE AI Results (AIR) Rev 1.1. The attributes of the class will be used to populate the Contributing Equipment Sequence in the DICOM IOD per IHE AIR Rev 1.1, Section General Result Encoding Requirements, as the following, The Creator shall describe each algorithm that was used to generate the results in the Contributing Equipment Sequence (0018,A001). Multiple items may be included. The Creator shall encode the following details in the Contributing Equipment Sequence, - Purpose of Reference Code Sequence (0040,A170) shall be (Newcode1, 99IHE, 1630 "Processing Algorithm") - Manufacturer (0008,0070) - Manufacturer's Model Name (0008,1090) - Software Versions (0018,1020) - Device UID (0018,1002) Each time an AI Model is modified, for example by training, it would be appropriate to update the Device UID. """ def __init__(self, creator: str = "", name: str = "", version: str = "", uid: str = ""): self.creator = creator if isinstance(creator, str) else "" = name if isinstance(name, str) else "" self.version = version if isinstance(version, str) else "" self.uid = uid if isinstance(uid, str) else ""
[docs]class EquipmentInfo(object): """Class encapsulating attributes required for DICOM Equipment Module.""" def __init__( self, manufacturer: str = "MONAI Deploy", manufacturer_model: str = "MONAI Deploy App SDK", series_number: str = "0000", software_version_number: str = "", ): self.manufacturer = manufacturer if isinstance(manufacturer, str) else "" self.manufacturer_model = manufacturer_model if isinstance(manufacturer_model, str) else "" self.series_number = series_number if isinstance(series_number, str) else "" if software_version_number: self.software_version_number = str(software_version_number)[0:15] else: try: version_str = get_sdk_semver() # SDK Version except Exception: version_str = "" # Fall back to the initial version self.software_version_number = version_str[0:15]
# Utility functions def random_with_n_digits(n): """Random number generator to generate n digit int, where 1 <= n <= 32.""" assert isinstance(n, int) and n <= 32, "Argument n must be an int, n <= 32." n = n if n >= 1 else 1 range_start = 10 ** (n - 1) range_end = (10**n) - 1 return randint(range_start, range_end) def save_dcm_file(data_set: Dataset, file_path: Path, validate_readable: bool = True): """Save a DICOM data set, in pydicom Dataset, to the provided file path.""" logging.debug(f"DICOM dataset to be written:{data_set}") if not isinstance(data_set, Dataset): raise ValueError("data_set is not the expected Dataset type.") if not str(file_path).strip(): raise ValueError("file_path to save dcm file not provided.") dcmwrite(str(file_path).strip(), data_set, write_like_original=False)"Finished writing DICOM instance to file {file_path}") if validate_readable: # Test reading back _ = dcmread(str(file_path)) def write_common_modules( dicom_series: Optional[DICOMSeries], copy_tags: bool, modality_type: str, sop_class_uid: str, model_info: Optional[ModelInfo] = None, equipment_info: Optional[EquipmentInfo] = None, ) -> Dataset: """Writes DICOM object common modules with or without a reference DCIOM Series Common modules include Study, Patient, Equipment, Series, and SOP common. Args: dicom_series (DicomSeries): DicomSeries object encapsulating the original series. copy_tags (bool): If true, dicom_series must be provided for copying tags. modality_type (str): DICOM Modality Type, e.g. SR. sop_class_uid (str): Media Storage SOP Class UID, e.g. "1.2.840.10008." for Comprehensive 3D SR IOD. model_info (MoelInfo): Object encapsulating model creator, name, version and UID. equipment_info(EquipmentInfo): Object encapsulating attributes for DICOM Equipment Module Returns: pydicom Dataset Raises: ValueError: When dicom_series is not a DICOMSeries object, and new_study is not True. """ if copy_tags: if not isinstance(dicom_series, DICOMSeries): raise ValueError("A DICOMSeries object is required for coping tags.") if len(dicom_series.get_sop_instances()) < 1: raise ValueError("DICOMSeries must have at least one SOP instance.") # Get one of the SOP instance's native sop instance dataset orig_ds = dicom_series.get_sop_instances()[0].get_native_sop_instance() logging.debug("Writing DICOM common modules...") # Get and format date and time per DICOM standards. dt_now = date_now_dcm = dt_now.strftime("%Y%m%d") time_now_dcm = dt_now.strftime("%H%M%S") offset_from_utc = dt_now.astimezone().isoformat()[-6:].replace(":", "") # '2022-09-27T22:36:20.143857-07:00' # Generate UIDs and descriptions my_sop_instance_uid = generate_uid() my_series_instance_uid = generate_uid() my_series_description = "CAUTION: Not for Diagnostic Use, for research use only." my_series_number = str(random_with_n_digits(4)) # 4 digit number to avoid conflict my_study_instance_uid = orig_ds.StudyInstanceUID if copy_tags else generate_uid() # File meta info data set file_meta = Dataset() file_meta.FileMetaInformationGroupLength = 198 file_meta.FileMetaInformationVersion = bytes("01", "utf-8") # '\x00\x01' file_meta.MediaStorageSOPClassUID = sop_class_uid file_meta.MediaStorageSOPInstanceUID = my_sop_instance_uid file_meta.TransferSyntaxUID = ImplicitVRLittleEndian # 1.2.840.10008.1.2, Little Endian Implicit VR file_meta.ImplementationClassUID = "" # Made up. Not registered. file_meta.ImplementationVersionName = equipment_info.software_version_number if equipment_info else "" # Write modules to data set ds = Dataset() ds.file_meta = file_meta ds.is_implicit_VR = True ds.is_little_endian = True # Content Date (0008,0023) and Content Time (0008,0033) are defined to be the date and time that # the document content creation started. In the context of analysis results, these may be considered # to be the date and time that the analysis that generated the result(s) started executing. # Use current time for now, but could potentially use the actual inference start time. ds.ContentDate = date_now_dcm ds.ContentTime = time_now_dcm ds.TimezoneOffsetFromUTC = offset_from_utc # The date and time that the original generation of the data in the document started. ds.AcquisitionDateTime = date_now_dcm + time_now_dcm # Result has just been created. # Patient Module, mandatory. # Copy over from the original DICOM metadata. ds.PatientName = orig_ds.get("PatientName", "") if copy_tags else "" ds.PatientID = orig_ds.get("PatientID", "") if copy_tags else "" ds.IssuerOfPatientID = orig_ds.get("IssuerOfPatientID", "") if copy_tags else "" ds.PatientBirthDate = orig_ds.get("PatientBirthDate", "") if copy_tags else "" ds.PatientSex = orig_ds.get("PatientSex", "") if copy_tags else "" # Study Module, mandatory # Copy over from the original DICOM metadata. ds.StudyDate = orig_ds.get("StudyDate", "") if copy_tags else date_now_dcm ds.StudyTime = orig_ds.get("StudyTime", "") if copy_tags else time_now_dcm ds.AccessionNumber = orig_ds.get("AccessionNumber", "") if copy_tags else "" ds.StudyDescription = orig_ds.get("StudyDescription", "") if copy_tags else "AI results." ds.StudyInstanceUID = my_study_instance_uid ds.StudyID = orig_ds.get("StudyID", "") if copy_tags else "1" ds.ReferringPhysicianName = orig_ds.get("ReferringPhysicianName", "") if copy_tags else "" # Equipment Module, mandatory if equipment_info: ds.Manufacturer = equipment_info.manufacturer ds.ManufacturerModelName = equipment_info.manufacturer_model ds.SeriesNumber = equipment_info.series_number ds.SoftwareVersions = equipment_info.software_version_number # SOP Common Module, mandatory ds.InstanceCreationDate = date_now_dcm ds.InstanceCreationTime = time_now_dcm ds.SOPClassUID = sop_class_uid ds.SOPInstanceUID = my_sop_instance_uid ds.InstanceNumber = "1" ds.SpecificCharacterSet = "ISO_IR 100" # Series Module, mandatory ds.Modality = modality_type ds.SeriesInstanceUID = my_series_instance_uid ds.SeriesNumber = my_series_number ds.SeriesDescription = my_series_description ds.SeriesDate = date_now_dcm ds.SeriesTime = time_now_dcm # Body part copied over, although not mandatory depending on modality ds.BodyPartExamined = orig_ds.get("BodyPartExamined", "") if copy_tags else "" ds.RequestedProcedureID = orig_ds.get("RequestedProcedureID", "") if copy_tags else "" # Contributing Equipment Sequence # The Creator shall describe each algorithm that was used to generate the results in the # Contributing Equipment Sequence (0018,A001). Multiple items may be included. The Creator # shall encode the following details in the Contributing Equipment Sequence: # • Purpose of Reference Code Sequence (0040,A170) shall be (Newcode1, 99IHE, 1630 "Processing Algorithm") # • Manufacturer (0008,0070) # • Manufacturer’s Model Name (0008,1090) # • Software Versions (0018,1020) # • Device UID (0018,1002) if model_info: # First create the Purpose of Reference Code Sequence seq_purpose_of_reference_code = Sequence() ds_purpose_of_reference_code = Dataset() ds_purpose_of_reference_code.CodeValue = "Newcode1" ds_purpose_of_reference_code.CodingSchemeDesignator = "99IHE" ds_purpose_of_reference_code.CodeMeaning = '"Processing Algorithm' seq_purpose_of_reference_code.append(ds_purpose_of_reference_code) seq_contributing_equipment = Sequence() ds_contributing_equipment = Dataset() ds_contributing_equipment.PurposeOfReferenceCodeSequence = seq_purpose_of_reference_code # '(121014, DCM, “Device Observer Manufacturer")' ds_contributing_equipment.Manufacturer = model_info.creator # u'(121015, DCM, “Device Observer Model Name")' ds_contributing_equipment.ManufacturerModelName = # u'(111003, DCM, “Algorithm Version")' ds_contributing_equipment.SoftwareVersions = model_info.version ds_contributing_equipment.DeviceUID = model_info.uid # u'(121012, DCM, “Device Observer UID")' seq_contributing_equipment.append(ds_contributing_equipment) ds.ContributingEquipmentSequence = seq_contributing_equipment logging.debug("DICOM common modules written:\n{}".format(ds)) return ds