mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-07-30 23:59:02 +03:00
Josh HawkinsandGitHub
7ed7ed56cf
CI / AMD64 Build (push) Canceled after 0s
CI / ARM Build (push) Canceled after 0s
CI / Jetson Jetpack 6 (push) Canceled after 0s
CI / AMD64 Extra Build (push) Canceled after 0s
CI / ARM Extra Build (push) Canceled after 0s
CI / Synaptics Build (push) Canceled after 0s
CI / Assemble and push default build (push) Canceled after 0s
* fix watchdog process restarts reverting to the boot config /api/config/set parses a new FrigateConfig and swaps the API and dispatcher onto it, but FrigateApp.config was never rebound, so the watchdog factories rebuilt a crashed process from the config as of startup. Fix is to read through a ConfigHolder that the swap updates. * fix birdseye camera overrides being clobbered by a global mode change A global birdseye save published only the global object, leaving the output process to infer which cameras were inheriting by comparing against the previous global mode. That cannot tell an inherited value from an explicit one that happens to match, so it overwrote the override until a restart. Publish the per-camera values the config parse already resolved instead. * Reject non-finite numbers in GenAI review descriptions A model returning NaN for confidence or potential_threat_level slipped through the model_construct fallback, which skips validation, and was written into the review segment's JSON data. NaN is not valid JSON, so every subsequent /review request failed with "Out of range float values are not JSON compliant", blanking the review page for any time range containing the poisoned row. * restore fused DetectionOutput in the OpenVINO SSD model conversion * fix rgb swap issue for face dataset testing script
776 lines
28 KiB
Python
776 lines
28 KiB
Python
"""
|
|
Face recognition investigation script.
|
|
|
|
Standalone replica of Frigate's ArcFace pipeline (see
|
|
frigate/data_processing/common/face/model.py and
|
|
frigate/embeddings/onnx/face_embedding.py) for analyzing a face collection
|
|
outside the running service. Useful for:
|
|
|
|
- Diagnosing why a person's collection produces false positives
|
|
- Finding outlier/contaminating training images
|
|
- Inspecting the effect of the shipped vector-wise outlier filter
|
|
|
|
Layout:
|
|
- Core pipeline: LandmarkAligner, ArcFaceEmbedder, arcface_preprocess,
|
|
similarity_to_confidence, blur_reduction — all mirroring the production
|
|
code exactly
|
|
- Default run: summarize positive and negative sets against a baseline
|
|
trim_mean class representation
|
|
- Optional diagnostics (flags): vector-outlier filter behavior, degenerate
|
|
"tiny crop" embedding clustering, and multi-identity contamination
|
|
|
|
Usage:
|
|
python3 face_investigate.py \\
|
|
--positive <positive_folder> \\
|
|
--negative <negative_folder> \\
|
|
[--model-cache /path/to/model_cache] \\
|
|
[--vector-outlier] [--degenerate] [--contamination]
|
|
|
|
The positive folder should contain training images for a single identity
|
|
(same layout as FACE_DIR/<name>/*.webp). The negative folder should contain
|
|
runtime crops to test against — a mix of true matches and misfires.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import onnxruntime as ort
|
|
from PIL import Image
|
|
from scipy import stats
|
|
|
|
ARCFACE_INPUT_SIZE = 112
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Replicated Frigate pipeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _bgr_to_rgb(frame: np.ndarray) -> np.ndarray:
|
|
"""Mirror BaseEmbedding._bgr_to_rgb."""
|
|
if isinstance(frame, np.ndarray) and frame.ndim == 3:
|
|
return np.ascontiguousarray(frame[:, :, ::-1])
|
|
|
|
return frame
|
|
|
|
|
|
def _process_image_frigate(image: np.ndarray) -> Image.Image:
|
|
"""Mirror BaseEmbedding._process_image for an ndarray input.
|
|
|
|
`Image.fromarray` does not reorder channels, so whatever order it is
|
|
handed is what reaches the model. Callers swap to RGB first, exactly as
|
|
ArcfaceEmbedding._preprocess_inputs does.
|
|
"""
|
|
return Image.fromarray(image)
|
|
|
|
|
|
def arcface_preprocess(image_bgr: np.ndarray) -> np.ndarray:
|
|
"""Mirror ArcfaceEmbedding._preprocess_inputs.
|
|
|
|
Face crops arrive BGR from cv2 and #23712 added the swap to RGB before
|
|
embedding, so this script has to do it too.
|
|
"""
|
|
pil = _process_image_frigate(_bgr_to_rgb(image_bgr))
|
|
|
|
width, height = pil.size
|
|
if width != ARCFACE_INPUT_SIZE or height != ARCFACE_INPUT_SIZE:
|
|
if width > height:
|
|
new_height = int(((height / width) * ARCFACE_INPUT_SIZE) // 4 * 4)
|
|
pil = pil.resize((ARCFACE_INPUT_SIZE, new_height))
|
|
else:
|
|
new_width = int(((width / height) * ARCFACE_INPUT_SIZE) // 4 * 4)
|
|
pil = pil.resize((new_width, ARCFACE_INPUT_SIZE))
|
|
|
|
og = np.array(pil).astype(np.float32)
|
|
og_h, og_w, channels = og.shape
|
|
|
|
frame = np.zeros(
|
|
(ARCFACE_INPUT_SIZE, ARCFACE_INPUT_SIZE, channels), dtype=np.float32
|
|
)
|
|
x_center = (ARCFACE_INPUT_SIZE - og_w) // 2
|
|
y_center = (ARCFACE_INPUT_SIZE - og_h) // 2
|
|
frame[y_center : y_center + og_h, x_center : x_center + og_w] = og
|
|
|
|
frame = (frame / 127.5) - 1.0
|
|
frame = np.transpose(frame, (2, 0, 1))
|
|
frame = np.expand_dims(frame, axis=0)
|
|
return frame
|
|
|
|
|
|
class LandmarkAligner:
|
|
"""Mirror FaceRecognizer.align_face."""
|
|
|
|
def __init__(self, landmark_model_path: str):
|
|
if not os.path.exists(landmark_model_path):
|
|
raise FileNotFoundError(landmark_model_path)
|
|
self.detector = cv2.face.createFacemarkLBF()
|
|
self.detector.loadModel(landmark_model_path)
|
|
|
|
def align(
|
|
self, image: np.ndarray, out_w: int, out_h: int
|
|
) -> tuple[np.ndarray, dict]:
|
|
land_image = (
|
|
cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) if image.ndim == 3 else image
|
|
)
|
|
_, lands = self.detector.fit(
|
|
land_image, np.array([(0, 0, land_image.shape[1], land_image.shape[0])])
|
|
)
|
|
landmarks = lands[0][0]
|
|
|
|
leftEyePts = landmarks[42:48]
|
|
rightEyePts = landmarks[36:42]
|
|
leftEyeCenter = leftEyePts.mean(axis=0).astype("int")
|
|
rightEyeCenter = rightEyePts.mean(axis=0).astype("int")
|
|
|
|
dY = rightEyeCenter[1] - leftEyeCenter[1]
|
|
dX = rightEyeCenter[0] - leftEyeCenter[0]
|
|
angle = np.degrees(np.arctan2(dY, dX)) - 180
|
|
dist = float(np.sqrt((dX**2) + (dY**2)))
|
|
|
|
desiredRightEyeX = 1.0 - 0.35
|
|
desiredDist = (desiredRightEyeX - 0.35) * out_w
|
|
scale = desiredDist / dist if dist > 0 else 1.0
|
|
|
|
eyesCenter = (
|
|
int((leftEyeCenter[0] + rightEyeCenter[0]) // 2),
|
|
int((leftEyeCenter[1] + rightEyeCenter[1]) // 2),
|
|
)
|
|
M = cv2.getRotationMatrix2D(eyesCenter, angle, scale)
|
|
tX = out_w * 0.5
|
|
tY = out_h * 0.35
|
|
M[0, 2] += tX - eyesCenter[0]
|
|
M[1, 2] += tY - eyesCenter[1]
|
|
|
|
aligned = cv2.warpAffine(image, M, (out_w, out_h), flags=cv2.INTER_CUBIC)
|
|
info = dict(
|
|
angle=float(angle),
|
|
eye_dist_px=dist,
|
|
scale=float(scale),
|
|
landmarks=landmarks,
|
|
)
|
|
return aligned, info
|
|
|
|
|
|
class ArcFaceEmbedder:
|
|
def __init__(self, model_path: str):
|
|
self.session = ort.InferenceSession(
|
|
model_path, providers=["CPUExecutionProvider"]
|
|
)
|
|
self.input_name = self.session.get_inputs()[0].name
|
|
|
|
def embed(self, image_bgr: np.ndarray) -> np.ndarray:
|
|
tensor = arcface_preprocess(image_bgr)
|
|
out = self.session.run(None, {self.input_name: tensor})[0]
|
|
return out.squeeze()
|
|
|
|
|
|
def similarity_to_confidence(
|
|
cos_sim: float,
|
|
median: float = 0.3,
|
|
range_width: float = 0.6,
|
|
slope_factor: float = 12,
|
|
) -> float:
|
|
slope = slope_factor / range_width
|
|
return float(1.0 / (1.0 + np.exp(-slope * (cos_sim - median))))
|
|
|
|
|
|
def laplacian_variance(image: np.ndarray) -> float:
|
|
return float(cv2.Laplacian(image, cv2.CV_64F).var())
|
|
|
|
|
|
def blur_reduction(variance: float) -> float:
|
|
if variance < 120:
|
|
return 0.06
|
|
elif variance < 160:
|
|
return 0.04
|
|
elif variance < 200:
|
|
return 0.02
|
|
elif variance < 250:
|
|
return 0.01
|
|
return 0.0
|
|
|
|
|
|
def cosine(a: np.ndarray, b: np.ndarray) -> float:
|
|
denom = np.linalg.norm(a) * np.linalg.norm(b)
|
|
if denom == 0:
|
|
return 0.0
|
|
return float(np.dot(a, b) / denom)
|
|
|
|
|
|
def l2(v: np.ndarray) -> np.ndarray:
|
|
return v / (np.linalg.norm(v) + 1e-9)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class FaceSample:
|
|
path: str
|
|
shape: tuple[int, int]
|
|
embedding: np.ndarray
|
|
blur_var: float
|
|
align_info: dict
|
|
|
|
|
|
def load_folder(
|
|
folder: str, aligner: LandmarkAligner, embedder: ArcFaceEmbedder
|
|
) -> list[FaceSample]:
|
|
samples: list[FaceSample] = []
|
|
names = sorted(os.listdir(folder))
|
|
for name in names:
|
|
if name.startswith("."):
|
|
continue
|
|
path = os.path.join(folder, name)
|
|
if not os.path.isfile(path):
|
|
continue
|
|
img = cv2.imread(path)
|
|
if img is None:
|
|
print(f" [skip unreadable] {name}")
|
|
continue
|
|
aligned, info = aligner.align(img, img.shape[1], img.shape[0])
|
|
emb = embedder.embed(aligned)
|
|
samples.append(
|
|
FaceSample(
|
|
path=path,
|
|
shape=(img.shape[1], img.shape[0]),
|
|
embedding=emb,
|
|
blur_var=laplacian_variance(img),
|
|
align_info=info,
|
|
)
|
|
)
|
|
return samples
|
|
|
|
|
|
def trimmed_mean(embs: Iterable[np.ndarray], trim: float = 0.15) -> np.ndarray:
|
|
arr = np.stack(list(embs), axis=0)
|
|
return stats.trim_mean(arr, trim, axis=0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Baseline analyses (always run)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def summarize_positive(samples: list[FaceSample], mean_emb: np.ndarray) -> None:
|
|
"""Summary of training set: per-sample cos to class mean, intra-class stats.
|
|
|
|
Outliers with cos far below the rest are likely degrading the mean —
|
|
they'd be the first candidates the shipped vector-outlier filter drops.
|
|
"""
|
|
print("\n" + "=" * 78)
|
|
print(f"POSITIVE SET ANALYSIS ({len(samples)} images)")
|
|
print("=" * 78)
|
|
|
|
rows = []
|
|
for s in samples:
|
|
cs = cosine(s.embedding, mean_emb)
|
|
conf = similarity_to_confidence(cs)
|
|
red = blur_reduction(s.blur_var)
|
|
rows.append(
|
|
dict(
|
|
name=os.path.basename(s.path),
|
|
shape=f"{s.shape[0]}x{s.shape[1]}",
|
|
eye_px=s.align_info["eye_dist_px"],
|
|
angle=s.align_info["angle"] + 180,
|
|
blur=s.blur_var,
|
|
cos=cs,
|
|
conf=conf,
|
|
red=red,
|
|
adj_conf=max(0.0, conf - red),
|
|
)
|
|
)
|
|
|
|
rows.sort(key=lambda r: r["cos"])
|
|
sims = np.array([r["cos"] for r in rows])
|
|
print(
|
|
f"\nCosine-to-trimmed-mean: mean={sims.mean():.3f} std={sims.std():.3f} "
|
|
f"min={sims.min():.3f} max={sims.max():.3f}"
|
|
)
|
|
|
|
print("\n-- Worst matches (bottom 10, most likely hurting the mean) --")
|
|
print(
|
|
f"{'cos':>6} {'conf':>6} {'blur':>7} {'eyes':>6} "
|
|
f"{'angle':>6} {'shape':>9} name"
|
|
)
|
|
for r in rows[:10]:
|
|
print(
|
|
f"{r['cos']:6.3f} {r['conf']:6.3f} {r['blur']:7.1f} "
|
|
f"{r['eye_px']:6.1f} {r['angle']:6.1f} {r['shape']:>9} {r['name']}"
|
|
)
|
|
|
|
print("\n-- Best matches (top 5) --")
|
|
for r in rows[-5:][::-1]:
|
|
print(
|
|
f"{r['cos']:6.3f} {r['conf']:6.3f} {r['blur']:7.1f} "
|
|
f"{r['eye_px']:6.1f} {r['angle']:6.1f} {r['shape']:>9} {r['name']}"
|
|
)
|
|
|
|
# Pairwise analysis — flags embeddings poorly correlated with the rest
|
|
print("\n-- Pairwise intra-class similarity (mean cos vs. other positives) --")
|
|
embs = np.stack([s.embedding for s in samples], axis=0)
|
|
norms = embs / (np.linalg.norm(embs, axis=1, keepdims=True) + 1e-9)
|
|
sim_matrix = norms @ norms.T
|
|
np.fill_diagonal(sim_matrix, np.nan)
|
|
mean_pairwise = np.nanmean(sim_matrix, axis=1)
|
|
names = [os.path.basename(s.path) for s in samples]
|
|
ordered = sorted(zip(names, mean_pairwise), key=lambda t: t[1])
|
|
print(f"{'mean_cos':>9} name")
|
|
for nm, mp in ordered[:10]:
|
|
print(f"{mp:9.3f} {nm}")
|
|
print(f"\n overall mean pairwise cos: {np.nanmean(sim_matrix):.3f}")
|
|
print(f" median pairwise cos: {np.nanmedian(sim_matrix):.3f}")
|
|
|
|
|
|
def summarize_negative(
|
|
neg_samples: list[FaceSample],
|
|
mean_emb: np.ndarray,
|
|
pos_samples: list[FaceSample],
|
|
) -> None:
|
|
"""Score each negative against the class mean, then show its top-3
|
|
nearest positives. High-scoring negatives that match specific outlier
|
|
positives hint at training-set contamination.
|
|
"""
|
|
print("\n" + "=" * 78)
|
|
print(f"NEGATIVE SET ANALYSIS ({len(neg_samples)} images)")
|
|
print("=" * 78)
|
|
print(
|
|
f"\n{'cos':>6} {'conf':>6} {'red':>5} {'adj':>5} "
|
|
f"{'blur':>7} {'eyes':>6} {'shape':>9} name"
|
|
)
|
|
for s in neg_samples:
|
|
cs = cosine(s.embedding, mean_emb)
|
|
conf = similarity_to_confidence(cs)
|
|
red = blur_reduction(s.blur_var)
|
|
print(
|
|
f"{cs:6.3f} {conf:6.3f} {red:5.2f} {max(0, conf - red):5.2f} "
|
|
f"{s.blur_var:7.1f} {s.align_info['eye_dist_px']:6.1f} "
|
|
f"{s.shape[0]}x{s.shape[1]:<5} {os.path.basename(s.path)}"
|
|
)
|
|
|
|
print("\n-- For each negative, top-3 most similar positives --")
|
|
pos_embs = np.stack([p.embedding for p in pos_samples])
|
|
pos_norm = pos_embs / (np.linalg.norm(pos_embs, axis=1, keepdims=True) + 1e-9)
|
|
for s in neg_samples:
|
|
v = s.embedding / (np.linalg.norm(s.embedding) + 1e-9)
|
|
sims = pos_norm @ v
|
|
idx = np.argsort(-sims)[:3]
|
|
print(f"\n {os.path.basename(s.path)}:")
|
|
for i in idx:
|
|
print(
|
|
f" {sims[i]:6.3f} {os.path.basename(pos_samples[i].path)} "
|
|
f"blur={pos_samples[i].blur_var:.1f} "
|
|
f"eyes={pos_samples[i].align_info['eye_dist_px']:.1f}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Optional diagnostics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def vector_outlier_test(
|
|
pos: list[FaceSample], neg: list[FaceSample], base_trim: float = 0.15
|
|
) -> None:
|
|
"""Measure the shipped vector-wise outlier filter at various thresholds.
|
|
|
|
The production filter at `build_class_mean` in
|
|
frigate/data_processing/common/face/model.py uses T=0.30. This test
|
|
sweeps T so you can see which images would be dropped on a new collection
|
|
and how that affects the negative scores.
|
|
|
|
Algorithm: iteratively recompute trim_mean on the kept set, drop any
|
|
embedding with cos < T to that mean, repeat until converged. Floor at
|
|
50% of the collection to avoid collapse.
|
|
"""
|
|
print("\n" + "=" * 78)
|
|
print("VECTOR-WISE OUTLIER PRE-FILTER — layered on trim_mean(0.15)")
|
|
print("=" * 78)
|
|
|
|
all_embs = np.stack([s.embedding for s in pos])
|
|
|
|
def iterative_mean(
|
|
embs: np.ndarray,
|
|
threshold: float,
|
|
iters: int = 3,
|
|
min_keep_frac: float = 0.5,
|
|
) -> tuple[np.ndarray, np.ndarray]:
|
|
keep = np.ones(len(embs), dtype=bool)
|
|
floor = max(5, int(np.ceil(min_keep_frac * len(embs))))
|
|
for _ in range(iters):
|
|
m = stats.trim_mean(embs[keep], base_trim, axis=0)
|
|
m_norm = m / (np.linalg.norm(m) + 1e-9)
|
|
e_norms = embs / (np.linalg.norm(embs, axis=1, keepdims=True) + 1e-9)
|
|
cos_to_mean = e_norms @ m_norm
|
|
new_keep = cos_to_mean >= threshold
|
|
if new_keep.sum() < floor:
|
|
top_idx = np.argsort(-cos_to_mean)[:floor]
|
|
new_keep = np.zeros_like(new_keep)
|
|
new_keep[top_idx] = True
|
|
if np.array_equal(new_keep, keep):
|
|
break
|
|
keep = new_keep
|
|
final = stats.trim_mean(embs[keep], base_trim, axis=0)
|
|
return final, keep
|
|
|
|
provisional = stats.trim_mean(all_embs, base_trim, axis=0)
|
|
p_norm = provisional / (np.linalg.norm(provisional) + 1e-9)
|
|
e_norms_all = all_embs / (np.linalg.norm(all_embs, axis=1, keepdims=True) + 1e-9)
|
|
cos_to_prov = e_norms_all @ p_norm
|
|
print("\nDistribution of cos(positive, provisional trim_mean):")
|
|
print(
|
|
f" min={cos_to_prov.min():.3f} p10={np.percentile(cos_to_prov, 10):.3f} "
|
|
f"p25={np.percentile(cos_to_prov, 25):.3f} "
|
|
f"median={np.median(cos_to_prov):.3f} "
|
|
f"p75={np.percentile(cos_to_prov, 75):.3f} max={cos_to_prov.max():.3f}"
|
|
)
|
|
|
|
baseline_mean = stats.trim_mean(all_embs, base_trim, axis=0)
|
|
baseline_pos = np.array([cosine(p.embedding, baseline_mean) for p in pos])
|
|
baseline_neg = (
|
|
np.array([cosine(n.embedding, baseline_mean) for n in neg])
|
|
if neg
|
|
else np.array([])
|
|
)
|
|
baseline_conf_neg = np.array([similarity_to_confidence(c) for c in baseline_neg])
|
|
|
|
print(
|
|
f"\nBaseline (trim_mean only, {len(pos)} images):"
|
|
f"\n pos cos min={baseline_pos.min():.3f} "
|
|
f"mean={baseline_pos.mean():.3f} max={baseline_pos.max():.3f}"
|
|
)
|
|
if len(neg):
|
|
print(
|
|
f" neg cos min={baseline_neg.min():.3f} "
|
|
f"mean={baseline_neg.mean():.3f} max={baseline_neg.max():.3f}"
|
|
)
|
|
print(
|
|
f" neg conf min={baseline_conf_neg.min():.3f} "
|
|
f"mean={baseline_conf_neg.mean():.3f} max={baseline_conf_neg.max():.3f}"
|
|
)
|
|
print(
|
|
f" margin (pos.min - neg.max): "
|
|
f"{baseline_pos.min() - baseline_neg.max():+.3f}"
|
|
)
|
|
|
|
print("\nIterative (refine mean → drop vectors with cos<T → repeat):")
|
|
print(
|
|
f"\n{'T':>5} {'kept':>6} {'pos min':>7} {'pos mean':>8} "
|
|
f"{'neg max':>7} {'neg mean':>8} {'neg conf.max':>12} {'margin':>7}"
|
|
)
|
|
for T in [0.15, 0.20, 0.25, 0.28, 0.30, 0.33, 0.36, 0.40]:
|
|
mean, keep = iterative_mean(all_embs, T)
|
|
pos_sims = np.array([cosine(p.embedding, mean) for p in pos])
|
|
neg_sims = (
|
|
np.array([cosine(n.embedding, mean) for n in neg]) if neg else np.array([])
|
|
)
|
|
neg_conf = np.array([similarity_to_confidence(c) for c in neg_sims])
|
|
margin = pos_sims.min() - (neg_sims.max() if len(neg_sims) else 0)
|
|
print(
|
|
f"{T:5.2f} {int(keep.sum()):>3}/{len(pos):<2} "
|
|
f"{pos_sims.min():7.3f} {pos_sims.mean():8.3f} "
|
|
f"{neg_sims.max() if len(neg_sims) else float('nan'):7.3f} "
|
|
f"{neg_sims.mean() if len(neg_sims) else float('nan'):8.3f} "
|
|
f"{neg_conf.max() if len(neg_conf) else float('nan'):12.3f} "
|
|
f"{margin:+7.3f}"
|
|
)
|
|
|
|
# Show which images get dropped at the shipped threshold + neighbors
|
|
for T_show in (0.25, 0.30, 0.33):
|
|
_, keep = iterative_mean(all_embs, T_show)
|
|
print(f"\nAt T={T_show}, the {int((~keep).sum())} dropped positives are:")
|
|
final_mean = stats.trim_mean(all_embs[keep], base_trim, axis=0)
|
|
m_n = final_mean / (np.linalg.norm(final_mean) + 1e-9)
|
|
for i, (p, k) in enumerate(zip(pos, keep)):
|
|
if not k:
|
|
e_n = p.embedding / (np.linalg.norm(p.embedding) + 1e-9)
|
|
cos_final = float(e_n @ m_n)
|
|
print(
|
|
f" cos_to_clean_mean={cos_final:6.3f} "
|
|
f"shape={p.shape[0]}x{p.shape[1]} "
|
|
f"eyes={p.align_info['eye_dist_px']:6.1f} "
|
|
f"blur={p.blur_var:7.1f} "
|
|
f"{os.path.basename(p.path)}"
|
|
)
|
|
|
|
|
|
def degenerate_embedding_test(pos: list[FaceSample], neg: list[FaceSample]) -> None:
|
|
"""Detect whether negatives and low-quality positives share a degenerate
|
|
'tiny/noisy face' region of the embedding space.
|
|
|
|
Signal: if neg-to-neg cos is higher than pos-to-pos cos, the negatives
|
|
aren't really per-identity embeddings — they're dominated by upsample /
|
|
low-resolution artifacts that all map to a similar corner of embedding
|
|
space regardless of who the face belongs to.
|
|
|
|
Also rebuilds the mean using only high-intra-similarity positives to
|
|
show whether a cleaner training set separates the negatives.
|
|
"""
|
|
print("\n" + "=" * 78)
|
|
print("DEGENERATE-EMBEDDING TEST")
|
|
print("=" * 78)
|
|
|
|
pos_embs = np.stack([l2(s.embedding) for s in pos])
|
|
neg_embs = np.stack([l2(s.embedding) for s in neg])
|
|
|
|
nn = neg_embs @ neg_embs.T
|
|
np.fill_diagonal(nn, np.nan)
|
|
pp = pos_embs @ pos_embs.T
|
|
np.fill_diagonal(pp, np.nan)
|
|
pn = pos_embs @ neg_embs.T
|
|
|
|
print(
|
|
f"\n neg<->neg mean cos : {np.nanmean(nn):.3f} "
|
|
f"(how tightly negatives cluster together)"
|
|
)
|
|
print(
|
|
f" pos<->pos mean cos : {np.nanmean(pp):.3f} (how tightly positives cluster)"
|
|
)
|
|
print(
|
|
f" pos<->neg mean cos : {pn.mean():.3f} "
|
|
f"(cross-class — should be low for a clean class)"
|
|
)
|
|
if np.nanmean(nn) > np.nanmean(pp):
|
|
print(
|
|
"\n >> neg<->neg > pos<->pos: negatives cluster more tightly than\n"
|
|
" positives. This is the degenerate-embedding signature —\n"
|
|
" upsampled tiny crops share a common 'face-like blob' region\n"
|
|
" regardless of identity."
|
|
)
|
|
|
|
mean_intra = np.nanmean(pp, axis=1)
|
|
for thresh in (0.30, 0.33, 0.36):
|
|
keep = mean_intra >= thresh
|
|
if keep.sum() < 5:
|
|
continue
|
|
clean_embs = [pos[i].embedding for i in range(len(pos)) if keep[i]]
|
|
clean_mean = stats.trim_mean(np.stack(clean_embs), 0.15, axis=0)
|
|
neg_scores = np.array([cosine(n.embedding, clean_mean) for n in neg])
|
|
neg_confs = np.array([similarity_to_confidence(c) for c in neg_scores])
|
|
pos_scores = np.array(
|
|
[cosine(pos[i].embedding, clean_mean) for i in range(len(pos)) if keep[i]]
|
|
)
|
|
print(
|
|
f"\n mean_intra >= {thresh}: keeping {int(keep.sum())}/{len(pos)} positives"
|
|
)
|
|
print(
|
|
f" pos cos vs mean : min={pos_scores.min():.3f} "
|
|
f"mean={pos_scores.mean():.3f} max={pos_scores.max():.3f}"
|
|
)
|
|
print(
|
|
f" neg cos vs mean : min={neg_scores.min():.3f} "
|
|
f"mean={neg_scores.mean():.3f} max={neg_scores.max():.3f}"
|
|
)
|
|
print(
|
|
f" neg conf : min={neg_confs.min():.3f} "
|
|
f"mean={neg_confs.mean():.3f} max={neg_confs.max():.3f}"
|
|
)
|
|
print(
|
|
f" margin (pos.min - neg.max): "
|
|
f"{pos_scores.min() - neg_scores.max():+.3f}"
|
|
)
|
|
|
|
|
|
def contamination_analysis(pos: list[FaceSample], neg: list[FaceSample]) -> None:
|
|
"""Check whether the positive collection contains a second identity.
|
|
|
|
Two signals:
|
|
(a) Per-positive: if an image is closer to at least one negative than
|
|
to the rest of the positive class, it's likely a mislabeled face.
|
|
(b) 2-means split of the positive embeddings: if one cluster center
|
|
lands close to the negative mean, that cluster is a contaminating
|
|
sub-identity that's pulling the class mean toward the negatives.
|
|
"""
|
|
print("\n" + "=" * 78)
|
|
print("CONTAMINATION ANALYSIS")
|
|
print("=" * 78)
|
|
|
|
pos_embs = np.stack([l2(s.embedding) for s in pos])
|
|
neg_embs = np.stack([l2(s.embedding) for s in neg])
|
|
pos_names = [os.path.basename(s.path) for s in pos]
|
|
|
|
pos_pos = pos_embs @ pos_embs.T
|
|
np.fill_diagonal(pos_pos, np.nan)
|
|
pos_neg = pos_embs @ neg_embs.T
|
|
|
|
mean_intra = np.nanmean(pos_pos, axis=1)
|
|
max_to_neg = pos_neg.max(axis=1)
|
|
mean_to_neg = pos_neg.mean(axis=1)
|
|
|
|
print(
|
|
"\nPositives closer to a negative than to their own class avg"
|
|
"\n(these are candidates for mislabeled images):"
|
|
)
|
|
print(f"\n{'max_neg':>7} {'mean_neg':>8} {'mean_intra':>10} {'delta':>6} name")
|
|
rows = list(zip(pos_names, max_to_neg, mean_to_neg, mean_intra))
|
|
rows.sort(key=lambda r: -(r[1] - r[3]))
|
|
for nm, mxn, mnn, mi in rows[:15]:
|
|
delta = mxn - mi
|
|
marker = " <<" if delta > 0 else ""
|
|
print(f"{mxn:7.3f} {mnn:8.3f} {mi:10.3f} {delta:6.3f} {nm}{marker}")
|
|
|
|
# 2-means in cosine space (no sklearn dependency).
|
|
print("\n2-means split of positive embeddings (cosine space):")
|
|
rng = np.random.default_rng(0)
|
|
best = None
|
|
for _ in range(5):
|
|
idx = rng.choice(len(pos_embs), 2, replace=False)
|
|
centers = pos_embs[idx].copy()
|
|
for _ in range(50):
|
|
sims = pos_embs @ centers.T
|
|
labels = np.argmax(sims, axis=1)
|
|
new_centers = np.stack(
|
|
[
|
|
l2(pos_embs[labels == k].mean(axis=0))
|
|
if np.any(labels == k)
|
|
else centers[k]
|
|
for k in range(2)
|
|
]
|
|
)
|
|
if np.allclose(new_centers, centers):
|
|
break
|
|
centers = new_centers
|
|
tight = float(np.mean([sims[i, labels[i]] for i in range(len(labels))]))
|
|
if best is None or tight > best[0]:
|
|
best = (tight, labels.copy(), centers.copy())
|
|
|
|
_, labels, centers = best
|
|
sizes = [int((labels == k).sum()) for k in range(2)]
|
|
neg_mean = l2(neg_embs.mean(axis=0))
|
|
print(
|
|
f" cluster 0: size={sizes[0]:>2} "
|
|
f"center<->other_center_cos={float(centers[0] @ centers[1]):.3f} "
|
|
f"center<->neg_mean_cos={float(centers[0] @ neg_mean):.3f}"
|
|
)
|
|
print(
|
|
f" cluster 1: size={sizes[1]:>2} "
|
|
f"center<->neg_mean_cos={float(centers[1] @ neg_mean):.3f}"
|
|
)
|
|
|
|
neg_aligned = 0 if centers[0] @ neg_mean > centers[1] @ neg_mean else 1
|
|
print(
|
|
f"\n cluster {neg_aligned} is more similar to the negatives — "
|
|
f"its members are the contamination candidates:"
|
|
)
|
|
for i, lbl in enumerate(labels):
|
|
if lbl == neg_aligned:
|
|
print(
|
|
f" max_to_neg={max_to_neg[i]:.3f} "
|
|
f"mean_intra={mean_intra[i]:.3f} {pos_names[i]}"
|
|
)
|
|
|
|
keep_mask = labels != neg_aligned
|
|
if keep_mask.sum() >= 3:
|
|
clean_embs = [pos[i].embedding for i in range(len(pos)) if keep_mask[i]]
|
|
clean_mean = stats.trim_mean(np.stack(clean_embs), 0.15, axis=0)
|
|
print(
|
|
f"\n Rebuilding class mean from the OTHER cluster "
|
|
f"({keep_mask.sum()} images):"
|
|
)
|
|
print(f" {'cos':>6} {'conf':>6} name")
|
|
for n in neg:
|
|
cs = cosine(n.embedding, clean_mean)
|
|
cf = similarity_to_confidence(cs)
|
|
print(f" {cs:6.3f} {cf:6.3f} {os.path.basename(n.path)}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(
|
|
description="Analyze a face recognition collection outside Frigate.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=__doc__,
|
|
)
|
|
ap.add_argument(
|
|
"--positive", required=True, help="Training folder for one identity"
|
|
)
|
|
ap.add_argument(
|
|
"--negative",
|
|
default=None,
|
|
help="Runtime-crop folder to score against (optional)",
|
|
)
|
|
ap.add_argument(
|
|
"--model-cache",
|
|
default="/config/model_cache",
|
|
help="Directory containing facedet/arcface.onnx and facedet/landmarkdet.yaml",
|
|
)
|
|
ap.add_argument(
|
|
"--trim",
|
|
type=float,
|
|
default=0.15,
|
|
help="trim_mean proportion (Frigate uses 0.15)",
|
|
)
|
|
ap.add_argument(
|
|
"--vector-outlier",
|
|
action="store_true",
|
|
help="Sweep the vector-wise outlier filter threshold",
|
|
)
|
|
ap.add_argument(
|
|
"--degenerate",
|
|
action="store_true",
|
|
help="Test whether negatives share a degenerate embedding region",
|
|
)
|
|
ap.add_argument(
|
|
"--contamination",
|
|
action="store_true",
|
|
help="Check whether the positive folder contains a second identity",
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
arcface_path = os.path.join(args.model_cache, "facedet", "arcface.onnx")
|
|
landmark_path = os.path.join(args.model_cache, "facedet", "landmarkdet.yaml")
|
|
for p in (arcface_path, landmark_path):
|
|
if not os.path.exists(p):
|
|
print(f"ERROR: model file not found: {p}")
|
|
return 1
|
|
|
|
print(f"Loading ArcFace from {arcface_path}")
|
|
embedder = ArcFaceEmbedder(arcface_path)
|
|
print(f"Loading landmark model from {landmark_path}")
|
|
aligner = LandmarkAligner(landmark_path)
|
|
|
|
print(f"\nLoading positives from {args.positive} ...")
|
|
pos = load_folder(args.positive, aligner, embedder)
|
|
print(f" {len(pos)} positives loaded")
|
|
|
|
neg: list[FaceSample] = []
|
|
if args.negative:
|
|
print(f"\nLoading negatives from {args.negative} ...")
|
|
neg = load_folder(args.negative, aligner, embedder)
|
|
print(f" {len(neg)} negatives loaded")
|
|
|
|
if not pos:
|
|
print("no positive samples — aborting")
|
|
return 1
|
|
|
|
mean_emb = trimmed_mean([s.embedding for s in pos], trim=args.trim)
|
|
summarize_positive(pos, mean_emb)
|
|
if neg:
|
|
summarize_negative(neg, mean_emb, pos)
|
|
|
|
if args.vector_outlier:
|
|
vector_outlier_test(pos, neg, args.trim)
|
|
if args.degenerate and neg:
|
|
degenerate_embedding_test(pos, neg)
|
|
if args.contamination and neg:
|
|
contamination_analysis(pos, neg)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|