|
| 1 | +# Copyright (c) MONAI Consortium |
| 2 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 3 | +# you may not use this file except in compliance with the License. |
| 4 | +# You may obtain a copy of the License at |
| 5 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 6 | +# Unless required by applicable law or agreed to in writing, software |
| 7 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 8 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 9 | +# See the License for the specific language governing permissions and |
| 10 | +# limitations under the License. |
| 11 | + |
| 12 | + |
| 13 | +import json |
| 14 | +import os |
| 15 | +from monai.apps.utils import extractall |
| 16 | +from typing import Sequence |
| 17 | + |
| 18 | +from monai.utils import ensure_tuple_rep |
| 19 | + |
| 20 | + |
| 21 | +def convert_body_region(body_region: str | Sequence[str]) -> Sequence[int]: |
| 22 | + """ |
| 23 | + Convert body region string to body region index. |
| 24 | + Args: |
| 25 | + body_region: list of input body region string. If single str, will be converted to list of str. |
| 26 | + Return: |
| 27 | + body_region_indices, list of input body region index. |
| 28 | + """ |
| 29 | + if type(body_region) is str: |
| 30 | + body_region = [body_region] |
| 31 | + |
| 32 | + # body region mapping for maisi |
| 33 | + region_mapping_maisi = { |
| 34 | + "head": 0, |
| 35 | + "chest": 1, |
| 36 | + "thorax": 1, |
| 37 | + "chest/thorax": 1, |
| 38 | + "abdomen": 2, |
| 39 | + "pelvis": 3, |
| 40 | + "lower": 3, |
| 41 | + "pelvis/lower": 3, |
| 42 | + } |
| 43 | + |
| 44 | + # perform mapping |
| 45 | + body_region_indices = [] |
| 46 | + for region in body_region: |
| 47 | + normalized_region = region.lower() # norm str to lower case |
| 48 | + if normalized_region not in region_mapping_maisi: |
| 49 | + raise ValueError(f"Invalid region: {normalized_region}") |
| 50 | + body_region_indices.append(region_mapping_maisi[normalized_region]) |
| 51 | + |
| 52 | + return body_region_indices |
| 53 | + |
| 54 | + |
| 55 | +def find_masks( |
| 56 | + body_region: str | Sequence[str], |
| 57 | + anatomy_list: int | Sequence[int], |
| 58 | + spacing: Sequence[float] | float = 1.0, |
| 59 | + output_size: Sequence[int] = [512, 512, 512], |
| 60 | + check_spacing_and_output_size: bool = False, |
| 61 | + database_filepath: str = "./data/database.json", |
| 62 | + mask_foldername: str = "./data/masks/", |
| 63 | +): |
| 64 | + """ |
| 65 | + Find candidate masks that fullfills all the requirements. |
| 66 | + They shoud contain all the body region in `body_region`, all the anatomies in `anatomy_list`. |
| 67 | + If there is no tumor specified in `anatomy_list`, we also expect the candidate masks to be tumor free. |
| 68 | + If check_spacing_and_output_size is True, the candidate masks need to have the expected `spacing` and `output_size`. |
| 69 | + Args: |
| 70 | + body_region: list of input body region string. If single str, will be converted to list of str. |
| 71 | + The found candidate mask will include these body regions. |
| 72 | + anatomy_list: list of input anatomy. The found candidate mask will include these anatomies. |
| 73 | + spacing: list of three floats, voxel spacing. If providing a single number, will use it for all the three dimensions. |
| 74 | + output_size: list of three int, expected candidate mask spatial size. |
| 75 | + check_spacing_and_output_size: whether we expect candidate mask to have spatial size of `output_size` and voxel size of `spacing`. |
| 76 | + database_filepath: path for the json file that stores the information of all the candidate masks. |
| 77 | + mask_foldername: directory that saves all the candidate masks. |
| 78 | + Return: |
| 79 | + candidate_masks, list of dict, each dict contains information of one candidate mask that fullfills all the requirements. |
| 80 | + """ |
| 81 | + # check and preprocess input |
| 82 | + body_region = convert_body_region(body_region) |
| 83 | + |
| 84 | + if isinstance(anatomy_list, int): |
| 85 | + anatomy_list = [anatomy_list] |
| 86 | + |
| 87 | + spacing = ensure_tuple_rep(spacing, 3) |
| 88 | + |
| 89 | + if not os.path.exists(mask_foldername): |
| 90 | + zip_file_path = mask_foldername + ".zip" |
| 91 | + |
| 92 | + if not os.path.isfile(zip_file_path): |
| 93 | + raise ValueError(f"Please download {zip_file_path} following the instruction in ./data/README.md.") |
| 94 | + |
| 95 | + print(f"Extracting {zip_file_path}...") |
| 96 | + extractall(filepath=mask_foldername, output_dir=mask_foldername) |
| 97 | + print(f"Unzipped {zip_file_path} to {mask_foldername}.") |
| 98 | + |
| 99 | + if not os.path.isfile(database_filepath): |
| 100 | + raise ValueError(f"Please download {database_filepath} following the instruction in ./data/README.md.") |
| 101 | + with open(database_filepath, "r") as f: |
| 102 | + db = json.load(f) |
| 103 | + |
| 104 | + # select candidate_masks |
| 105 | + candidate_masks = [] |
| 106 | + for _item in db: |
| 107 | + if not set(anatomy_list).issubset(_item["label_list"]): |
| 108 | + continue |
| 109 | + |
| 110 | + # extract region indice (top_index and bottom_index) for candidate mask |
| 111 | + top_index = [index for index, element in enumerate(_item["top_region_index"]) if element != 0] |
| 112 | + top_index = top_index[0] |
| 113 | + bottom_index = [index for index, element in enumerate(_item["bottom_region_index"]) if element != 0] |
| 114 | + bottom_index = bottom_index[0] |
| 115 | + |
| 116 | + # whether to keep this mask, default to be True. |
| 117 | + keep_mask = True |
| 118 | + |
| 119 | + # if candiate mask does not contain all the body_region, skip it |
| 120 | + for _idx in body_region: |
| 121 | + if _idx > bottom_index or _idx < top_index: |
| 122 | + keep_mask = False |
| 123 | + |
| 124 | + for tumor_label in [23, 24, 26, 27, 128]: |
| 125 | + # we skip those mask with tumors if users do not provide tumor label in anatomy_list |
| 126 | + if tumor_label not in anatomy_list and tumor_label in _item["label_list"]: |
| 127 | + keep_mask = False |
| 128 | + |
| 129 | + if check_spacing_and_output_size: |
| 130 | + # if the output_size and spacing are different with user's input, skip it |
| 131 | + for axis in range(3): |
| 132 | + if _item["dim"][axis] != output_size[axis] or _item["spacing"][axis] != spacing[axis]: |
| 133 | + keep_mask = False |
| 134 | + |
| 135 | + if keep_mask: |
| 136 | + # if decide to keep this mask, we pack the information of this mask and add to final output. |
| 137 | + candidate = { |
| 138 | + "pseudo_label": os.path.join(mask_foldername, _item["pseudo_label_filename"]), |
| 139 | + "spacing": _item["spacing"], |
| 140 | + "dim": _item["dim"], |
| 141 | + "top_region_index": _item["top_region_index"], |
| 142 | + "bottom_region_index": _item["bottom_region_index"], |
| 143 | + } |
| 144 | + |
| 145 | + # Conditionally add the label to the candidate dictionary |
| 146 | + if "label_filename" in _item: |
| 147 | + candidate["label"] = os.path.join(mask_foldername, _item["label_filename"]) |
| 148 | + |
| 149 | + candidate_masks.append(candidate) |
| 150 | + |
| 151 | + if len(candidate_masks) == 0 and not check_spacing_and_output_size: |
| 152 | + raise ValueError("Cannot find body region with given anatomy list.") |
| 153 | + |
| 154 | + return candidate_masks |
0 commit comments