Add Torch Inferencer and Update Openvino and Gradio Inferencers.

anomalib/data/inference.py

@@ -63,6 +63,6 @@ def __getitem__(self, index: int) -> Any:
         image_filename = self.image_filenames[index]
         image = read_image(path=image_filename)
         pre_processed = self.pre_process(image=image)
-        pre_processed["image_path"] = image_filename
+        pre_processed["image_path"] = str(image_filename)
 
         return pre_processed

anomalib/data/utils/__init__.py

@@ -16,6 +16,13 @@
 
 from .download import DownloadProgressBar, hash_check
 from .generators import random_2d_perlin
-from .image import get_image_filenames, read_image
+from .image import generate_output_image_filename, get_image_filenames, read_image
 
-__all__ = ["get_image_filenames", "hash_check", "random_2d_perlin", "read_image", "DownloadProgressBar"]
+__all__ = [
+    "generate_output_image_filename",
+    "get_image_filenames",
+    "hash_check",
+    "random_2d_perlin",
+    "read_image",
+    "DownloadProgressBar",
+]

anomalib/data/utils/image.py

@@ -15,6 +15,7 @@
 # and limitations under the License.
 
 import math
+import warnings
 from pathlib import Path
 from typing import List, Union
 
@@ -25,33 +26,132 @@
 from torchvision.datasets.folder import IMG_EXTENSIONS
 
 
-def get_image_filenames(path: Union[str, Path]) -> List[str]:
+def get_image_filenames(path: Union[str, Path]) -> List[Path]:
     """Get image filenames.
 
     Args:
         path (Union[str, Path]): Path to image or image-folder.
 
     Returns:
-        List[str]: List of image filenames
+        List[Path]: List of image filenames
 
     """
-    image_filenames: List[str]
+    image_filenames: List[Path]
 
     if isinstance(path, str):
         path = Path(path)
 
     if path.is_file() and path.suffix in IMG_EXTENSIONS:
-        image_filenames = [str(path)]
+        image_filenames = [path]
 
     if path.is_dir():
-        image_filenames = [str(p) for p in path.glob("**/*") if p.suffix in IMG_EXTENSIONS]
+        image_filenames = [p for p in path.glob("**/*") if p.suffix in IMG_EXTENSIONS]
 
     if len(image_filenames) == 0:
         raise ValueError(f"Found 0 images in {path}")
 
     return image_filenames
 
 
+def duplicate_filename(path: Union[str, Path]) -> Path:
+    """Check and duplicate filename.
+
+    This function checks the path and adds a suffix if it already exists on the file system.
+
+    Args:
+        path (Union[str, Path]): Input Path
+
+    Examples:
+        >>> path = Path("datasets/MVTec/bottle/test/broken_large/000.png")
+        >>> path.exists()
+        True
+
+        If we pass this to ``duplicate_filename`` function we would get the following:
+        >>> duplicate_filename(path)
+        PosixPath('datasets/MVTec/bottle/test/broken_large/000_1.png')
+
+    Returns:
+        Path: Duplicated output path.
+    """
+
+    if isinstance(path, str):
+        path = Path(path)
+
+    i = 0
+    while True:
+        duplicated_path = path if i == 0 else path.parent / (path.stem + f"_{i}" + path.suffix)
+        if not duplicated_path.exists():
+            break
+        i += 1
+
+    return duplicated_path
+
+
+def generate_output_image_filename(input_path: Union[str, Path], output_path: Union[str, Path]) -> Path:
+    """Generate an output filename to save the inference image.
+
+    This function generates an output filaname by checking the input and output filenames. Input path is
+    the input to infer, and output path is the path to save the output predictions specified by the user.
+
+    The function expects ``input_path`` to always be a file, not a directory. ``output_path`` could be a
+    filename or directory. If it is a filename, the function checks if the specified filename exists on
+    the file system. If yes, the function calls ``duplicate_filename`` to duplicate the filename to avoid
+    overwriting the existing file. If ``output_path`` is a directory, this function adds the parent and
+    filenames of ``input_path`` to ``output_path``.
+
+    Args:
+        input_path (Union[str, Path]): Path to the input image to infer.
+        output_path (Union[str, Path]): Path to output to save the predictions.
+            Could be a filename or a directory.
+
+    Examples:
+        >>> input_path = Path("datasets/MVTec/bottle/test/broken_large/000.png")
+        >>> output_path = Path("datasets/MVTec/bottle/test/broken_large/000.png")
+        >>> generate_output_image_filename(input_path, output_path)
+        PosixPath('datasets/MVTec/bottle/test/broken_large/000_1.png')
+
+        >>> input_path = Path("datasets/MVTec/bottle/test/broken_large/000.png")
+        >>> output_path = Path("results/images")
+        >>> generate_output_image_filename(input_path, output_path)
+        PosixPath('results/images/broken_large/000.png')
+
+    Raises:
+        ValueError: When the ``input_path`` is not a file.
+
+    Returns:
+        Path: The output filename to save the output predictions from the inferencer.
+    """
+
+    if isinstance(input_path, str):
+        input_path = Path(input_path)
+
+    if isinstance(output_path, str):
+        output_path = Path(output_path)
+
+    # This function expects an ``input_path`` that is a file. This is to check if output_path
+    if input_path.is_file() is False:
+        raise ValueError("input_path is expected to be a file to generate a proper output filename.")
+
+    file_path: Path
+    if output_path.suffix == "":
+        # If the output is a directory, then add parent directory name
+        # and filename to the path. This is to ensure we do not overwrite
+        # images and organize based on the categories.
+        file_path = output_path / input_path.parent.name / input_path.name
+    else:
+        file_path = output_path
+
+    # This new ``file_path`` might contain a directory path yet to be created.
+    # Create the parent directory to avoid such cases.
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+
+    if file_path.is_file():
+        warnings.warn(f"{output_path} already exists. Renaming the file to avoid overwriting.")
+        file_path = duplicate_filename(file_path)
+
+    return file_path
+
+
 def read_image(path: Union[str, Path]) -> np.ndarray:
     """Read image from disk in RGB format.
 

anomalib/deploy/inferencers/__init__.py

@@ -14,8 +14,8 @@
 # See the License for the specific language governing permissions
 # and limitations under the License.
 
-from .base_inference import Inferencer
-from .openvino_inference import OpenVINOInferencer
-from .torch_inference import TorchInferencer
+from .base_inferencer import Inferencer
+from .openvino_inferencer import OpenVINOInferencer
+from .torch_inferencer import TorchInferencer
 
 __all__ = ["Inferencer", "OpenVINOInferencer", "TorchInferencer"]

anomalib/deploy/inferencers/base_inference.py → anomalib/deploy/inferencers/base_inferencer.py

@@ -16,7 +16,7 @@
 
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Dict, Optional, Tuple, Union, cast
+from typing import Any, Dict, Optional, Tuple, Union, cast
 
 import cv2
 import numpy as np
@@ -26,7 +26,7 @@
 from torch import Tensor
 
 from anomalib.data.utils import read_image
-from anomalib.post_processing import compute_mask, superimpose_anomaly_map
+from anomalib.post_processing import ImageResult, compute_mask
 from anomalib.post_processing.normalization.cdf import normalize as normalize_cdf
 from anomalib.post_processing.normalization.cdf import standardize
 from anomalib.post_processing.normalization.min_max import (
@@ -57,18 +57,16 @@ def forward(self, image: Union[np.ndarray, Tensor]) -> Union[np.ndarray, Tensor]
 
     @abstractmethod
     def post_process(
-        self, predictions: Union[np.ndarray, Tensor], meta_data: Optional[Dict]
-    ) -> Tuple[np.ndarray, float]:
+        self, predictions: Union[np.ndarray, Tensor], meta_data: Optional[Dict[str, Any]]
+    ) -> Dict[str, Any]:
         """Post-Process."""
         raise NotImplementedError
 
     def predict(
         self,
         image: Union[str, np.ndarray, Path],
-        superimpose: bool = True,
-        meta_data: Optional[dict] = None,
-        overlay_mask: bool = False,
-    ) -> Tuple[np.ndarray, float]:
+        meta_data: Optional[Dict[str, Any]] = None,
+    ) -> ImageResult:
         """Perform a prediction for a given input image.
 
         The main workflow is (i) pre-processing, (ii) forward-pass, (iii) post-process.
