Face recognition improvements (#17387)

* Increase frequency of updates when internal face detection is used

* Adjust number of required faces based on detection type

* Adjust min_score config to unknown_score

* Only for person

* Improve typing

* Update face rec docs

* Cleanup ui colors

* Cleanup

frigate/camera/state.py

@@ -5,7 +5,7 @@ import logging
 import os
 import threading
 from collections import defaultdict
-from typing import Callable
+from typing import Any, Callable
 
 import cv2
 import numpy as np
@@ -53,8 +53,19 @@ class CameraState:
         self.callbacks = defaultdict(list)
         self.ptz_autotracker_thread = ptz_autotracker_thread
         self.prev_enabled = self.camera_config.enabled
+        self.requires_face_detection = (
+            self.config.face_recognition.enabled
+            and "face" not in self.config.objects.all_objects
+        )
 
-    def get_current_frame(self, draw_options={}):
+    def get_max_update_frequency(self, obj: TrackedObject) -> int:
+        return (
+            1
+            if self.requires_face_detection and obj.obj_data["label"] == "person"
+            else 5
+        )
+
+    def get_current_frame(self, draw_options: dict[str, Any] = {}):
         with self.current_frame_lock:
             frame_copy = np.copy(self._current_frame)
             frame_time = self.current_frame_time
@@ -283,11 +294,12 @@ class CameraState:
 
                 updated_obj.last_updated = frame_time
 
-            # if it has been more than 5 seconds since the last thumb update
+            # if it has been more than max_update_frequency seconds since the last thumb update
             # and the last update is greater than the last publish or
             # the object has changed significantly
             if (
-                frame_time - updated_obj.last_published > 5
+                frame_time - updated_obj.last_published
+                > self.get_max_update_frequency(updated_obj)
                 and updated_obj.last_updated > updated_obj.last_published
             ) or significant_update:
                 # call event handlers

frigate/config/classification.py

@@ -54,8 +54,8 @@ class FaceRecognitionConfig(FrigateBaseModel):
     model_size: str = Field(
         default="small", title="The size of the embeddings model used."
     )
-    min_score: float = Field(
-        title="Minimum face distance score required to save the attempt.",
+    unknown_score: float = Field(
+        title="Minimum face distance score required to be marked as a potential match.",
         default=0.8,
         gt=0.0,
         le=1.0,

frigate/data_processing/common/face/model.py

@@ -164,9 +164,7 @@ class LBPHRecognizer(FaceRecognizer):
             return
 
         self.recognizer: cv2.face.LBPHFaceRecognizer = (
-            cv2.face.LBPHFaceRecognizer_create(
-                radius=2, threshold=(1 - self.config.face_recognition.min_score) * 1000
-            )
+            cv2.face.LBPHFaceRecognizer_create(radius=2, threshold=400)
         )
         self.recognizer.train(faces, np.array(labels))
 
@@ -243,6 +241,8 @@ class ArcFaceRecognizer(FaceRecognizer):
         for name, embs in face_embeddings_map.items():
             self.mean_embs[name] = stats.trim_mean(embs, 0.15)
 
+        logger.debug("Finished building ArcFace model")
+
     def similarity_to_confidence(
         self, cosine_similarity: float, median=0.3, range_width=0.6, slope_factor=12
     ):
@@ -302,7 +302,4 @@ class ArcFaceRecognizer(FaceRecognizer):
                 score = confidence
                 label = name
 
-        if score < self.config.face_recognition.min_score:
-            return None
-
         return label, round(score * blur_factor, 2)

frigate/data_processing/real_time/face.py

@@ -36,36 +36,6 @@ MAX_DETECTION_HEIGHT = 1080
 MIN_MATCHING_FACES = 2
 
 
-def weighted_average_by_area(results_list: list[tuple[str, float, int]]):
-    if len(results_list) < 3:
-        return "unknown", 0.0
-
-    score_count = {}
-    weighted_scores = {}
-    total_face_areas = {}
-
-    for name, score, face_area in results_list:
-        if name not in weighted_scores:
-            score_count[name] = 1
-            weighted_scores[name] = 0.0
-            total_face_areas[name] = 0.0
-        else:
-            score_count[name] += 1
-
-        weighted_scores[name] += score * face_area
-        total_face_areas[name] += face_area
-
-    prominent_name = max(score_count)
-
-    # if a single name is not prominent in the history then we are not confident
-    if score_count[prominent_name] / len(results_list) < 0.65:
-        return "unknown", 0.0
-
-    return prominent_name, weighted_scores[prominent_name] / total_face_areas[
-        prominent_name
-    ]
-
-
 class FaceRealTimeProcessor(RealTimeProcessorApi):
     def __init__(
         self,
@@ -271,6 +241,9 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
 
         sub_label, score = res
 
+        if score < self.face_config.unknown_score:
+            sub_label = "unknown"
+
         logger.debug(
             f"Detected best face for person as: {sub_label} with probability {score}"
         )
@@ -288,7 +261,7 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
         self.person_face_history[id].append(
             (sub_label, score, face_frame.shape[0] * face_frame.shape[1])
         )
-        (weighted_sub_label, weighted_score) = weighted_average_by_area(
+        (weighted_sub_label, weighted_score) = self.weighted_average_by_area(
             self.person_face_history[id]
         )
 
@@ -415,3 +388,34 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
     def expire_object(self, object_id: str):
         if object_id in self.person_face_history:
             self.person_face_history.pop(object_id)
+
+    def weighted_average_by_area(self, results_list: list[tuple[str, float, int]]):
+        min_faces = 1 if self.requires_face_detection else 3
+
+        if len(results_list) < min_faces:
+            return "unknown", 0.0
+
+        score_count = {}
+        weighted_scores = {}
+        total_face_areas = {}
+
+        for name, score, face_area in results_list:
+            if name not in weighted_scores:
+                score_count[name] = 1
+                weighted_scores[name] = 0.0
+                total_face_areas[name] = 0.0
+            else:
+                score_count[name] += 1
+
+            weighted_scores[name] += score * face_area
+            total_face_areas[name] += face_area
+
+        prominent_name = max(score_count)
+
+        # if a single name is not prominent in the history then we are not confident
+        if score_count[prominent_name] / len(results_list) < 0.65:
+            return "unknown", 0.0
+
+        return prominent_name, weighted_scores[prominent_name] / total_face_areas[
+            prominent_name
+        ]