Moving baselines to bob.bio.face and make them as transformers

bob/bio/face/test/test_transformers.py

+import bob.bio.face
+import numpy as np
+
+
+def test_facenet():
+    from bob.bio.face.transformers import FaceNetSanderberg
+
+    np.random.seed(10)
+
+    transformer = FaceNetSanderberg()
+    data = np.random.rand(3, 160, 160).astype("uint8")
+    output = transformer.transform(data)
+    assert output.size == 128, output.shape
+
+
+def test_idiap_inceptionv2_msceleb():
+    from bob.bio.face.transformers import InceptionResnetv2_MsCeleb
+
+    np.random.seed(10)
+    transformer = InceptionResnetv2_MsCeleb()
+    data = np.random.rand(3, 160, 160).astype("uint8")
+    output = transformer.transform(data)
+    assert output.size == 128, output.shape
+
+
+def test_idiap_inceptionv2_casia():
+    from bob.bio.face.transformers import InceptionResnetv2_CasiaWebFace
+
+    np.random.seed(10)
+    transformer = InceptionResnetv2_CasiaWebFace()
+    data = np.random.rand(3, 160, 160).astype("uint8")
+    output = transformer.transform(data)
+    assert output.size == 128, output.shape
+
+
+def test_idiap_inceptionv1_msceleb():
+    from bob.bio.face.transformers import InceptionResnetv1_MsCeleb
+
+    np.random.seed(10)
+    transformer = InceptionResnetv1_MsCeleb()
+    data = np.random.rand(3, 160, 160).astype("uint8")
+    output = transformer.transform(data)
+    assert output.size == 128, output.shape
+
+
+def test_idiap_inceptionv1_casia():
+    from bob.bio.face.transformers import InceptionResnetv1_CasiaWebFace
+
+    np.random.seed(10)
+    transformer = InceptionResnetv1_CasiaWebFace()
+    data = np.random.rand(3, 160, 160).astype("uint8")
+    output = transformer.transform(data)
+    assert output.size == 128, output.shape

bob/bio/face/transformers/__init__.py

+from .facenet_sanderberg import FaceNetSanderberg
+from .idiap_inception_resnet import (
+    InceptionResnetv2_MsCeleb,
+    InceptionResnetv2_CasiaWebFace,
+    InceptionResnetv1_MsCeleb,
+    InceptionResnetv1_CasiaWebFace,
+)
+
+
+# gets sphinx autodoc done right - don't remove it
+def __appropriate__(*args):
+    """Says object was actually declared here, and not in the import module.
+  Fixing sphinx warnings of not being able to find classes, when path is shortened.
+  Parameters:
+
+    *args: An iterable of objects to modify
+
+  Resolves `Sphinx referencing issues
+  <https://github.com/sphinx-doc/sphinx/issues/3048>`
+  """
+
+    for obj in args:
+        obj.__module__ = __name__
+
+
+__appropriate__(
+    FaceNetSanderberg,
+    InceptionResnetv2_MsCeleb,
+    InceptionResnetv2_CasiaWebFace,
+    InceptionResnetv1_MsCeleb,
+    InceptionResnetv1_CasiaWebFace,
+)
+__all__ = [_ for _ in dir() if not _.startswith("_")]

bob/bio/face/transformers/facenet_sanderberg.py

+#!/usr/bin/env python
+# vim: set fileencoding=utf-8 :
+
+"""
+Wrapper for the free FaceNet variant:
+    https://github.com/davidsandberg/facenet
+
+
+    Model 20170512-110547    
+"""
+
+from __future__ import division
+
+from sklearn.base import TransformerMixin, BaseEstimator
+
+import os
+import re
+import logging
+import numpy
+import tensorflow as tf
+from bob.ip.color import gray_to_rgb
+from bob.io.image import to_matplotlib
+from bob.extension import rc
+import bob.extension.download
+import bob.io.base
+
+logger = logging.getLogger(__name__)
+
+FACENET_MODELPATH_KEY = "bob.bio.face.facenet_sanderberg_modelpath"
+
+
+def prewhiten(img):
+    mean = numpy.mean(img)
+    std = numpy.std(img)
+    std_adj = numpy.maximum(std, 1.0 / numpy.sqrt(img.size))
+    y = numpy.multiply(numpy.subtract(img, mean), 1 / std_adj)
+    return y
+
+
+def get_model_filenames(model_dir):
+    # code from https://github.com/davidsandberg/facenet
+    files = os.listdir(model_dir)
+    meta_files = [s for s in files if s.endswith(".meta")]
+    if len(meta_files) == 0:
+        raise ValueError("No meta file found in the model directory (%s)" % model_dir)
+    elif len(meta_files) > 1:
+        raise ValueError(
+            "There should not be more than one meta file in the model "
+            "directory (%s)" % model_dir
+        )
+    meta_file = meta_files[0]
+    max_step = -1
+    for f in files:
+        step_str = re.match(r"(^model-[\w\- ]+.ckpt-(\d+))", f)
+        if step_str is not None and len(step_str.groups()) >= 2:
+            step = int(step_str.groups()[1])
+            if step > max_step:
+                max_step = step
+                ckpt_file = step_str.groups()[0]
+    return meta_file, ckpt_file
+
+
+class FaceNetSanderberg(TransformerMixin, BaseEstimator):
+    """Wrapper for the free FaceNet variant:
+    https://github.com/davidsandberg/facenet
+
+    And for a preprocessor you can use::
+
+        from bob.bio.face.preprocessor import FaceCrop
+        # This is the size of the image that this model expects
+        CROPPED_IMAGE_HEIGHT = 160
+        CROPPED_IMAGE_WIDTH = 160
+        # eye positions for frontal images
+        RIGHT_EYE_POS = (46, 53)
+        LEFT_EYE_POS = (46, 107)
+        # Crops the face using eye annotations
+        preprocessor = FaceCrop(
+            cropped_image_size=(CROPPED_IMAGE_HEIGHT, CROPPED_IMAGE_WIDTH),
+            cropped_positions={'leye': LEFT_EYE_POS, 'reye': RIGHT_EYE_POS},
+            color_channel='rgb'
+        )
+
+    """
+
+    def __init__(
+        self,
+        model_path=rc[FACENET_MODELPATH_KEY],
+        image_size=160,
+        layer_name="embeddings:0",
+        **kwargs
+    ):
+        super(FaceNetSanderberg, self).__init__()
+        self.model_path = model_path
+        self.image_size = image_size
+        self.layer_name = layer_name
+        self.loaded = False
+        self._clean_unpicklables()
+
+    def _clean_unpicklables(self):
+        self.session = None
+        self.embeddings = None
+        self.graph = None
+        self.images_placeholder = None
+        self.phase_train_placeholder = None
+
+    def _check_feature(self, img):
+        img = numpy.ascontiguousarray(img)
+        if img.ndim == 2:
+            img = gray_to_rgb(img)
+        assert img.shape[-1] == self.image_size
+        assert img.shape[-2] == self.image_size
+        img = to_matplotlib(img)
+        img = prewhiten(img)
+        return img[None, ...]
+
+    def load_model(self):
+        tf.compat.v1.reset_default_graph()
+
+        session_conf = tf.compat.v1.ConfigProto(
+            intra_op_parallelism_threads=1, inter_op_parallelism_threads=1
+        )
+        self.graph = tf.Graph()
+        self.session = tf.compat.v1.Session(graph=self.graph, config=session_conf)
+
+        if self.model_path is None:
+            self.model_path = self.get_modelpath()
+        if not os.path.exists(self.model_path):
+            bob.io.base.create_directories_safe(FaceNetSanderberg.get_modelpath())
+            zip_file = os.path.join(FaceNetSanderberg.get_modelpath(), "20170512-110547.zip")
+            urls = [
+                # This link only works in Idiap CI to save bandwidth.
+                "http://www.idiap.ch/private/wheels/gitlab/"
+                "facenet_model2_20170512-110547.zip",
+                # this link to dropbox would work for everybody
+                "https://www.dropbox.com/s/"
+                "k7bhxe58q7d48g7/facenet_model2_20170512-110547.zip?dl=1",
+            ]
+            bob.extension.download.download_and_unzip(urls, zip_file)
+
+        # code from https://github.com/davidsandberg/facenet
+        model_exp = os.path.expanduser(self.model_path)
+
+        with self.graph.as_default():
+            if os.path.isfile(model_exp):
+                logger.info("Model filename: %s" % model_exp)
+                with tf.compat.v1.gfile.FastGFile(model_exp, "rb") as f:
+                    graph_def = tf.compat.v1.GraphDef()
+                    graph_def.ParseFromString(f.read())
+                    tf.import_graph_def(graph_def, name="")
+            else:
+                logger.info("Model directory: %s" % model_exp)
+                meta_file, ckpt_file = get_model_filenames(model_exp)
+
+                logger.info("Metagraph file: %s" % meta_file)
+                logger.info("Checkpoint file: %s" % ckpt_file)
+
+                saver = tf.compat.v1.train.import_meta_graph(
+                    os.path.join(model_exp, meta_file)
+                )
+                saver.restore(self.session, os.path.join(model_exp, ckpt_file))
+        # Get input and output tensors
+        self.images_placeholder = self.graph.get_tensor_by_name("input:0")
+        self.embeddings = self.graph.get_tensor_by_name(self.layer_name)
+        self.phase_train_placeholder = self.graph.get_tensor_by_name("phase_train:0")
+        logger.info("Successfully loaded the model.")
+        self.loaded = True
+
+    def transform(self, img):
+        images = self._check_feature(img)
+        if not self.loaded:
+            self.load_model()
+
+        feed_dict = {
+            self.images_placeholder: images,
+            self.phase_train_placeholder: False,
+        }
+        features = self.session.run(self.embeddings, feed_dict=feed_dict)
+        return features.flatten()
+
+    @staticmethod
+    def get_modelpath():
+        """
+        Get default model path.
+
+        First we try the to search this path via Global Configuration System.
+        If we can not find it, we set the path in the directory
+        `<project>/data`
+        """
+
+        # Priority to the RC path
+        model_path = rc[FACENET_MODELPATH_KEY]
+
+        if model_path is None:
+            import pkg_resources
+
+            model_path = pkg_resources.resource_filename(
+                __name__, "data/FaceNet/20170512-110547"
+            )
+
+        return model_path
+
+    def __setstate__(self, d):
+        # Handling unpicklable objects
+        self.__dict__ = d
+        self.loaded = False
+
+    def __getstate__(self):
+        # Handling unpicklable objects
+        d = self.__dict__
+        d.pop("session") if "session" in d else None
+        d.pop("embeddings") if "embeddings" in d else None
+        d.pop("graph") if "graph" in d else None
+        d.pop("images_placeholder") if "images_placeholder" in d else None
+        d.pop("phase_train_placeholder") if "phase_train_placeholder" in d else None
+        tf.compat.v1.reset_default_graph()
+        return d

bob/bio/face/transformers/idiap_inception_resnet.py

+#!/usr/bin/env python
+# vim: set fileencoding=utf-8 :
+
+import os
+from sklearn.base import TransformerMixin, BaseEstimator
+from .tensorflow_compat_v1 import TensorflowCompatV1
+
+
+class InceptionResnetv2_MsCeleb(TensorflowCompatV1):
+    """
+    Inception Restnet v2 model from https://gitlab.idiap.ch/bob/bob.learn.tensorflow/-/blob/1e40a68bfbbb3dd8813c48d50b2f23ff7a399956/bob/learn/tensorflow/network/InceptionResnetV2.py
+
+    This model was trained using the MsCeleb 1M dataset
+
+    The input shape of this model is :math:`3 \times 160 \times 160`
+    The output embedding is :math:`n \times 128`, where :math:`n` is the number of samples
+
+    """
+
+    def __init__(self):
+        from bob.learn.tensorflow.network import inception_resnet_v2_batch_norm
+
+        bob_rc_variable = "bob.bio.face.idiap_inception_resnet_v2_path"
+        urls = ["https://www.idiap.ch/software/bob/data/bob/bob.bio.face/master/inception-v2_batchnorm_rgb.tar.gz"]
+        model_subdirectory = "idiap_inception_resnet_v2_path"
+
+        
+        checkpoint_filename = self.get_modelpath(bob_rc_variable, model_subdirectory)
+        self.download_model(checkpoint_filename, urls)
+
+        input_shape = (1, 160, 160, 3)
+        architecture_fn = inception_resnet_v2_batch_norm
+
+        super().__init__(checkpoint_filename, input_shape, architecture_fn)
+
+
+class InceptionResnetv2_CasiaWebFace(TensorflowCompatV1):
+    """
+    Inception Restnet v2 model from https://gitlab.idiap.ch/bob/bob.learn.tensorflow/-/blob/1e40a68bfbbb3dd8813c48d50b2f23ff7a399956/bob/learn/tensorflow/network/InceptionResnetV2.py
+
+    This model was trained using the Casia WebFace
+
+    The input shape of this model is :math:`3 \times 160 \times 160`
+    The output embedding is :math:`n \times 128`, where :math:`n` is the number of samples
+
+    """
+
+    def __init__(self):
+        """Loads the tensorflow model
+        """
+        from bob.learn.tensorflow.network import inception_resnet_v2_batch_norm
+
+        bob_rc_variable = "bob.bio.face.idiap_inception_resnet_v2_casiawebface_path"
+        urls = ["https://www.idiap.ch/software/bob/data/bob/bob.bio.face/master/idiap_inception_resnet_v2_casiawebface_path.tar.gz"]
+        model_subdirectory = "idiap_inception_resnet_v2_casiawebface_path"
+
+
+        checkpoint_filename = self.get_modelpath(bob_rc_variable, model_subdirectory)
+        self.download_model(checkpoint_filename, urls)
+
+        input_shape = (1, 160, 160, 3)
+        architecture_fn = inception_resnet_v2_batch_norm
+
+        super().__init__(checkpoint_filename, input_shape, architecture_fn)
+
+
+class InceptionResnetv1_MsCeleb(TensorflowCompatV1):
+    """
+    Inception Restnet v1 model from https://gitlab.idiap.ch/bob/bob.learn.tensorflow/-/blob/1e40a68bfbbb3dd8813c48d50b2f23ff7a399956/bob/learn/tensorflow/network/InceptionResnetV1.py
+
+    This model was trained using the MsCeleb 1M dataset
+
+    The input shape of this model is :math:`3 \times 160 \times 160`
+    The output embedding is :math:`n \times 128`, where :math:`n` is the number of samples
+
+    """
+
+    def __init__(self):
+        from bob.learn.tensorflow.network import inception_resnet_v1_batch_norm
+
+        bob_rc_variable = "bob.bio.face.idiap_inception_resnet_v1_msceleb_path"
+        urls = ["https://www.idiap.ch/software/bob/data/bob/bob.bio.face/master/idiap_inception_resnet_v1_msceleb_path.tar.gz"]
+        model_subdirectory = "idiap_inception_resnet_v1_msceleb_path"
+
+        
+        checkpoint_filename = self.get_modelpath(bob_rc_variable, model_subdirectory)
+        self.download_model(checkpoint_filename, urls)
+
+        input_shape = (1, 160, 160, 3)
+        architecture_fn = inception_resnet_v1_batch_norm
+
+        super().__init__(checkpoint_filename, input_shape, architecture_fn)
+
+
+
+class InceptionResnetv1_CasiaWebFace(TensorflowCompatV1):
+    """
+    Inception Restnet v1 model from https://gitlab.idiap.ch/bob/bob.learn.tensorflow/-/blob/1e40a68bfbbb3dd8813c48d50b2f23ff7a399956/bob/learn/tensorflow/network/InceptionResnetV1.py
+
+    This model was trained using the Casia WebFace
+
+    The input shape of this model is :math:`3 \times 160 \times 160`
+    The output embedding is :math:`n \times 128`, where :math:`n` is the number of samples
+
+    """
+
+    def __init__(self):
+        """Loads the tensorflow model
+        """
+        from bob.learn.tensorflow.network import inception_resnet_v1_batch_norm
+
+        bob_rc_variable = "bob.bio.face.idiap_inception_resnet_v1_casiawebface_path"
+        urls = ["https://www.idiap.ch/software/bob/data/bob/bob.bio.face/master/idiap_inception_resnet_v1_casiawebface_path.tar.gz"]
+        model_subdirectory = "idiap_inception_resnet_v1_casiawebface_path"
+
+
+        checkpoint_filename = self.get_modelpath(bob_rc_variable, model_subdirectory)
+        self.download_model(checkpoint_filename, urls)
+
+        input_shape = (1, 160, 160, 3)
+        architecture_fn = inception_resnet_v1_batch_norm
+
+        super().__init__(checkpoint_filename, input_shape, architecture_fn)

