testing minimum version

app.py

@@ -2,291 +2,414 @@ import streamlit as st
 import time
 from typing import List
 from streamlit_webrtc import webrtc_streamer, WebRtcMode
-import logging
 import av
-import queue
-from streamlit_toggle import st_toggle_switch
-import pandas as pd
-from tools.nametypes import Stats, Detection, Identity, Match
-from tools.utils import get_ice_servers, rgb, format_dflist
-from tools.face_detection import FaceDetection
-from tools.face_recognition import FaceRecognition
-from tools.annotation import Annotation
-from tools.gallery import init_gallery
-from tools.pca import pca
+import numpy as np
+import onnxruntime as rt
+import threading
+import mediapipe as mp
+import os
+from twilio.rest import Client
+import cv2
+from skimage.transform import SimilarityTransform
+from types import SimpleNamespace
+from sklearn.metrics.pairwise import cosine_distances
 
 
-# Set logging level to error (To avoid getting spammed by queue warnings etc.)
-logger = logging.getLogger(__name__)
-logging.basicConfig(level=logging.ERROR)
+class Detection(SimpleNamespace):
+    bbox: List[List[float]] = None
+    landmarks: List[List[float]] = None
 
 
+class Identity(SimpleNamespace):
+    detection: Detection = Detection()
+    name: str = None
+    embedding: np.ndarray = None
+    face: np.ndarray = None
+
+
+class Match(SimpleNamespace):
+    subject_id: Identity = Identity()
+    gallery_id: Identity = Identity()
+    distance: float = None
+    name: str = None
+
+
+class Grabber(object):
+    def __init__(self, video_receiver) -> None:
+        self.currentFrame = None
+        self.capture = video_receiver
+        self.thread = threading.Thread(target=self.update_frame)
+        self.thread.daemon = True
+
+    def update_frame(self) -> None:
+        while True:
+            self.currentFrame = self.capture.get_frame()
+
+    def get_frame(self) -> av.VideoFrame:
+        return self.currentFrame
+
+
+# Similarity threshold for face matching
+SIMILARITY_THRESHOLD = 1.2
+
+# Get twilio ice server configuration using twilio credentials from environment variables (set in streamlit secrets)
+# Ref: https://www.twilio.com/docs/stun-turn/api
+ICE_SERVERS = Client(os.environ["TWILIO_ACCOUNT_SID"], os.environ["TWILIO_AUTH_TOKEN"]).tokens.create().ice_servers
+
 # Set page layout for streamlit to wide
-st.set_page_config(layout="wide", page_title="FaceID App Demo", page_icon=":sunglasses:")
-with st.sidebar:
-    st.markdown("# Settings")
-    face_rec_on = st_toggle_switch(
-        "Live Face Recognition",
-        key="activate_face_rec",
-        default_value=True,
-        active_color=rgb(255, 75, 75),
-        track_color=rgb(50, 50, 50),
-        label_after=True,
-    )
+st.set_page_config(layout="wide", page_title="Live Face Recognition", page_icon=":sunglasses:")
 
-    with st.expander("Advanced Settings", expanded=False):
-        st.markdown("## Webcam & Stream")
-        resolution = st.selectbox(
-            "Webcam Resolution",
-            [(1920, 1080), (1280, 720), (640, 360)],
-            index=2,
-        )
-        st.markdown("Note: To change the resolution, you have to restart the stream.")
+# Streamlit app
+st.title("Live Webcam Face Recognition")
 
-        ice_server = st.selectbox("ICE Server", ["twilio", "metered"], index=1)
-        st.markdown(
-            "Note: metered is a free server with limited bandwidth, and can take a while to connect. Twilio is a paid service and is payed by me, so please don't abuse it."
-        )
-        st.markdown("---")
-        st.markdown("## Face Detection")
-        detection_min_face_size = st.slider("Min Face Size", min_value=5, max_value=120, value=40)
-        detection_scale_factor = st.slider("Scale Factor", min_value=0.1, max_value=1.0, value=0.7)
-        detection_confidence = st.slider("Min Detection Confidence", min_value=0.5, max_value=1.0, value=0.9)
-        st.markdown("---")
-        st.markdown("## Face Recognition")
-        similarity_threshold = st.slider("Similarity Threshold", min_value=0.0, max_value=2.0, value=0.67)
-        st.markdown(
-            "This sets a maximum distance for the cosine similarity between the embeddings of the detected face and the gallery images. If the distance is below the threshold, the face is recognized as the gallery image with the lowest distance. If the distance is above the threshold, the face is not recognized."
-        )
-        model_name = st.selectbox("Model", ["mobileNet", "resNet"], index=0)
-        st.markdown(
-            "Note: The mobileNet model is smaller and faster, but less accurate. The resNet50 model is bigger and slower, but more accurate."
+st.markdown("**Live Stream**")
+ctx_container = st.container()
+stream_container = st.empty()
+
+st.markdown("**Matches**")
+matches_container = st.info("No matches found yet ...")
+
+
+# Init face detector and face recognizer
+face_recognizer = rt.InferenceSession("model.fixed.onnx", providers=rt.get_available_providers())
+face_detector = mp.solutions.face_mesh.FaceMesh(
+    refine_landmarks=True,
+    min_detection_confidence=0.5,
+    min_tracking_confidence=0.5,
+    max_num_faces=5,
+)
+
+
+def detect_faces(frame: np.ndarray) -> List[Detection]:
+    # Process the frame with the face detector
+    result = face_detector.process(frame)
+
+    # Initialize an empty list to store the detected faces
+    detections = []
+
+    # Check if any faces were detected
+    if result.multi_face_landmarks:
+        # Iterate over each detected face
+        for count, detection in enumerate(result.multi_face_landmarks):
+            # Select 5 Landmarks
+            five_landmarks = np.asarray(detection.landmark)[[470, 475, 1, 57, 287]]
+
+            # Extract the x and y coordinates of the landmarks of interest
+            landmarks = [[landmark.x * frame.shape[1], landmark.y * frame.shape[0]] for landmark in five_landmarks]
+
+            # Extract the x and y coordinates of all landmarks
+            all_x_coords = [landmark.x * frame.shape[1] for landmark in detection.landmark]
+            all_y_coords = [landmark.y * frame.shape[0] for landmark in detection.landmark]
+
+            # Compute the bounding box of the face
+            x_min, x_max = int(min(all_x_coords)), int(max(all_x_coords))
+            y_min, y_max = int(min(all_y_coords)), int(max(all_y_coords))
+            bbox = [[x_min, y_min], [x_max, y_max]]
+
+            # Create a Detection object for the face
+            detection = Detection(
+                idx=count,
+                bbox=bbox,
+                landmarks=landmarks,
+                confidence=None,
+            )
+
+            # Add the detection to the list
+            detections.append(detection)
+
+    # Return the list of detections
+    return detections
+
+
+def recognize_faces(frame: np.ndarray, detections: List[Detection]) -> List[Identity]:
+    if not detections:
+        return []
+
+    identities = []
+    for detection in detections:
+        # ALIGNMENT -----------------------------------------------------------
+        # Target landmark coordinates (as used in training)
+        landmarks_target = 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(detection.landmarks, landmarks_target)
+        tmatrix = tform.params[0:2, :]
+        face_aligned = cv2.warpAffine(frame, tmatrix, (112, 112), borderValue=0.0)
+        # ---------------------------------------------------------------------
+
+        # INFERENCE -----------------------------------------------------------
+        # Inference face embeddings with onnxruntime
+        input_image = (np.asarray([face_aligned]).astype(np.float32) / 255.0).clip(0.0, 1.0)
+        embedding = face_recognizer.run(None, {"input_image": input_image})[0][0]
+        # ---------------------------------------------------------------------
+
+        # Create Identity object
+        identities.append(Identity(detection=detection, embedding=embedding, face=face_aligned))
+
+    return identities
+
+
+def match_faces(subjects: List[Identity], gallery: List[Identity]) -> List[Match]:
+    if len(gallery) == 0 or len(subjects) == 0:
+        return []
+
+    # Get Embeddings
+    embs_gal = np.asarray([identity.embedding for identity in gallery])
+    embs_det = np.asarray([identity.embedding for identity in subjects])
+
+    # Calculate Cosine Distances
+    cos_distances = cosine_distances(embs_det, embs_gal)
+
+    # Find Matches
+    matches = []
+    for ident_idx, identity in enumerate(subjects):
+        dists_to_identity = cos_distances[ident_idx]
+        idx_min = np.argmin(dists_to_identity)
+        if dists_to_identity[idx_min] < SIMILARITY_THRESHOLD:
+            matches.append(
+                Match(
+                    subject_id=identity,
+                    gallery_id=gallery[idx_min],
+                    distance=dists_to_identity[idx_min],
+                )
+            )
 
-    st.markdown("# Face Gallery")
-    files = st.sidebar.file_uploader(
-        "Upload images to gallery",
-        type=["png", "jpg", "jpeg"],
-        accept_multiple_files=True,
-        label_visibility="collapsed",
+    # Sort Matches by identity_idx
+    matches = sorted(matches, key=lambda match: match.gallery_id.name)
+
+    return matches
+
+
+def draw_annotations(frame: np.ndarray, detections: List[Detection], matches: List[Match]) -> np.ndarray:
+    global timestamp
+    shape = np.asarray(frame.shape[:2][::-1])
+
+    # Upscale frame to 1080p for better visualization of drawn annotations
+    frame = cv2.resize(frame, (1920, 1080))
+    upscale_factor = np.asarray([1920 / shape[0], 1080 / shape[1]])
+    shape = np.asarray(frame.shape[:2][::-1])
+
+    # Make frame writeable (for better performance)
+    frame.flags.writeable = True
+
+    fps = 1 / (time.time() - timestamp)
+    timestamp = time.time()
+
+    # Draw FPS
+    cv2.putText(
+        frame,
+        f"FPS: {fps:.1f}",
+        (20, 40),
+        cv2.FONT_HERSHEY_SIMPLEX,
+        1,
+        (0, 255, 0),
+        2,
     )
 
-    with st.expander("Uploaded Images", expanded=True):
-        if files:
-            st.image(files, width=112, caption=files)
-        else:
-            st.info("No images uploaded yet.")
+    # Draw Detections
+    for detection in detections:
+        # Draw Landmarks
+        for landmark in detection.landmarks:
+            cv2.circle(
+                frame,
+                (landmark * upscale_factor).astype(int),
+                2,
+                (255, 255, 255),
+                -1,
+            )
 
+        # Draw Bounding Box
+        cv2.rectangle(
+            frame,
+            (detection.bbox[0] * upscale_factor).astype(int),
+            (detection.bbox[1] * upscale_factor).astype(int),
+            (255, 0, 0),
+            2,
+        )
 
-gallery = init_gallery(
-    files,
-    min_detections_conf=detection_confidence,
-    min_similarity=similarity_threshold,
-    model_name=model_name,
-)
+        # Draw Index
+        cv2.putText(
+            frame,
+            str(detection.idx),
+            (
+                ((detection.bbox[1][0] + 2) * upscale_factor[0]).astype(int),
+                ((detection.bbox[1][1] + 2) * upscale_factor[1]).astype(int),
+            ),
+            cv2.LINE_AA,
+            0.5,
+            (0, 0, 0),
+            2,
+        )
 
-face_detector = FaceDetection(
-    min_detections_conf=detection_confidence,
-    min_face_size=detection_min_face_size,
-    scale_factor=detection_scale_factor,
-)
-face_recognizer = FaceRecognition(model_name=model_name, min_similarity=similarity_threshold)
-annotator = Annotation()
+    # Draw Matches
+    for match in matches:
+        detection = match.subject_id.detection
+        name = match.gallery_id.name
+
+        # Draw Bounding Box in green
+        cv2.rectangle(
+            frame,
+            (detection.bbox[0] * upscale_factor).astype(int),
+            (detection.bbox[1] * upscale_factor).astype(int),
+            (0, 255, 0),
+            2,
+        )
 
-transfer_queue: "queue.Queue[Stats, List[Detection], List[Identity], List[Match]]" = queue.Queue()
+        # Draw Banner
+        cv2.rectangle(
+            frame,
+            (
+                (detection.bbox[0][0] * upscale_factor[0]).astype(int),
+                (detection.bbox[0][1] * upscale_factor[1] - (shape[1] // 25)).astype(int),
+            ),
+            (
+                (detection.bbox[1][0] * upscale_factor[0]).astype(int),
+                (detection.bbox[0][1] * upscale_factor[1]).astype(int),
+            ),
+            (255, 255, 255),
+            -1,
+        )
 
+        # Draw Name
+        cv2.putText(
+            frame,
+            name,
+            (
+                ((detection.bbox[0][0] + shape[0] // 400) * upscale_factor[0]).astype(int),
+                ((detection.bbox[0][1] - shape[1] // 50) * upscale_factor[1]).astype(int),
+            ),
+            cv2.LINE_AA,
+            0.7,
+            (0, 0, 0),
+            2,
+        )
 
-def video_frame_callback(frame: av.VideoFrame) -> av.VideoFrame:
-    # Initialize detections
-    detections, identities, matches = [], [], []
+        # Draw Distance
+        cv2.putText(
+            frame,
+            f" Distance: {match.distance:.2f}",
+            (
+                ((detection.bbox[0][0] + shape[0] // 400) * upscale_factor[0]).astype(int),
+                ((detection.bbox[0][1] - shape[1] // 350) * upscale_factor[1]).astype(int),
+            ),
+            cv2.LINE_AA,
+            0.5,
+            (0, 0, 0),
+            2,
+        )
 
-    # Initialize stats
-    stats = Stats()
+    return frame
 
-    # Start timer for FPS calculation
-    frame_start = time.time()
 
+def video_frame_callback(frame: av.VideoFrame) -> av.VideoFrame:
     # Convert frame to numpy array
     frame = frame.to_ndarray(format="rgb24")
 
-    # Get frame resolution and add to stats
-    resolution = frame.shape
-    stats = stats._replace(resolution=resolution)
-
-    if face_rec_on:
-        # Run face detection
-        start = time.time()
-        frame, detections = face_detector(frame)
-        stats = stats._replace(num_faces=len(detections) if detections else 0)
-        stats = stats._replace(detection=(time.time() - start) * 1000)
+    # Run face detection
+    detections = detect_faces(frame)
 
-        # Run face recognition
-        start = time.time()
-        identities = face_recognizer(frame, detections)
-        stats = stats._replace(recognition=(time.time() - start) * 1000)
+    # Run face recognition
+    subjects = recognize_faces(frame, detections)
 
-        # Do matching
-        start = time.time()
-        matches = face_recognizer.find_matches(identities, gallery)
-        stats = stats._replace(matching=(time.time() - start) * 1000)
+    # Run face matching
+    matches = match_faces(subjects, gallery)
 
-        # Draw annotations
-        start = time.time()
-        frame = annotator(frame, detections, identities, matches, gallery)
-        stats = stats._replace(annotation=(time.time() - start) * 1000)
+    # Draw annotations
+    frame = draw_annotations(frame, detections, matches)
 
     # Convert frame back to av.VideoFrame
     frame = av.VideoFrame.from_ndarray(frame, format="rgb24")
 
-    # Calculate FPS and add to stats
-    stats = stats._replace(fps=1 / (time.time() - frame_start))
+    return frame, matches
 
-    # Send data to other thread
-    transfer_queue.put_nowait([stats, detections, identities, matches])
 
-    return frame
+# Sidebar for face gallery
+with st.sidebar:
+    st.markdown("# Face Gallery")
+    files = st.sidebar.file_uploader(
+        "Upload images to gallery",
+        type=["png", "jpg", "jpeg"],
+        accept_multiple_files=True,
+        label_visibility="collapsed",
+    )
 
+    # Init gallery
+    gallery = []
+    for file in files:
+        # Read file bytes
+        file_bytes = np.asarray(bytearray(file.read()), dtype=np.uint8)
 
-# Streamlit app
-st.title("Live Webcam Face Recognition")
+        # Decode image and convert from BGR to RGB
+        img = cv2.cvtColor(cv2.imdecode(file_bytes, cv2.IMREAD_COLOR), cv2.COLOR_BGR2RGB)
 
-st.markdown("**Stream Stats**")
-disp_stats = st.info("No streaming statistics yet, please start the stream.")
-
-ctx = webrtc_streamer(
-    key="FaceIDAppDemo",
-    mode=WebRtcMode.SENDRECV,
-    rtc_configuration={"iceServers": get_ice_servers(name=ice_server)},
-    video_frame_callback=video_frame_callback,
-    media_stream_constraints={
-        "video": {
-            "width": {
-                "min": resolution[0],
-                "ideal": resolution[0],
-                "max": resolution[0],
-            },
-            "height": {
-                "min": resolution[1],
-                "ideal": resolution[1],
-                "max": resolution[1],
-            },
-        },
-        "audio": False,
-    },
-    async_processing=True,
-)
+        # Detect faces
+        detections = detect_faces(img)
 
-tab_recognition, tab_metrics, tab_pca = st.tabs(["Recognized Identities", "Recognition Metrics", "Live PCAs"])
-
-
-with tab_recognition:
-    # Display Gallery and Recognized Identities
-    col1, col2 = st.columns(2)
-    col1.markdown("**Gallery Identities**")
-    disp_identities_gal = col1.info("No gallery images uploaded yet ...")
-    col2.markdown("**Recognized Identities**")
-    disp_identities_rec = col2.info("No recognized identities yet ...")
-
-with tab_metrics:
-    # Display Detections and Identities
-    st.markdown("**Detection Metrics**")
-    disp_detection_metrics = st.info("No detected faces yet ...")
-
-    # Display Recognition Metrics
-    st.markdown("**Recognition Metrics**")
-    disp_recognition_metrics = st.info("No recognized identities yet ...")
-
-with tab_pca:
-    # Display 2D and 3D PCA
-    col1, col2 = st.columns(2)
-    col1.markdown("**PCA 2D**")
-    disp_pca3d = col1.info("Only available if more than 1 recognized face ...")
-    col2.markdown("**PCA 3D**")
-    disp_pca2d = col2.info("Only available if more than 1 recognized face ...")
-    freeze_pcas = st.button("Freeze PCAs for Interaction", key="reset_pca")
-
-    # Show PCAs
-    if freeze_pcas and gallery:
-        col1, col2 = st.columns(2)
-        if len(st.session_state.matches) > 1:
-            col1.plotly_chart(
-                pca(
-                    st.session_state.matches,
-                    st.session_state.identities,
-                    gallery,
-                    dim=3,
-                ),
-                use_container_width=True,
-            )
-            col2.plotly_chart(
-                pca(
-                    st.session_state.matches,
-                    st.session_state.identities,
-                    gallery,
-                    dim=2,
-                ),
-                use_container_width=True,
+        if detections:
+            # recognize faces
+            subjects = recognize_faces(img, detections[:1])
+
+            # Add subjects to gallery
+            gallery.append(
+                Identity(
+                    name=os.path.splitext(file.name)[0],
+                    embedding=subjects[0].embedding,
+                    face=subjects[0].face,
+                )
             )
 
+    # Show gallery images
+    st.image(
+        image=[identity.face for identity in gallery],
+        caption=[identity.name for identity in gallery],
+    )
 
-# Show Gallery Identities
-if gallery:
-    disp_identities_gal.image(
-        image=[identity.face_aligned for identity in gallery],
-        caption=[match.name for match in gallery],
+# Start streaming component
+with ctx_container:
+    ctx = webrtc_streamer(
+        key="LiveFaceRecognition",
+        mode=WebRtcMode.SENDONLY,
+        rtc_configuration={"iceServers": ICE_SERVERS},
+        media_stream_constraints={"video": {"width": 1920}, "audio": False},
     )
-else:
-    disp_identities_gal.info("No gallery images uploaded yet ...")
 
+# Initialize frame grabber
+grabber = Grabber(ctx.video_receiver)
 
-# Display Live Stats
 if ctx.state.playing:
-    while True:
-        # Retrieve data from other thread
-        stats, detections, identities, matches = transfer_queue.get()
-
-        # Save for PCA Snapshot
-        st.session_state.identities = identities
-        st.session_state.matches = matches
-
-        # Show Stats
-        disp_stats.dataframe(
-            pd.DataFrame([stats]).applymap(lambda x: (format_dflist(x))),
-            use_container_width=True,
-        )
-
-        # Show Detections Metrics
-        if detections:
-            disp_detection_metrics.dataframe(
-                pd.DataFrame(detections).applymap(lambda x: (format_dflist(x))),
-                use_container_width=True,
-            )
-        else:
-            disp_detection_metrics.info("No detected faces yet ...")
-
-        # Show Match Metrics
-        if matches:
-            disp_recognition_metrics.dataframe(
-                pd.DataFrame(matches).applymap(lambda x: (format_dflist(x))),
-                use_container_width=True,
-            )
-        else:
-            disp_recognition_metrics.info("No recognized identities yet ...")
-
-        if len(matches) > 1:
-            disp_pca3d.plotly_chart(pca(matches, identities, gallery, dim=3), use_container_width=True)
-            disp_pca2d.plotly_chart(pca(matches, identities, gallery, dim=2), use_container_width=True)
-        else:
-            disp_pca3d.info("Only available if more than 1 recognized face ...")
-            disp_pca2d.info("Only available if more than 1 recognized face ...")
-
-        # Show Recognized Identities
-        if matches:
-            disp_identities_rec.image(
-                image=[identities[match.identity_idx].face_aligned for match in matches],
-                caption=[gallery[match.gallery_idx].name for match in matches],
-            )
-        else:
-            disp_identities_rec.info("No recognized identities yet ...")
+    # Start frame grabber in background thread
+    grabber.thread.start()
+    timestamp = time.time()
 
-# BUG Recognized Identity Image is not updating on cloud version? (works on local!!!)
+    # Start main loop
+    while True:
+        frame = grabber.get_frame()
+        if frame is not None:
+            # Print frame timestamp to streamlit
+            st.write(f"Frame timestamp: {frame.time}")
+
+            # Run face detection and recognition
+            frame, matches = video_frame_callback(frame)
+
+            # Convert frame to numpy array
+            frame = frame.to_ndarray(format="rgb24")
+
+            # Show Stream
+            stream_container.image(frame, channels="RGB")
+
+            # Show Matches
+            if matches:
+                matches_container.image(
+                    image=[match.subject_id.face for match in matches],
+                    caption=[match.gallery_id.name for match in matches],
+                )
+            else:
+                matches_container.info("No matches found yet ...")

requirements.txt

@@ -1,13 +1,9 @@
 streamlit
 scikit-image
 scikit-learn
-mediapipe
 opencv-python-headless
 watchdog
 streamlit-webrtc
-matplotlib
-streamlit-toggle-switch
-tflite-runtime
 twilio
-tqdm
-plotly
+onnxruntime
+mediapipe

tools/annotation.py

@@ -1,107 +0,0 @@
-import numpy as np
-import cv2
-
-
-class Annotation:
-    def __init__(self, draw_bbox=True, draw_landmarks=True, draw_name=True, upscale=True):
-        self.bbox = draw_bbox
-        self.landmarks = draw_landmarks
-        self.name = draw_name
-        self.upscale = upscale
-
-    def __call__(self, frame, detections, identities, matches, gallery):
-        shape = np.asarray(frame.shape[:2][::-1])
-        if self.upscale:
-            frame = cv2.resize(frame, (1920, 1080))
-            upscale_factor = np.asarray([1920 / shape[0], 1080 / shape[1]])
-            shape = np.asarray(frame.shape[:2][::-1])
-        else:
-            upscale_factor = np.asarray([1, 1])
-
-        frame.flags.writeable = True
-
-        for detection in detections:
-            # Draw Landmarks
-            if self.landmarks:
-                for landmark in detection.landmarks:
-                    cv2.circle(
-                        frame,
-                        (landmark * upscale_factor).astype(int),
-                        2,
-                        (255, 255, 255),
-                        -1,
-                    )
-
-            # Draw Bounding Box
-            if self.bbox:
-                cv2.rectangle(
-                    frame,
-                    (detection.bbox[0] * upscale_factor).astype(int),
-                    (detection.bbox[1] * upscale_factor).astype(int),
-                    (255, 0, 0),
-                    2,
-                )
-
-            # Draw Index
-            cv2.putText(
-                frame,
-                str(detection.idx),
-                (
-                    ((detection.bbox[1][0] + 2) * upscale_factor[0]).astype(int),
-                    ((detection.bbox[1][1] + 2) * upscale_factor[1]).astype(int),
-                ),
-                cv2.LINE_AA,
-                0.5,
-                (0, 0, 0),
-                2,
-            )
-
-        # Draw Name
-        if self.name:
-            for match in matches:
-                try:
-                    detection = detections[identities[match.identity_idx].detection_idx]
-                except:
-                    print("Identity IDX: ", match.identity_idx)
-                    print("Len(Detections): ", len(detections))
-                    print("Len(Identites): ", len(identities))
-                    print("Detection IDX: ", identities[match.identity_idx].detection_idx)
-
-                    # print("Detections: ", detections)
-
-                cv2.rectangle(
-                    frame,
-                    (detection.bbox[0] * upscale_factor).astype(int),
-                    (detection.bbox[1] * upscale_factor).astype(int),
-                    (0, 255, 0),
-                    2,
-                )
-
-                cv2.rectangle(
-                    frame,
-                    (
-                        (detection.bbox[0][0] * upscale_factor[0]).astype(int),
-                        (detection.bbox[0][1] * upscale_factor[1] - (shape[1] // 25)).astype(int),
-                    ),
-                    (
-                        (detection.bbox[1][0] * upscale_factor[0]).astype(int),
-                        (detection.bbox[0][1] * upscale_factor[1]).astype(int),
-                    ),
-                    (255, 255, 255),
-                    -1,
-                )
-
-                cv2.putText(
-                    frame,
-                    gallery[match.gallery_idx].name,
-                    (
-                        ((detection.bbox[0][0] + shape[0] // 400) * upscale_factor[0]).astype(int),
-                        ((detection.bbox[0][1] - shape[1] // 100) * upscale_factor[1]).astype(int),
-                    ),
-                    cv2.LINE_AA,
-                    0.5,
-                    (0, 0, 0),
-                    2,
-                )
-
-        return frame

tools/face_detection.py

@@ -1,481 +0,0 @@
-import tflite_runtime.interpreter as tflite
-import cv2
-import numpy as np
-from .utils import tflite_inference
-from .nametypes import Detection
-from .utils import get_file
-
-
-BASE_URL = "https://github.com/Martlgap/FaceIDLight/releases/download/v.0.1/"
-
-FILE_HASHES = {
-    "o_net": "768385d570300648b7b881acbd418146522b79b4771029bb2e684bdd8c764b9f",
-    "p_net": "530183192e24f7cc86b6706e1eb600482c4ed4306399ac939c472e3957bae15e",
-    "r_net": "5ec33b065eb2802bc4c2575d21feff1a56958d854785bc3e2907d3b7ace861a2",
-}
-
-
-class StageStatus:
-    """
-    Keeps status between MTCNN stages
-    """
-
-    def __init__(self, pad_result: tuple = None, width=0, height=0):
-        self.width = width
-        self.height = height
-        self.dy = self.edy = self.dx = self.edx = self.y = self.ey = self.x = self.ex = self.tmp_w = self.tmp_h = []
-
-        if pad_result is not None:
-            self.update(pad_result)
-
-    def update(self, pad_result: tuple):
-        s = self
-        s.dy, s.edy, s.dx, s.edx, s.y, s.ey, s.x, s.ex, s.tmp_w, s.tmp_h = pad_result
-
-
-class FaceDetection:
-    """
-    Allows to perform MTCNN Detection ->
-        a) Detection of faces (with the confidence probability)
-        b) Detection of keypoints (left eye, right eye, nose, mouth_left, mouth_right)
-    """
-
-    def __init__(
-        self,
-        min_face_size: int = 40,
-        steps_threshold: list = None,
-        scale_factor: float = 0.7,
-        min_detections_conf: float = 0.9,
-    ):
-        """
-        Initializes the MTCNN.
-        :param min_face_size: minimum size of the face to detect
-        :param steps_threshold: step's thresholds values
-        :param scale_factor: scale factor
-        """
-        if steps_threshold is None:
-            steps_threshold = [0.6, 0.7, 0.7]  # original mtcnn values [0.6, 0.7, 0.7]
-        self._min_face_size = min_face_size
-        self._steps_threshold = steps_threshold
-        self._scale_factor = scale_factor
-        self.min_detections_conf = min_detections_conf
-        self.p_net = tflite.Interpreter(model_path=get_file(BASE_URL + "p_net.tflite", FILE_HASHES["p_net"]))
-        self.r_net = tflite.Interpreter(model_path=get_file(BASE_URL + "r_net.tflite", FILE_HASHES["r_net"]))
-        self.o_net = tflite.Interpreter(model_path=get_file(BASE_URL + "o_net.tflite", FILE_HASHES["o_net"]))
-
-    def __call__(self, frame):
-        """
-        Detects bounding boxes from the specified image.
-        :param img: image to process
-        :return: list containing all the bounding boxes detected with their keypoints.
-
-        From MTCNN:
-        # Total boxes (bBoxes for faces)
-        # 1. dim -> Number of found Faces
-        # 2. dim -> x_min, y_min, x_max, y_max, score
-
-        # Points (Landmarks left eye, right eye, nose, left mouth, right mouth)
-        # 1. dim -> Number of found Faces
-        # 2. dim -> x1, x2, x3, x4, x5, y2, y2, y3, y4, y5 Coordinates
-        """
-
-        height, width, _ = frame.shape
-        stage_status = StageStatus(width=width, height=height)
-        m = 12 / self._min_face_size
-        min_layer = np.amin([height, width]) * m
-        scales = self.__compute_scale_pyramid(m, min_layer)
-
-        # We pipe here each of the stages
-        total_boxes, stage_status = self.__stage1(frame, scales, stage_status)
-        total_boxes, stage_status = self.__stage2(frame, total_boxes, stage_status)
-        bboxes, points = self.__stage3(frame, total_boxes, stage_status)
-
-        # Sort by location (to prevent flickering)
-        sort_idx = np.argsort(bboxes[:, 0])
-        bboxes = bboxes[sort_idx]
-        points = points[sort_idx]
-
-        # Transform to better shape and points now inside bbox
-        detections = []
-        cnt = 0
-        for i in range(bboxes.shape[0]):
-            conf = bboxes[i, -1].astype(np.float32)
-            if conf > self.min_detections_conf:
-                bboxes_c = np.reshape(bboxes[i, :-1], [2, 2]).astype(np.float32)
-                points_c = np.reshape(points[i], [2, 5]).transpose().astype(np.float32)
-                detections.append(
-                    Detection(
-                        idx=cnt,
-                        bbox=list(bboxes_c),
-                        landmarks=list(points_c),
-                        confidence=conf,
-                    )
-                )
-                cnt += 1
-        return frame, detections
-
-    def __compute_scale_pyramid(self, m, min_layer):
-        scales = []
-        factor_count = 0
-
-        while min_layer >= 12:
-            scales += [m * np.power(self._scale_factor, factor_count)]
-            min_layer = min_layer * self._scale_factor
-            factor_count += 1
-
-        return scales
-
-    @staticmethod
-    def __scale_image(image, scale: float):
-        """
-        Scales the image to a given scale.
-        :param image:
-        :param scale:
-        :return:
-        """
-        height, width, _ = image.shape
-
-        width_scaled = int(np.ceil(width * scale))
-        height_scaled = int(np.ceil(height * scale))
-
-        im_data = cv2.resize(image, (width_scaled, height_scaled), interpolation=cv2.INTER_AREA)
-
-        # Normalize the image's pixels
-        im_data_normalized = (im_data - 127.5) * 0.0078125
-
-        return im_data_normalized
-
-    @staticmethod
-    def __generate_bounding_box(imap, reg, scale, t):
-        # use heatmap to generate bounding boxes
-        stride = 2
-        cellsize = 12
-
-        imap = np.transpose(imap)
-        dx1 = np.transpose(reg[:, :, 0])
-        dy1 = np.transpose(reg[:, :, 1])
-        dx2 = np.transpose(reg[:, :, 2])
-        dy2 = np.transpose(reg[:, :, 3])
-
-        y, x = np.where(imap >= t)
-
-        if y.shape[0] == 1:
-            dx1 = np.flipud(dx1)
-            dy1 = np.flipud(dy1)
-            dx2 = np.flipud(dx2)
-            dy2 = np.flipud(dy2)
-
-        score = imap[(y, x)]
-        reg = np.transpose(np.vstack([dx1[(y, x)], dy1[(y, x)], dx2[(y, x)], dy2[(y, x)]]))
-
-        if reg.size == 0:
-            reg = np.empty(shape=(0, 3))
-
-        bb = np.transpose(np.vstack([y, x]))
-
-        q1 = np.fix((stride * bb + 1) / scale)
-        q2 = np.fix((stride * bb + cellsize) / scale)
-        boundingbox = np.hstack([q1, q2, np.expand_dims(score, 1), reg])
-
-        return boundingbox, reg
-
-    @staticmethod
-    def __nms(boxes, threshold, method):
-        """
-        Non Maximum Suppression.
-
-        :param boxes: np array with bounding boxes.
-        :param threshold:
-        :param method: NMS method to apply. Available values ('Min', 'Union')
-        :return:
-        """
-        if boxes.size == 0:
-            return np.empty((0, 3))
-
-        x1 = boxes[:, 0]
-        y1 = boxes[:, 1]
-        x2 = boxes[:, 2]
-        y2 = boxes[:, 3]
-        s = boxes[:, 4]
-
-        area = (x2 - x1 + 1) * (y2 - y1 + 1)
-        sorted_s = np.argsort(s)
-
-        pick = np.zeros_like(s, dtype=np.int16)
-        counter = 0
-        while sorted_s.size > 0:
-            i = sorted_s[-1]
-            pick[counter] = i
-            counter += 1
-            idx = sorted_s[0:-1]
-
-            xx1 = np.maximum(x1[i], x1[idx])
-            yy1 = np.maximum(y1[i], y1[idx])
-            xx2 = np.minimum(x2[i], x2[idx])
-            yy2 = np.minimum(y2[i], y2[idx])
-
-            w = np.maximum(0.0, xx2 - xx1 + 1)
-            h = np.maximum(0.0, yy2 - yy1 + 1)
-
-            inter = w * h
-
-            if method == "Min":
-                o = inter / np.minimum(area[i], area[idx])
-            else:
-                o = inter / (area[i] + area[idx] - inter)
-
-            sorted_s = sorted_s[np.where(o <= threshold)]
-
-        pick = pick[0:counter]
-
-        return pick
-
-    @staticmethod
-    def __pad(total_boxes, w, h):
-        # compute the padding coordinates (pad the bounding boxes to square)
-        tmp_w = (total_boxes[:, 2] - total_boxes[:, 0] + 1).astype(np.int32)
-        tmp_h = (total_boxes[:, 3] - total_boxes[:, 1] + 1).astype(np.int32)
-        numbox = total_boxes.shape[0]
-
-        dx = np.ones(numbox, dtype=np.int32)
-        dy = np.ones(numbox, dtype=np.int32)
-        edx = tmp_w.copy().astype(np.int32)
-        edy = tmp_h.copy().astype(np.int32)
-
-        x = total_boxes[:, 0].copy().astype(np.int32)
-        y = total_boxes[:, 1].copy().astype(np.int32)
-        ex = total_boxes[:, 2].copy().astype(np.int32)
-        ey = total_boxes[:, 3].copy().astype(np.int32)
-
-        tmp = np.where(ex > w)
-        edx.flat[tmp] = np.expand_dims(-ex[tmp] + w + tmp_w[tmp], 1)
-        ex[tmp] = w
-
-        tmp = np.where(ey > h)
-        edy.flat[tmp] = np.expand_dims(-ey[tmp] + h + tmp_h[tmp], 1)
-        ey[tmp] = h
-
-        tmp = np.where(x < 1)
-        dx.flat[tmp] = np.expand_dims(2 - x[tmp], 1)
-        x[tmp] = 1
-
-        tmp = np.where(y < 1)
-        dy.flat[tmp] = np.expand_dims(2 - y[tmp], 1)
-        y[tmp] = 1
-
-        return dy, edy, dx, edx, y, ey, x, ex, tmp_w, tmp_h
-
-    @staticmethod
-    def __rerec(bbox):
-        # convert bbox to square
-        height = bbox[:, 3] - bbox[:, 1]
-        width = bbox[:, 2] - bbox[:, 0]
-        max_side_length = np.maximum(width, height)
-        bbox[:, 0] = bbox[:, 0] + width * 0.5 - max_side_length * 0.5
-        bbox[:, 1] = bbox[:, 1] + height * 0.5 - max_side_length * 0.5
-        bbox[:, 2:4] = bbox[:, 0:2] + np.transpose(np.tile(max_side_length, (2, 1)))
-        return bbox
-
-    @staticmethod
-    def __bbreg(boundingbox, reg):
-        # calibrate bounding boxes
-        if reg.shape[1] == 1:
-            reg = np.reshape(reg, (reg.shape[2], reg.shape[3]))
-
-        w = boundingbox[:, 2] - boundingbox[:, 0] + 1
-        h = boundingbox[:, 3] - boundingbox[:, 1] + 1
-        b1 = boundingbox[:, 0] + reg[:, 0] * w
-        b2 = boundingbox[:, 1] + reg[:, 1] * h
-        b3 = boundingbox[:, 2] + reg[:, 2] * w
-        b4 = boundingbox[:, 3] + reg[:, 3] * h
-        boundingbox[:, 0:4] = np.transpose(np.vstack([b1, b2, b3, b4]))
-        return boundingbox
-
-    def __stage1(self, image, scales: list, stage_status: StageStatus):
-        """
-        First stage of the MTCNN.
-        :param image:
-        :param scales:
-        :param stage_status:
-        :return:
-        """
-        total_boxes = np.empty((0, 9))
-        status = stage_status
-
-        for scale in scales:
-            scaled_image = self.__scale_image(image, scale)
-
-            img_x = np.expand_dims(scaled_image, 0)
-            img_y = np.transpose(img_x, (0, 2, 1, 3))
-
-            out = tflite_inference(self.p_net, img_y)
-
-            out0 = np.transpose(out[0], (0, 2, 1, 3))
-            out1 = np.transpose(out[1], (0, 2, 1, 3))
-
-            boxes, _ = self.__generate_bounding_box(
-                out1[0, :, :, 1].copy(),
-                out0[0, :, :, :].copy(),
-                scale,
-                self._steps_threshold[0],
-            )
-
-            # inter-scale nms
-            pick = self.__nms(boxes.copy(), 0.5, "Union")
-            if boxes.size > 0 and pick.size > 0:
-                boxes = boxes[pick, :]
-                total_boxes = np.append(total_boxes, boxes, axis=0)
-
-        numboxes = total_boxes.shape[0]
-
-        if numboxes > 0:
-            pick = self.__nms(total_boxes.copy(), 0.7, "Union")
-            total_boxes = total_boxes[pick, :]
-
-            regw = total_boxes[:, 2] - total_boxes[:, 0]
-            regh = total_boxes[:, 3] - total_boxes[:, 1]
-
-            qq1 = total_boxes[:, 0] + total_boxes[:, 5] * regw
-            qq2 = total_boxes[:, 1] + total_boxes[:, 6] * regh
-            qq3 = total_boxes[:, 2] + total_boxes[:, 7] * regw
-            qq4 = total_boxes[:, 3] + total_boxes[:, 8] * regh
-
-            total_boxes = np.transpose(np.vstack([qq1, qq2, qq3, qq4, total_boxes[:, 4]]))
-            total_boxes = self.__rerec(total_boxes.copy())
-
-            total_boxes[:, 0:4] = np.fix(total_boxes[:, 0:4]).astype(np.int32)
-            status = StageStatus(
-                self.__pad(total_boxes.copy(), stage_status.width, stage_status.height),
-                width=stage_status.width,
-                height=stage_status.height,
-            )
-
-        return total_boxes, status
-
-    def __stage2(self, img, total_boxes, stage_status: StageStatus):
-        """
-        Second stage of the MTCNN.
-        :param img:
-        :param total_boxes:
-        :param stage_status:
-        :return:
-        """
-
-        num_boxes = total_boxes.shape[0]
-        if num_boxes == 0:
-            return total_boxes, stage_status
-
-        # second stage
-        tempimg = np.zeros(shape=(24, 24, 3, num_boxes))
-
-        for k in range(0, num_boxes):
-            tmp = np.zeros((int(stage_status.tmp_h[k]), int(stage_status.tmp_w[k]), 3))
-
-            tmp[
-                stage_status.dy[k] - 1 : stage_status.edy[k],
-                stage_status.dx[k] - 1 : stage_status.edx[k],
-                :,
-            ] = img[
-                stage_status.y[k] - 1 : stage_status.ey[k],
-                stage_status.x[k] - 1 : stage_status.ex[k],
-                :,
-            ]
-
-            if tmp.shape[0] > 0 and tmp.shape[1] > 0 or tmp.shape[0] == 0 and tmp.shape[1] == 0:
-                tempimg[:, :, :, k] = cv2.resize(tmp, (24, 24), interpolation=cv2.INTER_AREA)
-
-            else:
-                return np.empty(shape=(0,)), stage_status
-
-        tempimg = (tempimg - 127.5) * 0.0078125
-        tempimg1 = np.transpose(tempimg, (3, 1, 0, 2))
-
-        out = tflite_inference(self.r_net, tempimg1)
-
-        out0 = np.transpose(out[0])
-        out1 = np.transpose(out[1])
-
-        score = out1[1, :]
-
-        ipass = np.where(score > self._steps_threshold[1])
-
-        total_boxes = np.hstack([total_boxes[ipass[0], 0:4].copy(), np.expand_dims(score[ipass].copy(), 1)])
-
-        mv = out0[:, ipass[0]]
-
-        if total_boxes.shape[0] > 0:
-            pick = self.__nms(total_boxes, 0.7, "Union")
-            total_boxes = total_boxes[pick, :]
-            total_boxes = self.__bbreg(total_boxes.copy(), np.transpose(mv[:, pick]))
-            total_boxes = self.__rerec(total_boxes.copy())
-
-        return total_boxes, stage_status
-
-    def __stage3(self, img, total_boxes, stage_status: StageStatus):
-        """
-        Third stage of the MTCNN.
-
-        :param img:
-        :param total_boxes:
-        :param stage_status:
-        :return:
-        """
-        num_boxes = total_boxes.shape[0]
-        if num_boxes == 0:
-            return total_boxes, np.empty(shape=(0,))
-
-        total_boxes = np.fix(total_boxes).astype(np.int32)
-
-        status = StageStatus(
-            self.__pad(total_boxes.copy(), stage_status.width, stage_status.height),
-            width=stage_status.width,
-            height=stage_status.height,
-        )
-
-        tempimg = np.zeros((48, 48, 3, num_boxes))
-
-        for k in range(0, num_boxes):
-            tmp = np.zeros((int(status.tmp_h[k]), int(status.tmp_w[k]), 3))
-
-            tmp[status.dy[k] - 1 : status.edy[k], status.dx[k] - 1 : status.edx[k], :] = img[
-                status.y[k] - 1 : status.ey[k], status.x[k] - 1 : status.ex[k], :
-            ]
-
-            if tmp.shape[0] > 0 and tmp.shape[1] > 0 or tmp.shape[0] == 0 and tmp.shape[1] == 0:
-                tempimg[:, :, :, k] = cv2.resize(tmp, (48, 48), interpolation=cv2.INTER_AREA)
-            else:
-                return np.empty(shape=(0,)), np.empty(shape=(0,))
-
-        tempimg = (tempimg - 127.5) * 0.0078125
-        tempimg1 = np.transpose(tempimg, (3, 1, 0, 2))
-
-        out = tflite_inference(self.o_net, tempimg1)
-        out0 = np.transpose(out[0])
-        out1 = np.transpose(out[1])
-        out2 = np.transpose(out[2])
-
-        score = out2[1, :]
-
-        points = out1
-
-        ipass = np.where(score > self._steps_threshold[2])
-
-        points = points[:, ipass[0]]
-
-        total_boxes = np.hstack([total_boxes[ipass[0], 0:4].copy(), np.expand_dims(score[ipass].copy(), 1)])
-
-        mv = out0[:, ipass[0]]
-
-        w = total_boxes[:, 2] - total_boxes[:, 0] + 1
-        h = total_boxes[:, 3] - total_boxes[:, 1] + 1
-
-        points[0:5, :] = np.tile(w, (5, 1)) * points[0:5, :] + np.tile(total_boxes[:, 0], (5, 1)) - 1
-        points[5:10, :] = np.tile(h, (5, 1)) * points[5:10, :] + np.tile(total_boxes[:, 1], (5, 1)) - 1
-
-        if total_boxes.shape[0] > 0:
-            total_boxes = self.__bbreg(total_boxes.copy(), np.transpose(mv))
-            pick = self.__nms(total_boxes.copy(), 0.7, "Min")
-            total_boxes = total_boxes[pick, :]
-            points = points[:, pick]
-
-        return total_boxes, points.transpose()

tools/face_recognition.py

@@ -1,114 +0,0 @@
-from .utils import tflite_inference
-from .nametypes import Identity, Match
-from sklearn.metrics.pairwise import cosine_distances
-import numpy as np
-import cv2
-from skimage.transform import SimilarityTransform
-from .utils import get_file
-import tflite_runtime.interpreter as tflite
-from typing import Literal
-
-
-BASE_URL = "https://github.com/Martlgap/FaceIDLight/releases/download/v.0.1/"
-
-FILE_HASHES = {
-    "mobileNet": "6c19b789f661caa8da735566490bfd8895beffb2a1ec97a56b126f0539991aa6",
-    "resNet": "f4d8b0194957a3ad766135505fc70a91343660151a8103bbb6c3b8ac34dbb4e2",
-}
-
-
-class FaceRecognition:
-    def __init__(
-        self,
-        min_similarity: float = 0.67,
-        model_name: Literal["mobileNet", "resNet50"] = "mobileNet",
-    ):
-        self.min_similarity = min_similarity
-        self.model = tflite.Interpreter(model_path=get_file(BASE_URL + f"{model_name}.tflite", FILE_HASHES[model_name]))
-
-    def __call__(self, frame, detections):
-        # Align Faces
-        faces, faces_aligned = [], []
-        for detection in detections:
-            face = frame[
-                int(detection.bbox[0][1]) : int(detection.bbox[1][1]),
-                int(detection.bbox[0][0]) : int(detection.bbox[1][0]),
-            ]
-            try:
-                face = cv2.resize(face, (112, 112))
-            except:
-                face = np.zeros((112, 112, 3))
-
-            faces.append(face)
-            faces_aligned.append(self.align(frame, detection.landmarks))
-
-        # Do Inference
-        if len(faces_aligned) == 0:
-            return []
-
-        # Normalize images from [0, 255] to [0, 1]
-        faces_aligned_norm = np.asarray(faces_aligned).astype(np.float32) / 255.0
-
-        embs_det = tflite_inference(self.model, faces_aligned_norm)
-        embs_det = np.asarray(embs_det[0])
-
-        # Save Identities
-        identities = []
-        for idx, detection in enumerate(detections):
-            identities.append(
-                Identity(
-                    detection_idx=detection.idx,
-                    embedding=embs_det[idx],
-                    face_aligned=faces_aligned[idx],
-                )
-            )
-        return identities
-
-    def find_matches(self, identities, gallery):
-        if len(gallery) == 0 or len(identities) == 0:
-            return []
-
-        # Get Embeddings
-        embs_gal = np.asarray([identity.embedding for identity in gallery])
-        embs_det = np.asarray([identity.embedding for identity in identities])
-
-        # Calculate Cosine Distances
-        cos_distances = cosine_distances(embs_det, embs_gal)
-
-        # Find Matches
-        matches = []
-        for ident_idx, identity in enumerate(identities):
-            dist_to_identity = cos_distances[ident_idx]
-            idx_min = np.argmin(dist_to_identity)
-            if dist_to_identity[idx_min] < self.min_similarity:
-                matches.append(
-                    Match(
-                        identity_idx=identity.detection_idx,
-                        gallery_idx=idx_min,
-                        distance=dist_to_identity[idx_min],
-                        name=gallery[idx_min].name,
-                    )
-                )
-
-        # Sort Matches by identity_idx
-        matches = sorted(matches, key=lambda match: match.gallery_idx)
-
-        return matches
-
-    @staticmethod
-    def align(img, landmarks_source, target_size=(112, 112)):
-        landmarks_target = 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(landmarks_source, landmarks_target)
-        tmatrix = tform.params[0:2, :]
-        face_aligned = cv2.warpAffine(img, tmatrix, target_size, borderValue=0.0)
-        return face_aligned

tools/gallery.py

@@ -1,37 +0,0 @@
-from .face_detection import FaceDetection
-from .face_recognition import FaceRecognition
-from .nametypes import Identity
-import cv2
-import os
-import numpy as np
-
-
-def init_gallery(files, min_detections_conf=0.8, min_similarity=0.67, model_name="mobileNet"):
-    face_detector = FaceDetection(min_detections_conf=min_detections_conf)
-    face_recognizer = FaceRecognition(model_name=model_name, min_similarity=min_similarity)
-
-    gallery = []
-    for file in files:
-        file_bytes = np.asarray(bytearray(file.read()), dtype=np.uint8)
-        img = cv2.cvtColor(cv2.imdecode(file_bytes, cv2.IMREAD_COLOR), cv2.COLOR_BGR2RGB)
-        # Face Detection
-        img, detections = face_detector(img)
-
-        if detections == []:
-            continue
-        elif len(detections) > 1:
-            detections = detections[:1]
-
-        # Face Recognition
-        identities = face_recognizer(img, detections)
-
-        # Add to gallery
-        gallery.append(
-            Identity(
-                name=os.path.splitext(file.name)[0],
-                embedding=identities[0].embedding,
-                face_aligned=identities[0].face_aligned,
-            )
-        )
-
-    return gallery

tools/nametypes.py

@@ -1,33 +0,0 @@
-from typing import NamedTuple, List
-import numpy as np
-
-
-class Detection(NamedTuple):
-    idx: int = None
-    bbox: List[List[float]] = None
-    landmarks: List[List[float]] = None
-    confidence: float = None
-
-
-class Identity(NamedTuple):
-    detection_idx: int = None
-    name: str = None
-    embedding: np.ndarray = None
-    face_aligned: np.ndarray = None
-
-
-class Stats(NamedTuple):
-    fps: float = 0
-    resolution: List[int] = [None, None, None]
-    num_faces: int = 0
-    detection: float = None
-    recognition: float = None
-    matching: float = None
-    annotation: float = None
-
-
-class Match(NamedTuple):
-    identity_idx: int = None
-    gallery_idx: int = None
-    distance: float = None
-    name: str = None

tools/pca.py

@@ -1,59 +0,0 @@
-from sklearn.decomposition import PCA
-import numpy as np
-import plotly.express as px
-
-
-def pca(matches, identities, gallery, dim=3):
-    """
-    Perform PCA on embeddings.
-    Args:
-        embeddings: np.array of shape (n_embeddings, 512)
-    Returns:
-        embeddings_pca: np.array of shape (n_embeddings, 3)
-    """
-
-    # Get Gallery and Detection Embeddings and stich them together in groups
-    embeddings = np.concatenate(
-        [[gallery[match.gallery_idx].embedding, identities[match.identity_idx].embedding] for match in matches],
-        axis=0,
-    )
-
-    # Get Identity Names and stich them together in groups
-    identity_names = np.concatenate(
-        [[gallery[match.gallery_idx].name, gallery[match.gallery_idx].name] for match in matches],
-        axis=0,
-    )
-
-    # Do 3D PCA
-    pca = PCA(n_components=dim)
-    pca.fit(embeddings)
-    embeddings_pca = pca.transform(embeddings)
-
-    if dim == 3:
-        fig = px.scatter_3d(
-            embeddings_pca,
-            x=0,
-            y=1,
-            z=2,
-            opacity=0.7,
-            color=identity_names,
-            color_discrete_sequence=px.colors.qualitative.Vivid,
-        )
-        fig.update_traces(marker=dict(size=4))
-    elif dim == 2:
-        fig = px.scatter(
-            embeddings_pca,
-            x=0,
-            y=1,
-            opacity=0.7,
-            color=identity_names,
-            color_discrete_sequence=px.colors.qualitative.Vivid,
-        )
-        fig.update_traces(marker=dict(size=4))
-        fig.update_xaxes(showgrid=True)
-        fig.update_yaxes(showgrid=True)
-    else:
-        raise ValueError("dim must be either 2 or 3")
-    fig.update_layout(margin=dict(l=0, r=0, b=0, t=0))
-
-    return fig

tools/utils.py

@@ -1,164 +0,0 @@
-import logging
-import os
-import streamlit as st
-from twilio.rest import Client
-import os
-import numpy as np
-import hashlib
-import tempfile
-import os
-import hashlib
-from tqdm import tqdm
-from zipfile import ZipFile
-from urllib.request import urlopen
-
-
-logger = logging.getLogger(__name__)
-
-
-@st.cache_data
-def get_ice_servers(name="twilio"):
-    """Get ICE servers from Twilio.
-    Returns:
-        List of ICE servers.
-    """
-    if name == "twilio":
-        # Ref: https://www.twilio.com/docs/stun-turn/api
-        try:
-            account_sid = os.environ["TWILIO_ACCOUNT_SID"]
-            auth_token = os.environ["TWILIO_AUTH_TOKEN"]
-        except KeyError:
-            logger.warning("Twilio credentials are not set. Fallback to a free STUN server from Google.")
-            return [{"urls": ["stun:stun.l.google.com:19302"]}]
-
-        client = Client(account_sid, auth_token)
-
-        token = client.tokens.create()
-
-        return token.ice_servers
-
-    elif name == "metered":
-        try:
-            username = os.environ["METERED_USERNAME"]
-            credential = os.environ["METERED_CREDENTIAL"]
-        except KeyError:
-            logger.warning("Metered credentials are not set. Fallback to a free STUN server from Google.")
-            return [{"urls": ["stun:stun.l.google.com:19302"]}]
-
-        ice_servers = [
-            {"url": "stun:a.relay.metered.ca:80", "urls": "stun:a.relay.metered.ca:80"},
-            {
-                "url": "turn:a.relay.metered.ca:80",
-                "username": username,
-                "urls": "turn:a.relay.metered.ca:80",
-                "credential": credential,
-            },
-            {
-                "url": "turn:a.relay.metered.ca:80?transport=tcp",
-                "username": username,
-                "urls": "turn:a.relay.metered.ca:80?transport=tcp",
-                "credential": credential,
-            },
-            {
-                "url": "turn:a.relay.metered.ca:443",
-                "username": username,
-                "urls": "turn:a.relay.metered.ca:443",
-                "credential": credential,
-            },
-            {
-                "url": "turn:a.relay.metered.ca:443?transport=tcp",
-                "username": username,
-                "urls": "turn:a.relay.metered.ca:443?transport=tcp",
-                "credential": credential,
-            },
-        ]
-        return ice_servers
-    else:
-        raise ValueError(f"Unknown name: {name}")
-
-
-# Function to format floats within a list
-def format_dflist(val):
-    if isinstance(val, list):
-        return [format_dflist(num) for num in val]
-    if isinstance(val, np.ndarray):
-        return np.asarray([format_dflist(num) for num in val])
-    if isinstance(val, np.float32):
-        return f"{val:.2f}"
-    if isinstance(val, float):
-        return f"{val:.2f}"
-    else:
-        return val
-
-
-def rgb(r, g, b):
-    return "#{:02x}{:02x}{:02x}".format(r, g, b)
-
-
-def tflite_inference(model, img):
-    """Inferences an image through the model with tflite interpreter on CPU
-    :param model: a tflite.Interpreter loaded with a model
-    :param img: image
-    :return: list of outputs of the model
-    """
-    # Check if img is np.ndarray
-    if not isinstance(img, np.ndarray):
-        img = np.asarray(img)
-
-    # Check if dim is 4
-    if len(img.shape) == 3:
-        img = np.expand_dims(img, axis=0)
-
-    input_details = model.get_input_details()
-    output_details = model.get_output_details()
-    model.resize_tensor_input(input_details[0]["index"], img.shape)
-    model.allocate_tensors()
-    model.set_tensor(input_details[0]["index"], img.astype(np.float32))
-    model.invoke()
-    return [model.get_tensor(elem["index"]) for elem in output_details]
-
-
-def get_file(origin, file_hash, is_zip=False):
-    tmp_file = os.path.join(tempfile.gettempdir(), "FaceIDLight", origin.split("/")[-1])
-    os.makedirs(os.path.dirname(tmp_file), exist_ok=True)
-    if not os.path.exists(tmp_file):
-        download = True
-    else:
-        hasher = hashlib.sha256()
-        with open(tmp_file, "rb") as file:
-            for chunk in iter(lambda: file.read(65535), b""):
-                hasher.update(chunk)
-        if not hasher.hexdigest() == file_hash:
-            print(
-                "A local file was found, but it seems to be incomplete or outdated because the file hash does not "
-                "match the original value of " + file_hash + " so data will be downloaded."
-            )
-            download = True
-        else:
-            download = False
-
-    if download:
-        response = urlopen(origin)
-        with tqdm.wrapattr(
-            open(tmp_file, "wb"),
-            "write",
-            miniters=1,
-            desc="Downloading " + origin.split("/")[-1] + " to: " + tmp_file,
-            total=getattr(response, "length", None),
-        ) as file:
-            for chunk in response:
-                file.write(chunk)
-            file.close()
-    if is_zip:
-        with ZipFile(tmp_file, "r") as zipObj:
-            zipObj.extractall(tmp_file.split(".")[0])
-        tmp_file = os.path.join(tmp_file.split(".")[0])
-    return tmp_file
-
-
-def get_hash(filepath):
-    hasher = hashlib.sha256()
-    with open(filepath, "rb") as file:
-        for chunk in iter(lambda: file.read(65535), b""):
-            hasher.update(chunk)
-    return hasher.hexdigest()