Adding the pipeline for the task explanation and Llm

src/anomalib/__init__.py

@@ -22,3 +22,4 @@ class TaskType(str, Enum):
     CLASSIFICATION = "classification"
     DETECTION = "detection"
     SEGMENTATION = "segmentation"
+    EXPLANATION = "explanation"

src/anomalib/callbacks/metrics.py

@@ -75,7 +75,7 @@ def setup(
         pixel_metric_names: list[str] | dict[str, dict[str, Any]]
         if self.pixel_metric_names is None:
             pixel_metric_names = []
-        elif self.task == TaskType.CLASSIFICATION:
+        elif self.task in (TaskType.CLASSIFICATION, TaskType.EXPLANATION):
             pixel_metric_names = []
             logger.warning(
                 "Cannot perform pixel-level evaluation when task type is classification. "
@@ -88,14 +88,23 @@ def setup(
             )
 
         if isinstance(pl_module, AnomalyModule):
-            pl_module.image_metrics = create_metric_collection(image_metric_names, "image_")
-            if hasattr(pl_module, "pixel_metrics"):  # incase metrics are loaded from model checkpoint
+            pl_module.image_metrics = create_metric_collection(
+                image_metric_names,
+                "image_",
+            )
+            if hasattr(
+                pl_module,
+                "pixel_metrics",
+            ):  # incase metrics are loaded from model checkpoint
                 new_metrics = create_metric_collection(pixel_metric_names)
                 for name in new_metrics:
                     if name not in pl_module.pixel_metrics:
                         pl_module.pixel_metrics.add_metrics(new_metrics[name])
             else:
-                pl_module.pixel_metrics = create_metric_collection(pixel_metric_names, "pixel_")
+                pl_module.pixel_metrics = create_metric_collection(
+                    pixel_metric_names,
+                    "pixel_",
+                )
             self._set_threshold(pl_module)
 
     def on_validation_epoch_start(
@@ -121,7 +130,11 @@ def on_validation_batch_end(
 
         if outputs is not None:
             self._outputs_to_device(outputs)
-            self._update_metrics(pl_module.image_metrics, pl_module.pixel_metrics, outputs)
+            self._update_metrics(
+                pl_module.image_metrics,
+                pl_module.pixel_metrics,
+                outputs,
+            )
 
     def on_validation_epoch_end(
         self,
@@ -156,7 +169,11 @@ def on_test_batch_end(
 
         if outputs is not None:
             self._outputs_to_device(outputs)
-            self._update_metrics(pl_module.image_metrics, pl_module.pixel_metrics, outputs)
+            self._update_metrics(
+                pl_module.image_metrics,
+                pl_module.pixel_metrics,
+                outputs,
+            )
 
     def on_test_epoch_end(
         self,
@@ -181,7 +198,10 @@ def _update_metrics(
         image_metric.update(output["pred_scores"], output["label"].int())
         if "mask" in output and "anomaly_maps" in output:
             pixel_metric.to(self.device)
-            pixel_metric.update(torch.squeeze(output["anomaly_maps"]), torch.squeeze(output["mask"].int()))
+            pixel_metric.update(
+                torch.squeeze(output["anomaly_maps"]),
+                torch.squeeze(output["mask"].int()),
+            )
 
     def _outputs_to_device(self, output: STEP_OUTPUT) -> STEP_OUTPUT | dict[str, Any]:
         if isinstance(output, dict):

src/anomalib/data/base/dataset.py

@@ -20,9 +20,11 @@
 from anomalib.data.utils import LabelName, masks_to_boxes, read_image, read_mask
 
 _EXPECTED_COLUMNS_CLASSIFICATION = ["image_path", "split"]
+_EXPECTED_COLUMNS_EXPLANATION = ["image_path", "split"]
 _EXPECTED_COLUMNS_SEGMENTATION = [*_EXPECTED_COLUMNS_CLASSIFICATION, "mask_path"]
 _EXPECTED_COLUMNS_PERTASK = {
     "classification": _EXPECTED_COLUMNS_CLASSIFICATION,
+    "explanation": _EXPECTED_COLUMNS_EXPLANATION,
     "segmentation": _EXPECTED_COLUMNS_SEGMENTATION,
     "detection": _EXPECTED_COLUMNS_SEGMENTATION,
 }
@@ -61,7 +63,11 @@ class AnomalibDataset(Dataset, ABC):
             Defaults to ``None``.
     """
 
-    def __init__(self, task: TaskType | str, transform: Transform | None = None) -> None:
+    def __init__(
+        self,
+        task: TaskType | str,
+        transform: Transform | None = None,
+    ) -> None:
         super().__init__()
         self.task = TaskType(task)
         self.transform = transform
@@ -83,7 +89,11 @@ def __len__(self) -> int:
         """Get length of the dataset."""
         return len(self.samples)
 
-    def subsample(self, indices: Sequence[int], inplace: bool = False) -> "AnomalibDataset":
+    def subsample(
+        self,
+        indices: Sequence[int],
+        inplace: bool = False,
+    ) -> "AnomalibDataset":
         """Subsamples the dataset at the provided indices.
 
         Args:
@@ -169,7 +179,7 @@ def __getitem__(self, index: int) -> dict[str, str | torch.Tensor]:
         image = read_image(image_path, as_tensor=True)
         item = {"image_path": image_path, "label": label_index}
 
-        if self.task == TaskType.CLASSIFICATION:
+        if self.task in (TaskType.CLASSIFICATION, TaskType.EXPLANATION):
             item["image"] = self.transform(image) if self.transform else image
         elif self.task in (TaskType.DETECTION, TaskType.SEGMENTATION):
             # Only Anomalous (1) images have masks in anomaly datasets
@@ -204,5 +214,8 @@ def __add__(self, other_dataset: "AnomalibDataset") -> "AnomalibDataset":
             msg = "Cannot concatenate datasets that are not of the same type."
             raise TypeError(msg)
         dataset = copy.deepcopy(self)
-        dataset.samples = pd.concat([self.samples, other_dataset.samples], ignore_index=True)
+        dataset.samples = pd.concat(
+            [self.samples, other_dataset.samples],
+            ignore_index=True,
+        )
         return dataset

src/anomalib/data/base/depth.py

@@ -46,9 +46,13 @@ def __getitem__(self, index: int) -> dict[str, str | torch.Tensor]:
 
         image = to_tensor(Image.open(image_path))
         depth_image = to_tensor(read_depth_image(depth_path))
-        item = {"image_path": image_path, "depth_path": depth_path, "label": label_index}
+        item = {
+            "image_path": image_path,
+            "depth_path": depth_path,
+            "label": label_index,
+        }
 
-        if self.task == TaskType.CLASSIFICATION:
+        if self.task in (TaskType.CLASSIFICATION, TaskType.EXPLANATION):
             item["image"], item["depth_image"] = (
                 self.transform(image, depth_image) if self.transform else (image, depth_image)
             )

src/anomalib/deploy/inferencers/openvino_inferencer.py

@@ -27,7 +27,9 @@
     if TYPE_CHECKING:
         from openvino import CompiledModel
 else:
-    logger.warning("OpenVINO is not installed. Please install OpenVINO to use OpenVINOInferencer.")
+    logger.warning(
+        "OpenVINO is not installed. Please install OpenVINO to use OpenVINOInferencer.",
+    )
 
 
 class OpenVINOInferencer(Inferencer):
@@ -110,7 +112,10 @@
 
         self.task = TaskType(task) if task else TaskType(self.metadata["task"])
 
-    def load_model(self, path: str | Path | tuple[bytes, bytes]) -> tuple[Any, Any, "CompiledModel"]:
+    def load_model(
+        self,
+        path: str | Path | tuple[bytes, bytes],
+    ) -> tuple[Any, Any, "CompiledModel"]:
         """Load the OpenVINO model.
 
         Args:
@@ -143,7 +148,11 @@
         cache_folder.mkdir(exist_ok=True)
         core.set_property({"CACHE_DIR": cache_folder})
 
-        compile_model = core.compile_model(model=model, device_name=self.device, config=self.config)
+        compile_model = core.compile_model(
+            model=model,
+            device_name=self.device,
+            config=self.config,
+        )
 
         input_blob = compile_model.input(0)
         output_blob = compile_model.output(0)
@@ -238,7 +247,11 @@
         """
         return self.model(image)
 
-    def post_process(self, predictions: np.ndarray, metadata: dict | DictConfig | None = None) -> dict[str, Any]:
+    def post_process(
+        self,
+        predictions: np.ndarray,
+        metadata: dict | DictConfig | None = None,
+    ) -> dict[str, Any]:
         """Post process the output predictions.
 
         Args:
@@ -277,11 +290,13 @@
             pred_idx = pred_score >= metadata["image_threshold"]
             pred_label = LabelName.ABNORMAL if pred_idx else LabelName.NORMAL
 
-        if task == TaskType.CLASSIFICATION:
+        if task in (TaskType.CLASSIFICATION, TaskType.EXPLANATION):
             _, pred_score = self._normalize(pred_scores=pred_score, metadata=metadata)
         elif task in (TaskType.SEGMENTATION, TaskType.DETECTION):
             if "pixel_threshold" in metadata:
-                pred_mask = (anomaly_map >= metadata["pixel_threshold"]).astype(np.uint8)
+                pred_mask = (anomaly_map >= metadata["pixel_threshold"]).astype(
+                    np.uint8,
+                )
 
             anomaly_map, pred_score = self._normalize(
                 pred_scores=pred_score,