mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-09-24 18:26:51 +03:00
Revamp Face Recognition (#24236)
* Rename face model to face recognizer * Refactor face detection into own module * Return face and face landmarks * Align faces with 5 points instead of just the eyes * Add landmark validation to throw out images which do not fit a landmark * Fix circular import and lock face detector for concurrent runs across threads * Fix mypy
This commit is contained in:
Nicolas Mowen
@@ -0,0 +1,278 @@
|
||||
"""Handle face detection."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from frigate.const import MODEL_CACHE_DIR
|
||||
from frigate.log import redirect_output_to_logger
|
||||
from frigate.util.image import area
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MAX_DETECTION_HEIGHT = 1080
|
||||
|
||||
FACE_DET_DIR = os.path.join(MODEL_CACHE_DIR, "facedet")
|
||||
|
||||
# 5 point template the arcface models are trained on, defined against a 112x112
|
||||
# crop and scaled to whatever size the embedding model takes
|
||||
FACE_TEMPLATE_SIZE = 112
|
||||
FACE_TEMPLATE = 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,
|
||||
)
|
||||
|
||||
|
||||
# landmarks further than this from a plausible face shape are not trusted. on a
|
||||
# sample of camera face crops every set that failed a basic eye, nose, and mouth
|
||||
# ordering check scored above 9.5 and every set that passed scored below 9.2
|
||||
MAX_LANDMARK_FIT_ERROR = 9.0
|
||||
|
||||
|
||||
def landmark_fit_error(landmarks: tuple[tuple[float, float], ...]) -> float:
|
||||
"""Mean distance in template pixels once landmarks are fit to the template.
|
||||
|
||||
Scale, rotation, and position are fit out, so this measures only how far
|
||||
the landmarks are from a plausible face shape.
|
||||
"""
|
||||
src = np.array(landmarks, dtype=np.float32)
|
||||
matrix, _ = cv2.estimateAffinePartial2D(src, FACE_TEMPLATE, method=cv2.LMEDS)
|
||||
|
||||
if matrix is None:
|
||||
return float("inf") # type: ignore[unreachable]
|
||||
|
||||
fit = src @ matrix[:, :2].T + matrix[:, 2]
|
||||
return float(np.linalg.norm(fit - FACE_TEMPLATE, axis=1).mean())
|
||||
|
||||
|
||||
@dataclass
|
||||
class DetectionResult:
|
||||
"""A face detected by the face detector."""
|
||||
|
||||
# (x1, y1, x2, y2)
|
||||
face: tuple[int, int, int, int]
|
||||
|
||||
# eyes, nose tip, and mouth corners as (x, y) pairs, each pair ordered left
|
||||
# to right in image coordinates to match the arcface template. kept as
|
||||
# floats for sub pixel alignment accuracy
|
||||
landmarks: tuple[tuple[float, float], ...]
|
||||
|
||||
|
||||
class FaceDetector:
|
||||
"""Face detection runner."""
|
||||
|
||||
def __init__(self, on_ready: Callable[[], None] | None = None) -> None:
|
||||
self.detector: cv2.FaceDetectorYN | None = None
|
||||
self.landmark_detector: cv2.face.Facemark | None = None
|
||||
self.on_ready = on_ready
|
||||
|
||||
# both models hold internal state across a call, and the recognizer
|
||||
# builds its class means on a background thread while frames are
|
||||
# still being processed, so calls into them are serialized
|
||||
self.lock = threading.Lock()
|
||||
|
||||
GITHUB_ENDPOINT = os.environ.get("GITHUB_ENDPOINT", "https://github.com")
|
||||
|
||||
self.model_files = {
|
||||
"facedet.onnx": f"{GITHUB_ENDPOINT}/NickM-27/facenet-onnx/releases/download/v1.0/facedet.onnx",
|
||||
"landmarkdet.yaml": f"{GITHUB_ENDPOINT}/NickM-27/facenet-onnx/releases/download/v1.0/landmarkdet.yaml",
|
||||
}
|
||||
|
||||
if not all(
|
||||
os.path.exists(os.path.join(FACE_DET_DIR, n))
|
||||
for n in self.model_files.keys()
|
||||
):
|
||||
# conditionally import ModelDownloader
|
||||
from frigate.util.downloader import ModelDownloader
|
||||
|
||||
self.downloader = ModelDownloader(
|
||||
model_name="facedet",
|
||||
download_path=FACE_DET_DIR,
|
||||
file_names=list(self.model_files.keys()),
|
||||
download_func=self.__download_models,
|
||||
complete_func=self.__build_detector,
|
||||
)
|
||||
self.downloader.ensure_model_files()
|
||||
else:
|
||||
self.__build_detector()
|
||||
|
||||
def __download_models(self, path: str) -> None:
|
||||
try:
|
||||
file_name = os.path.basename(path)
|
||||
# conditionally import ModelDownloader
|
||||
from frigate.util.downloader import ModelDownloader
|
||||
|
||||
ModelDownloader.download_from_url(self.model_files[file_name], path)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to download {path}: {e}")
|
||||
|
||||
def __build_detector(self) -> None:
|
||||
self.detector = cv2.FaceDetectorYN.create(
|
||||
os.path.join(FACE_DET_DIR, "facedet.onnx"),
|
||||
config="",
|
||||
input_size=(320, 320),
|
||||
score_threshold=0.5,
|
||||
nms_threshold=0.3,
|
||||
)
|
||||
self.__init_landmark_detector()
|
||||
|
||||
if self.on_ready is not None:
|
||||
self.on_ready()
|
||||
|
||||
@property
|
||||
def is_ready(self) -> bool:
|
||||
"""Whether both the detection and landmark models are loaded."""
|
||||
return self.detector is not None and self.landmark_detector is not None
|
||||
|
||||
@redirect_output_to_logger(logger, logging.DEBUG)
|
||||
def __init_landmark_detector(self) -> None:
|
||||
landmark_model = os.path.join(FACE_DET_DIR, "landmarkdet.yaml")
|
||||
|
||||
if os.path.exists(landmark_model):
|
||||
landmark_detector = cv2.face.createFacemarkLBF()
|
||||
landmark_detector.loadModel(landmark_model)
|
||||
self.landmark_detector = landmark_detector
|
||||
|
||||
def detect(self, input: np.ndarray, threshold: float) -> DetectionResult | None:
|
||||
"""Detect the largest face in the input image.
|
||||
|
||||
Args:
|
||||
input: The image to run detection on
|
||||
threshold: Minimum detection confidence to accept a face
|
||||
|
||||
Returns:
|
||||
The largest detected face with its landmarks, or None
|
||||
"""
|
||||
if not self.detector:
|
||||
return None
|
||||
|
||||
height, width = input.shape[:2]
|
||||
|
||||
# YN face detector fails at extreme definitions
|
||||
# this rescales to a size that can properly detect faces
|
||||
# still retaining plenty of detail
|
||||
if height > MAX_DETECTION_HEIGHT:
|
||||
scale_factor = MAX_DETECTION_HEIGHT / height
|
||||
new_width = int(scale_factor * width)
|
||||
input = cv2.resize(input, (new_width, MAX_DETECTION_HEIGHT))
|
||||
else:
|
||||
scale_factor = 1
|
||||
|
||||
with self.lock:
|
||||
self.detector.setInputSize((input.shape[1], input.shape[0]))
|
||||
faces = self.detector.detect(input)
|
||||
|
||||
if faces is None or faces[1] is None:
|
||||
return None # type: ignore[unreachable]
|
||||
|
||||
best: DetectionResult | None = None
|
||||
best_area = 0
|
||||
|
||||
for potential_face in faces[1]:
|
||||
if potential_face[-1] < threshold:
|
||||
continue
|
||||
|
||||
# YuNet reports floats outside of the image for cut off faces, the
|
||||
# far edges are derived before clamping so they don't move with the
|
||||
# clamped near edges
|
||||
raw_x = float(potential_face[0]) / scale_factor
|
||||
raw_y = float(potential_face[1]) / scale_factor
|
||||
bbox = (
|
||||
max(int(raw_x), 0),
|
||||
max(int(raw_y), 0),
|
||||
min(int(raw_x + float(potential_face[2]) / scale_factor), width),
|
||||
min(int(raw_y + float(potential_face[3]) / scale_factor), height),
|
||||
)
|
||||
bbox_area = area(bbox)
|
||||
|
||||
if bbox_area <= best_area:
|
||||
continue
|
||||
|
||||
# landmarks are left unclamped for a more accurate alignment fit
|
||||
best = DetectionResult(
|
||||
face=bbox,
|
||||
landmarks=tuple(
|
||||
(float(x) / scale_factor, float(y) / scale_factor)
|
||||
for x, y in potential_face[4:14].reshape(5, 2)
|
||||
),
|
||||
)
|
||||
best_area = bbox_area
|
||||
|
||||
return best
|
||||
|
||||
def get_face_landmarks(
|
||||
self, input: np.ndarray, threshold: float = 0.5
|
||||
) -> tuple[tuple[float, float], ...] | None:
|
||||
"""Get the alignment landmarks for an image that is already a face crop.
|
||||
|
||||
Args:
|
||||
input: The face crop to get landmarks for
|
||||
threshold: Minimum detection confidence to accept a face
|
||||
|
||||
Returns:
|
||||
Eye, nose, and mouth landmarks, or None
|
||||
"""
|
||||
detection = self.detect(input, threshold)
|
||||
|
||||
if (
|
||||
detection is not None
|
||||
and landmark_fit_error(detection.landmarks) <= MAX_LANDMARK_FIT_ERROR
|
||||
):
|
||||
return detection.landmarks
|
||||
|
||||
# detection either failed, which is common on a crop that is already
|
||||
# tight around the face, or returned landmarks that are not shaped like
|
||||
# a face, so the landmark model is given the whole crop as the face
|
||||
landmarks = self.__fit_landmarks(input)
|
||||
|
||||
if landmarks is None or landmark_fit_error(landmarks) > MAX_LANDMARK_FIT_ERROR:
|
||||
return None
|
||||
|
||||
return landmarks
|
||||
|
||||
def __fit_landmarks(
|
||||
self, input: np.ndarray
|
||||
) -> tuple[tuple[float, float], ...] | None:
|
||||
"""Derive the 5 alignment landmarks from the 68 point landmark model."""
|
||||
if self.landmark_detector is None:
|
||||
return None
|
||||
|
||||
# the landmark model runs on grayscale
|
||||
gray = cv2.cvtColor(input, cv2.COLOR_BGR2GRAY) if input.ndim == 3 else input
|
||||
|
||||
try:
|
||||
with self.lock:
|
||||
success, faces = self.landmark_detector.fit(
|
||||
gray, np.array([(0, 0, gray.shape[1], gray.shape[0])])
|
||||
)
|
||||
except cv2.error:
|
||||
logger.debug("Failed to fit landmarks")
|
||||
return None
|
||||
|
||||
if not success or not len(faces):
|
||||
return None
|
||||
|
||||
points = faces[0][0]
|
||||
|
||||
# each eye is the mean of the 6 points around it
|
||||
return tuple(
|
||||
(float(p[0]), float(p[1]))
|
||||
for p in (
|
||||
points[36:42].mean(axis=0),
|
||||
points[42:48].mean(axis=0),
|
||||
points[30],
|
||||
points[48],
|
||||
points[54],
|
||||
)
|
||||
)
|
||||
+68
-87
@@ -9,9 +9,18 @@ import numpy as np
|
||||
from scipy import stats
|
||||
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.const import FACE_DIR, MODEL_CACHE_DIR
|
||||
from frigate.embeddings.onnx.face_embedding import ArcfaceEmbedding, FaceNetEmbedding
|
||||
from frigate.log import redirect_output_to_logger
|
||||
from frigate.const import FACE_DIR
|
||||
from frigate.data_processing.common.face.detector import (
|
||||
FACE_TEMPLATE,
|
||||
FACE_TEMPLATE_SIZE,
|
||||
FaceDetector,
|
||||
)
|
||||
from frigate.embeddings.onnx.face_embedding import (
|
||||
ARCFACE_INPUT_SIZE,
|
||||
FACENET_INPUT_SIZE,
|
||||
ArcfaceEmbedding,
|
||||
FaceNetEmbedding,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -19,10 +28,9 @@ logger = logging.getLogger(__name__)
|
||||
class FaceRecognizer(ABC):
|
||||
"""Face recognition runner."""
|
||||
|
||||
def __init__(self, config: FrigateConfig) -> None:
|
||||
def __init__(self, config: FrigateConfig, detector: FaceDetector) -> None:
|
||||
self.config = config
|
||||
self.landmark_detector: cv2.face.Facemark | None = None
|
||||
self.init_landmark_detector()
|
||||
self.detector = detector
|
||||
|
||||
@abstractmethod
|
||||
def build(self) -> None:
|
||||
@@ -38,79 +46,38 @@ class FaceRecognizer(ABC):
|
||||
def classify(self, face_image: np.ndarray) -> tuple[str, float] | None:
|
||||
pass
|
||||
|
||||
@redirect_output_to_logger(logger, logging.DEBUG) # type: ignore[misc]
|
||||
def init_landmark_detector(self) -> None:
|
||||
landmark_model = os.path.join(MODEL_CACHE_DIR, "facedet/landmarkdet.yaml")
|
||||
def align_face(self, image: np.ndarray, output_size: int) -> np.ndarray | None:
|
||||
"""Warp a face onto the template the embedding model was trained on.
|
||||
|
||||
if os.path.exists(landmark_model):
|
||||
landmark_detector = cv2.face.createFacemarkLBF()
|
||||
landmark_detector.loadModel(landmark_model)
|
||||
self.landmark_detector = landmark_detector
|
||||
Args:
|
||||
image: The face crop to align
|
||||
output_size: Width and height of the model input
|
||||
|
||||
def align_face(
|
||||
self,
|
||||
image: np.ndarray,
|
||||
output_width: int,
|
||||
output_height: int,
|
||||
) -> np.ndarray:
|
||||
if not self.landmark_detector:
|
||||
raise ValueError("Landmark detector not initialized")
|
||||
Returns:
|
||||
The aligned face, or None if it could not be aligned
|
||||
"""
|
||||
landmarks = self.detector.get_face_landmarks(image)
|
||||
|
||||
# landmark is run on grayscale images
|
||||
if image.ndim == 3:
|
||||
land_image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
|
||||
else:
|
||||
land_image = image
|
||||
if landmarks is None:
|
||||
return None
|
||||
|
||||
_, lands = self.landmark_detector.fit(
|
||||
land_image, np.array([(0, 0, land_image.shape[1], land_image.shape[0])])
|
||||
# fitting all 5 points constrains rotation, scale, and position, an eye
|
||||
# line alone leaves them free to slip on the small faces from a camera
|
||||
matrix, _ = cv2.estimateAffinePartial2D(
|
||||
np.array(landmarks, dtype=np.float32),
|
||||
FACE_TEMPLATE * (output_size / FACE_TEMPLATE_SIZE),
|
||||
method=cv2.LMEDS,
|
||||
)
|
||||
landmarks: np.ndarray = lands[0][0]
|
||||
|
||||
# get landmarks for eyes
|
||||
leftEyePts = landmarks[42:48]
|
||||
rightEyePts = landmarks[36:42]
|
||||
# the fit fails on degenerate landmarks even though the stub says
|
||||
# otherwise, for example when every point collapses onto one pixel
|
||||
if matrix is None:
|
||||
return None # type: ignore[unreachable]
|
||||
|
||||
# compute the center of mass for each eye
|
||||
leftEyeCenter = leftEyePts.mean(axis=0).astype("int")
|
||||
rightEyeCenter = rightEyePts.mean(axis=0).astype("int")
|
||||
|
||||
# compute the angle between the eye centroids
|
||||
dY = rightEyeCenter[1] - leftEyeCenter[1]
|
||||
dX = rightEyeCenter[0] - leftEyeCenter[0]
|
||||
angle = np.degrees(np.arctan2(dY, dX)) - 180
|
||||
|
||||
# compute the desired right eye x-coordinate based on the
|
||||
# desired x-coordinate of the left eye
|
||||
desiredRightEyeX = 1.0 - 0.35
|
||||
|
||||
# determine the scale of the new resulting image by taking
|
||||
# the ratio of the distance between eyes in the *current*
|
||||
# image to the ratio of distance between eyes in the
|
||||
# *desired* image
|
||||
dist = np.sqrt((dX**2) + (dY**2))
|
||||
desiredDist = desiredRightEyeX - 0.35
|
||||
desiredDist *= output_width
|
||||
scale = desiredDist / dist
|
||||
|
||||
# compute center (x, y)-coordinates (i.e., the median point)
|
||||
# between the two eyes in the input image
|
||||
# grab the rotation matrix for rotating and scaling the face
|
||||
eyesCenter = (
|
||||
int((leftEyeCenter[0] + rightEyeCenter[0]) // 2),
|
||||
int((leftEyeCenter[1] + rightEyeCenter[1]) // 2),
|
||||
)
|
||||
M = cv2.getRotationMatrix2D(eyesCenter, angle, scale)
|
||||
|
||||
# update the translation component of the matrix
|
||||
tX = output_width * 0.5
|
||||
tY = output_height * 0.35
|
||||
M[0, 2] += tX - eyesCenter[0]
|
||||
M[1, 2] += tY - eyesCenter[1]
|
||||
|
||||
# apply the affine transformation
|
||||
# the output is already the model input size, so the embedder's resize
|
||||
# and letterbox padding are a no op
|
||||
return cv2.warpAffine(
|
||||
image, M, (output_width, output_height), flags=cv2.INTER_CUBIC
|
||||
image, matrix, (output_size, output_size), flags=cv2.INTER_CUBIC
|
||||
)
|
||||
|
||||
def get_blur_confidence_reduction(self, input: np.ndarray) -> float:
|
||||
@@ -217,8 +184,8 @@ def similarity_to_confidence(
|
||||
|
||||
|
||||
class FaceNetRecognizer(FaceRecognizer):
|
||||
def __init__(self, config: FrigateConfig):
|
||||
super().__init__(config)
|
||||
def __init__(self, config: FrigateConfig, detector: FaceDetector):
|
||||
super().__init__(config, detector)
|
||||
self.mean_embs: dict[str, np.ndarray] = {}
|
||||
self.face_embedder: FaceNetEmbedding = FaceNetEmbedding()
|
||||
self.model_builder_queue: queue.Queue | None = None
|
||||
@@ -250,8 +217,12 @@ class FaceNetRecognizer(FaceRecognizer):
|
||||
if img is None:
|
||||
continue # type: ignore[unreachable]
|
||||
|
||||
img = self.align_face(img, img.shape[1], img.shape[0])
|
||||
emb = self.face_embedder([img])[0].squeeze()
|
||||
aligned = self.align_face(img, FACENET_INPUT_SIZE)
|
||||
|
||||
if aligned is None:
|
||||
continue
|
||||
|
||||
emb = self.face_embedder([aligned])[0].squeeze()
|
||||
face_embeddings_map[name].append(emb)
|
||||
|
||||
idx += 1
|
||||
@@ -263,8 +234,7 @@ class FaceNetRecognizer(FaceRecognizer):
|
||||
thread.start()
|
||||
|
||||
def build(self) -> None:
|
||||
if not self.landmark_detector:
|
||||
self.init_landmark_detector()
|
||||
if not self.detector.is_ready:
|
||||
return None
|
||||
|
||||
if self.model_builder_queue is not None:
|
||||
@@ -289,7 +259,7 @@ class FaceNetRecognizer(FaceRecognizer):
|
||||
logger.debug("Finished building ArcFace model")
|
||||
|
||||
def classify(self, face_image: np.ndarray) -> tuple[str, float] | None:
|
||||
if not self.landmark_detector:
|
||||
if not self.detector.is_ready:
|
||||
return None
|
||||
|
||||
if not self.mean_embs:
|
||||
@@ -304,7 +274,11 @@ class FaceNetRecognizer(FaceRecognizer):
|
||||
blur_reduction = self.get_blur_confidence_reduction(face_image)
|
||||
|
||||
# align face and run recognition
|
||||
img = self.align_face(face_image, face_image.shape[1], face_image.shape[0])
|
||||
img = self.align_face(face_image, FACENET_INPUT_SIZE)
|
||||
|
||||
if img is None:
|
||||
return None
|
||||
|
||||
embedding = self.face_embedder([img])[0].squeeze()
|
||||
|
||||
score: float = 0
|
||||
@@ -328,8 +302,8 @@ class FaceNetRecognizer(FaceRecognizer):
|
||||
|
||||
|
||||
class ArcFaceRecognizer(FaceRecognizer):
|
||||
def __init__(self, config: FrigateConfig):
|
||||
super().__init__(config)
|
||||
def __init__(self, config: FrigateConfig, detector: FaceDetector):
|
||||
super().__init__(config, detector)
|
||||
self.mean_embs: dict[str, np.ndarray] = {}
|
||||
self.face_embedder: ArcfaceEmbedding = ArcfaceEmbedding(config.face_recognition)
|
||||
self.model_builder_queue: queue.Queue | None = None
|
||||
@@ -361,8 +335,12 @@ class ArcFaceRecognizer(FaceRecognizer):
|
||||
if img is None:
|
||||
continue # type: ignore[unreachable]
|
||||
|
||||
img = self.align_face(img, img.shape[1], img.shape[0])
|
||||
emb = self.face_embedder([img])[0].squeeze() # type: ignore[arg-type]
|
||||
aligned = self.align_face(img, ARCFACE_INPUT_SIZE)
|
||||
|
||||
if aligned is None:
|
||||
continue
|
||||
|
||||
emb = self.face_embedder([aligned])[0].squeeze() # type: ignore[arg-type]
|
||||
face_embeddings_map[name].append(emb)
|
||||
|
||||
idx += 1
|
||||
@@ -374,8 +352,7 @@ class ArcFaceRecognizer(FaceRecognizer):
|
||||
thread.start()
|
||||
|
||||
def build(self) -> None:
|
||||
if not self.landmark_detector:
|
||||
self.init_landmark_detector()
|
||||
if not self.detector.is_ready:
|
||||
return None
|
||||
|
||||
if self.model_builder_queue is not None:
|
||||
@@ -400,7 +377,7 @@ class ArcFaceRecognizer(FaceRecognizer):
|
||||
logger.debug("Finished building ArcFace model")
|
||||
|
||||
def classify(self, face_image: np.ndarray) -> tuple[str, float] | None:
|
||||
if not self.landmark_detector:
|
||||
if not self.detector.is_ready:
|
||||
return None
|
||||
|
||||
if not self.mean_embs:
|
||||
@@ -415,7 +392,11 @@ class ArcFaceRecognizer(FaceRecognizer):
|
||||
blur_reduction = self.get_blur_confidence_reduction(face_image)
|
||||
|
||||
# align face and run recognition
|
||||
img = self.align_face(face_image, face_image.shape[1], face_image.shape[0])
|
||||
img = self.align_face(face_image, ARCFACE_INPUT_SIZE)
|
||||
|
||||
if img is None:
|
||||
return None
|
||||
|
||||
embedding = self.face_embedder([img])[0].squeeze() # type: ignore[arg-type]
|
||||
|
||||
score: float = 0
|
||||
@@ -19,8 +19,9 @@ from frigate.comms.event_metadata_updater import (
|
||||
)
|
||||
from frigate.comms.inter_process import InterProcessRequestor
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.const import FACE_DIR, MODEL_CACHE_DIR
|
||||
from frigate.data_processing.common.face.model import (
|
||||
from frigate.const import FACE_DIR
|
||||
from frigate.data_processing.common.face.detector import FaceDetector
|
||||
from frigate.data_processing.common.face.recognizer import (
|
||||
ArcFaceRecognizer,
|
||||
FaceNetRecognizer,
|
||||
FaceRecognizer,
|
||||
@@ -36,7 +37,6 @@ from .api import RealTimeProcessorApi
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
MAX_DETECTION_HEIGHT = 1080
|
||||
MAX_FACES_ATTEMPTS_AFTER_REC = 6
|
||||
MAX_FACE_ATTEMPTS = 12
|
||||
|
||||
@@ -53,7 +53,6 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
self.face_config = config.face_recognition
|
||||
self.requestor = requestor
|
||||
self.sub_label_publisher = sub_label_publisher
|
||||
self.face_detector: cv2.FaceDetectorYN | None = None
|
||||
self.requires_face_detection = "face" not in self.config.objects.all_objects
|
||||
self.person_face_history: dict[str, list[tuple[str, float, int]]] = {}
|
||||
self.camera_current_people: dict[str, list[str]] = {}
|
||||
@@ -61,38 +60,14 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
self.faces_per_second = EventsPerSecond()
|
||||
self.inference_speed = InferenceSpeed(self.metrics.face_rec_speed)
|
||||
|
||||
GITHUB_ENDPOINT = os.environ.get("GITHUB_ENDPOINT", "https://github.com")
|
||||
|
||||
download_path = os.path.join(MODEL_CACHE_DIR, "facedet")
|
||||
self.model_files = {
|
||||
"facedet.onnx": f"{GITHUB_ENDPOINT}/NickM-27/facenet-onnx/releases/download/v1.0/facedet.onnx",
|
||||
"landmarkdet.yaml": f"{GITHUB_ENDPOINT}/NickM-27/facenet-onnx/releases/download/v1.0/landmarkdet.yaml",
|
||||
}
|
||||
|
||||
if not all(
|
||||
os.path.exists(os.path.join(download_path, n))
|
||||
for n in self.model_files.keys()
|
||||
):
|
||||
# conditionally import ModelDownloader
|
||||
from frigate.util.downloader import ModelDownloader
|
||||
|
||||
self.downloader = ModelDownloader(
|
||||
model_name="facedet",
|
||||
download_path=download_path,
|
||||
file_names=list(self.model_files.keys()),
|
||||
download_func=self.__download_models,
|
||||
complete_func=self.__build_detector,
|
||||
)
|
||||
self.downloader.ensure_model_files()
|
||||
else:
|
||||
self.__build_detector()
|
||||
self.face_detector = FaceDetector(on_ready=self.faces_per_second.start)
|
||||
|
||||
self.label_map: dict[int, str] = {}
|
||||
|
||||
if self.face_config.model_size == "small":
|
||||
self.recognizer = FaceNetRecognizer(self.config)
|
||||
self.recognizer = FaceNetRecognizer(self.config, self.face_detector)
|
||||
else:
|
||||
self.recognizer = ArcFaceRecognizer(self.config)
|
||||
self.recognizer = ArcFaceRecognizer(self.config, self.face_detector)
|
||||
|
||||
self.recognizer.build()
|
||||
|
||||
@@ -113,67 +88,6 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
|
||||
logger.debug("Face recognition config updated dynamically")
|
||||
|
||||
def __download_models(self, path: str) -> None:
|
||||
try:
|
||||
file_name = os.path.basename(path)
|
||||
# conditionally import ModelDownloader
|
||||
from frigate.util.downloader import ModelDownloader
|
||||
|
||||
ModelDownloader.download_from_url(self.model_files[file_name], path)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to download {path}: {e}")
|
||||
|
||||
def __build_detector(self) -> None:
|
||||
self.face_detector = cv2.FaceDetectorYN.create(
|
||||
os.path.join(MODEL_CACHE_DIR, "facedet/facedet.onnx"),
|
||||
config="",
|
||||
input_size=(320, 320),
|
||||
score_threshold=0.5,
|
||||
nms_threshold=0.3,
|
||||
)
|
||||
self.faces_per_second.start()
|
||||
|
||||
def __detect_face(
|
||||
self, input: np.ndarray, threshold: float
|
||||
) -> tuple[int, int, int, int] | None:
|
||||
"""Detect faces in input image."""
|
||||
if not self.face_detector:
|
||||
return None
|
||||
|
||||
# YN face detector fails at extreme definitions
|
||||
# this rescales to a size that can properly detect faces
|
||||
# still retaining plenty of detail
|
||||
if input.shape[0] > MAX_DETECTION_HEIGHT:
|
||||
scale_factor = MAX_DETECTION_HEIGHT / input.shape[0]
|
||||
new_width = int(scale_factor * input.shape[1])
|
||||
input = cv2.resize(input, (new_width, MAX_DETECTION_HEIGHT))
|
||||
else:
|
||||
scale_factor = 1
|
||||
|
||||
self.face_detector.setInputSize((input.shape[1], input.shape[0]))
|
||||
faces = self.face_detector.detect(input)
|
||||
|
||||
if faces is None or faces[1] is None:
|
||||
return None # type: ignore[unreachable]
|
||||
|
||||
face = None
|
||||
|
||||
for _, potential_face in enumerate(faces[1]):
|
||||
if potential_face[-1] < threshold:
|
||||
continue
|
||||
|
||||
raw_bbox = potential_face[0:4].astype(np.uint16)
|
||||
x: int = int(max(raw_bbox[0], 0) / scale_factor)
|
||||
y: int = int(max(raw_bbox[1], 0) / scale_factor)
|
||||
w: int = int(raw_bbox[2] / scale_factor)
|
||||
h: int = int(raw_bbox[3] / scale_factor)
|
||||
bbox = (x, y, x + w, y + h)
|
||||
|
||||
if face is None or area(bbox) > area(face): # type: ignore[unreachable]
|
||||
face = bbox
|
||||
|
||||
return face
|
||||
|
||||
def __update_metrics(self, duration: float) -> None:
|
||||
self.faces_per_second.update()
|
||||
self.inference_speed.update(duration)
|
||||
@@ -221,6 +135,7 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
return
|
||||
|
||||
face: dict[str, Any] | None = None
|
||||
face_box: tuple[int, int, int, int]
|
||||
|
||||
if self.requires_face_detection:
|
||||
logger.debug("Running manual face detection.")
|
||||
@@ -234,12 +149,15 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
bgr = cv2.cvtColor(frame, cv2.COLOR_YUV2BGR_I420)
|
||||
left, top, right, bottom = person_box
|
||||
person = bgr[top:bottom, left:right]
|
||||
face_box = self.__detect_face(person, self.face_config.detection_threshold)
|
||||
detection = self.face_detector.detect(
|
||||
person, self.face_config.detection_threshold
|
||||
)
|
||||
|
||||
if not face_box:
|
||||
if detection is None:
|
||||
logger.debug("Detected no faces for person object.")
|
||||
return
|
||||
|
||||
face_box = detection.face
|
||||
face_frame = person[
|
||||
max(0, face_box[1]) : min(frame.shape[0], face_box[3]),
|
||||
max(0, face_box[0]) : min(frame.shape[1], face_box[2]),
|
||||
@@ -271,17 +189,19 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
logger.debug(f"No face attributes found for {id}")
|
||||
return
|
||||
|
||||
face_box = face.get("box")
|
||||
attr_box = face.get("box")
|
||||
|
||||
# check that face is valid
|
||||
if (
|
||||
not face_box
|
||||
or area(face_box)
|
||||
not attr_box
|
||||
or area(attr_box)
|
||||
< self.config.cameras[camera].face_recognition.min_area
|
||||
):
|
||||
logger.debug(f"Invalid face box {face}")
|
||||
return
|
||||
|
||||
face_box = attr_box
|
||||
|
||||
face_frame = cv2.cvtColor(frame, cv2.COLOR_YUV2BGR_I420)
|
||||
|
||||
face_frame = face_frame[
|
||||
@@ -364,11 +284,12 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
|
||||
# detect faces with lower confidence since we expect the face
|
||||
# to be visible in uploaded images
|
||||
face_box = self.__detect_face(img, 0.5)
|
||||
detection = self.face_detector.detect(img, 0.5)
|
||||
|
||||
if not face_box:
|
||||
if detection is None:
|
||||
return {"message": "No face was detected.", "success": False}
|
||||
|
||||
face_box = detection.face
|
||||
face = img[face_box[1] : face_box[3], face_box[0] : face_box[2]]
|
||||
res = self.recognizer.classify(face)
|
||||
|
||||
@@ -396,14 +317,15 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
|
||||
|
||||
# detect faces with lower confidence since we expect the face
|
||||
# to be visible in uploaded images
|
||||
face_box = self.__detect_face(img, 0.5)
|
||||
detection = self.face_detector.detect(img, 0.5)
|
||||
|
||||
if not face_box:
|
||||
if detection is None:
|
||||
return {
|
||||
"message": "No face was detected.",
|
||||
"success": False,
|
||||
}
|
||||
|
||||
face_box = detection.face
|
||||
face = img[face_box[1] : face_box[3], face_box[0] : face_box[2]]
|
||||
_, thumbnail = cv2.imencode(
|
||||
".webp", face, [int(cv2.IMWRITE_WEBP_QUALITY), 100]
|
||||
|
||||
@@ -24,7 +24,6 @@ from frigate.util.classification import kickoff_model_training
|
||||
from frigate.util.path import safe_join
|
||||
from frigate.util.process import FrigateProcess
|
||||
|
||||
from .maintainer import EmbeddingMaintainer
|
||||
from .util import ZScoreNormalization
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -47,6 +46,10 @@ class EmbeddingProcess(FrigateProcess):
|
||||
self.metrics = metrics
|
||||
|
||||
def run(self) -> None:
|
||||
# imported here so that importing this package does not pull in the
|
||||
# processors, which import back into it and form a cycle
|
||||
from .maintainer import EmbeddingMaintainer
|
||||
|
||||
self.pre_run_setup(self.config.logger)
|
||||
maintainer = EmbeddingMaintainer(
|
||||
self.config,
|
||||
|
||||
+7
-4
@@ -13,10 +13,13 @@ from functools import wraps
|
||||
from logging.handlers import QueueHandler, QueueListener
|
||||
from multiprocessing.managers import SyncManager
|
||||
from queue import Empty, Queue
|
||||
from typing import Any
|
||||
from typing import Any, TypeVar, cast
|
||||
|
||||
from frigate.util.builtin import clean_camera_user_pass
|
||||
|
||||
# lets a decorator keep the signature of the function it wraps
|
||||
_F = TypeVar("_F", bound=Callable[..., Any])
|
||||
|
||||
LOG_HANDLER = logging.StreamHandler()
|
||||
LOG_HANDLER.setFormatter(
|
||||
logging.Formatter(
|
||||
@@ -242,10 +245,10 @@ def __redirect_fd_to_queue(queue: Queue[str]) -> Generator[None, None, None]:
|
||||
pass
|
||||
|
||||
|
||||
def redirect_output_to_logger(logger: logging.Logger, level: int) -> Any:
|
||||
def redirect_output_to_logger(logger: logging.Logger, level: int) -> Callable[[_F], _F]:
|
||||
"""Decorator to redirect both Python sys.stdout/stderr and C-level stdout to logger."""
|
||||
|
||||
def decorator(func: Callable) -> Callable:
|
||||
def decorator(func: _F) -> _F:
|
||||
@wraps(func)
|
||||
def wrapper(*args: Any, **kwargs: Any) -> Any:
|
||||
queue: Queue[str] = Queue()
|
||||
@@ -275,7 +278,7 @@ def redirect_output_to_logger(logger: logging.Logger, level: int) -> Any:
|
||||
|
||||
return result
|
||||
|
||||
return wrapper
|
||||
return cast(_F, wrapper)
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
"""Tests for face detection results, landmark selection, and face alignment."""
|
||||
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from frigate.data_processing.common.face.detector import (
|
||||
FACE_TEMPLATE,
|
||||
FACE_TEMPLATE_SIZE,
|
||||
MAX_LANDMARK_FIT_ERROR,
|
||||
FaceDetector,
|
||||
landmark_fit_error,
|
||||
)
|
||||
from frigate.data_processing.common.face.recognizer import FaceRecognizer
|
||||
|
||||
# a real detection from a 27x30 crop where the left mouth corner landed above
|
||||
# the eye line, the case the fit error check exists to catch
|
||||
BROKEN_LANDMARKS = (
|
||||
(7.07, 8.98),
|
||||
(16.98, 9.59),
|
||||
(10.79, 15.78),
|
||||
(6.76, 6.50),
|
||||
(15.43, 21.05),
|
||||
)
|
||||
|
||||
|
||||
def _yunet_row(x: float, y: float, w: float, h: float, score: float = 0.9):
|
||||
"""Build a YuNet detection row of box, 5 landmarks, and score."""
|
||||
landmarks = [
|
||||
x + w * 0.3, y + h * 0.35,
|
||||
x + w * 0.7, y + h * 0.35,
|
||||
x + w * 0.5, y + h * 0.55,
|
||||
x + w * 0.35, y + h * 0.75,
|
||||
x + w * 0.65, y + h * 0.75,
|
||||
] # fmt: skip
|
||||
return np.array([x, y, w, h, *landmarks, score], dtype=np.float32)
|
||||
|
||||
|
||||
def _row_with_landmarks(landmarks, x=0.0, y=0.0, w=30.0, h=30.0):
|
||||
"""Build a YuNet detection row carrying specific landmarks."""
|
||||
flat = [v for point in landmarks for v in point]
|
||||
return np.array([x, y, w, h, *flat, 0.9], dtype=np.float32)
|
||||
|
||||
|
||||
def _detector(rows=None, lbf_points=None) -> FaceDetector:
|
||||
"""Build a detector with stubbed models, bypassing model downloads."""
|
||||
detector = FaceDetector.__new__(FaceDetector)
|
||||
detector.lock = threading.Lock()
|
||||
detector.detector = MagicMock()
|
||||
detector.detector.detect.return_value = (
|
||||
1,
|
||||
None if rows is None else np.array(rows, dtype=np.float32),
|
||||
)
|
||||
|
||||
if lbf_points is None:
|
||||
detector.landmark_detector = None
|
||||
else:
|
||||
detector.landmark_detector = MagicMock()
|
||||
detector.landmark_detector.fit.return_value = (
|
||||
True,
|
||||
[np.array([lbf_points], dtype=np.float32)],
|
||||
)
|
||||
|
||||
return detector
|
||||
|
||||
|
||||
class TestFaceBox(unittest.TestCase):
|
||||
"""The reported bug: YuNet returns floats outside of the image."""
|
||||
|
||||
def test_face_cut_off_at_near_edge_stays_inside_image(self):
|
||||
detector = _detector([_yunet_row(-6.4, -3.2, 50, 60)])
|
||||
|
||||
result = detector.detect(np.zeros((200, 200, 3), np.uint8), 0.5)
|
||||
|
||||
assert result is not None
|
||||
# clamping the near edges must not drag the far edges out with them
|
||||
self.assertEqual(result.face, (0, 0, 43, 56))
|
||||
|
||||
def test_face_past_far_edge_is_clamped_to_image(self):
|
||||
detector = _detector([_yunet_row(80, 70, 50, 60)])
|
||||
|
||||
result = detector.detect(np.zeros((100, 100, 3), np.uint8), 0.5)
|
||||
|
||||
assert result is not None
|
||||
self.assertEqual(result.face, (80, 70, 100, 100))
|
||||
|
||||
def test_box_and_landmarks_are_scaled_back_to_full_resolution(self):
|
||||
"""Tall images are downscaled for detection before being reported."""
|
||||
detector = _detector([_yunet_row(100, 200, 50, 60)])
|
||||
|
||||
result = detector.detect(np.zeros((2160, 400, 3), np.uint8), 0.5)
|
||||
|
||||
assert result is not None
|
||||
# detection runs at 1080 height, so results come back at half scale
|
||||
self.assertEqual(result.face, (200, 400, 300, 520))
|
||||
self.assertEqual(result.landmarks[0], (230.0, 442.0))
|
||||
|
||||
def test_largest_face_is_returned(self):
|
||||
detector = _detector([_yunet_row(0, 0, 20, 20), _yunet_row(50, 50, 60, 60)])
|
||||
|
||||
result = detector.detect(np.zeros((200, 200, 3), np.uint8), 0.5)
|
||||
|
||||
assert result is not None
|
||||
self.assertEqual(result.face, (50, 50, 110, 110))
|
||||
|
||||
|
||||
class TestLandmarkFitError(unittest.TestCase):
|
||||
def test_error_ignores_scale_rotation_and_position(self):
|
||||
"""The metric must only measure shape, so the threshold is meaningful."""
|
||||
angle = np.radians(20)
|
||||
rotate = np.array(
|
||||
[[np.cos(angle), -np.sin(angle)], [np.sin(angle), np.cos(angle)]],
|
||||
dtype=np.float32,
|
||||
)
|
||||
moved = (FACE_TEMPLATE * 3.7) @ rotate.T + np.array([250.0, -40.0])
|
||||
|
||||
self.assertLess(landmark_fit_error(tuple(map(tuple, FACE_TEMPLATE))), 0.01)
|
||||
self.assertLess(landmark_fit_error(tuple(map(tuple, moved))), 0.01)
|
||||
|
||||
def test_implausible_landmarks_score_above_the_threshold(self):
|
||||
self.assertGreater(landmark_fit_error(BROKEN_LANDMARKS), MAX_LANDMARK_FIT_ERROR)
|
||||
|
||||
|
||||
class TestLandmarkSelection(unittest.TestCase):
|
||||
"""get_face_landmarks prefers detection and falls back to the landmark model."""
|
||||
|
||||
def _lbf_points(self):
|
||||
# only the points the 5 point mapping reads have to be real
|
||||
points = np.zeros((68, 2), dtype=np.float32)
|
||||
points[36:42] = [10.0, 20.0]
|
||||
points[42:48] = [30.0, 20.0]
|
||||
points[30] = [20.0, 30.0]
|
||||
points[48] = [12.0, 40.0]
|
||||
points[54] = [28.0, 40.0]
|
||||
return points
|
||||
|
||||
def test_plausible_detection_landmarks_are_used(self):
|
||||
detector = _detector([_yunet_row(10, 20, 40, 50)], self._lbf_points())
|
||||
|
||||
result = detector.get_face_landmarks(np.zeros((200, 200, 3), np.uint8))
|
||||
|
||||
self.assertEqual(result[0], (22.0, 37.5))
|
||||
detector.landmark_detector.fit.assert_not_called()
|
||||
|
||||
def test_implausible_detection_landmarks_fall_back_to_landmark_model(self):
|
||||
detector = _detector(
|
||||
[_row_with_landmarks(BROKEN_LANDMARKS)], self._lbf_points()
|
||||
)
|
||||
|
||||
result = detector.get_face_landmarks(np.zeros((30, 27, 3), np.uint8))
|
||||
|
||||
# the 68 point model maps to eyes, nose tip, then mouth corners
|
||||
self.assertEqual(
|
||||
result,
|
||||
((10.0, 20.0), (30.0, 20.0), (20.0, 30.0), (12.0, 40.0), (28.0, 40.0)),
|
||||
)
|
||||
|
||||
def test_no_usable_landmarks_returns_none(self):
|
||||
detector = _detector([_row_with_landmarks(BROKEN_LANDMARKS)])
|
||||
|
||||
self.assertIsNone(detector.get_face_landmarks(np.zeros((30, 27, 3), np.uint8)))
|
||||
|
||||
|
||||
class _StubRecognizer(FaceRecognizer):
|
||||
"""Concrete recognizer so align_face can be exercised without a model."""
|
||||
|
||||
def build(self) -> None:
|
||||
pass
|
||||
|
||||
def clear(self) -> None:
|
||||
pass
|
||||
|
||||
def classify(self, face_image):
|
||||
return None
|
||||
|
||||
|
||||
class TestAlignFace(unittest.TestCase):
|
||||
def _recognizer(self, landmarks):
|
||||
detector = MagicMock()
|
||||
detector.get_face_landmarks.return_value = landmarks
|
||||
return _StubRecognizer(MagicMock(), detector)
|
||||
|
||||
def _assert_lands_on_template(self, output_size: int):
|
||||
# place the landmarks as a scaled and shifted copy of the template, so
|
||||
# a correct warp puts them back onto the template exactly
|
||||
source = tuple(map(tuple, FACE_TEMPLATE * 2.0 + np.array([60.0, 25.0])))
|
||||
recognizer = self._recognizer(source)
|
||||
|
||||
image = np.zeros((300, 300, 3), np.uint8)
|
||||
for x, y in source:
|
||||
cv2.circle(image, (int(round(x)), int(round(y))), 4, (255, 255, 255), -1)
|
||||
|
||||
aligned = recognizer.align_face(image, output_size)
|
||||
|
||||
assert aligned is not None
|
||||
self.assertEqual(aligned.shape, (output_size, output_size, 3))
|
||||
|
||||
gray = cv2.cvtColor(aligned, cv2.COLOR_BGR2GRAY)
|
||||
for x, y in FACE_TEMPLATE * (output_size / FACE_TEMPLATE_SIZE):
|
||||
window = gray[int(y) - 2 : int(y) + 3, int(x) - 2 : int(x) + 3]
|
||||
self.assertGreater(
|
||||
window.max(),
|
||||
200,
|
||||
f"no landmark near ({x:.0f}, {y:.0f}) at {output_size}",
|
||||
)
|
||||
|
||||
def test_landmarks_are_warped_onto_the_template(self):
|
||||
self._assert_lands_on_template(FACE_TEMPLATE_SIZE)
|
||||
|
||||
def test_template_is_scaled_to_the_model_input_size(self):
|
||||
"""The facenet model takes 160px, the template has to scale with it."""
|
||||
self._assert_lands_on_template(160)
|
||||
|
||||
def test_missing_landmarks_return_none(self):
|
||||
self.assertIsNone(
|
||||
self._recognizer(None).align_face(np.zeros((60, 60, 3), np.uint8), 112)
|
||||
)
|
||||
Reference in New Issue
Block a user