administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-08-02 17:12:16 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Refactor face library page (#17424)

Browse Source 
* Section faces by event id

* Make score keeping more robust

* layout improvements

* Cleanup dialog

* Fix clicking behavior

* Add view in explore option

* math.round

* Don't require events

* Cleanup

* Remove selection

* Don't require

* Change dialog size with snapshot

* Use filename as key

* fix key

* Rework layout for mobile

* Handle mobile landscape

* Fix train issue

* Match logic

* Move deletion logic

* Fix reprocessing

* Support creating a new face

* Translations

* Do sorting in frontend

* Adjust unknown

* Cleanup

* Set max limit to faces to recognize

* Fix sorting

* Fix
This commit is contained in:
Nicolas Mowen
2025-03-28 13:52:12 -05:00
committed by GitHub
parent 37e0b9b904
commit b14abffea3
6 changed files with 325 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/data_processing/real_time/face.py
+71 -26
View File
@@ -33,7 +33,8 @@ logger = logging.getLogger(__name__)
MAX_DETECTION_HEIGHT = 1080
MIN_MATCHING_FACES = 2
MAX_FACES_ATTEMPTS_AFTER_REC = 6
MAX_FACE_ATTEMPTS = 12
class FaceRealTimeProcessor(RealTimeProcessorApi):
@@ -170,6 +171,23 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
)
return
# check if we have hit limits
if (
id in self.person_face_history
and len(self.person_face_history[id]) >= MAX_FACES_ATTEMPTS_AFTER_REC
):
# if we are at max attempts after rec and we have a rec
if obj_data.get("sub_label"):
logger.debug(
"Not processing due to hitting max attempts after true recognition."
)
return
# if we don't have a rec and are at max attempts
if len(self.person_face_history[id]) >= MAX_FACE_ATTEMPTS:
logger.debug("Not processing due to hitting max rec attempts.")
return
face: Optional[dict[str, any]] = None
if self.requires_face_detection:
@@ -241,7 +259,7 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
sub_label, score = res
if score < self.face_config.unknown_score:
if score <= self.face_config.unknown_score:
sub_label = "unknown"
logger.debug(
@@ -255,13 +273,23 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
os.makedirs(folder, exist_ok=True)
cv2.imwrite(file, face_frame)
files = sorted(
filter(lambda f: (f.endswith(".webp")), os.listdir(folder)),
key=lambda f: os.path.getctime(os.path.join(folder, f)),
reverse=True,
)
# delete oldest face image if maximum is reached
if len(files) > self.config.face_recognition.save_attempts:
os.unlink(os.path.join(folder, files[-1]))
if id not in self.person_face_history:
self.person_face_history[id] = []
self.person_face_history[id].append(
(sub_label, score, face_frame.shape[0] * face_frame.shape[1])
)
(weighted_sub_label, weighted_score) = self.weighted_average_by_area(
(weighted_sub_label, weighted_score) = self.weighted_average(
self.person_face_history[id]
)
@@ -297,6 +325,9 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
sub_label, score = res
if score <= self.face_config.unknown_score:
sub_label = "unknown"
return {"success": True, "score": score, "face_name": sub_label}
elif topic == EmbeddingsRequestEnum.register_face.value:
rand_id = "".join(
@@ -366,6 +397,9 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
sub_label, score = res
if score <= self.face_config.unknown_score:
sub_label = "unknown"
if self.config.face_recognition.save_attempts:
# write face to library
folder = os.path.join(FACE_DIR, "train")
@@ -375,38 +409,49 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
)
shutil.move(current_file, new_file)
files = sorted(
filter(lambda f: (f.endswith(".webp")), os.listdir(folder)),
key=lambda f: os.path.getctime(os.path.join(folder, f)),
reverse=True,
)
# delete oldest face image if maximum is reached
if len(files) > self.config.face_recognition.save_attempts:
os.unlink(os.path.join(folder, files[-1]))
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]]):
score_count = {}
def weighted_average(
self, results_list: list[tuple[str, float, int]], max_weight: int = 4000
):
"""
Calculates a robust weighted average, capping the area weight and giving more weight to higher scores.
Args:
results_list: A list of tuples, where each tuple contains (name, score, face_area).
max_weight: The maximum weight to apply based on face area.
Returns:
A tuple containing the prominent name and its weighted average score, or (None, 0.0) if the list is empty.
"""
if not results_list:
return None, 0.0
weighted_scores = {}
total_face_areas = {}
total_weights = {}
for name, score, face_area in results_list:
if name == "unknown":
continue
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
total_weights[name] = 0.0
weighted_scores[name] += score * face_area
total_face_areas[name] += face_area
# Capped weight based on face area
weight = min(face_area, max_weight)
prominent_name = max(score_count)
# Score-based weighting (higher scores get more weight)
weight *= (score - self.face_config.unknown_score) * 10
weighted_scores[name] += score * weight
total_weights[name] += weight
return prominent_name, weighted_scores[prominent_name] / total_face_areas[
prominent_name
]
if not weighted_scores:
return None, 0.0
best_name = max(weighted_scores, key=weighted_scores.get)
weighted_average = weighted_scores[best_name] / total_weights[best_name]
return best_name, weighted_average