<jupyter_start><jupyter_text>Create the Inception Model and Load the pre-trained Weights<jupyter_code>import warnings

# Suppress warnings
warnings.filterwarnings('ignore')

from model import create_model

InceptionModel = create_model()
InceptionModel.load_weights('data/weights/nn4.small2.v1.h5')<jupyter_output>Using TensorFlow backend.<jupyter_text>Define a Triplet Loss Layer and Create Model<jupyter_code>from keras import backend as K
from keras.models import Model
from keras.layers import Input, Layer

# Input for anchor, positive and negative images
in_a = Input(shape=(96, 96, 3))
in_p = Input(shape=(96, 96, 3))
in_n = Input(shape=(96, 96, 3))

# Output for anchor, positive and negative embedding vectors
# The InceptionModel model instance is shared (Siamese network)
emb_a = InceptionModel(in_a)
emb_p = InceptionModel(in_p)
emb_n = InceptionModel(in_n)

loss = 'lossful'

# Lossless Triplet Loss Layer
# (https://towardsdatascience.com/lossless-triplet-loss-7e932f990b24)
class LosslessTripletLossLayer(Layer):

    def __init__(self, N=3, beta=3, epsilon=1e-8, **kwargs):
        self.N = N
        self.beta = beta
        self.epsilon = epsilon
        super(LosslessTripletLossLayer, self).__init__(**kwargs)
    
    def lossless_triplet_loss(self, inputs):
        a, p, n = inputs
        p_dist = K.sum(K.square(a - p), axis=-1)
        n_dist = K.sum(K.square(a - n), axis=-1)
        pos_dist = - K.log(-(p_dist / self.beta) + 1 + self.epsilon)
        neg_dist = - K.log(- ((self.N - n_dist) / self.beta) + 1 + self.epsilon)
        return neg_dist + pos_dist
    
    def call(self, inputs):
        loss = self.lossless_triplet_loss(inputs)
        self.add_loss(loss)
        return loss

# Triplet Loss Layer
class TripletLossLayer(Layer):
    
    def __init__(self, alpha=0.2, **kwargs):
        self.alpha = alpha
        super(TripletLossLayer, self).__init__(**kwargs)
    
    def triplet_loss(self, inputs):
        a, p, n = inputs
        p_dist = K.sum(K.square(a - p), axis=-1)
        n_dist = K.sum(K.square(a - n), axis=-1)
        return K.sum(K.maximum(p_dist - n_dist + self.alpha, 0), axis=0)
    
    def call(self, inputs):
        loss = self.triplet_loss(inputs)
        self.add_loss(loss)
        return loss<jupyter_output><empty_output><jupyter_text>Load the Data Generator for triplet pairs and Fit the Model<jupyter_code>from data import triplet_generator

if loss == 'lossless':
    # Layer that computes the lossless triplet loss from anchor, positive and negative embedding vectors
    triplet_loss_layer = LosslessTripletLossLayer(name='lossless_triplet_loss_layer')([emb_a, emb_p, emb_n])
else:
    # Layer that computes the triplet loss from anchor, positive and negative embedding vectors
    triplet_loss_layer = TripletLossLayer(alpha=0.2, name='triplet_loss_layer')([emb_a, emb_p, emb_n])

# Model that can be trained with anchor, positive negative images
SiameseTripletNet = Model([in_a, in_p, in_n], triplet_loss_layer)
SiameseTripletNet.compile(optimizer='adam', loss=None, metrics=['accuracy'])

batch_size = 4

# triplet_generator() creates a generator that continuously returns 
# ([a_batch, p_batch, n_batch], None) tuples where a_batch, p_batch 
# and n_batch are batches of anchor, positive and negative RGB images 
# each having a shape of (batch_size, 96, 96, 3).
generator = triplet_generator(batch_size)

SiameseTripletNet.fit_generator(generator, epochs=10, steps_per_epoch=100)

# Please note that the current implementation of the generator only generates 
# random image data. The main goal of this code snippet is to demonstrate 
# the general setup for model training. In the following, we will anyway 
# use a pre-trained model so we don't need a generator here that operates 
# on real training data. I'll maybe provide a fully functional generator
# later.<jupyter_output><empty_output><jupyter_text>Create Metadata for image files<jupyter_code>import numpy as np
import os
import cv2
from tqdm import tqdm
from skimage.io import imread
from skimage.transform import resize

def load_image(path):
    img = cv2.imread(path, 1)
    # OpenCV loads images with color channels
    # in BGR order. So we need to reverse them
    return img[...,::-1]

class Image():
    def __init__(self, base, name, file):
        # dataset base directory
        self.base = base
        # identity name
        self.name = name
        # image file name
        self.file = file

    def __repr__(self):
        return self.image_path()
    
    def show_name(self):
        return self.name.replace('_', ' ')

    def image_path(self):
        return os.path.join(self.base, self.name, self.file) 
    
def load_metadata(path):
    metadata = []
    classes = os.listdir(path)
    images = [(image, class_folder) for class_folder in classes for image in os.listdir(os.path.join(path, class_folder))]
    for image, class_folder in tqdm(images, desc="Images", total=len(images), unit="files"):
        metadata.append(Image(path, class_folder, image))
    return np.array(metadata), classes

train_metadata, classes = load_metadata('data/images/train')
test_metadata, _ = load_metadata('data/images/test')

print('Loaded {} training images of {} classes'
      .format(len(train_metadata), len(classes)))
print('Loaded {} testing images of {} classes'
      .format(len(test_metadata), len(classes)))<jupyter_output>Images: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 209/209 [00:00<00:00, 209464.64files/s]
Images: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 33/33 [00:00<?, ?files/s]<jupyter_text>Test the Face Alignment feature<jupyter_code>import matplotlib.pyplot as plt
import matplotlib.patches as patches

# Download DLib 19.8.1 for windows from: https://github.com/charlielito/install-dlib-python-windows
from align import AlignDlib

%matplotlib inline

# Initialize the OpenFace face alignment utility
alignment = AlignDlib('data/model/landmarks.dat')

def align_image(img):
    return alignment.align(96, img, alignment.getLargestFaceBoundingBox(img), 
                           landmarkIndices=AlignDlib.OUTER_EYES_AND_NOSE)

def show_aligned_pair(index, metadata=train_metadata):
    
    print(metadata[index].show_name())
    
    # Load original image
    jc_orig = load_image(metadata[index].image_path())
    
    # Show original image
    plt.subplot(131)
    plt.imshow(jc_orig)
    
    # Detect face and return bounding box
    bb = alignment.getLargestFaceBoundingBox(jc_orig)
    
    # Show original image with bounding box
    plt.subplot(132)
    plt.imshow(jc_orig)
    plt.gca().add_patch(patches.Rectangle((bb.left(), bb.top()), bb.width(), bb.height(), fill=False, color='red'))

    # Show aligned image
    jc_aligned = align_image(jc_orig)
    plt.subplot(133)
    plt.imshow(jc_aligned);
    
# Load an image of Madonna
show_aligned_pair(142)
def image_to_embedding(image):
    img = align_image(image)
    # check for NoneType returned error
    if img is None:
        return None
    # scale RGB values to interval [0,1]
    img = (img / 255.).astype(np.float32)
    # obtain embedding vector for image
    embedding = InceptionModel.predict(np.expand_dims(img, axis=0))[0]
    return embedding

def get_embeddings(metadata=train_metadata):
    
    embedded = np.zeros((metadata.shape[0], 128))
    errors = []
    
    pbar = tqdm(enumerate(metadata), desc="Embeddings", total=len(metadata))
    
    for i, m in pbar:
        img = load_image(m.image_path())
        emb = image_to_embedding(img)
        if emb is None:
            errors.append(m)
        embedded[i] = emb
        
    pbar.close()

    if len(errors) is not 0:
        print('Errors detected in the following files: ')
        print(errors)
        
    return embedded, errors
# embedded_train, _ = get_embeddings(train_metadata)
# embedded_test, _ = get_embeddings(test_metadata)

def distance(emb1, emb2):
    return np.sum(np.square(emb1 - emb2))

def show_pair(idx1, idx2, embedded=embedded_train, metadata=train_metadata):
    plt.figure(figsize=(8,3))
    plt.suptitle('Distance = {:.4f}'.format(distance(embedded[idx1], embedded[idx2])))
    plt.subplot(121)
    plt.imshow(load_image(metadata[idx1].image_path()))
    plt.subplot(122)
    plt.imshow(load_image(metadata[idx2].image_path()));    

show_pair(10, 3)
show_pair(21, 22)<jupyter_output><empty_output><jupyter_text>Calculate Optimal Threshold Value based on F1 Scores<jupyter_code>from sklearn.metrics import f1_score, accuracy_score

def calculate_optimal_threshold(embedded=embedded_train, metadata=train_metadata):

    distances = [] # squared L2 distance between pairs
    identical = [] # 1 if same identity, 0 otherwise

    num = len(metadata) # number of samples
    
    pbar = tqdm(total=(num - 1)**2, desc="Computing", unit="samples")

    # Iterate over all embeddings to 
    # calculate distances between them
    for i in range(num - 1):
        for j in range(1, num):
            # Save the distance between embeddings
            distances.append(distance(embedded[i], embedded[j]))
            # 1 if same person, 0 otherwise
            identical.append(1 if metadata[i].name == metadata[j].name else 0)
            pbar.update(1)

    distances = np.array(distances)
    identical = np.array(identical)

    thresholds = np.arange(0.2, 1.0, 0.01)

    f1_scores = [f1_score(identical, distances < t) for t in thresholds]
    acc_scores = [accuracy_score(identical, distances < t) for t in thresholds]

    opt_idx = np.argmax(f1_scores)
    # Threshold at maximal F1 score
    opt_tau = thresholds[opt_idx]
    # Accuracy at maximal F1 score
    opt_acc = accuracy_score(identical, distances < opt_tau)

    # Plot F1 score and accuracy as function of distance threshold
    plt.plot(thresholds, f1_scores, label='F1 score');
    plt.plot(thresholds, acc_scores, label='Accuracy');
    plt.axvline(x=opt_tau, linestyle='--', lw=1, c='lightgrey', label='Threshold')
    plt.title('Accuracy at threshold {:.2f} = {:.2f} %'.format(opt_tau, opt_acc * 100));
    plt.xlabel('Distance threshold')
    plt.legend();
    
    return opt_tau, opt_acc, distances, identical
    
# set the value of optimal threshold
optimal_threshold, optimal_accuracy, distances, identical = calculate_optimal_threshold(embedded_test, test_metadata)<jupyter_output>Computing: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 1024/1024 [00:00<00:00, 15316.41samples/s]<jupyter_text>Plot Distance Distributions of Positive and Negative Pairs<jupyter_code>def plot_distance_distros(distances, identical, threshold):
    dist_pos = distances[identical == 1]
    dist_neg = distances[identical == 0]

    plt.figure(figsize=(12,4))

    plt.subplot(121)
    plt.hist(dist_pos)
    plt.axvline(x=threshold, linestyle='--', lw=1, c='lightgrey', label='Threshold')
    plt.title('Distances (pos. pairs)')
    plt.legend();

    plt.subplot(122)
    plt.hist(dist_neg)
    plt.axvline(x=threshold, linestyle='--', lw=1, c='lightgrey', label='Threshold')
    plt.title('Distances (neg. pairs)')
    plt.legend();

plot_distance_distros(distances, identical, optimal_threshold)
from sklearn.preprocessing import LabelEncoder
from sklearn.neighbors import KNeighborsClassifier
from sklearn.svm import LinearSVC

targets_train = np.array([m.name for m in train_metadata])
targets_test = np.array([m.name for m in test_metadata])

encoder = LabelEncoder()
encoder.fit(classes)

# Numerical encoding of identities/labels/classes
y_train = encoder.transform(targets_train)
y_test = encoder.transform(targets_test)

train_idx = np.arange(train_metadata.shape[0])
test_idx = np.arange(test_metadata.shape[0])

# 182 train examples of 15 identities (~12 examples each)
X_train = embedded_train
# 30 test examples of 15 identities (2 examples each)
X_test = embedded_test

knn = KNeighborsClassifier(n_neighbors=1, metric='euclidean')
svc = LinearSVC()

knn.fit(X_train, y_train)
svc.fit(X_train, y_train)

acc_knn = accuracy_score(y_test, knn.predict(X_test))
acc_svc = accuracy_score(y_test, svc.predict(X_test))

print('KNN accuracy = {:.2f}, SVM accuracy = {:.2f}'.format(acc_knn, acc_svc))
def create_database(embeddings=X_train, labels=y_train, encoder=encoder):

    labels = encoder.classes_
    
    class_to_idx = {}
    idx_to_class = {}
    database = {}
    
    for i, label in enumerate(labels):
        class_to_idx[label] = i
        idx_to_class[i] = label
        
    for label in labels:
        database[label] = []
    
    for i, emb in enumerate(X_train):
        database[idx_to_class[y_train[i]]].append(emb)
        
    return database, class_to_idx, idx_to_class
    
database, class_to_idx, idx_to_class = create_database()
from sklearn.manifold import TSNE

X_embedded = TSNE(n_components=2).fit_transform(embedded_train)

for i, t in enumerate(set(targets_train)):
    idx = targets_train == t
    plt.scatter(X_embedded[idx, 0], X_embedded[idx, 1], label=t)   

plt.legend(bbox_to_anchor=(1, 1));
import warnings
# Suppress warnings
warnings.filterwarnings('ignore')

import time

DEBUG = True

def debug(message=''):
    debug_msg = 'DEBUG:'
    if DEBUG:
        print('{} {}'.format(debug_msg, message))

confidence_threshold = 0.4
avg_dist_threshold = optimal_threshold

def who_is_it(image, database=database, clf=knn, encoder=encoder):
    
    start = time.process_time()

    embedding = image_to_embedding(image)
    if embedding is None:
        return None, None, None
    prediction = clf.predict([embedding])
    identity = idx_to_class[prediction[0]]
    
    verified = False
    dist = 0
    num = 0
    
    embeddings_to_verify = database[identity]
    total = len(embeddings_to_verify)
    
    for m in embeddings_to_verify:
        m_dist = distance(m, embedding)
        if m_dist < optimal_threshold:
            verified = True
            num += 1
            dist += m_dist
            
    debug('Time: {:.4f} secs'.format(time.process_time() - start))
            
    if verified:
        confidence = num / total
        avg_dist = dist / num

        debug('Avg dist: {:.4f} Confidence: {:.2f}% with {}'.format(avg_dist, confidence * 100, identity.replace('_', ' ')))

        if confidence > confidence_threshold and avg_dist < avg_dist_threshold:
            debug('Recognized as {}\n'.format(identity.replace('_', ' ')))
            return identity, confidence, avg_dist
    return None, None, None

val_image = load_image(test_metadata[28].image_path())
identity, confidence, avg_dist = who_is_it(val_image)

plt.imshow(val_image)
if identity is not None:
    print('Avg Distance: {:.4f} Confidence: {:.2f}%'.format(avg_dist, confidence * 100))
    plt.title('Recognized as {}'.format(identity.replace('_', ' ')))
else:
    plt.title('Not recognized')
num = 0
total = len(test_metadata)
for i in range(total):
    val_image = load_image(test_metadata[i].image_path())
    identity, confidence, avg_dist = who_is_it(val_image)
    if identity is not None:
        if class_to_idx[identity] == y_test[i]:
            num += 1
    
print('Accuracy: {:.2f}'.format(num / total))
import time
from multiprocessing.dummy import Pool
# import win32com.client as wincl

PADDING = 20
ready_to_detect_identity = True
# windows10_voice_interface = wincl.Dispatch("SAPI.SpVoice")

def webcam_face_recognizer():
    """
    Runs a loop that extracts images from the computer's webcam and determines whether or not
    it contains the face of a person in our database.
    If it contains a face, an audio message will be played welcoming the user.
    If not, the program will process the next frame from the webcam
    """
    global ready_to_detect_identity
    
    print("Press ESC to exit\n")

    cv2.namedWindow("preview")
    vc = cv2.VideoCapture(0)

    face_cascade = cv2.CascadeClassifier('data/model/haarcascade_frontalface_default.xml')
    
    while vc.isOpened():
        _, frame = vc.read()
        image = frame

        # We do not want to detect a new identity while the program 
        # is in the process of identifying another person
        if ready_to_detect_identity:
            image = process_frame(image, frame, face_cascade)   
        
        key = cv2.waitKey(10)
        cv2.imshow("preview", image)

        if key == 27: # exit on ESC
            break
    cv2.destroyWindow("preview")

def process_frame(image, frame, face_cascade):
    """
    Determine whether the current frame contains the faces of people from our database
    """
    global ready_to_detect_identity
    
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    faces = face_cascade.detectMultiScale(gray, 1.3, 5)

    # Loop through all the faces detected and determine whether or not they are in the database
    results = []
    
    for (x, y, w, h) in faces:
        x1 = x - PADDING
        y1 = y - PADDING
        x2 = x + w + PADDING
        y2 = y + h + PADDING

        identity, confidence, avg_dist = find_identity(frame, x1, y1, x2, y2)
        
        if identity is not None:
            cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 0, 255), 2)
            cv2.putText(image, '{} - {:.3f}'.format(identity, avg_dist), 
                        (x1, y1 - 10), cv2.FONT_HERSHEY_PLAIN, 1, (0, 0, 255), 2)
            
    return image

def find_identity(frame, x1, y1, x2, y2):
    """
    Determine whether the face contained within the bounding box exists in our database
    x1,y1_____________
    |                 |
    |                 |
    |_________________x2,y2
    """
    height, width, channels = frame.shape
    
    # The padding is necessary since the OpenCV face detector creates the bounding box around the face and not the head
    part_image = frame[max(0, y1):min(height, y2), max(0, x1):min(width, x2)]
    
    identity, confidence, avg_dist = who_is_it(part_image)
    if identity is None:
        return None, None, None
    else:
        return identity.replace('_', ' '), confidence, avg_dist

#     windows10_voice_interface.Speak(welcome_message)
    
webcam_face_recognizer()<jupyter_output>Press ESC to exit

DEBUG: Time: 0.1250 secs
DEBUG: Avg dist: 0.3895 Confidence: 6.67% with Prabhsimran Singh
DEBUG: Time: 0.0625 secs
DEBUG: Avg dist: 0.5133 Confidence: 20.00% with <NAME>
DEBUG: Time: 0.0312 secs
DEBUG: Avg dist: 0.5253 Confidence: 13.33% with <NAME>
DEBUG: Time: 0.0312 secs
DEBUG: Avg dist: 0.5254 Confidence: 6.67% with <NAME>
DEBUG: Time: 0.0469 secs
DEBUG: Avg dist: 0.5042 Confidence: 13.33% with Prabhsimran Singh
DEBUG: Time: 0.0312 secs
DEBUG: Avg dist: 0.4223 Confidence: 13.33% with Prabhsimran Singh
DEBUG: Time: 0.0469 secs
DEBUG: Avg dist: 0.5059 Confidence: 66.67% with Prabhsimran Singh
DEBUG: Recognized as <NAME>

DEBUG: Time: 0.0469 secs
DEBUG: Avg dist: 0.4954 Confidence: 73.33% with Prabhsimran Singh
DEBUG: Recognized as <NAME>

DEBUG: Time: 0.0312 secs
DEBUG: Avg dist: 0.5076 Confidence: 73.33% with <NAME>
DEBUG: Recognized as <NAME>

DEBUG: Time: 0.0312 secs
DEBUG: Avg dist: 0.5013 Confidence: 73.3[...]
{'fraction_non_alphanumeric': 0.12554653341661462, 'fraction_numerical': 0.024984384759525295, 'mean_word_length': 3.6251805488685602, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 33, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '364999', 'license': 'MIT', 'n_tokens_mistral': 6919, 'n_tokens_neox': 6305, 'n_words': 1639, 'path': 'face_recog.ipynb', 'repo_name': 'pskrunner14/inception-face-recognition', 'size': '19268'}