bob/bio/face/transformers/tensorflow_compat_v1.py

+#!/usr/bin/env python
+# vim: set fileencoding=utf-8 :
+
+import tensorflow as tf
+import os
+from tensorflow.python import debug as tf_debug
+import pkg_resources
+import bob.extension.download
+from bob.extension import rc
+from sklearn.base import TransformerMixin, BaseEstimator
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class TensorflowCompatV1(TransformerMixin, BaseEstimator):
+    """
+    Tensorflow v1 compatible set of transformers.
+
+    Parameters
+    ----------
+    checkpoint_filename: str
+        Path of your checkpoint. If the .meta file is providede the last checkpoint will be loaded.
+
+    input_shape: tuple
+        input_shape: Input shape for the tensorflow neural network
+
+    architecture_fn :
+        A tf.Tensor containing the operations to be executed
+    """
+
+    def __init__(self, checkpoint_filename, input_shape, architecture_fn):
+        """Loads the tensorflow model
+        """
+
+        self.checkpoint_filename = checkpoint_filename
+        self.input_shape = input_shape
+        self.architecture_fn = architecture_fn
+        self.loaded = False
+
+    def transform(self, data):
+        """
+        Forward the data with the loaded neural network
+
+        Parameters
+        ----------
+        image : numpy.ndarray
+            Input Data
+
+        Returns
+        -------
+        numpy.ndarray
+            The features.
+
+        """
+        if not self.loaded:
+            self.load_model()
+
+        return self.session.run(self.embedding, feed_dict={self.input_tensor: data.reshape(self.input_shape)})
+
+    def load_model(self):
+        logger.info(f"Loading model `{self.checkpoint_filename}`")
+
+        tf.compat.v1.reset_default_graph()
+
+        self.input_tensor = tf.placeholder(tf.float32, shape=self.input_shape)
+
+        # Taking the embedding
+        prelogits = self.architecture_fn(
+            tf.stack(
+                [
+                    tf.image.per_image_standardization(i)
+                    for i in tf.unstack(self.input_tensor)
+                ]
+            ),
+            mode=tf.estimator.ModeKeys.PREDICT,
+        )[0]
+        self.embedding = tf.nn.l2_normalize(prelogits, dim=1, name="embedding")
+
+        # Initializing the variables of the current architecture_fn
+        self.session = tf.compat.v1.Session()
+        self.session.run(tf.compat.v1.global_variables_initializer())
+
+        # Loading the last checkpoint and overwriting the current variables
+        saver = tf.compat.v1.train.Saver()        
+        if os.path.splitext(self.checkpoint_filename)[1] == ".meta":
+            saver.restore(
+                self.session,
+                tf.train.latest_checkpoint(os.path.dirname(self.checkpoint_filename)),
+            )
+        elif os.path.isdir(self.checkpoint_filename):
+            saver.restore(self.session, tf.train.latest_checkpoint(self.checkpoint_filename))
+        else:
+            saver.restore(self.session, self.checkpoint_filename)
+
+        self.loaded = True
+
+    def __setstate__(self, d):
+        # Handling unpicklable objects
+        self.__dict__ = d
+        self.loaded = False
+
+    def __getstate__(self):
+        # Handling unpicklable objects
+        d = self.__dict__
+        d.pop("session", None)
+        d.pop("input_tensor", None)
+        d.pop("embedding", None)
+        tf.compat.v1.reset_default_graph()
+        return d
+
+    #def __del__(self):
+    #    tf.compat.v1.reset_default_graph()
+
+    def get_modelpath(self, bob_rc_variable, model_subdirectory):
+        """
+        Get default model path.
+
+        First we try the to search this path via Global Configuration System.
+        If we can not find it, we set the path in the directory
+        `<project>/data`
+        """
+
+        # Priority to the RC path
+        model_path = rc[bob_rc_variable]
+
+        if model_path is None:
+
+            model_path = pkg_resources.resource_filename(
+                __name__, os.path.join("data", model_subdirectory)
+            )
+
+        return model_path
+
+    def download_model(self, model_path, urls, zip_file="model.tar.gz"):
+        """
+        Download and unzip a model from some URL.
+
+        Parameters
+        ----------
+
+        model_path: str
+            Path where the model is supposed to be stored
+
+        urls: list
+            List of paths where the model is stored
+
+        zip_file: str
+            File name after the download
+
+        """
+
+        if not os.path.exists(model_path):
+            bob.io.base.create_directories_safe(model_path)
+            zip_file = os.path.join(model_path, zip_file)
+            bob.extension.download.download_and_unzip(urls, zip_file)