🚀from_config API: Create a path between API & configuration file (CLI)

CHANGELOG.md

@@ -11,6 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - 🚀 Update OpenVINO and ONNX export to support fixed input shape by @adrianboguszewski in https://github.com/openvinotoolkit/anomalib/pull/2006
 - Add data_path argument to predict entrypoint and add properties for retrieving model path by @djdameln in https://github.com/openvinotoolkit/anomalib/pull/2018
 - 🚀 Add compression and quantization for OpenVINO export by @adrianboguszewski in https://github.com/openvinotoolkit/anomalib/pull/2052
+- 🚀from_config API: Create a path between API & configuration file (CLI) by @harimkang in https://github.com/openvinotoolkit/anomalib/pull/2065
 
 ### Changed
 

configs/model/reverse_distillation.yaml

@@ -6,8 +6,6 @@ model:
       - layer1
       - layer2
       - layer3
-    beta1: 0.5
-    beta2: 0.999
     anomaly_map_mode: ADD
     pre_trained: true
 

src/anomalib/cli/cli.py

@@ -50,7 +50,7 @@ class AnomalibCLI:
     ``SaveConfigCallback`` overwrites the config if it already exists.
     """
 
-    def __init__(self, args: Sequence[str] | None = None) -> None:
+    def __init__(self, args: Sequence[str] | None = None, run: bool = True) -> None:
         self.parser = self.init_parser()
         self.subcommand_parsers: dict[str, ArgumentParser] = {}
         self.subcommand_method_arguments: dict[str, list[str]] = {}
@@ -60,7 +60,8 @@ def __init__(self, args: Sequence[str] | None = None) -> None:
         if _LIGHTNING_AVAILABLE:
             self.before_instantiate_classes()
             self.instantiate_classes()
-        self._run_subcommand()
+        if run:
+            self._run_subcommand()
 
     def init_parser(self, **kwargs) -> ArgumentParser:
         """Method that instantiates the argument parser."""

src/anomalib/data/base/datamodule.py

@@ -6,6 +6,7 @@
 
 import logging
 from abc import ABC, abstractmethod
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from lightning.pytorch import LightningDataModule
@@ -289,3 +290,51 @@ def eval_transform(self) -> Transform:
         if self.image_size:
             return Resize(self.image_size, antialias=True)
         return None
+
+    @classmethod
+    def from_config(
+        cls: type["AnomalibDataModule"],
+        config_path: str | Path,
+        **kwargs,
+    ) -> "AnomalibDataModule":
+        """Create a datamodule instance from the configuration.
+
+        Args:
+            config_path (str | Path): Path to the data configuration file.
+            **kwargs (dict): Additional keyword arguments.
+
+        Returns:
+            AnomalibDataModule: Datamodule instance.
+
+        Example:
+            The following example shows how to get datamodule from mvtec.yaml:
+
+            .. code-block:: python
+                >>> data_config = "configs/data/mvtec.yaml"
+                >>> datamodule = AnomalibDataModule.from_config(config_path=data_config)
+
+            The following example shows overriding the configuration file with additional keyword arguments:
+
+            .. code-block:: python
+                >>> override_kwargs = {"data.train_batch_size": 8}
+                >>> datamodule = AnomalibDataModule.from_config(config_path=data_config, **override_kwargs)
+        """
+        from jsonargparse import ArgumentParser
+
+        if not Path(config_path).exists():
+            msg = f"Configuration file not found: {config_path}"
+            raise FileNotFoundError(msg)
+
+        data_parser = ArgumentParser()
+        data_parser.add_subclass_arguments(AnomalibDataModule, "data", required=False, fail_untyped=False)
+        args = ["--data", str(config_path)]
+        for key, value in kwargs.items():
+            args.extend([f"--{key}", str(value)])
+        config = data_parser.parse_args(args=args)
+        instantiated_classes = data_parser.instantiate_classes(config)
+        datamodule = instantiated_classes.get("data")
+        if isinstance(datamodule, AnomalibDataModule):
+            return datamodule
+
+        msg = f"Datamodule is not an instance of AnomalibDataModule: {datamodule}"
+        raise ValueError(msg)

src/anomalib/engine/engine.py

@@ -962,3 +962,52 @@ def export(
         if exported_model_path:
             logging.info(f"Exported model to {exported_model_path}")
         return exported_model_path
+
+    @classmethod
+    def from_config(
+        cls: type["Engine"],
+        config_path: str | Path,
+        **kwargs,
+    ) -> tuple["Engine", AnomalyModule, AnomalibDataModule]:
+        """Create an Engine instance from a configuration file.
+
+        Args:
+            config_path (str | Path): Path to the full configuration file.
+            **kwargs (dict): Additional keyword arguments.
+
+        Returns:
+            tuple[Engine, AnomalyModule, AnomalibDataModule]: Engine instance.
+
+        Example:
+            The following example shows training with full configuration file:
+
+            .. code-block:: python
+                >>> config_path = "anomalib_full_config.yaml"
+                >>> engine, model, datamodule = Engine.from_config(config_path=config_path)
+                >>> engine.fit(datamodule=datamodule, model=model)
+
+            The following example shows overriding the configuration file with additional keyword arguments:
+
+            .. code-block:: python
+                >>> override_kwargs = {"data.train_batch_size": 8}
+                >>> engine, model, datamodule = Engine.from_config(config_path=config_path, **override_kwargs)
+                >>> engine.fit(datamodule=datamodule, model=model)
+        """
+        from anomalib.cli.cli import AnomalibCLI
+
+        if not Path(config_path).exists():
+            msg = f"Configuration file not found: {config_path}"
+            raise FileNotFoundError(msg)
+
+        args = [
+            "fit",
+            "--config",
+            str(config_path),
+        ]
+        for key, value in kwargs.items():
+            args.extend([f"--{key}", str(value)])
+        anomalib_cli = AnomalibCLI(
+            args=args,
+            run=False,
+        )
+        return anomalib_cli.engine, anomalib_cli.model, anomalib_cli.datamodule

src/anomalib/models/components/base/anomaly_module.py

@@ -7,6 +7,7 @@
 import logging
 from abc import ABC, abstractmethod
 from collections import OrderedDict
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 import lightning.pytorch as pl
@@ -275,3 +276,65 @@ def on_load_checkpoint(self, checkpoint: dict[str, Any]) -> None:
         """
         self._transform = checkpoint["transform"]
         self.setup("load_checkpoint")
+
+    @classmethod
+    def from_config(
+        cls: type["AnomalyModule"],
+        config_path: str | Path,
+        **kwargs,
+    ) -> "AnomalyModule":
+        """Create a model instance from the configuration.
+
+        Args:
+            config_path (str | Path): Path to the model configuration file.
+            **kwargs (dict): Additional keyword arguments.
+
+        Returns:
+            AnomalyModule: model instance.
+
+        Example:
+            The following example shows how to get model from patchcore.yaml:
+
+            .. code-block:: python
+                >>> model_config = "configs/model/patchcore.yaml"
+                >>> model = AnomalyModule.from_config(config_path=model_config)
+
+            The following example shows overriding the configuration file with additional keyword arguments:
+
+            .. code-block:: python
+                >>> override_kwargs = {"model.pre_trained": False}
+                >>> model = AnomalyModule.from_config(config_path=model_config, **override_kwargs)
+        """
+        from jsonargparse import ActionConfigFile, ArgumentParser
+        from lightning.pytorch import Trainer
+
+        from anomalib import TaskType
+
+        if not Path(config_path).exists():
+            msg = f"Configuration file not found: {config_path}"
+            raise FileNotFoundError(msg)
+
+        model_parser = ArgumentParser()
+        model_parser.add_argument(
+            "-c",
+            "--config",
+            action=ActionConfigFile,
+            help="Path to a configuration file in json or yaml format.",
+        )
+        model_parser.add_subclass_arguments(AnomalyModule, "model", required=False, fail_untyped=False)
+        model_parser.add_argument("--task", type=TaskType | str, default=TaskType.SEGMENTATION)
+        model_parser.add_argument("--metrics.image", type=list[str] | str | None, default=["F1Score", "AUROC"])
+        model_parser.add_argument("--metrics.pixel", type=list[str] | str | None, default=None, required=False)
+        model_parser.add_argument("--metrics.threshold", type=BaseThreshold | str, default="F1AdaptiveThreshold")
+        model_parser.add_class_arguments(Trainer, "trainer", fail_untyped=False, instantiate=False, sub_configs=True)
+        args = ["--config", str(config_path)]
+        for key, value in kwargs.items():
+            args.extend([f"--{key}", str(value)])
+        config = model_parser.parse_args(args=args)
+        instantiated_classes = model_parser.instantiate_classes(config)
+        model = instantiated_classes.get("model")
+        if isinstance(model, AnomalyModule):
+            return model
+
+        msg = f"Model is not an instance of AnomalyModule: {model}"
+        raise ValueError(msg)

tests/unit/data/base/base.py

@@ -23,3 +23,19 @@ def test_datamodule_has_dataloader_attributes(self, datamodule: AnomalibDataModu
         dataloader = f"{subset}_dataloader"
         assert hasattr(datamodule, dataloader)
         assert isinstance(getattr(datamodule, dataloader)(), DataLoader)
+
+    def test_datamodule_from_config(self, fxt_data_config_path: str) -> None:
+        # 1. Wrong file path:
+        with pytest.raises(FileNotFoundError):
+            AnomalibDataModule.from_config(config_path="wrong_configs.yaml")
+
+        # 2. Correct file path:
+        datamodule = AnomalibDataModule.from_config(config_path=fxt_data_config_path)
+        assert datamodule is not None
+        assert isinstance(datamodule, AnomalibDataModule)
+
+        # 3. Override batch_size & num_workers
+        override_kwargs = {"data.train_batch_size": 1, "data.num_workers": 1}
+        datamodule = AnomalibDataModule.from_config(config_path=fxt_data_config_path, **override_kwargs)
+        assert datamodule.train_batch_size == 1
+        assert datamodule.num_workers == 1

tests/unit/data/image/test_btech.py

@@ -31,3 +31,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> BTech:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/btech.yaml"

tests/unit/data/image/test_folder.py

@@ -41,3 +41,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> Folder:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/folder.yaml"

tests/unit/data/image/test_folder_3d.py

@@ -38,3 +38,20 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> Folder3D:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/folder_3d.yaml"
+
+    def test_datamodule_from_config(self, fxt_data_config_path: str) -> None:
+        """Test method to create a datamodule from a configuration file.
+
+        Args:
+            fxt_data_config_path (str): The path to the configuration file.
+
+        Returns:
+            None
+        """
+        pytest.skip("The configuration file does not exist.")
+        _ = fxt_data_config_path

tests/unit/data/image/test_kolektor.py

@@ -30,3 +30,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> Kolektor:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/kolektor.yaml"

tests/unit/data/image/test_mvtec.py

@@ -29,3 +29,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> MVTec:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/mvtec.yaml"

tests/unit/data/image/test_mvtec_3d.py

@@ -31,3 +31,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> MVTec3D:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/mvtec_3d.yaml"

tests/unit/data/image/test_visa.py

@@ -31,3 +31,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType) -> Visa:
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/visa.yaml"

tests/unit/data/video/test_avenue.py

@@ -38,3 +38,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType, clip_length_in_fra
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/avenue.yaml"

tests/unit/data/video/test_shanghaitech.py

@@ -39,3 +39,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType, clip_length_in_fra
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/shanghaitec.yaml"

tests/unit/data/video/test_ucsdped.py

@@ -37,3 +37,8 @@ def datamodule(self, dataset_path: Path, task_type: TaskType, clip_length_in_fra
         _datamodule.setup()
 
         return _datamodule
+
+    @pytest.fixture()
+    def fxt_data_config_path(self) -> str:
+        """Return the path to the test data config."""
+        return "configs/data/ucsd_ped.yaml"