diff --git a/frigate/data_processing/common/face/detector.py b/frigate/data_processing/common/face/detector.py new file mode 100644 index 0000000000..3d890ca3e4 --- /dev/null +++ b/frigate/data_processing/common/face/detector.py @@ -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], + ) + ) diff --git a/frigate/data_processing/common/face/model.py b/frigate/data_processing/common/face/recognizer.py similarity index 72% rename from frigate/data_processing/common/face/model.py rename to frigate/data_processing/common/face/recognizer.py index 87293f7f02..3714995209 100644 --- a/frigate/data_processing/common/face/model.py +++ b/frigate/data_processing/common/face/recognizer.py @@ -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 diff --git a/frigate/data_processing/real_time/face.py b/frigate/data_processing/real_time/face.py index a55c2566e1..3ccd4327fd 100644 --- a/frigate/data_processing/real_time/face.py +++ b/frigate/data_processing/real_time/face.py @@ -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] diff --git a/frigate/embeddings/__init__.py b/frigate/embeddings/__init__.py index 61e9f45fb5..a5f9f708d2 100644 --- a/frigate/embeddings/__init__.py +++ b/frigate/embeddings/__init__.py @@ -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, diff --git a/frigate/log.py b/frigate/log.py index 3c5f8ec8b8..693d70e8c9 100644 --- a/frigate/log.py +++ b/frigate/log.py @@ -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 diff --git a/frigate/test/test_face_detector.py b/frigate/test/test_face_detector.py new file mode 100644 index 0000000000..2447f5b246 --- /dev/null +++ b/frigate/test/test_face_detector.py @@ -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) + )