Source code for monai.deploy.operators.dicom_series_selector_operator

# Copyright 2021-2023 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
#     http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
import numbers
import re
from json import loads as json_loads
from typing import List

from monai.deploy.core import ConditionType, Fragment, Operator, OperatorSpec
from monai.deploy.core.domain.dicom_series import DICOMSeries
from monai.deploy.core.domain.dicom_series_selection import SelectedSeries, StudySelectedSeries
from monai.deploy.core.domain.dicom_study import DICOMStudy


[docs]class DICOMSeriesSelectorOperator(Operator): """This operator selects a list of DICOM Series in a DICOM Study for a given set of selection rules. Named input: dicom_study_list: A list of DICOMStudy objects. Named output: study_selected_series_list: A list of StudySelectedSeries objects. Downstream receiver optional. This class can be considered a base class, and a derived class can override the 'filer' function to with custom logics. In its default implementation, this class 1. selects a series or all matched series within the scope of a study in a list of studies 2. uses rules defined in JSON string, see below for details 3. supports DICOM Study and Series module attribute matching, including regex for string types 4. supports multiple named selections, in the scope of each DICOM study 5. outputs a list of SutdySelectedSeries objects, as well as a flat list of SelectedSeries (to be deprecated) The selection rules are defined in JSON, 1. attribute "selections" value is a list of selections 2. each selection has a "name", and its "conditions" value is a list of matching criteria 3. each condition has uses the implicit equal operator, except for regex for str type and union for set type 4. DICOM attribute keywords are used, and only for those defined as DICOMStudy and DICOMSeries properties An example selection rules: { "selections": [ { "name": "CT Series 1", "conditions": { "StudyDescription": "(?i)^Spleen", "Modality": "(?i)CT", "SeriesDescription": "(?i)^No series description|(.*?)" } }, { "name": "CT Series 2", "conditions": { "Modality": "CT", "BodyPartExamined": "Abdomen", "SeriesDescription" : "Not to be matched. For illustration only." } } ] } """
[docs] def __init__(self, fragment: Fragment, *args, rules: str = "", all_matched: bool = False, **kwargs) -> None: """Instantiate an instance. Args: fragment (Fragment): An instance of the Application class which is derived from Fragment. rules (Text): Selection rules in JSON string. all_matched (bool): Gets all matched series in a study. Defaults to False for first match only. """ # rules: Text = "", all_matched: bool = False, # Delay loading the rules as JSON string till compute time. self._rules_json_str = rules if rules and rules.strip() else None self._all_matched = all_matched # all_matched self.input_name_study_list = "dicom_study_list" self.output_name_selected_series = "study_selected_series_list" super().__init__(fragment, *args, **kwargs)
[docs] def setup(self, spec: OperatorSpec): spec.input(self.input_name_study_list) spec.output(self.output_name_selected_series).condition(ConditionType.NONE) # Receiver optional
# Can use the config file to alter the selection rules per app run # spec.param("selection_rules")
[docs] def compute(self, op_input, op_output, context): """Performs computation for this operator.""" dicom_study_list = op_input.receive(self.input_name_study_list) selection_rules = self._load_rules() if self._rules_json_str else None study_selected_series = self.filter(selection_rules, dicom_study_list, self._all_matched) op_output.emit(study_selected_series, self.output_name_selected_series)
[docs] def filter(self, selection_rules, dicom_study_list, all_matched: bool = False) -> List[StudySelectedSeries]: """Selects the series with the given matching rules. If rules object is None, all series will be returned with series instance UID as the selection name. Simplistic matching is used for demonstration: Number: exactly matches String: matches case insensitive, if fails then tries RegEx search String array matches as subset, case insensitive Args: selection_rules (object): JSON object containing the matching rules. dicom_study_list (list): A list of DICOMStudiy objects. all_matched (bool): Gets all matched series in a study. Defaults to False for first match only. Returns: list: A list of objects of type StudySelectedSeries. Raises: ValueError: If the selection_rules object does not contain "selections" attribute. """ if not dicom_study_list or len(dicom_study_list) < 1: return [] if not selection_rules: # Return all series if no selection rules are supplied logging.warn("No selection rules given; select all series.") return self._select_all_series(dicom_study_list) selections = selection_rules.get("selections", None) # TODO type is not json now. # If missing selections in the rules then it is an error. if not selections: raise ValueError('Expected "selections" not found in the rules.') study_selected_series_list = [] # List of StudySelectedSeries objects for study in dicom_study_list: study_selected_series = StudySelectedSeries(study) for selection in selections: # Get the selection name. Blank name will be handled by the SelectedSeries selection_name = selection.get("name", "").strip() logging.info(f"Finding series for Selection named: {selection_name}") # Skip if no selection conditions are provided. conditions = selection.get("conditions", None) if not conditions: continue # Select only the first series that matches the conditions, list of one series_list = self._select_series(conditions, study, all_matched) if series_list and len(series_list) > 0: for series in series_list: selected_series = SelectedSeries(selection_name, series, None) # No Image obj yet. study_selected_series.add_selected_series(selected_series) if len(study_selected_series.selected_series) > 0: study_selected_series_list.append(study_selected_series) return study_selected_series_list
def _load_rules(self): return json_loads(self._rules_json_str) if self._rules_json_str else None def _select_all_series(self, dicom_study_list: List[DICOMStudy]) -> List[StudySelectedSeries]: """Select all series in studies Returns: list: list of StudySelectedSeries objects """ study_selected_series_list = [] for study in dicom_study_list: logging.info(f"Working on study, instance UID: {study.StudyInstanceUID}") study_selected_series = StudySelectedSeries(study) for series in study.get_all_series(): logging.info(f"Working on series, instance UID: {str(series.SeriesInstanceUID)}") selected_series = SelectedSeries("", series, None) # No selection name or Image obj. study_selected_series.add_selected_series(selected_series) study_selected_series_list.append(study_selected_series) return study_selected_series_list def _select_series(self, attributes: dict, study: DICOMStudy, all_matched=False) -> List[DICOMSeries]: """Finds series whose attributes match the given attributes. Args: attributes (dict): Dictionary of attributes for matching all_matched (bool): Gets all matched series in a study. Defaults to False for first match only. Returns: List of DICOMSeries. At most one element if all_matched is False. """ assert isinstance(attributes, dict), '"attributes" must be a dict.' logging.info(f"Searching study, : {study.StudyInstanceUID}\n # of series: {len(study.get_all_series())}") study_attr = self._get_instance_properties(study) found_series = [] for series in study.get_all_series(): logging.info(f"Working on series, instance UID: {series.SeriesInstanceUID}") # Combine Study and current Series properties for matching series_attr = self._get_instance_properties(series) series_attr.update(study_attr) matched = True # Simple matching on attribute value for key, value_to_match in attributes.items(): logging.info(f"On attribute: {key!r} to match value: {value_to_match!r}") # Ignore None if not value_to_match: continue # Try getting the attribute value from Study and current Series prop dict attr_value = series_attr.get(key, None) logging.info(f" Series attribute {key} value: {attr_value}") # If not found, try the best at the native instance level for string VR # This is mainly for attributes like ImageType if not attr_value: try: # Can use some enhancements, especially multi-value where VM > 1 elem = series.get_sop_instances()[0].get_native_sop_instance()[key] if elem.VM > 1: attr_value = [elem.repval] # repval: str representation of the element’s value else: attr_value = elem.value # element's value series_attr.update({key: attr_value}) except Exception: logging.info(f" Attribute {key} not at instance level either.") if not attr_value: matched = False elif isinstance(attr_value, numbers.Number): matched = value_to_match == attr_value elif isinstance(attr_value, str): matched = attr_value.casefold() == (value_to_match.casefold()) if not matched: # For str, also try RegEx search to check for a match anywhere in the string # unless the user constrains it in the expression. logging.info("Series attribute string value did not match. Try regEx.") if re.search(value_to_match, attr_value, re.IGNORECASE): matched = True elif isinstance(attr_value, list): # Assume multi value string attributes meta_data_list = str(attr_value).lower() if isinstance(value_to_match, list): value_set = {str(element).lower() for element in value_to_match} matched = all(val in meta_data_list for val in value_set) elif isinstance(value_to_match, (str, numbers.Number)): matched = str(value_to_match).lower() in meta_data_list else: raise NotImplementedError(f"Not support for matching on this type: {type(value_to_match)}") if not matched: logging.info("This series does not match the selection conditions.") break if matched: logging.info(f"Selected Series, UID: {series.SeriesInstanceUID}") found_series.append(series) if not all_matched: return found_series return found_series @staticmethod def _get_instance_properties(obj: object): if not obj: return {} else: return {x: getattr(obj, x, None) for x in type(obj).__dict__ if isinstance(type(obj).__dict__[x], property)}
# Module functions # Helper function to get console output of the selection content when testing the script def _print_instance_properties(obj: object, pre_fix: str = "", print_val=True): print(f"{pre_fix}Instance of {type(obj)}") for attribute in [x for x in type(obj).__dict__ if isinstance(type(obj).__dict__[x], property)]: attr_val = getattr(obj, attribute, None) print(f"{pre_fix} {attribute}: {type(attr_val)} {attr_val if print_val else ''}") def test(): from pathlib import Path from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator current_file_dir = Path(__file__).parent.resolve() data_path = current_file_dir.joinpath("../../../inputs/spleen_ct/dcm").absolute() fragment = Fragment() loader = DICOMDataLoaderOperator(fragment, name="loader_op") selector = DICOMSeriesSelectorOperator(fragment, name="selector_op") study_list = loader.load_data_to_studies(data_path) sample_selection_rule = json_loads(Sample_Rules_Text) print(f"Selection rules in JSON:\n{sample_selection_rule}") study_selected_seriee_list = selector.filter(sample_selection_rule, study_list) for sss_obj in study_selected_seriee_list: _print_instance_properties(sss_obj, pre_fix="", print_val=False) study = sss_obj.study pre_fix = " " print(f"{pre_fix}==== Details of the study ====") _print_instance_properties(study, pre_fix, print_val=False) print(f"{pre_fix}==============================") # The following commented code block accesses and prints the flat list of all selected series. # for ss_obj in sss_obj.selected_series: # pre_fix = " " # _print_instance_properties(ss_obj, pre_fix, print_val=False) # pre_fix = " " # print(f"{pre_fix}==== Details of the series ====") # _print_instance_properties(ss_obj, pre_fix) # print(f"{pre_fix}===============================") # The following block uses hierarchical grouping by selection name, and prints the list of series for each. for selection_name, ss_list in sss_obj.series_by_selection_name.items(): pre_fix = " " print(f"{pre_fix}Selection name: {selection_name}") for ss_obj in ss_list: pre_fix = " " _print_instance_properties(ss_obj, pre_fix, print_val=False) print(f"{pre_fix}==== Details of the series ====") _print_instance_properties(ss_obj, pre_fix) print(f"{pre_fix}===============================") print(f" A total of {len(sss_obj.selected_series)} series selected for study {study.StudyInstanceUID}") # Sample rule used for testing Sample_Rules_Text = """ { "selections": [ { "name": "CT Series 1", "conditions": { "StudyDescription": "(?i)^Spleen", "Modality": "(?i)CT", "SeriesDescription": "(?i)^No series description|(.*?)" } }, { "name": "CT Series 2", "conditions": { "Modality": "CT", "BodyPartExamined": "Abdomen", "SeriesDescription" : "Not to be matched" } } ] } """ if __name__ == "__main__": test()