279 embeddings

.gitignore

@@ -34,3 +34,6 @@ venv
 venv3
 
 .python-version
+
+# testing jpgs
+embedding_WIP/*.jpg*

dianna/__init__.py

@@ -24,6 +24,7 @@
 import importlib
 import logging
 from . import utils
+from .methods.distance import DistanceExplainer
 
 
 logging.getLogger(__name__).addHandler(logging.NullHandler())
@@ -75,6 +76,12 @@ def explain_text(model_or_function, input_data, method, labels=(1,), **kwargs):
     explain_text_kwargs = utils.get_kwargs_applicable_to_function(explainer.explain_text, kwargs)
     return explainer.explain_text(model_or_function, input_data, labels, **explain_text_kwargs)
 
+def explain_image_distance(model_or_function, input_data, embedded_reference, **kwargs):
+    method_kwargs = utils.get_kwargs_applicable_to_function(DistanceExplainer.__init__, kwargs)
+    explainer = DistanceExplainer(**method_kwargs)
+    explain_distance_kwargs = utils.get_kwargs_applicable_to_function(explainer.explain_image_distance, kwargs)
+    return explainer.explain_image_distance(model_or_function, input_data, embedded_reference, **explain_distance_kwargs)
+
 
 def _get_explainer(method, kwargs):
     method_submodule = importlib.import_module(f'dianna.methods.{method.lower()}')

dianna/methods/distance.py

@@ -0,0 +1,156 @@
+import os
+import random
+from urllib.parse import urlparse
+import numpy as np
+import tensorflow as tf
+from keras import backend as K
+from keras.preprocessing import image
+from matplotlib import pyplot as plt
+from requests import get
+from skimage.transform import resize
+from tensorflow.keras import backend as K
+from tensorflow.keras.applications.resnet50 import ResNet50
+from tensorflow.keras.applications.resnet50 import decode_predictions
+from tensorflow.keras.applications.resnet50 import preprocess_input
+from tqdm import tqdm
+from dianna.methods.rise import generate_masks_for_images
+from dianna import utils
+from sklearn.metrics import pairwise_distances
+
+
+class DistanceExplainer:
+    # axis labels required to be present in input image data
+    required_labels = ('channels',)
+
+    def __init__(self, n_masks=1000, feature_res=8, p_keep=.5,  # pylint: disable=too-many-arguments
+                 mask_selection_range_max=0.2, mask_selection_range_min=0, mask_selection_negative_range_max=1,
+                 mask_selection_negative_range_min=0.8, axis_labels=None, batch_size=10,
+                 preprocess_function=None):
+        self.n_masks = n_masks
+        self.feature_res = feature_res
+        self.p_keep = p_keep
+        self.preprocess_function = preprocess_function
+        self.masks = None
+        self.predictions = None
+        self.axis_labels = axis_labels if axis_labels is not None else []
+        self.mask_selection_range_max = mask_selection_range_max
+        self.mask_selection_range_min = mask_selection_range_min
+        self.mask_selection_negative_range_max = mask_selection_negative_range_max
+        self.mask_selection_negative_range_min = mask_selection_negative_range_min
+        self.batch_size = batch_size
+
+    def explain_image_distance(self, model_or_function, input_data, embedded_reference, **explain_distance_kwargs):
+        """
+
+        :param model_or_function:
+        :param input_data:
+        :param embedded_reference:
+        :param explain_distance_kwargs:
+        :return: saliency map and the neutral value within the saliency map which indicates the parts of the image that
+        neither bring the image closer nor further away from the embedded reference.
+        """
+        full_preprocess_function, input_data = self._prepare_input_data(input_data)
+        runner = utils.get_function(model_or_function, preprocess_function=full_preprocess_function)
+        active_p_keep = 0.5 if self.p_keep is None else self.p_keep  # Could autotune here (See #319)
+
+        # data shape without batch axis and channel axis
+        img_shape = input_data.shape[1:3]
+        # Expose masks for to make user inspection possible
+        self.masks = generate_masks_for_images(img_shape, active_p_keep, self.n_masks, self.feature_res)
+        # Make sure multiplication is being done for correct axes
+        masked = input_data * self.masks
+
+        batch_predictions = []
+
+        for i in tqdm(range(0, self.n_masks, self.batch_size), desc='Explaining'):
+            new_predictions = runner(masked[i:i + self.batch_size])
+            batch_predictions.append(new_predictions)
+
+        self.predictions = np.concatenate(batch_predictions)
+
+        lowest_distances_masks, lowest_mask_weights = self._get_lowest_distance_masks_and_weights(embedded_reference,
+                                                                                           self.predictions, self.masks,
+                                                                                           self.mask_selection_range_min,
+                                                                                           self.mask_selection_range_max)
+        highest_distances_masks, highest_mask_weights = self._get_lowest_distance_masks_and_weights(embedded_reference,
+                                                                                           self.predictions, self.masks,
+                                                                                           self.mask_selection_negative_range_min,
+                                                                                           self.mask_selection_negative_range_max)
+
+        def describe(x, name):
+            return f'Description of {name}\nmean:{np.mean(x)}\nstd:{np.std(x)}\nmin:{np.min(x)}\nmax:{np.max(x)}'
+        self.statistics = describe(highest_mask_weights, 'highest_mask_weights') +'\n' + describe(lowest_mask_weights, 'lowest_mask_weights')
+
+        unnormalized_sal_lowest = np.mean(lowest_distances_masks, axis=0)
+        unnormalized_sal_highest = np.mean(highest_distances_masks, axis=0)
+        unnormalized_sal = unnormalized_sal_lowest - unnormalized_sal_highest
+
+        saliency = unnormalized_sal
+
+        input_prediction = runner(input_data)
+        input_distance = pairwise_distances(input_prediction, embedded_reference, metric='cosine') / 2
+        neutral_value = np.exp(-input_distance)
+
+        return saliency, neutral_value
+
+    @staticmethod
+    def _get_lowest_distance_masks_and_weights(embedded_reference, predictions, masks, mask_selection_range_min,
+                                               mask_selection_range_max):
+        distances = pairwise_distances(predictions, embedded_reference,
+                                       metric='cosine') / 2  # divide by 2 to have [0.1] output range
+        lowest_distances_indices = np.argsort(distances, axis=0)[
+                                   int(len(predictions) * mask_selection_range_min)
+                                   :int(len(predictions) * mask_selection_range_max)]
+        mask_weights = np.exp(-distances[lowest_distances_indices])
+        lowest_distances_masks = masks[lowest_distances_indices]
+        return lowest_distances_masks, mask_weights
+
+    def _prepare_input_data(self, input_data):
+        input_data_xarray = utils.to_xarray(input_data, self.axis_labels, DistanceExplainer.required_labels)
+        input_data_xarray_expanded = input_data_xarray.expand_dims('batch', 0)
+        # ensure channels axis is last and keep track of where it was so we can move it back
+        channels_axis_index = input_data_xarray_expanded.dims.index('channels')
+        prepared_input_data = utils.move_axis(input_data_xarray_expanded, 'channels', -1)
+        # create preprocessing function that puts model input generated by RISE into the right shape and dtype,
+        # followed by running the user's preprocessing function
+        full_preprocess_function = self._get_full_preprocess_function(channels_axis_index, prepared_input_data.dtype)
+        return full_preprocess_function, prepared_input_data
+
+    def _prepare_image_data(self, input_data):
+        """Transforms the data to be of the shape and type RISE expects.
+
+        Args:
+            input_data (xarray): Data to be explained
+
+        Returns:
+            transformed input data, preprocessing function to use with utils.get_function()
+        """
+        # ensure channels axis is last and keep track of where it was so we can move it back
+        channels_axis_index = input_data.dims.index('channels')
+        input_data = utils.move_axis(input_data, 'channels', -1)
+        # create preprocessing function that puts model input generated by RISE into the right shape and dtype,
+        # followed by running the user's preprocessing function
+        full_preprocess_function = self._get_full_preprocess_function(channels_axis_index, input_data.dtype)
+        return input_data, full_preprocess_function
+
+    def _get_full_preprocess_function(self, channel_axis_index, dtype):
+        """Creates a full preprocessing function.
+
+        Creates a preprocessing function that incorporates both the (optional) user's
+        preprocessing function, as well as any needed dtype and shape conversions
+
+        Args:
+            channel_axis_index (int): Axis index of the channels in the input data
+            dtype (type): Data type of the input data (e.g. np.float32)
+
+        Returns:
+            Function that first ensures the data has the same shape and type as the input data,
+            then runs the users' preprocessing function
+        """
+
+        def moveaxis_function(data):
+            return utils.move_axis(data, 'channels', channel_axis_index).astype(dtype).values
+
+        if self.preprocess_function is None:
+            return moveaxis_function
+        return lambda data: self.preprocess_function(moveaxis_function(data))

dianna/methods/rise.py

@@ -13,6 +13,33 @@ def _upscale(grid_i, up_size):
     return resize(grid_i, up_size, order=1, mode='reflect', anti_aliasing=False)
 
 
+def generate_masks_for_images(input_size, p_keep, n_masks, feature_res):
+    """Generates a set of random masks to mask the input data.
+
+    Args:
+        input_size (int): Size of a single sample of input data, for images without the channel axis.
+
+    Returns:
+        The generated masks (np.ndarray)
+    """
+    cell_size = np.ceil(np.array(input_size) / feature_res)
+    up_size = (feature_res + 1) * cell_size
+
+    grid = np.random.choice(a=(True, False), size=(n_masks, feature_res, feature_res),
+                            p=(p_keep, 1 - p_keep))
+    grid = grid.astype('float32')
+
+    masks = np.empty((n_masks, *input_size), dtype=np.float32)
+
+    for i in range(n_masks):
+        y = np.random.randint(0, cell_size[0])
+        x = np.random.randint(0, cell_size[1])
+        # Linear upsampling and cropping
+        masks[i, :, :] = _upscale(grid[i], up_size)[y:y + input_size[0], x:x + input_size[1]]
+    masks = masks.reshape(-1, *input_size, 1)
+    return masks
+
+
 class RISE:
     """RISE implementation based on https://github.com/eclique/RISE/blob/master/Easy_start.ipynb."""
     # axis labels required to be present in input image data
@@ -149,7 +176,7 @@ def explain_image(self, model_or_function, input_data, labels=None, batch_size=1
         # data shape without batch axis and channel axis
         img_shape = input_data.shape[1:3]
         # Expose masks for to make user inspection possible
-        self.masks = self.generate_masks_for_images(img_shape, active_p_keep, self.n_masks)
+        self.masks = generate_masks_for_images(img_shape, active_p_keep, self.n_masks, self.feature_res)
 
         # Make sure multiplication is being done for correct axes
         masked = input_data * self.masks
@@ -180,7 +207,7 @@ def _determine_p_keep_for_images(self, input_data, runner, n_masks=100):
     def _calculate_mean_class_std_for_images(self, p_keep, runner, input_data, n_masks):
         batch_size = 50
         img_shape = input_data.shape[1:3]
-        masks = self.generate_masks_for_images(img_shape, p_keep, n_masks)
+        masks = generate_masks_for_images(img_shape, p_keep, n_masks, self.feature_res)
         masked = input_data * masks
         predictions = []
         for i in range(0, n_masks, batch_size):
@@ -189,33 +216,7 @@ def _calculate_mean_class_std_for_images(self, p_keep, runner, input_data, n_mas
             predictions.append(current_predictions.max(axis=1))
         predictions = np.concatenate(predictions)
         std_per_class = predictions.std()
-        return np.mean(std_per_class)
-
-    def generate_masks_for_images(self, input_size, p_keep, n_masks):
-        """Generates a set of random masks to mask the input data.
-
-        Args:
-            input_size (int): Size of a single sample of input data, for images without the channel axis.
-
-        Returns:
-            The generated masks (np.ndarray)
-        """
-        cell_size = np.ceil(np.array(input_size) / self.feature_res)
-        up_size = (self.feature_res + 1) * cell_size
-
-        grid = np.random.choice(a=(True, False), size=(n_masks, self.feature_res, self.feature_res),
-                                p=(p_keep, 1 - p_keep))
-        grid = grid.astype('float32')
-
-        masks = np.empty((n_masks, *input_size), dtype=np.float32)
-
-        for i in range(n_masks):
-            y = np.random.randint(0, cell_size[0])
-            x = np.random.randint(0, cell_size[1])
-            # Linear upsampling and cropping
-            masks[i, :, :] = _upscale(grid[i], up_size)[y:y + input_size[0], x:x + input_size[1]]
-        masks = masks.reshape(-1, *input_size, 1)
-        return masks
+        return np.mean(std_per_class)    
 
     def _prepare_image_data(self, input_data):
         """Transforms the data to be of the shape and type RISE expects.

dianna/utils/misc.py

@@ -1,5 +1,8 @@
 import inspect
 
+import PIL
+import numpy as np
+
 
 def get_function(model_or_function, preprocess_function=None):
     """Converts input to callable function.
@@ -36,9 +39,20 @@ def get_kwargs_applicable_to_function(function, kwargs):
             if key in inspect.getfullargspec(function).args}
 
 
+def _get_num_dims(data):
+    if hasattr(data, 'ndim'):
+        return data.ndim
+    if hasattr(data, 'shape'):
+        return len(data.shape)
+    raise TypeError('Unsupported data type. Supported types are numpy arrays or PIL images and similar.')
+
+
 def to_xarray(data, axis_labels, required_labels=None):
     """Converts numpy data and axes labels to an xarray object."""
-    if isinstance(axis_labels, dict):
+    if isinstance(data, PIL.Image.Image):
+        data = np.array(data)
+        labels = ['dim_0', 'dim_1', 'channels']
+    elif isinstance(axis_labels, dict):
         # key = axis index, value = label
         # not all axes have to be present in the input, but we need to provide
         # a name for each axis
@@ -47,7 +61,7 @@ def to_xarray(data, axis_labels, required_labels=None):
         for index in indices:
             if index < 0:
                 axis_labels[data.ndim + index] = axis_labels.pop(index)
-        labels = [axis_labels[index] if index in axis_labels else f'dim_{index}' for index in range(data.ndim)]
+        labels = [axis_labels[index] if index in axis_labels else f'dim_{index}' for index in range(_get_num_dims(data))]
     else:
         labels = list(axis_labels)