Source code for monai.deploy.operators.dicom_text_sr_writer_operator

# Copyright 2021 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 Dict, List, Text, Union

from monai.deploy.utils.importutil import optional_import
from monai.deploy.utils.version import get_sdk_semver

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")

import monai.deploy.core as md
from monai.deploy.core import DataPath, ExecutionContext, InputContext, IOType, Operator, OutputContext
from monai.deploy.core.domain.dicom_series import DICOMSeries
from monai.deploy.core.domain.dicom_series_selection import StudySelectedSeries
from monai.deploy.exceptions import ItemNotExistsError

# Utility classes considered to be moved into Domain module
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 ""

class EquipmentInfo(object):
    """Class encapsulating attributes required for DICOM Equipment Module."""

    def __init__(
        manufacturer: str = "MONAI Deploy",
        manufacturer_model: str = "MONAI Deploy App SDK",
        series_number: str = "0000",
        software_version_number: str = "0.2",

        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 ""
            version_str = get_sdk_semver()  # SDK Version
        except Exception:
            version_str = "0.2"  # Fall back to the initial version
        self.software_version_number = (
            software_version_number if isinstance(software_version_number, str) else version_str

# The SR writer operator class
[docs]@md.input("classification_result", Text, IOType.IN_MEMORY) @md.input("classification_result_file", DataPath, IOType.DISK) @md.input("study_selected_series_list", List[StudySelectedSeries], IOType.IN_MEMORY) @md.output("dicom_instance", DataPath, IOType.DISK) @md.env(pip_packages=["pydicom >= 1.4.2"]) class DICOMTextSRWriterOperator(Operator): DCM_EXTENSION = ".dcm"
[docs] def __init__( self, copy_tags: bool, model_info: ModelInfo, equipment_info: Union[EquipmentInfo, None] = None, custom_tags: Union[Dict[str, str], None] = None, *args, **kwargs, ): """Class to write DICOM SR SOP Instance for AI textual result in memory or in a file. Args: copy_tags (bool): True for copying DICOM attributes from a provided DICOMSeries. model_info (ModelInfo): Object encapsulating model creator, name, version and UID. equipment_info (EquipmentInfo, optional): Object encapsulating info for DICOM Equipment Module. Defaults to None. custom_tags (Dict[str, str], optional): Dictionary for setting custom DICOM tags using Keywords and str values only. Defaults to None. Raises: ValueError: If copy_tags is true and no DICOMSeries object provided, or if result cannot be found either in memory or from file. """ super().__init__(*args, **kwargs) self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__)) self.copy_tags = copy_tags self.model_info = model_info if model_info else ModelInfo() self.equipment_info = equipment_info if equipment_info else EquipmentInfo() self.custom_tags = custom_tags # Set own Modality and SOP Class UID # Modality, e.g., # "OT" for PDF # "SR" for Structured Report. # Media Storage SOP Class UID, e.g., # "1.2.840.10008." for Encapsulated PDF Storage, # "1.2.840.10008." for Comprehensive 3D SR IOD # "1.2.840.10008." for Segmentation Storage self.modality_type = "SR" self.sop_class_uid = "1.2.840.10008." self.implementation_version_name = "MONAI Deploy App SDK 0.2" self.operators_name = f"AI Algorithm {}"
[docs] def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext): """Performs computation for this operator and handles I/O. For now, only a single image object or result content is supported and the selected DICOM series for inference is required, because the generated IOD needs to refer to original instance. When there are multiple selected series in the input, the first series' containing study will be used for retrieving DICOM Study module attributes, e.g. StudyInstanceUID. Raises: FileNotFoundError: When result object not in the input, and result file not found either. ValueError: Neither image object nor image file's folder is in the input, or no selected series. """ # Gets the input, prepares the output folder, and then delegates the processing. result_text = "" try: result_text = str(op_input.get("classification_result")).strip() except ItemNotExistsError: try: file_path = op_input.get("classification_result_file") except ItemNotExistsError: raise ValueError("None of the named inputs for result can be found.") from None # Read file, and if exception, let it bubble up with open(file_path.path, "r") as f: result_text = if not result_text: raise IOError("Input is read but blank.") try: study_selected_series_list = op_input.get("study_selected_series_list") except ItemNotExistsError: study_selected_series_list = None dicom_series = None # It can be None of copy_tags is false. if self.copy_tags: # Get the first DICOM Series, as for now, only expecting this. if not study_selected_series_list or len(study_selected_series_list) < 1: raise ValueError("Missing input, list of 'StudySelectedSeries'.") for study_selected_series in study_selected_series_list: if not isinstance(study_selected_series, StudySelectedSeries): raise ValueError("Element in input is not expected type, 'StudySelectedSeries'.") for selected_series in study_selected_series.selected_series: dicom_series = selected_series.series output_dir = op_output.get().path output_dir.mkdir(parents=True, exist_ok=True) # Now ready to starting writing the DICOM instance self.write(result_text, dicom_series, output_dir)
[docs] def write(self, content_text, dicom_series: Union[DICOMSeries, None], output_dir: Path): """Writes DICOM object Args: content_file (str): file containing the contents dicom_series (DicomSeries): DicomSeries object encapsulating the original series. model_info (MoelInfo): Object encapsulating model creator, name, version and UID. Returns: PyDicom Dataset """ self._logger.debug("Writing DICOM object...\n{}") output_dir.mkdir(parents=True, exist_ok=True) # Just in case ds = DICOMTextSRWriterOperator.write_common_modules( dicom_series, self.copy_tags, self.modality_type, self.sop_class_uid, self.model_info, self.equipment_info ) # SR specific ds.VerificationFlag = "UNVERIFIED" # Not attested by a legally accountable person. # Per recommendation of IHE Radiology Technical Framework Supplement # AI Results (AIR) Rev1.1 - Trial Implementation # Specifically for Qualitative Findings, # Qualitative findings shall be encoded in an instance of the DICOM Comprehensive 3D SR SOP # Class using TID 1500 (Measurement Report) as the root template. # DICOM PS3.16: TID 1500 Measurement Report # # The value for Procedure Reported (121058, DCM, "Procedure reported") shall describe the # imaging procedure analyzed, not the algorithm used. # Use text value for example ds.ValueType = "TEXT" # ConceptNameCode Sequence seq_concept_name_code = Sequence() ds_concept_name_code = Dataset() ds_concept_name_code.CodeValue = "18748-4" ds_concept_name_code.CodingSchemeDesignator = "LN" ds_concept_name_code.CodeMeaning = "Diagnostic Imaging Report" seq_concept_name_code.append(ds_concept_name_code) ds.ConceptNameCodeSequence = seq_concept_name_code ds.TextValue = content_text # self._content_to_string(content_file) # For now, only allow str Keywords and str value if self.custom_tags: for k, v in self.custom_tags.items(): if isinstance(k, str) and isinstance(v, str): try: ds.update({k: v}) except Exception as ex: # Best effort for now. logging.warning(f"Tag {k} was not written, due to {ex}") # Create the dcm file name, based on series instance UID, then save it. file_name = f"{ds.SeriesInstanceUID}_{ds.Modality}{DICOMTextSRWriterOperator.DCM_EXTENSION}" file_path = output_dir.joinpath(file_name) self.save_dcm_file(ds, file_path)
@staticmethod def save_dcm_file(data_set, file_path: Path, validate_readable: bool = True): logging.debug(f"DICOM dataset to be written:{data_set}") # Write out the DCM file if file_path: dcmwrite(str(file_path), data_set, write_like_original=False)"Finished writing DICOM instance to file {file_path}") if validate_readable: # Test reading back _ = dcmread(str(file_path)) # TODO: The following function can be moved into Domain module as it's common.
[docs] @staticmethod def write_common_modules( dicom_series: Union[DICOMSeries, None], copy_tags: bool, modality_type: str, sop_class_uid: str, model_info: Union[ModelInfo, None] = None, equipment_info: Union[EquipmentInfo, None] = None, ): """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") # 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(DICOMTextSRWriterOperator.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 = "MONAI Deploy App SDK 0.2" # 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 # 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.ManufacturerModel = equipment_info.manufacturer_model ds.SeriesNumber = equipment_info.series_number ds.SoftwareVersionNumber = 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.ManufacturerModel = # u'(111003, DCM, “Algorithm Version")' ds_contributing_equipment.SoftwareVersionNumber = 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
[docs] @staticmethod 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 test(): from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator from monai.deploy.operators.dicom_series_selector_operator import DICOMSeriesSelectorOperator current_file_dir = Path(__file__).parent.resolve() data_path = current_file_dir.joinpath("../../../examples/ai_spleen_seg_data/dcm") out_path = current_file_dir.joinpath("../../../examples/output_sr_op") test_report_text = "Dummy AI classification resutls." test_copy_tags = True loader = DICOMDataLoaderOperator() series_selector = DICOMSeriesSelectorOperator() sr_writer = DICOMTextSRWriterOperator( copy_tags=test_copy_tags, model_info=None, custom_tags={"SeriesDescription": "New AI Series"} ) # Testing with the main entry functions dicom_series = None if test_copy_tags: study_list = loader.load_data_to_studies(Path(data_path).absolute()) study_selected_series_list = series_selector.filter(None, study_list) # Get the first DICOM Series, as for now, only expecting this. if not study_selected_series_list or len(study_selected_series_list) < 1: raise ValueError("Missing input, list of 'StudySelectedSeries'.") for study_selected_series in study_selected_series_list: if not isinstance(study_selected_series, StudySelectedSeries): raise ValueError("Element in input is not expected type, 'StudySelectedSeries'.") for selected_series in study_selected_series.selected_series: print(type(selected_series)) dicom_series = selected_series.series print(type(dicom_series)) sr_writer.write(test_report_text, dicom_series, Path(out_path).absolute()) if __name__ == "__main__": test()