working on perturbation detectors

giskard_vision/core/dataloaders/hf.py

@@ -31,7 +31,11 @@ class HFDataLoader(DataIteratorBase):
     """
 
     def __init__(
-        self, hf_id: str, hf_config: Optional[str] = None, hf_split: str = "test", name: Optional[str] = None
+        self,
+        hf_id: str,
+        hf_config: Optional[str] = None,
+        hf_split: str = "test",
+        name: Optional[str] = None,
     ) -> None:
         """
         Initializes the general HuggingFace Datasets instance.

giskard_vision/core/detectors/metrics.py

@@ -0,0 +1,9 @@
+from giskard_vision.image_classification.tests.performance import Accuracy
+from giskard_vision.landmark_detection.tests.performance import NMEMean
+from giskard_vision.object_detection.tests.performance import IoU
+
+detector_metrics = {
+    "image_classification": Accuracy,
+    "landmark": NMEMean,
+    "object_detection": IoU,
+}

giskard_vision/core/detectors/perturbation.py

@@ -0,0 +1,114 @@
+import os
+from abc import abstractmethod
+from pathlib import Path
+from typing import Any, Sequence
+
+import cv2
+
+from giskard_vision.core.dataloaders.wrappers import FilteredDataLoader
+from giskard_vision.core.detectors.base import (
+    DetectorVisionBase,
+    IssueGroup,
+    ScanResult,
+)
+from giskard_vision.landmark_detection.tests.base import TestDiff
+from giskard_vision.utils.errors import GiskardImportError
+
+from .metrics import detector_metrics
+
+Cropping = IssueGroup(
+    "Cropping", description="Cropping involves evaluating the landmark detection model on specific face areas."
+)
+
+Ethical = IssueGroup(
+    "Ethical",
+    description="The data are filtered by ethnicity to detect ethical biases in the landmark detection model.",
+)
+
+Pose = IssueGroup(
+    "Head Pose",
+    description="The data are filtered by head pose to detect biases in the landmark detection model.",
+)
+
+Robustness = IssueGroup(
+    "Robustness",
+    description="Images from the dataset are blurred, recolored and resized to test the robustness of the model to transformations.",
+)
+
+
+class PerturbationBaseDetector(DetectorVisionBase):
+    """
+    Abstract class for Landmark Detection Detectors
+
+    Methods:
+        get_dataloaders(dataset: Any) -> Sequence[Any]:
+            Abstract method that returns a list of dataloaders corresponding to
+            slices or transformations
+
+        get_results(model: Any, dataset: Any) -> Sequence[ScanResult]:
+            Returns a list of ScanResult containing the evaluation results
+
+        get_scan_result(self, test_result) -> ScanResult:
+            Convert TestResult to ScanResult
+    """
+
+    @abstractmethod
+    def get_dataloaders(self, dataset: Any) -> Sequence[Any]: ...
+
+    def get_results(self, model: Any, dataset: Any) -> Sequence[ScanResult]:
+        dataloaders = self.get_dataloaders(dataset)
+
+        results = []
+        for dl in dataloaders:
+            test_result = TestDiff(metric=detector_metrics[model.model_type], threshold=1).run(
+                model=model,
+                dataloader=dl,
+                dataloader_ref=dataset,
+            )
+
+            # Save example images from dataloader and dataset
+            current_path = str(Path())
+            os.makedirs(f"{current_path}/examples_images", exist_ok=True)
+            filename_examples = []
+
+            index_worst = 0 if test_result.indexes_examples is None else test_result.indexes_examples[0]
+
+            if isinstance(dl, FilteredDataLoader):
+                filename_example_dataloader_ref = str(Path() / "examples_images" / f"{dataset.name}_{index_worst}.png")
+                cv2.imwrite(
+                    filename_example_dataloader_ref, cv2.resize(dataset[index_worst][0][0], (0, 0), fx=0.3, fy=0.3)
+                )
+                filename_examples.append(filename_example_dataloader_ref)
+
+            filename_example_dataloader = str(Path() / "examples_images" / f"{dl.name}_{index_worst}.png")
+            cv2.imwrite(filename_example_dataloader, cv2.resize(dl[index_worst][0][0], (0, 0), fx=0.3, fy=0.3))
+            filename_examples.append(filename_example_dataloader)
+            results.append(self.get_scan_result(test_result, filename_examples, dl.name, len(dl)))
+
+        return results
+
+    def get_scan_result(self, test_result, filename_examples, name, size_data) -> ScanResult:
+        try:
+            from giskard.scanner.issues import IssueLevel
+        except (ImportError, ModuleNotFoundError) as e:
+            raise GiskardImportError(["giskard"]) from e
+
+        relative_delta = (test_result.metric_value_test - test_result.metric_value_ref) / test_result.metric_value_ref
+
+        if relative_delta > self.issue_level_threshold + self.deviation_threshold:
+            issue_level = IssueLevel.MAJOR
+        elif relative_delta > self.issue_level_threshold:
+            issue_level = IssueLevel.MEDIUM
+        else:
+            issue_level = IssueLevel.MINOR
+
+        return ScanResult(
+            name=name,
+            metric_name=test_result.metric_name,
+            metric_value=test_result.metric_value_test,
+            metric_reference_value=test_result.metric_value_ref,
+            issue_level=issue_level,
+            slice_size=size_data,
+            filename_examples=filename_examples,
+            relative_delta=relative_delta,
+        )

