Resolve conflicts

Signed-off-by: Samet Akcay <[email protected]>

Author: samet-akcay

notebooks/700_metrics/701b_aupimo_advanced_i.ipynb

@@ -125,7 +125,7 @@ def to_onnx(
         dynamic_axes = (
             {"input": {0: "batch_size"}, "output": {0: "batch_size"}}
             if input_size
-            else {"input": {0: "batch_size", 2: "height", 3: "width"}, "output": {0: "batch_size"}}
+            else {"input": {0: "batch_size", 2: "height", 3: "weight"}, "output": {0: "batch_size"}}
         )
         onnx_path = export_root / "model.onnx"
         # apply pass through the model to get the output names

notebooks/700_metrics/701c_aupimo_advanced_ii.ipynb

@@ -0,0 +1,12 @@
+"""Tiled ensemble pipelines."""
+
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from .test_pipeline import EvalTiledEnsemble
+from .train_pipeline import TrainTiledEnsemble
+
+__all__ = [
+    "TrainTiledEnsemble",
+    "EvalTiledEnsemble",
+]

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

@@ -0,0 +1,30 @@
+"""Tiled ensemble pipeline components."""
+
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from .merging import MergeJobGenerator
+from .metrics_calculation import MetricsCalculationJobGenerator
+from .model_training import TrainModelJobGenerator
+from .normalization import NormalizationJobGenerator
+from .prediction import PredictJobGenerator
+from .smoothing import SmoothingJobGenerator
+from .stats_calculation import StatisticsJobGenerator
+from .thresholding import ThresholdingJobGenerator
+from .utils import NormalizationStage, PredictData, ThresholdStage
+from .visualization import VisualizationJobGenerator
+
+__all__ = [
+    "NormalizationStage",
+    "ThresholdStage",
+    "PredictData",
+    "TrainModelJobGenerator",
+    "PredictJobGenerator",
+    "MergeJobGenerator",
+    "SmoothingJobGenerator",
+    "StatisticsJobGenerator",
+    "NormalizationJobGenerator",
+    "ThresholdingJobGenerator",
+    "VisualizationJobGenerator",
+    "MetricsCalculationJobGenerator",
+]

src/anomalib/pipelines/tiled_ensemble/__init__.py

@@ -0,0 +1,110 @@
+"""Tiled ensemble - prediction merging job."""
+
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+from collections.abc import Generator
+from typing import Any
+
+from tqdm import tqdm
+
+from anomalib.pipelines.components import Job, JobGenerator
+from anomalib.pipelines.types import GATHERED_RESULTS, RUN_RESULTS
+
+from .utils.ensemble_tiling import EnsembleTiler
+from .utils.helper_functions import get_ensemble_tiler
+from .utils.prediction_data import EnsemblePredictions
+from .utils.prediction_merging import PredictionMergingMechanism
+
+logger = logging.getLogger(__name__)
+
+
+class MergeJob(Job):
+    """Job for merging tile-level predictions into image-level predictions.
+
+    Args:
+        predictions (EnsemblePredictions): Object containing ensemble predictions.
+        tiler (EnsembleTiler): Ensemble tiler used for untiling.
+    """
+
+    name = "Merge"
+
+    def __init__(self, predictions: EnsemblePredictions, tiler: EnsembleTiler) -> None:
+        super().__init__()
+        self.predictions = predictions
+        self.tiler = tiler
+
+    def run(self, task_id: int | None = None) -> list[Any]:
+        """Run merging job that merges all batches of tile-level predictions into image-level predictions.
+
+        Args:
+            task_id: Not used in this case.
+
+        Returns:
+            list[Any]: List of merged predictions.
+        """
+        del task_id  # not needed here
+
+        merger = PredictionMergingMechanism(self.predictions, self.tiler)
+
+        logger.info("Merging predictions.")
+
+        # merge all batches
+        merged_predictions = [
+            merger.merge_tile_predictions(batch_idx)
+            for batch_idx in tqdm(range(merger.num_batches), desc="Prediction merging")
+        ]
+
+        return merged_predictions  # noqa: RET504
+
+    @staticmethod
+    def collect(results: list[RUN_RESULTS]) -> GATHERED_RESULTS:
+        """Nothing to collect in this job.
+
+        Returns:
+            list[Any]: List of predictions.
+        """
+        # take the first element as result is list of lists here
+        return results[0]
+
+    @staticmethod
+    def save(results: GATHERED_RESULTS) -> None:
+        """Nothing to save in this job."""
+
+
+class MergeJobGenerator(JobGenerator):
+    """Generate MergeJob."""
+
+    def __init__(self, tiling_args: dict, data_args: dict) -> None:
+        super().__init__()
+        self.tiling_args = tiling_args
+        self.data_args = data_args
+
+    @property
+    def job_class(self) -> type:
+        """Return the job class."""
+        return MergeJob
+
+    def generate_jobs(
+        self,
+        args: dict | None = None,
+        prev_stage_result: EnsemblePredictions | None = None,
+    ) -> Generator[MergeJob, None, None]:
+        """Return a generator producing a single merging job.
+
+        Args:
+            args (dict): Tiled ensemble pipeline args.
+            prev_stage_result (EnsemblePredictions): Ensemble predictions from predict step.
+
+        Returns:
+            Generator[MergeJob, None, None]: MergeJob generator
+        """
+        del args  # args not used here
+
+        tiler = get_ensemble_tiler(self.tiling_args, self.data_args)
+        if prev_stage_result is not None:
+            yield MergeJob(prev_stage_result, tiler)
+        else:
+            msg = "Merging job requires tile level predictions from previous step."
+            raise ValueError(msg)

src/anomalib/pipelines/tiled_ensemble/components/__init__.py

@@ -0,0 +1,217 @@
+"""Tiled ensemble - metrics calculation job."""
+
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+from collections.abc import Generator
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+from tqdm import tqdm
+
+from anomalib import TaskType
+from anomalib.metrics import AnomalibMetricCollection, create_metric_collection
+from anomalib.pipelines.components import Job, JobGenerator
+from anomalib.pipelines.types import GATHERED_RESULTS, PREV_STAGE_RESULT, RUN_RESULTS
+
+from .utils import NormalizationStage
+from .utils.helper_functions import get_threshold_values
+
+logger = logging.getLogger(__name__)
+
+
+class MetricsCalculationJob(Job):
+    """Job for image and pixel metrics calculation.
+
+    Args:
+        accelerator (str): Accelerator (device) to use.
+        predictions (list[Any]): List of batch predictions.
+        root_dir (Path): Root directory to save checkpoints, stats and images.
+        image_metrics (AnomalibMetricCollection): Collection of all image-level metrics.
+        pixel_metrics (AnomalibMetricCollection): Collection of all pixel-level metrics.
+    """
+
+    name = "Metrics"
+
+    def __init__(
+        self,
+        accelerator: str,
+        predictions: list[Any] | None,
+        root_dir: Path,
+        image_metrics: AnomalibMetricCollection,
+        pixel_metrics: AnomalibMetricCollection,
+    ) -> None:
+        super().__init__()
+        self.accelerator = accelerator
+        self.predictions = predictions
+        self.root_dir = root_dir
+        self.image_metrics = image_metrics
+        self.pixel_metrics = pixel_metrics
+
+    def run(self, task_id: int | None = None) -> dict:
+        """Run a job that calculates image and pixel level metrics.
+
+        Args:
+            task_id: Not used in this case.
+
+        Returns:
+            dict[str, float]: Dictionary containing calculated metric values.
+        """
+        del task_id  # not needed here
+
+        logger.info("Starting metrics calculation.")
+
+        # add predicted data to metrics
+        for data in tqdm(self.predictions, desc="Calculating metrics"):
+            self.image_metrics.update(data["pred_scores"], data["label"].int())
+            if "mask" in data and "anomaly_maps" in data:
+                self.pixel_metrics.update(data["anomaly_maps"], data["mask"].int())
+
+        # compute all metrics on specified accelerator
+        metrics_dict = {}
+        for name, metric in self.image_metrics.items():
+            metric.to(self.accelerator)
+            metrics_dict[name] = metric.compute().item()
+            metric.cpu()
+
+        if self.pixel_metrics.update_called:
+            for name, metric in self.pixel_metrics.items():
+                metric.to(self.accelerator)
+                metrics_dict[name] = metric.compute().item()
+                metric.cpu()
+
+        for name, value in metrics_dict.items():
+            print(f"{name}: {value:.4f}")
+
+        # save path used in `save` method
+        metrics_dict["save_path"] = self.root_dir / "metric_results.csv"
+
+        return metrics_dict
+
+    @staticmethod
+    def collect(results: list[RUN_RESULTS]) -> GATHERED_RESULTS:
+        """Nothing to collect in this job.
+
+        Returns:
+            list[Any]: list of predictions.
+        """
+        # take the first element as result is list of dict here
+        return results[0]
+
+    @staticmethod
+    def save(results: GATHERED_RESULTS) -> None:
+        """Save metrics values to csv."""
+        logger.info("Saving metrics to csv.")
+
+        # get and remove path from stats dict
+        results_path: Path = results.pop("save_path")
+        results_path.parent.mkdir(parents=True, exist_ok=True)
+
+        df_dict = {k: [v] for k, v in results.items()}
+        metrics_df = pd.DataFrame(df_dict)
+        metrics_df.to_csv(results_path, index=False)
+
+
+class MetricsCalculationJobGenerator(JobGenerator):
+    """Generate MetricsCalculationJob.
+
+    Args:
+        root_dir (Path): Root directory to save checkpoints, stats and images.
+    """
+
+    def __init__(
+        self,
+        accelerator: str,
+        root_dir: Path,
+        task: TaskType,
+        metrics: dict,
+        normalization_stage: NormalizationStage,
+    ) -> None:
+        self.accelerator = accelerator
+        self.root_dir = root_dir
+        self.task = task
+        self.metrics = metrics
+        self.normalization_stage = normalization_stage
+
+    @property
+    def job_class(self) -> type:
+        """Return the job class."""
+        return MetricsCalculationJob
+
+    def configure_ensemble_metrics(
+        self,
+        image_metrics: list[str] | dict[str, dict[str, Any]] | None = None,
+        pixel_metrics: list[str] | dict[str, dict[str, Any]] | None = None,
+    ) -> tuple[AnomalibMetricCollection, AnomalibMetricCollection]:
+        """Configure image and pixel metrics and put them into a collection.
+
+        Args:
+            image_metrics (list[str] | None): List of image-level metric names.
+            pixel_metrics (list[str] | None): List of pixel-level metric names.
+
+        Returns:
+            tuple[AnomalibMetricCollection, AnomalibMetricCollection]:
+                Image-metrics collection and pixel-metrics collection
+        """
+        image_metrics = [] if image_metrics is None else image_metrics
+
+        if pixel_metrics is None:
+            pixel_metrics = []
+        elif self.task == TaskType.CLASSIFICATION:
+            pixel_metrics = []
+            logger.warning(
+                "Cannot perform pixel-level evaluation when task type is classification. "
+                "Ignoring the following pixel-level metrics: %s",
+                pixel_metrics,
+            )
+
+        # if a single metric is passed, transform to list to fit the creation function
+        if isinstance(image_metrics, str):
+            image_metrics = [image_metrics]
+        if isinstance(pixel_metrics, str):
+            pixel_metrics = [pixel_metrics]
+
+        image_metrics_collection = create_metric_collection(image_metrics, "image_")
+        pixel_metrics_collection = create_metric_collection(pixel_metrics, "pixel_")
+
+        return image_metrics_collection, pixel_metrics_collection
+
+    def generate_jobs(
+        self,
+        args: dict | None = None,
+        prev_stage_result: PREV_STAGE_RESULT = None,
+    ) -> Generator[MetricsCalculationJob, None, None]:
+        """Make a generator that yields a single metrics calculation job.
+
+        Args:
+            args: ensemble run config.
+            prev_stage_result: ensemble predictions from previous step.
+
+        Returns:
+            Generator[MetricsCalculationJob, None, None]: MetricsCalculationJob generator
+        """
+        del args  # args not used here
+
+        image_metrics_config = self.metrics.get("image", None)
+        pixel_metrics_config = self.metrics.get("pixel", None)
+
+        image_threshold, pixel_threshold = get_threshold_values(self.normalization_stage, self.root_dir)
+
+        image_metrics, pixel_metrics = self.configure_ensemble_metrics(
+            image_metrics=image_metrics_config,
+            pixel_metrics=pixel_metrics_config,
+        )
+
+        # set thresholds for metrics that need it
+        image_metrics.set_threshold(image_threshold)
+        pixel_metrics.set_threshold(pixel_threshold)
+
+        yield MetricsCalculationJob(
+            accelerator=self.accelerator,
+            predictions=prev_stage_result,
+            root_dir=self.root_dir,
+            image_metrics=image_metrics,
+            pixel_metrics=pixel_metrics,
+        )