|
15 | 15 | from pathlib import Path |
16 | 16 | from typing import Dict, Optional, Union |
17 | 17 |
|
| 18 | +from monai.deploy.core import ConditionType, Fragment, Operator, OperatorSpec |
| 19 | +from monai.deploy.core.domain.dicom_series import DICOMSeries |
| 20 | +from monai.deploy.core.domain.dicom_series_selection import StudySelectedSeries |
| 21 | +from monai.deploy.operators.dicom_utils import EquipmentInfo, ModelInfo, save_dcm_file, write_common_modules |
| 22 | +from monai.deploy.utils.version import get_sdk_semver |
18 | 23 | from monai.deploy.utils.importutil import optional_import |
19 | 24 |
|
20 | 25 | dcmread, _ = optional_import("pydicom", name="dcmread") |
|
24 | 29 | Dataset, _ = optional_import("pydicom.dataset", name="Dataset") |
25 | 30 | FileDataset, _ = optional_import("pydicom.dataset", name="FileDataset") |
26 | 31 | Sequence, _ = optional_import("pydicom.sequence", name="Sequence") |
27 | | -PdfReader, _ = optional_import("PyPDF2", name="PdfReader") |
28 | | - |
29 | | -from monai.deploy.core import ConditionType, Fragment, Operator, OperatorSpec |
30 | | -from monai.deploy.core.domain.dicom_series import DICOMSeries |
31 | | -from monai.deploy.core.domain.dicom_series_selection import StudySelectedSeries |
32 | | -from monai.deploy.operators.dicom_utils import EquipmentInfo, ModelInfo, save_dcm_file, write_common_modules |
33 | | -from monai.deploy.utils.version import get_sdk_semver |
| 32 | +PdfReader, _ = optional_import("pypdf", name="PdfReader") |
34 | 33 |
|
35 | 34 |
|
36 | 35 | # @md.env(pip_packages=["pydicom >= 1.4.2", "PyPDF2 >= 2.11.1", "monai"]) |
@@ -58,7 +57,7 @@ def __init__( |
58 | 57 | fragment: Fragment, |
59 | 58 | *args, |
60 | 59 | output_folder: Union[str, Path], |
61 | | - model_info: ModelInfo, |
| 60 | + model_info: Optional[ModelInfo] = None, |
62 | 61 | equipment_info: Optional[EquipmentInfo] = None, |
63 | 62 | copy_tags: bool = True, |
64 | 63 | custom_tags: Optional[Dict[str, str]] = None, |
@@ -249,52 +248,51 @@ def _is_pdf_bytes(self, content: bytes): |
249 | 248 | return True |
250 | 249 |
|
251 | 250 |
|
252 | | -# Commenting out the following as pttype complains about the constructor for no reason |
253 | | -# def test(test_copy_tags: bool = True): |
254 | | -# from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator |
255 | | -# from monai.deploy.operators.dicom_series_selector_operator import DICOMSeriesSelectorOperator |
256 | | - |
257 | | -# current_file_dir = Path(__file__).parent.resolve() |
258 | | -# dcm_folder = current_file_dir.joinpath("../../../inputs/livertumor_ct/dcm/1-CT_series_liver_tumor_from_nii014") |
259 | | -# pdf_file = current_file_dir.joinpath("../../../inputs/pdf/TestPDF.pdf") |
260 | | -# out_path = Path("output_pdf_op").absolute() |
261 | | -# pdf_bytes = b"Not PDF bytes." |
262 | | - |
263 | | -# fragment = Fragment() |
264 | | -# loader = DICOMDataLoaderOperator(fragment, name="loader_op") |
265 | | -# series_selector = DICOMSeriesSelectorOperator(fragment, name="selector_op") |
266 | | -# sr_writer = DICOMEncapsulatedPDFWriterOperator( |
267 | | -# fragment, |
268 | | -# output_folder=out_path, |
269 | | -# copy_tags=test_copy_tags, |
270 | | -# model_info=None, |
271 | | -# equipment_info=EquipmentInfo(), |
272 | | -# custom_tags={"SeriesDescription": "Report from AI algorithm. Not for clinical use."}, |
273 | | -# name="writer_op", |
274 | | -# ) |
275 | | - |
276 | | -# # Testing with the main entry functions |
277 | | -# dicom_series = None |
278 | | -# if test_copy_tags: |
279 | | -# study_list = loader.load_data_to_studies(Path(dcm_folder).absolute()) |
280 | | -# study_selected_series_list = series_selector.filter(None, study_list) |
281 | | -# # Get the first DICOM Series, as for now, only expecting this. |
282 | | -# if not study_selected_series_list or len(study_selected_series_list) < 1: |
283 | | -# raise ValueError("Missing input, list of 'StudySelectedSeries'.") |
284 | | -# for study_selected_series in study_selected_series_list: |
285 | | -# if not isinstance(study_selected_series, StudySelectedSeries): |
286 | | -# raise ValueError("Element in input is not expected type, 'StudySelectedSeries'.") |
287 | | -# for selected_series in study_selected_series.selected_series: |
288 | | -# print(type(selected_series)) |
289 | | -# dicom_series = selected_series.series |
290 | | -# print(type(dicom_series)) |
291 | | - |
292 | | -# with open(pdf_file, "rb") as f: |
293 | | -# pdf_bytes = f.read() |
294 | | - |
295 | | -# sr_writer.write(pdf_bytes, dicom_series, out_path) |
296 | | - |
297 | | - |
298 | | -# if __name__ == "__main__": |
299 | | -# test(test_copy_tags=True) |
300 | | -# test(test_copy_tags=False) |
| 251 | +def test(test_copy_tags: bool = True): |
| 252 | + from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator |
| 253 | + from monai.deploy.operators.dicom_series_selector_operator import DICOMSeriesSelectorOperator |
| 254 | + |
| 255 | + current_file_dir = Path(__file__).parent.resolve() |
| 256 | + dcm_folder = current_file_dir.joinpath("../../../inputs/livertumor_ct/dcm/1-CT_series_liver_tumor_from_nii014") |
| 257 | + pdf_file = current_file_dir.joinpath("../../../inputs/pdf/TestPDF.pdf") |
| 258 | + out_path = Path("output_pdf_op").absolute() |
| 259 | + pdf_bytes = b"Not PDF bytes." |
| 260 | + |
| 261 | + fragment = Fragment() |
| 262 | + loader = DICOMDataLoaderOperator(fragment, name="loader_op") |
| 263 | + series_selector = DICOMSeriesSelectorOperator(fragment, name="selector_op") |
| 264 | + sr_writer = DICOMEncapsulatedPDFWriterOperator( |
| 265 | + fragment, |
| 266 | + output_folder=out_path, |
| 267 | + copy_tags=test_copy_tags, |
| 268 | + model_info=None, |
| 269 | + equipment_info=EquipmentInfo(), |
| 270 | + custom_tags={"SeriesDescription": "Report from AI algorithm. Not for clinical use."}, |
| 271 | + name="writer_op", |
| 272 | + ) |
| 273 | + |
| 274 | + # Testing with the main entry functions |
| 275 | + dicom_series = None |
| 276 | + if test_copy_tags: |
| 277 | + study_list = loader.load_data_to_studies(Path(dcm_folder).absolute()) |
| 278 | + study_selected_series_list = series_selector.filter(None, study_list) |
| 279 | + # Get the first DICOM Series, as for now, only expecting this. |
| 280 | + if not study_selected_series_list or len(study_selected_series_list) < 1: |
| 281 | + raise ValueError("Missing input, list of 'StudySelectedSeries'.") |
| 282 | + for study_selected_series in study_selected_series_list: |
| 283 | + if not isinstance(study_selected_series, StudySelectedSeries): |
| 284 | + raise ValueError("Element in input is not expected type, 'StudySelectedSeries'.") |
| 285 | + for selected_series in study_selected_series.selected_series: |
| 286 | + print(type(selected_series)) |
| 287 | + dicom_series = selected_series.series |
| 288 | + print(type(dicom_series)) |
| 289 | + |
| 290 | + with open(pdf_file, "rb") as f: |
| 291 | + pdf_bytes = f.read() |
| 292 | + |
| 293 | + sr_writer.write(pdf_bytes, dicom_series, out_path) |
| 294 | + |
| 295 | + |
| 296 | +if __name__ == "__main__": |
| 297 | + test(test_copy_tags=True) |
| 298 | + test(test_copy_tags=False) |
0 commit comments