diff --git a/monai/bundle/workflows.py b/monai/bundle/workflows.py index ccdb08e208..aadaf5676a 100644 --- a/monai/bundle/workflows.py +++ b/monai/bundle/workflows.py @@ -11,6 +11,7 @@ from __future__ import annotations +import importlib import json import os import sys @@ -48,7 +49,12 @@ class BundleWorkflow(ABC): or "infer", "inference", "eval", "evaluation" for a inference workflow, other unsupported string will raise a ValueError. default to `None` for common workflow. - properties_path: the path to the JSON file of properties. + properties_path: the path to the file of properties. It can be a JSON file or a Python file. + If the file is a JSON file, it should be a dictionary of properties. + If the file is a Python file, it should define the properties in specific dictionaries: + - `TrainProperties` for training workflows. + - `InferProperties` for inference or evaluation workflows. + - `MetaProperties` is optional and can be provided for both types. meta_file: filepath of the metadata file, if this is a list of file paths, their contents will be merged in order. logging_file: config file for `logging` module in the program. for more details: https://docs.python.org/3/library/logging.config.html#logging.config.fileConfig. @@ -105,11 +111,42 @@ def __init__( properties_path = Path(properties_path) if not properties_path.is_file(): raise ValueError(f"Property file {properties_path} does not exist.") - with open(properties_path) as json_file: - self.properties = json.load(json_file) - self.workflow_type = None - self.meta_file = meta_file - return + if properties_path.suffix == ".json": + with open(properties_path) as json_file: + self.properties = json.load(json_file) + self.workflow_type = None + self.meta_file = meta_file + return + elif properties_path.suffix == ".py": + module_name = properties_path.stem + spec = importlib.util.spec_from_file_location(module_name, properties_path) + if spec is None: + raise ImportError(f"Cannot find module specification for {module_name}") + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + if spec.loader is None: + raise ImportError(f"Cannot find loader for {module_name}") + spec.loader.exec_module(module) + self.properties = {} + if workflow_type.lower() in self.supported_train_type: # type: ignore[union-attr] + if not hasattr(module, "TrainProperties"): + raise ValueError("TrainProperties is not defined in the properties file.") + self.properties.update(module.TrainProperties) + self.workflow_type = "train" + self.meta_file = meta_file + elif workflow_type.lower() in self.supported_infer_type: # type: ignore[union-attr] + if not hasattr(module, "InferProperties"): + raise ValueError("InferProperties is not defined in the properties file.") + self.properties.update(module.InferProperties) + self.workflow_type = "infer" + self.meta_file = meta_file + else: + raise ValueError(f"Unsupported workflow type: '{workflow_type}'.") + if hasattr(module, "MetaProperties"): + self.properties.update(module.MetaProperties) + return + else: + raise ValueError(f"Unsupported file type: {properties_path.suffix}. Only support .json or .py.") if workflow_type.lower() in self.supported_train_type: # type: ignore[union-attr] self.properties = {**TrainProperties, **MetaProperties} self.workflow_type = "train"