giskard_vision/core/detectors/transformation_blurring_detector.py

@@ -0,0 +1,24 @@
+from giskard_vision.core.dataloaders.wrappers import BlurredDataLoader
+
+from ...core.detectors.decorator import maybe_detector
+from .perturbation import PerturbationBaseDetector, Robustness
+
+
+@maybe_detector("blurring", tags=["vision", "robustness", "image_classification", "landmark", "object_detection"])
+class TransformationBlurringDetectorLandmark(PerturbationBaseDetector):
+    """
+    Detector that evaluates models performance on blurred images
+    """
+
+    issue_group = Robustness
+
+    def __init__(self, kernel_size=(11, 11), sigma=(3, 3)):
+        self.kernel_size = kernel_size
+        self.sigma = sigma
+
+    def get_dataloaders(self, dataset):
+        dl = BlurredDataLoader(dataset, self.kernel_size, self.sigma)
+
+        dls = [dl]
+
+        return dls

giskard_vision/core/detectors/transformation_color_detector.py

@@ -0,0 +1,20 @@
+from giskard_vision.core.dataloaders.wrappers import ColoredDataLoader
+
+from ...core.detectors.decorator import maybe_detector
+from .perturbation import PerturbationBaseDetector, Robustness
+
+
+@maybe_detector("coloring", tags=["vision", "robustness", "image_classification", "landmark", "object_detection"])
+class TransformationColorDetectorLandmark(PerturbationBaseDetector):
+    """
+    Detector that evaluates models performance depending on images in grayscale
+    """
+
+    issue_group = Robustness
+
+    def get_dataloaders(self, dataset):
+        dl = ColoredDataLoader(dataset)
+
+        dls = [dl]
+
+        return dls

giskard_vision/core/models/base.py

@@ -18,15 +18,37 @@ class ModelBase(ABC):
     prediction_result_cls = TypesBase.prediction_result
 
     @abstractmethod
-    def predict_image(self, image: np.ndarray) -> Any:
-        """abstract method that takes one image as input and outputs the prediction
+    def predict_rgb_image(self, image: np.ndarray) -> Any:
+        """abstract method that takes one RGB image as input and outputs the prediction
 
         Args:
             image (np.ndarray): input image
         """
 
         ...
 
+    def predict_gray_image(self, image: np.ndarray) -> Any:
+        """abstract method that takes one gray image as input and outputs the prediction
+
+        Args:
+            image (np.ndarray): input image
+        """
+
+        raise NotImplementedError("predict_gray_image method is not implemented")
+
+    def predict_image(self, image: np.ndarray) -> Any:
+        """abstract method that takes one image as input and outputs the prediction
+
+        Args:
+            image (np.ndarray): input image
+        """
+        if image.shape[-1] == 3:
+            return self.predict_rgb_image(image)
+        elif image.shape[-1] == 1 or len(image.shape) == 2:
+            return self.predict_gray_image(image)
+        else:
+            raise ValueError("predict_image: image shape not supported.")
+
     def predict_batch(self, idx: int, images: List[np.ndarray]) -> np.ndarray:
         """method that should be implemented if the passed dataloader has batch_size != 1
 
@@ -40,7 +62,7 @@ def predict_batch(self, idx: int, images: List[np.ndarray]) -> np.ndarray:
             except Exception:
                 res.append(None)
                 logger.warning(
-                    f"{self.__class__.__name__}: Face not detected in processed image of batch {idx} and index {i}."
+                    f"{self.__class__.__name__}: Prediction failed in processed image of batch {idx} and index {i}."
                 )
                 # logger.warning(e) # OpenCV's exception is very misleading
 

giskard_vision/core/models/hf_pipeline.py

@@ -19,9 +19,9 @@ def __init__(
         """init method that accepts a model object, number of landmarks and dimensions
 
         Args:
-            model_id (str): Hugging Face model ID
-            name (Optional[str]): name of the model
-            pipeline_task (HFPipelineTask): HuggingFace pipeline task
+            model_id (str): Hugging Face model ID.
+            name (Optional[str]): name of the model.
+            pipeline_task (HFPipelineTask): HuggingFace pipeline task.
 
         Raises:
             GiskardImportError: If there are missing Hugging Face dependencies.

giskard_vision/image_classification/dataloaders/loaders.py

@@ -45,7 +45,7 @@ def get_image(self, idx: int) -> np.ndarray:
         Returns:
             np.ndarray: The image data.
         """
-        return self.get_row(idx)["image"]
+        return np.array(self.get_row(idx)["image"])
 
     def get_labels(self, idx: int) -> Optional[np.ndarray]:
         """
@@ -124,7 +124,7 @@ def get_image(self, idx: int) -> Any:
         Returns:
             np.ndarray: The image data.
         """
-        return self.ds[idx]["image"]
+        return np.array(self.ds[idx]["image"])
 
     def get_labels(self, idx: int) -> Optional[np.ndarray]:
         """
@@ -201,7 +201,7 @@ def get_image(self, idx: int) -> Any:
         Returns:
             np.ndarray: The image data.
         """
-        return self.ds[idx]["img"]
+        return np.array(self.ds[idx]["img"])
 
     def get_labels(self, idx: int) -> Optional[np.ndarray]:
         """

giskard_vision/image_classification/models/base.py

@@ -1,6 +1,7 @@
 from typing import Optional
 
 import numpy as np
+from PIL import Image
 
 from giskard_vision.core.models.hf_pipeline import HFPipelineModelBase, HFPipelineTask
 from giskard_vision.image_classification.types import Types
@@ -10,9 +11,10 @@ class ImageClassificationHFModel(HFPipelineModelBase):
     """Hugging Face pipeline wrapper class that serves as a template for image classification predictions
 
     Args:
-        model_id (str): Hugging Face model ID
-        name (Optional[str]): name of the model
-        device (str): device to run the model on
+        model_id (str): Hugging Face model ID.
+        name (Optional[str]): name of the model.
+        device (str): device to run the model on.
+        mode (str): The mode to convert the numpy image data to PIL image, defaulting to "RGB".
 
     Attributes:
         classification_labels: list of classification labels, where the position of the label corresponds to the class index
@@ -21,13 +23,14 @@ class ImageClassificationHFModel(HFPipelineModelBase):
     model_type = "image_classification"
     prediction_result_cls = Types.prediction_result
 
-    def __init__(self, model_id: str, name: Optional[str] = None, device: str = "cpu"):
+    def __init__(self, model_id: str, name: Optional[str] = None, device: str = "cpu", mode: str = "RGB"):
         """init method that accepts a model id, name and device
 
         Args:
-            model_id (str): Hugging Face model ID
-            name (Optional[str]): name of the model
-            device (str): device to run the model on
+            model_id (str): Hugging Face model ID.
+            name (Optional[str]): name of the model.
+            device (str): device to run the model on.
+            mode (str): The mode to convert the numpy image data to PIL image, defaulting to "RGB".
         """
 
         super().__init__(
@@ -38,6 +41,7 @@ def __init__(self, model_id: str, name: Optional[str] = None, device: str = "cpu
         )
 
         self._classification_labels = list(self.pipeline.model.config.id2label.values())
+        self._mode = mode
 
     @property
     def classification_labels(self):
@@ -57,25 +61,29 @@ class SingleLabelImageClassificationHFModelWrapper(ImageClassificationHFModel):
         classification_labels: list of classification labels, where the position of the label corresponds to the class index
     """
 
-    def predict_probas(self, image: np.ndarray) -> np.ndarray:
+    def predict_probas(self, image: np.ndarray, mode=None) -> np.ndarray:
         """method that takes one image as input and outputs the prediction of probabilities for each class
 
         Args:
             image (np.ndarray): input image
+            mode (str): mode of the image
         """
+        m = mode or self._mode
+        pil_image = Image.fromarray(image, mode=m)
+
+        # Pipeline takes a PIL image as input
         _raw_prediction = self.pipeline(
-            image,
+            pil_image,
             top_k=len(self.classification_labels),  # Get probabilities for all labels
         )
         _prediction = {p["label"]: p["score"] for p in _raw_prediction}
 
         return np.array([_prediction[label] for label in self.classification_labels])
 
-    def predict_image(self, image) -> Types.label:
-        """method that takes one image as input and outputs one class label
+    def predict_rgb_image(self, image: np.ndarray) -> Types.label:
+        probas = self.predict_probas(image, mode=None)
+        return self.classification_labels[np.argmax(probas)]
 
-        Args:
-            image (np.ndarray): input image
-        """
-        probas = self.predict_probas(image)
+    def predict_gray_image(self, image: np.ndarray) -> Types.label:
+        probas = self.predict_probas(image, mode="L")
         return self.classification_labels[np.argmax(probas)]

giskard_vision/landmark_detection/models/wrappers.py

@@ -29,7 +29,7 @@ def __init__(self, model):
         super().__init__(n_landmarks=68, n_dimensions=2, name="FaceAlignment")
         self.model = model
 
-    def predict_image(self, image):
+    def predict_rgb_image(self, image):
         """
         Predict facial landmarks for a given image using the wrapped face alignment model.
 
@@ -100,7 +100,7 @@ def __init__(self):
         self.landmark_detector = cv2.face.createFacemarkLBF()
         self.landmark_detector.loadModel(LBFmodel)
 
-    def predict_image(self, image):
+    def predict_rgb_image(self, image):
         """
         Predict facial landmarks for a given image using the wrapped OpenCV face landmarks model.
 

giskard_vision/object_detection/models/wrappers.py

@@ -97,7 +97,7 @@ def shape_rescale(self, image, boxes):
     def positive_constraint(self, boxes):
         return np.clip(boxes, 0, None)
 
-    def predict_image(self, image: np.ndarray):
+    def predict_rgb_image(self, image: np.ndarray):
         try:
             from keras.applications.mobilenet import preprocess_input
         except ImportError: