[WIP] initial pass at convolutional layer visualization

kviz/conv.py

@@ -0,0 +1,179 @@
+"""
+Copyright 2021 Lance Galletti
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+"""
+
+
+import numpy as np
+from PIL import Image as im
+import matplotlib.pyplot as plt
+from tensorflow.keras import models
+
+
+class ConvGraph():
+    """
+        Class for creating and rendering visualization of Keras
+        Sequential Model with Convolutional Layers
+
+        Attributes:
+            model : tf.keras.Model
+                a compiled keras sequential model
+
+        Methods:
+            render :
+                Shows all the convolution activations
+
+    """
+
+    def __init__(self, model):
+        self.model = model
+
+
+    def _snap_layer(self, display_grid, scale, filename, xticks=None, yticks=None):
+        fig, ax = plt.subplots(figsize=(int(scale * display_grid.shape[1]), int(scale * display_grid.shape[0])))
+        ax.set_xticks(xticks)
+        ax.set_yticks(yticks)
+        ax.grid(True)
+        ax.axis('off')
+        ax.imshow(display_grid, aspect='auto')
+        fig.savefig(filename + '.png', transparent=True, bbox_inches='tight', pad_inches=0)
+        plt.close()
+        return np.asarray(im.open(filename + '.png'))
+
+
+    def animate(self, X=None, filename='conv_animation'):
+        """
+        Render animation of a Convolutional layers based on a stream
+        of input.
+
+        Parameters:
+            X : ndarray
+                input to a Keras model - ideally of the same class
+            filename : str
+                name of file to which visualization will be saved
+
+        Returns:
+            None
+        """
+
+        layer_outputs = [layer.output for layer in self.model.layers]
+        # Creates a model that will return these outputs, given the model input
+        activation_model = models.Model(inputs=self.model.input, outputs=layer_outputs)
+        images_per_row = 8
+
+        for i in range(len(self.model.layers)):
+            # Ignore non-conv2d layers
+            layer_name = self.model.layers[i].name
+            if not layer_name.startswith("conv2d"):
+                continue
+
+            images = []
+            heat = []
+            for j in range(len(X)):
+                activations = activation_model.predict(X[j])
+                # Number of features in the feature map
+                n_features = activations[i].shape[-1]
+                # The feature map has shape (1, size, size, n_features).
+                size = activations[i].shape[1]
+                # Tiles the activation channels in this matrix
+                n_cols = n_features // images_per_row
+                display_grid = np.zeros((size * n_cols, images_per_row * size))
+                # Tiles each filter into a big horizontal grid
+                for col in range(n_cols):
+                    for row in range(images_per_row):
+                        # Displays the grid
+                        display_grid[
+                            col * size: (col + 1) * size,
+                            row * size: (row + 1) * size] = activations[i][0, :, :, col * images_per_row + row]
+
+                snapped = self._snap_layer(
+                    display_grid, 1.0 / size,
+                    filename + "_" + layer_name,
+                    xticks=np.linspace(-1, display_grid.shape[1], images_per_row + 1),
+                    yticks=np.linspace(-1, display_grid.shape[0], n_cols + 1))
+                heat.append(snapped)
+                images.append(im.fromarray(snapped))
+
+            images[0].save(
+                filename + "_" + layer_name + '.gif',
+                optimize=False,  # important for transparent background
+                save_all=True,
+                append_images=images[1:],
+                loop=0,
+                duration=100,
+                transparency=255,  # prevent PIL from making background black
+                disposal=2
+            )
+
+            heatmap = heat[0]
+            for i in range(1, len(heat)):
+                heatmap = np.where(heatmap < heat[i], heat[i], heatmap)
+            fig, ax = plt.subplots()
+            ax.axis('off')
+            ax.imshow(heatmap)
+            fig.savefig(filename + "_" + layer_name + '_heatmap.png', transparent=True, bbox_inches='tight', pad_inches=0)
+
+        return
+
+
+    def render(self, X=None, filename='conv_filters'):
+        """
+        Render visualization of a Convolutional keras model
+
+        Parameters:
+            X : ndarray
+                input to a Keras model
+            filename : str
+                name of file to which visualization will be saved
+
+        Returns:
+            None
+        """
+
+        layer_outputs = [layer.output for layer in self.model.layers]
+        # Creates a model that will return these outputs, given the model input
+        activation_model = models.Model(inputs=self.model.input, outputs=layer_outputs)
+        images_per_row = 8
+
+        for j in range(len(X)):
+            activations = activation_model.predict(X[j])
+
+            for i in range(len(activations)):
+                # Ignore non-conv2d layers
+                layer_name = self.model.layers[i].name
+                if not layer_name.startswith("conv2d"):
+                    continue
+
+                # Number of features in the feature map
+                n_features = activations[i].shape[-1]
+                # The feature map has shape (1, size, size, n_features).
+                size = activations[i].shape[1]
+                # Tiles the activation channels in this matrix
+                n_cols = n_features // images_per_row
+                display_grid = np.zeros((size * n_cols, images_per_row * size))
+                # Tiles each filter into a big horizontal grid
+                for col in range(n_cols):
+                    for row in range(images_per_row):
+                        # Displays the grid
+                        display_grid[
+                            col * size: (col + 1) * size,
+                            row * size: (row + 1) * size] = activations[i][0, :, :, col * images_per_row + row]
+
+                self._snap_layer(
+                    display_grid, 1. / size,
+                    filename + "_" + str(j) + "_" + layer_name,
+                    xticks=np.linspace(-1, display_grid.shape[1], images_per_row + 1),
+                    yticks=np.linspace(-1, display_grid.shape[0], n_cols + 1))
+
+        return

kviz/dense.py

@@ -49,6 +49,10 @@ class DenseGraph():
                 is provided, show a GIF of the activations of each
                 Neuron based on the input provided.
 
+            animate_learning:
+                Make GIF from snapshots of decision boundary at
+                given snap_freq
+
     """
 
     def __init__(self, model):

tests/test_conv.py

@@ -0,0 +1,84 @@
+import numpy as np
+from tensorflow import keras
+from tensorflow.keras import layers, utils
+from tensorflow.keras.datasets import mnist
+
+from kviz.conv import ConvGraph
+
+
+def test_conv_input():
+    (X_train, y_train), (X_test, y_test) = mnist.load_data()
+
+    X_train = X_train.reshape(X_train.shape[0], 28, 28, 1)
+    X_train = X_train.astype('float32')
+    X_train /= 255
+
+    X_test = X_test.reshape(X_test.shape[0], 28, 28, 1)
+    X_test = X_test.astype('float32')
+    X_test /= 255
+
+    number_of_classes = 10
+    Y_train = utils.to_categorical(y_train, number_of_classes)
+    Y_test = utils.to_categorical(y_test, number_of_classes)
+
+    ACTIVATION = "relu"
+    model = keras.models.Sequential()
+    model.add(layers.Conv2D(32, 5, input_shape=(28, 28, 1), activation=ACTIVATION))
+    model.add(layers.MaxPooling2D())
+    model.add(layers.Conv2D(64, 5, activation=ACTIVATION))
+    model.add(layers.MaxPooling2D())
+    model.add(layers.Flatten())
+    model.add(layers.Dense(100, activation=ACTIVATION))
+    model.add(layers.Dense(10, activation="softmax"))
+    model.compile(loss="categorical_crossentropy", metrics=['accuracy'])
+
+    model.fit(X_train, Y_train, batch_size=100, epochs=5)
+
+    score = model.evaluate(X_test, Y_test, verbose=0)
+    print("Test loss:", score[0])
+    print("Test accuracy:", score[1])
+
+    dg = ConvGraph(model)
+    X = []
+    for i in range(number_of_classes):
+        X.append(np.expand_dims(X_train[np.where(y_train == i)[0][0]], axis=0))
+    dg.render(X, filename='test_input_mnist')
+
+
+def test_conv_animate():
+    (X_train, y_train), (X_test, y_test) = mnist.load_data()
+
+    X_train = X_train.reshape(X_train.shape[0], 28, 28, 1)
+    X_train = X_train.astype('float32')
+    X_train /= 255
+
+    X_test = X_test.reshape(X_test.shape[0], 28, 28, 1)
+    X_test = X_test.astype('float32')
+    X_test /= 255
+
+    number_of_classes = 10
+    Y_train = utils.to_categorical(y_train, number_of_classes)
+    Y_test = utils.to_categorical(y_test, number_of_classes)
+
+    ACTIVATION = "relu"
+    model = keras.models.Sequential()
+    model.add(layers.Conv2D(32, 5, input_shape=(28, 28, 1), activation=ACTIVATION))
+    model.add(layers.MaxPooling2D())
+    model.add(layers.Conv2D(64, 5, activation=ACTIVATION))
+    model.add(layers.MaxPooling2D())
+    model.add(layers.Flatten())
+    model.add(layers.Dense(100, activation=ACTIVATION))
+    model.add(layers.Dense(10, activation="softmax"))
+    model.compile(loss="categorical_crossentropy", metrics=['accuracy'])
+
+    model.fit(X_train, Y_train, batch_size=100, epochs=5)
+
+    score = model.evaluate(X_test, Y_test, verbose=0)
+    print("Test loss:", score[0])
+    print("Test accuracy:", score[1])
+
+    dg = ConvGraph(model)
+    X = []
+    for i in range(min(50, len(np.where(y_train == 0)[0]))):
+        X.append(np.expand_dims(X_train[np.where(y_train == 0)[0][i]], axis=0))
+    dg.animate(X, filename='test_animate_mnist')

tests/test_viz.py → tests/test_dense.py

@@ -35,7 +35,7 @@ def test_dense_input_xor():
         [1, 1]])
     Y = np.array([x[0] ^ x[1] for x in X])
 
-    model.fit(X, Y, batch_size=4, epochs=1000)
+    model.fit(X, Y, batch_size=4, epochs=100)
 
     colors = np.array(['b', 'g'])
     fig, ax = plt.subplots()
@@ -71,7 +71,7 @@ def test_dense_input_line():
     X = np.array(t)
     Y = np.array([1 if x[0] - x[1] >= 0 else 0 for x in X])
 
-    model.fit(X, Y, batch_size=50, epochs=100)
+    model.fit(X, Y, batch_size=50, epochs=10)
 
     # see which nodes activate for a given class
     X0 = X[X[:, 0] - X[:, 1] <= 0]