6834 Add MetaProterties and update load API

monai/bundle/__init__.py

@@ -16,6 +16,7 @@
 from .properties import InferProperties, TrainProperties
 from .reference_resolver import ReferenceResolver
 from .scripts import (
+    BundleManager,
     ckpt_export,
     download,
     get_all_bundles_list,

monai/bundle/scripts.py

@@ -1539,3 +1539,126 @@ def init_bundle(
         copyfile(str(ckpt_file), str(models_dir / "model.pt"))
     elif network is not None:
         save_state(network, str(models_dir / "model.pt"))
+
+
+def _find_bundle_file(root_dir: Path, file_name: str, suffix: Sequence[str] = ("json", "yaml", "yml")) -> str | None:
+    # find bundle file with possible suffix
+    for _suffix in suffix:
+        full_name = f"{file_name}.{_suffix}"
+        if full_name in os.listdir(root_dir):
+            return full_name
+    return None
+
+
+class BundleManager:
+    """
+    The `BundleManager` class facilitates the automatic downloading and instantiation of bundles.
+    It allows users to retrieve bundle properties and meta information.
+
+    Typical usage examples:
+
+    .. code-block:: python
+
+        from monai.bundle import BundleManager
+
+        # Create a BundleManager instance for the 'spleen_ct_segmentation' bundle
+        bundle = BundleManager("spleen_ct_segmentation")
+
+        # Get properties defined in `TrainProperties` or `InferProperties`
+        train_preprocessing = bundle.get("train_preprocessing")
+        print(train_preprocessing)
+
+        # Also support to retrieve meta information from the "metadata.json" file
+        version = bundle.get(meta="version")
+        print(version)
+
+    Args:
+        bundle_name: the name of the bundle. Defaults to None.
+        config_path: file path of the bundle. If a list of file paths is provided,
+            their contents will be merged. Defaults to None.
+        bundle_dir: the target directory to store downloaded bundle.
+            Defaults to the 'bundle' subfolder under `torch.hub.get_dir()`.
+        target directory to store the downloaded data.
+            Default is `bundle` subfolder under `torch.hub.get_dir()`.
+        configs: The name of the config file(s), supporting multiple names.
+            Defaults to "train".
+        kwargs: Additional arguments for download or workflow class instantiation.
+    """
+
+    def __init__(
+        self,
+        bundle_name: str | None = None,
+        config_path: str | Sequence[str] | None = None,
+        bundle_dir: PathLike | None = None,
+        configs: str | Sequence[str] = "train",
+        **kwargs: Any,
+    ) -> None:
+        if bundle_name is None and config_path is None:
+            raise ValueError("Must specify bundle_name or config_path.")
+        configs = ensure_tuple(configs)
+        if "train" in configs:
+            workflow = "train"
+        else:
+            workflow = "infer"
+
+        if config_path is not None:
+            _config_path = Path(ensure_tuple(config_path)[0])
+            if _config_path.is_file():
+                config_file = config_path
+                config_root_path = _config_path.parent
+            else:
+                raise FileNotFoundError(f"Cannot find the config file: {config_path}.")
+        else:
+            download_args = {
+                "version": kwargs.pop("version", None),
+                "args_file": kwargs.pop("args_file", None),
+                "source": kwargs.pop("source", download_source),
+            }
+            download(bundle_name, bundle_dir=bundle_dir, **download_args)
+            bundle_dir = _process_bundle_dir(bundle_dir)
+            config_root_path = bundle_dir / bundle_name / "configs"  # type: ignore
+            if len(configs) > 0:
+                config_file = [
+                    str(config_root_path / _find_bundle_file(config_root_path, _config)) for _config in configs  # type: ignore
+                ]
+            else:
+                config_file = str(config_root_path / _find_bundle_file(config_root_path, configs[0]))  # type: ignore
+
+        logging_file = config_root_path / "logging.conf"
+        self.meta_file = config_root_path / "metadata.json"
+
+        self.workflow = ConfigWorkflow(
+            config_file=config_file,
+            meta_file=str(self.meta_file),
+            logging_file=str(logging_file),
+            workflow=workflow,
+            **kwargs,
+        )
+        self.workflow.initialize()
+
+    def get(self, property: str | None = None, meta: str | None = None) -> Any:
+        """
+        Get information from the bundle.
+
+        Args:
+            property: the target property, defined in `TrainProperties` or `InferProperties`.
+            meta: meta information retrieved from the "metadata.json" file, such as version, changelog, etc.
+
+        """
+        if property is not None and meta is not None:
+            raise ValueError("Incompatible values: both property and meta are specified.")
+        if property is not None:
+            if property in self.workflow.properties:  # type: ignore
+                return getattr(self.workflow, property)
+            raise ValueError(f"Missing property {property} in the bundle.")
+        if meta is not None:
+            metadata = ConfigParser.load_config_files(files=self.meta_file)
+            if meta.lower() in metadata.keys():
+                return {meta: metadata[meta.lower()]}
+            raise ValueError(f"Missing meta {meta} informtation in metadata.json.")
+
+    def train(self):
+        pass
+
+    def predict(self):
+        pass