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. "

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,
 }
@@ -169,7 +171,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

src/anomalib/data/base/depth.py

@@ -48,7 +48,7 @@ def __getitem__(self, index: int) -> dict[str, str | torch.Tensor]:
         depth_image = to_tensor(read_depth_image(depth_path))
         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

@@ -277,7 +277,7 @@ def post_process(self, predictions: np.ndarray, metadata: dict | DictConfig | No
             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:

src/anomalib/utils/visualization/image.py

@@ -1,8 +1,10 @@
 """Image/video generator."""
 
+
 # Copyright (C) 2022-2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
 
+import textwrap
 from collections.abc import Iterator
 from enum import Enum
 from pathlib import Path
@@ -39,6 +41,7 @@ def __init__(
         image: np.ndarray,
         pred_score: float,
         pred_label: str,
+        text_descr: str | None = None,
         anomaly_map: np.ndarray | None = None,
         gt_mask: np.ndarray | None = None,
         pred_mask: np.ndarray | None = None,
@@ -47,6 +50,7 @@ def __init__(
         box_labels: np.ndarray | None = None,
         normalize: bool = False,
     ) -> None:
+        self.text_descr = text_descr
         self.anomaly_map = anomaly_map
         self.box_labels = box_labels
         self.gt_boxes = gt_boxes
@@ -93,6 +97,7 @@ def __repr__(self) -> str:
         repr_str += f", segmentations={self.segmentations}" if self.segmentations is not None else ""
         repr_str += f", normal_boxes={self.normal_boxes}" if self.normal_boxes is not None else ""
         repr_str += f", anomalous_boxes={self.anomalous_boxes}" if self.anomalous_boxes is not None else ""
+        repr_str += f", text_descr={self.text_descr}" if self.text_descr is not None else ""
         repr_str += ")"
         return repr_str
 
@@ -236,6 +241,13 @@ def _visualize_full(self, image_result: ImageResult) -> np.ndarray:
             else:
                 image_classified = add_normal_label(image_result.image, 1 - image_result.pred_score)
             image_grid.add_image(image=image_classified, title="Prediction")
+        elif self.task == TaskType.EXPLANATION:
+            description = ""
+            if image_result.text_descr:
+                description = image_result.text_descr
+
+            image_classified = add_normal_label(image_result.image, 1 - image_result.pred_score)
+            image_grid.add_image(image_classified, title="Explanation of Image", description=description)
 
         return image_grid.generate()
 
@@ -274,6 +286,22 @@ def _visualize_simple(self, image_result: ImageResult) -> np.ndarray:
             else:
                 image_classified = add_normal_label(image_result.image, 1 - image_result.pred_score)
             return image_classified
+
+        if self.task == TaskType.EXPLANATION:
+            image_grid = _ImageGrid()
+            description = ""
+            if image_result.text_descr:
+                description = image_result.text_descr
+
+            if image_result.pred_label:
+                image_classified = add_anomalous_label(image_result.image, image_result.pred_score)
+            else:
+                image_classified = add_normal_label(image_result.image, 1 - image_result.pred_score)
+
+            image_grid.add_image(image_classified, title="Explanation of Image", description=description)
+
+            return image_grid.generate()
+
         msg = f"Unknown task type: {self.task}"
         raise ValueError(msg)
 
@@ -290,15 +318,22 @@ def __init__(self) -> None:
         self.figure: matplotlib.figure.Figure | None = None
         self.axis: Axes | np.ndarray | None = None
 
-    def add_image(self, image: np.ndarray, title: str | None = None, color_map: str | None = None) -> None:
+    def add_image(
+        self,
+        image: np.ndarray,
+        title: str | None = None,
+        color_map: str | None = None,
+        description: str | None = None,
+    ) -> None:
         """Add an image to the grid.
 
         Args:
           image (np.ndarray): Image which should be added to the figure.
           title (str): Image title shown on the plot.
+          description (str): Description of the image shown on the plot TaskType.EXPLANATION.
           color_map (str | None): Name of matplotlib color map used to map scalar data to colours. Defaults to None.
         """
-        image_data = {"image": image, "title": title, "color_map": color_map}
+        image_data = {"image": image, "title": title, "color_map": color_map, "descr": description}
         self.images.append(image_data)
 
     def generate(self) -> np.ndarray:
@@ -324,6 +359,16 @@ def generate(self) -> np.ndarray:
             axis.imshow(image_dict["image"], image_dict["color_map"], vmin=0, vmax=255)
             if image_dict["title"] is not None:
                 axis.title.set_text(image_dict["title"])
+            if image_dict["descr"] is not None:
+                wrapped_text = textwrap.fill(
+                    image_dict["descr"],
+                    width=70 // num_cols,
+                )  # Adjust 'width' based on your subplot size and preference
+
+                axis.set_title(wrapped_text, fontsize=10)
+
+                self.figure.subplots_adjust(top=0.7)
+
         self.figure.canvas.draw()
         # convert canvas to numpy array to prepare for visualization with opencv
         img = np.frombuffer(self.figure.canvas.tostring_rgb(), dtype=np.uint8)