@@ -77,14 +75,10 @@ def predict(
             image (Union[str, np.ndarray]): Input image whose output is to be predicted.
                 It could be either a path to image or numpy array itself.
 
-            superimpose (bool): If this is set to True, output predictions
-                will be superimposed onto the original image. If false, `predict`
-                method will return the raw heatmap.
-
-            overlay_mask (bool): If this is set to True, output segmentation mask on top of image.
+            meta_data: Meta-data information such as shape, threshold.
 
         Returns:
-            np.ndarray: Output predictions to be visualized.
+            ImageResult: Prediction results to be visualized.
         """
         if meta_data is None:
             if hasattr(self, "meta_data"):
@@ -99,16 +93,15 @@ def predict(
 
         processed_image = self.pre_process(image_arr)
         predictions = self.forward(processed_image)
-        anomaly_map, pred_scores = self.post_process(predictions, meta_data=meta_data)
-
-        # Overlay segmentation mask using raw predictions
-        if overlay_mask and meta_data is not None:
-            image_arr = self._superimpose_segmentation_mask(meta_data, anomaly_map, image_arr)
-
-        if superimpose is True:
-            anomaly_map = superimpose_anomaly_map(anomaly_map, image_arr)
+        output = self.post_process(predictions, meta_data=meta_data)
 
-        return anomaly_map, pred_scores
+        return ImageResult(
+            image=image_arr,
+            pred_score=output["pred_score"],
+            pred_label=output["pred_label"],
+            anomaly_map=output["anomaly_map"],
+            pred_mask=output["pred_mask"],
+        )
 
     def _superimpose_segmentation_mask(self, meta_data: dict, anomaly_map: np.ndarray, image: np.ndarray):
         """Superimpose segmentation mask on top of image.
@@ -130,14 +123,14 @@ def _superimpose_segmentation_mask(self, meta_data: dict, anomaly_map: np.ndarra
         image[outlines] = [255, 0, 0]
         return image
 
-    def __call__(self, image: np.ndarray) -> Tuple[np.ndarray, float]:
+    def __call__(self, image: np.ndarray) -> ImageResult:
         """Call predict on the Image.
 
         Args:
             image (np.ndarray): Input Image
 
         Returns:
-            np.ndarray: Output predictions to be visualized
+            ImageResult: Prediction results to be visualized.
         """
         return self.predict(image)
 
@@ -185,9 +178,7 @@ def _normalize(
 
         return anomaly_maps, float(pred_scores)
 
-    def _load_meta_data(
-        self, path: Optional[Union[str, Path]] = None
-    ) -> Union[DictConfig, Dict[str, Union[float, np.ndarray, Tensor]]]:
+    def _load_meta_data(self, path: Optional[Union[str, Path]] = None) -> Union[DictConfig, Dict]:
         """Loads the meta data from the given path.
 
         Args: