initial trial

app.py

@@ -0,0 +1,299 @@
+import streamlit as st
+import streamlit_toggle as tog
+import time
+import numpy as np
+import cv2
+from tools.annotation import draw_mesh, draw_landmarks, draw_bounding_box, draw_text
+from tools.alignment import align_faces
+from tools.identification import load_identification_model, inference, identify
+from tools.utils import show_images, show_faces, rgb
+from tools.detection import load_detection_model, detect_faces
+from tools.webcam import init_webcam
+import logging
+
+
+# Set logging level to error (To avoid getting spammed by queue warnings etc.)
+logging.basicConfig(level=logging.ERROR)
+
+
+# Set page layout for streamlit to wide
+st.set_page_config(layout="wide")
+
+
+# Initialize the Face Detection and Identification Models
+detection_model = load_detection_model(max_faces=2, detection_confidence=0.5, tracking_confidence=0.9)
+identification_model = load_identification_model(name="MobileNet")
+
+
+# Gallery Processing
+@st.cache_data
+def gallery_processing(gallery_files):
+    """Process the gallery images (Complete Face Recognition Pipeline)
+
+    Args:
+        gallery_files (_type_): Files uploaded by the user
+
+    Returns:
+        _type_: Gallery Images, Gallery Embeddings, Gallery Names
+    """
+    gallery_images, gallery_embs, gallery_names = [], [], []
+    if gallery_files is not None:
+        for file in gallery_files:
+            file_bytes = np.asarray(bytearray(file.read()), dtype=np.uint8)
+            img = cv2.cvtColor(
+                cv2.imdecode(file_bytes, cv2.IMREAD_COLOR), cv2.COLOR_BGR2RGB
+            )
+            gallery_names.append(
+                file.name.split(".jpg")[0].split(".png")[0].split(".jpeg")[0]
+            )
+            detections = detect_faces(img, detection_model)
+            aligned_faces = align_faces(img, np.asarray([detections[0]]))
+            gallery_images.append(aligned_faces[0])
+            gallery_embs.append(inference(aligned_faces, identification_model)[0])
+    return gallery_images, gallery_embs, gallery_names
+
+
+class SideBar:
+    """A class to handle the sidebar
+    """
+    def __init__(self):
+        with st.sidebar:
+            st.markdown("# Preferences")
+            self.on_face_recognition = tog.st_toggle_switch(
+                "Face Recognition", key="activate_face_rec", default_value=True, active_color=rgb(255, 75, 75), track_color=rgb(50, 50, 50) 
+            )
+
+            st.markdown("---")
+
+            st.markdown("## Webcam")
+            self.resolution = st.selectbox(
+                "Webcam Resolution",
+                [(1920, 1080), (1280, 720), (640, 360)],
+                index=2,
+            )
+            st.markdown("To change webcam resolution: Please refresh page and select resolution before starting webcam stream.")
+
+            st.markdown("---")
+            st.markdown("## Face Detection")
+            self.max_faces = st.number_input(
+                "Maximum Number of Faces", value=2, min_value=1
+            )
+            self.detection_confidence = st.slider(
+                "Min Detection Confidence", min_value=0.0, max_value=1.0, value=0.5
+            )
+            self.tracking_confidence = st.slider(
+                "Min Tracking Confidence", min_value=0.0, max_value=1.0, value=0.9
+            )
+            switch1, switch2 = st.columns(2)
+            with switch1:
+                self.on_bounding_box = tog.st_toggle_switch(
+                    "Show Bounding Box", key="show_bounding_box", default_value=True, active_color=rgb(255, 75, 75), track_color=rgb(50, 50, 50) 
+                )
+            with switch2:
+                self.on_five_landmarks = tog.st_toggle_switch(
+                    "Show Five Landmarks", key="show_five_landmarks", default_value=True, active_color=rgb(255, 75, 75),
+                    track_color=rgb(50, 50, 50) 
+                )
+            switch3, switch4 = st.columns(2)
+            with switch3:
+                self.on_mesh = tog.st_toggle_switch(
+                    "Show Mesh", key="show_mesh", default_value=True, active_color=rgb(255, 75, 75),
+                    track_color=rgb(50, 50, 50) 
+                )
+            with switch4:
+                self.on_text = tog.st_toggle_switch(
+                    "Show Text", key="show_text", default_value=True, active_color=rgb(255, 75, 75),
+                    track_color=rgb(50, 50, 50) 
+                )
+            st.markdown("---")
+
+            st.markdown("## Face Recognition")
+            self.similarity_threshold = st.slider(
+                "Similarity Threshold", min_value=0.0, max_value=2.0, value=0.67
+            )
+
+            self.on_show_faces = tog.st_toggle_switch(
+                "Show Recognized Faces", key="show_recognized_faces", default_value=True, active_color=rgb(255, 75, 75), track_color=rgb(50, 50, 50) 
+            )
+            
+            self.model_name = st.selectbox(
+                "Model",
+                ["MobileNet", "ResNet"],
+                index=0,
+            )
+            st.markdown("---")
+
+            st.markdown("## Gallery")
+            self.uploaded_files = st.file_uploader(
+                "Choose multiple images to upload", accept_multiple_files=True
+            )
+
+            self.gallery_images, self.gallery_embs, self.gallery_names= gallery_processing(self.uploaded_files)
+
+            st.markdown("**Gallery Faces**")
+            show_images(self.gallery_images, self.gallery_names, 3)
+            st.markdown("---")
+
+
+class KPI:
+    """Class for displaying KPIs in a row
+    Args:
+        keys (list): List of KPI names
+    """
+    def __init__(self, keys):
+        self.kpi_texts = []
+        row = st.columns(len(keys))
+        for kpi, key in zip(row, keys):
+            with kpi:
+                item_row = st.columns(2)
+                item_row[0].markdown(f"**{key}**:")
+                self.kpi_texts.append(item_row[1].markdown("-"))
+
+    def update_kpi(self, kpi_values):
+        for kpi_text, kpi_value in zip(self.kpi_texts, kpi_values):
+            kpi_text.write(
+                f"<h5 style='text-align: center; color: red;'>{kpi_value:.2f}</h5>"
+                if isinstance(kpi_value, float)
+                else f"<h5 style='text-align: center; color: red;'>{kpi_value}</h5>",
+                unsafe_allow_html=True,
+            )
+
+# -----------------------------------------------------------------------------------------------
+# Streamlit App
+st.title("FaceID App Demonstration")
+
+# Sidebar
+sb = SideBar()
+
+# Get Access to Webcam
+webcam = init_webcam(width=sb.resolution[0])
+
+# KPI Section
+st.markdown("**Stats**")
+kpi = KPI([
+    "**FrameRate**",
+    "**Detected Faces**",
+    "**Image Dims**",
+    "**Detection [ms]**",
+    "**Normalization [ms]**",
+    "**Inference [ms]**",
+    "**Recognition [ms]**",
+    "**Annotations [ms]**",
+    "**Show Faces [ms]**",
+])
+st.markdown("---")
+
+# Live Stream Display
+stream_display = st.empty()
+st.markdown("---")
+
+# Display Detected Faces
+st.markdown("**Detected Faces**")
+face_window = st.empty()
+st.markdown("---")
+
+
+if webcam:
+    prevTime = 0
+    while True:
+        # Init times to "-" to show something if face recognition is turned off
+        time_detection = "-"
+        time_alignment = "-"
+        time_inference = "-"
+        time_identification = "-"
+        time_annotations = "-"
+        time_show_faces = "-"
+
+        try:
+            # Get Frame from Webcam
+            frame = webcam.get_frame(timeout=1)
+
+            # Convert to OpenCV Image
+            frame = frame.to_ndarray(format="rgb24")
+        except:
+            continue
+        
+        # ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+        # FACE RECOGNITION PIPELINE
+        if sb.on_face_recognition:
+            # FACE DETECTION ---------------------------------------------------------
+            start_time = time.time()
+            detections = detect_faces(frame, detection_model)
+            time_detection = (time.time() - start_time) * 1000
+
+            # FACE ALIGNMENT ---------------------------------------------------------
+            start_time = time.time()
+            aligned_faces = align_faces(frame, detections)
+            time_alignment = (time.time() - start_time) * 1000
+
+            # INFERENCE --------------------------------------------------------------
+            start_time = time.time()
+            if len(sb.gallery_embs) > 0: 
+                faces_embs = inference(aligned_faces, identification_model)
+            else:
+                faces_embs = []
+            time_inference = (time.time() - start_time) * 1000
+
+            # FACE IDENTIFCATION -----------------------------------------------------
+            start_time = time.time()
+            if len(faces_embs) > 0 and len(sb.gallery_embs) > 0:
+                ident_names, ident_dists, ident_imgs = identify(faces_embs, sb.gallery_embs, sb.gallery_names, sb.gallery_images, thresh=sb.similarity_threshold)
+            else:
+                ident_names, ident_dists, ident_imgs = [], [], []
+            time_identification = (time.time() - start_time) * 1000
+
+            # ANNOTATIONS ------------------------------------------------------------
+            start_time = time.time()
+            frame = cv2.resize(frame, (1920, 1080)) # to make annotation in HD
+            frame.flags.writeable = True  # (hack to make annotations faster)
+            if sb.on_mesh:
+                frame = draw_mesh(frame, detections)
+            if sb.on_five_landmarks:
+                frame = draw_landmarks(frame, detections)
+            if sb.on_bounding_box:
+                frame = draw_bounding_box(frame, detections, ident_names)
+            if sb.on_text:
+                frame = draw_text(frame, detections, ident_names)
+            time_annotations = (time.time() - start_time) * 1000
+
+            # DISPLAY DETECTED FACES -------------------------------------------------
+            start_time = time.time()
+            if sb.on_show_faces:
+                show_faces(
+                aligned_faces,
+                ident_names,
+                ident_dists,
+                ident_imgs, 
+                num_cols=3,
+                channels="RGB",
+                display=face_window,
+            )
+            time_show_faces = (time.time() - start_time) * 1000
+        # ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+
+
+        # DISPLAY THE LIVE STREAM --------------------------------------------------
+        stream_display.image(
+            frame, channels="RGB", caption="Live-Stream", use_column_width=True
+        )
+
+        # CALCULATE FPS -----------------------------------------------------------
+        currTime = time.time()
+        fps = 1 / (currTime - prevTime)
+        prevTime = currTime
+
+        # UPDATE KPIS -------------------------------------------------------------
+        kpi.update_kpi(
+            [
+                fps,
+                len(detections),
+                sb.resolution,
+                time_detection,
+                time_alignment,
+                time_inference,
+                time_identification,
+                time_annotations,
+                time_show_faces,
+            ]
+        )

models/mobileNet.tflite

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6c19b789f661caa8da735566490bfd8895beffb2a1ec97a56b126f0539991aa6
+size 8210384

models/resNet.tflite

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f4d8b0194957a3ad766135505fc70a91343660151a8103bbb6c3b8ac34dbb4e2
+size 40946048

requirements.txt

@@ -0,0 +1,10 @@
+streamlit
+scikit-image
+scikit-learn
+mediapipe
+opencv-python-headless
+watchdog
+streamlit-webrtc
+matplotlib
+streamlit-toggle-switch
+tflite-runtime

tools/alignment.py

@@ -0,0 +1,39 @@
+import numpy as np
+import cv2
+from skimage.transform import SimilarityTransform
+
+
+FIVE_LANDMARKS = [470, 475, 1, 57, 287]
+
+
+def align(img, landmarks, target_size=(112, 112)):
+    dst = np.array(
+        [
+            [
+                landmarks.landmark[i].x * img.shape[1],
+                landmarks.landmark[i].y * img.shape[0],
+            ]
+            for i in FIVE_LANDMARKS
+        ],
+    )
+
+    src = np.array(
+        [
+            [38.2946, 51.6963],
+            [73.5318, 51.5014],
+            [56.0252, 71.7366],
+            [41.5493, 92.3655],
+            [70.7299, 92.2041],
+        ],
+        dtype=np.float32,
+    )
+    tform = SimilarityTransform()
+    tform.estimate(dst, src)
+    tmatrix = tform.params[0:2, :]
+    return cv2.warpAffine(img, tmatrix, target_size, borderValue=0.0)
+
+
+
+def align_faces(img, detections):
+    aligned_faces = [align(img, detection.multi_face_landmarks) for detection in detections]
+    return aligned_faces

tools/annotation.py

@@ -0,0 +1,121 @@
+import cv2
+import mediapipe as mp
+import streamlit as st
+
+
+FIVE_LANDMARKS = [470, 475, 1, 57, 287]
+FACE_CONNECTIONS = mp.solutions.face_mesh_connections.FACEMESH_TESSELATION
+
+
+
+def draw_bounding_box(img, detections, ident_names, margin=10):
+    # Draw the bounding box on the original frame
+    for detection, name in zip(detections, ident_names):
+        
+        color = (255, 0, 0) if name == "Unknown" else (0, 255, 0)
+
+        x_coords = [
+            landmark.x * img.shape[1] for landmark in detection.multi_face_landmarks.landmark
+        ]
+        y_coords = [
+            landmark.y * img.shape[0] for landmark in detection.multi_face_landmarks.landmark
+        ]
+
+        x_min, x_max = int(min(x_coords) - margin), int(max(x_coords) + margin)
+        y_min, y_max = int(min(y_coords) - margin), int(max(y_coords) + margin)
+
+        cv2.rectangle(img, (x_min, y_min), (x_max, y_max), color, 2)
+        cv2.rectangle(img, (x_min, y_min - img.shape[0] // 25), (x_max, y_min), color, -1)
+
+    return img
+
+
+def draw_text(
+    img,
+    detections,
+    ident_names,
+    margin=10,
+    font_scale=1,
+    font_color=(0, 0, 0),
+    font=cv2.FONT_HERSHEY_SIMPLEX,
+):
+    
+    font_scale = img.shape[0] / 1000
+    for detection, name in zip(detections, ident_names):
+        x_coords = [
+            landmark.x * img.shape[1] for landmark in detection.multi_face_landmarks.landmark
+        ]
+        y_coords = [
+            landmark.y * img.shape[0] for landmark in detection.multi_face_landmarks.landmark
+        ]
+
+        x_min = int(min(x_coords) - margin)
+        y_min = int(min(y_coords) - margin)
+        
+        cv2.putText(
+            img,
+            name,
+            (x_min + img.shape[0] // 400, y_min - img.shape[0] // 100),
+            font,
+            font_scale,
+            font_color,
+            2,
+        )
+
+    return img
+
+
+def draw_mesh(img, detections):
+    for detection in detections:
+        # Draw the connections
+        for connection in FACE_CONNECTIONS:
+            cv2.line(
+                img,
+                (
+                    int(detection.multi_face_landmarks.landmark[connection[0]].x * img.shape[1]),
+                    int(detection.multi_face_landmarks.landmark[connection[0]].y * img.shape[0]),
+                ),
+                (
+                    int(detection.multi_face_landmarks.landmark[connection[1]].x * img.shape[1]),
+                    int(detection.multi_face_landmarks.landmark[connection[1]].y * img.shape[0]),
+                ),
+                (255, 255, 255),
+                1,
+            )
+
+        # Draw the landmarks
+        for points in detection.multi_face_landmarks.landmark:
+            cv2.circle(
+                img,
+                (
+                    int(points.x * img.shape[1]),
+                    int(points.y * img.shape[0]),
+                ),
+                1,
+                (0, 255, 0),
+                -1,
+            )
+    return img
+
+
+def draw_landmarks(img, detections):
+    # Draw the face landmarks on the original frame
+    for points in FIVE_LANDMARKS:
+        for detection in detections:
+            cv2.circle(
+                img,
+                (
+                    int(
+                        detection.multi_face_landmarks.landmark[points].x
+                        * img.shape[1]
+                    ),
+                    int(
+                        detection.multi_face_landmarks.landmark[points].y
+                        * img.shape[0]
+                    ),
+                ),
+                5,
+                (0, 0, 255),
+                -1,
+            )
+    return img

tools/detection.py

@@ -0,0 +1,44 @@
+import mediapipe as mp
+import streamlit as st
+
+
+class Detection:
+    multi_face_bboxes = []
+    multi_face_landmarks = []
+
+
+#@st.cache_resource
+def load_detection_model(max_faces=2, detection_confidence=0.5, tracking_confidence=0.5):
+    model = mp.solutions.face_mesh.FaceMesh(
+            refine_landmarks=True,
+            min_detection_confidence=detection_confidence,
+            min_tracking_confidence=tracking_confidence,
+            max_num_faces=max_faces,
+        )
+    return model
+
+
+def detect_faces(frame, model):
+    
+    # Process the frame with MediaPipe Face Mesh
+    results = model.process(frame)
+
+    # Get the Bounding Boxes from the detected faces
+    detections = []
+    if results.multi_face_landmarks:
+        for landmarks in results.multi_face_landmarks:
+            x_coords = [
+                landmark.x * frame.shape[1] for landmark in landmarks.landmark
+            ]
+            y_coords = [
+                landmark.y * frame.shape[0] for landmark in landmarks.landmark
+            ]
+
+            x_min, x_max = int(min(x_coords)), int(max(x_coords))
+            y_min, y_max = int(min(y_coords)), int(max(y_coords))
+
+            detection = Detection()
+            detection.multi_face_bboxes=[x_min, y_min, x_max, y_max]
+            detection.multi_face_landmarks=landmarks
+            detections.append(detection)
+    return detections

tools/identification.py

@@ -0,0 +1,47 @@
+import numpy as np
+import tflite_runtime.interpreter as tflite
+from sklearn.metrics.pairwise import cosine_distances
+import streamlit as st
+import time
+
+
+MODEL_PATHS = {
+    "MobileNet": "./models/mobileNet.tflite",
+    "ResNet": "./models/resNet.tflite",
+}
+
+
+#@st.cache_resource
+def load_identification_model(name="MobileNet"):
+    model = tflite.Interpreter(model_path=MODEL_PATHS[name])
+    return model
+
+
+def inference(imgs, model):
+    if len(imgs) > 0:
+        imgs = np.asarray(imgs).astype(np.float32) / 255
+        model.resize_tensor_input(model.get_input_details()[0]["index"], imgs.shape)
+        model.allocate_tensors()
+        model.set_tensor(model.get_input_details()[0]["index"], imgs)
+        model.invoke()
+        embs = [model.get_tensor(elem["index"]) for elem in model.get_output_details()]
+        return embs[0]
+    else:
+        return []
+
+
+def identify(embs_src, embs_gal, labels_gal, imgs_gal, thresh=None):
+    all_dists = cosine_distances(embs_src, embs_gal)
+    ident_names, ident_dists, ident_imgs = [], [], []
+    for dists in all_dists:
+        idx_min = np.argmin(dists)
+        if thresh and dists[idx_min] > thresh:
+            dist = dists[idx_min]
+            pred = None
+        else:
+            dist = dists[idx_min]
+            pred = idx_min
+        ident_names.append(labels_gal[pred] if pred is not None else "Unknown")
+        ident_dists.append(dist)
+        ident_imgs.append(imgs_gal[pred] if pred is not None else None)
+    return ident_names, ident_dists, ident_imgs

tools/utils.py

@@ -0,0 +1,66 @@
+import streamlit as st
+import cv2
+
+def rgb(r, g, b):
+    return '#{:02x}{:02x}{:02x}'.format(r, g, b)
+
+
+def show_images(images, names, num_cols, channels="RGB"):
+    num_images = len(images)
+
+    # Calculate the number of rows and columns
+    num_rows = -(
+        -num_images // num_cols
+    )  # This also handles the case when num_images is not a multiple of num_cols
+
+    for row in range(num_rows):
+        # Create the columns
+        cols = st.sidebar.columns(num_cols)
+
+        for i, col in enumerate(cols):
+            idx = row * num_cols + i
+
+            if idx < num_images:
+                img = images[idx]
+                if len(names) == 0:
+                    names = ["Unknown"] * len(images)
+                name = names[idx]
+                col.image(img, caption=name, channels=channels, width=112)
+
+
+def show_faces(images, names, distances, gal_images, num_cols, channels="RGB", display=st):
+    if len(images) == 0 or len(names) == 0:
+        display.write("No faces detected, or gallery empty!")
+        return
+    # Calculate the number of rows and columns
+    num_rows = -(
+        -len(images) // num_cols
+    )  # This also handles the case when num_images is not a multiple of num_cols
+
+    for row in range(num_rows):
+        # Create the columns
+        cols = display.columns(num_cols)
+
+        for i, col in enumerate(cols):
+            idx = row * num_cols + i
+
+            if idx < len(images):
+                img = images[idx]
+                name = names[idx]
+                dist = distances[idx]
+                col.image(img, channels=channels, width=112)
+                
+                if gal_images[idx] is not None:
+                    col.text("  ⬍ matching ⬍")
+                    col.image(gal_images[idx], caption=name, channels=channels, width=112)
+                else:
+                    col.markdown("")
+                    col.write("No match found")
+                col.markdown(
+                    f"**Distance: {dist:.4f}**" if dist else f"**Distance: -**"
+                )
+            else:
+                col.empty()
+                col.markdown("")
+                col.empty()
+                col.markdown("")

tools/webcam.py

@@ -0,0 +1,23 @@
+import streamlit as st
+from streamlit_webrtc import webrtc_streamer, WebRtcMode
+
+
+@st.cache_resource(experimental_allow_widgets=True)
+def init_webcam(width=680):
+    ctx = webrtc_streamer(
+        key="FaceIDAppDemo",
+        mode=WebRtcMode.SENDONLY,
+        media_stream_constraints={
+            "video": {
+                "width": {
+                    "min": width,
+                    "ideal": width,
+                    "max": width,
+                },
+            },
+            "audio": False,
+        },
+        video_receiver_size=1,
+        async_processing=True,
+    )
+    return ctx.video_receiver