Switch attendance recognition backbone from glintr100 to AdaFace IR-101

Same content as origin/main's cd43f50, applied directly on top of this
Space's existing history (not origin's full history, to avoid dragging in
old already-removed binary blobs from origin's deep history).

activity_web/backend/attendance_service.py

@@ -11,21 +11,31 @@ import cv2
 import numpy as np
 
 from insightface.app import FaceAnalysis
+from insightface.utils import face_align
 from .startup import ensure_insightface_models_flat
 from .config import ATTENDANCE_DIR
+from utils.adaface_backbone import AdaFaceWrapper, DEFAULT_CKPT_PATH as ADAFACE_CKPT_PATH
 
 
 UPLOAD_DIR = ATTENDANCE_DIR / "uploads"
 MARKED_DIR = ATTENDANCE_DIR / "marked"
 STORE_PATH = ATTENDANCE_DIR / "attendance_store.json"
 PHOTOS_PER_VIDEO = 32
-FACE_SIMILARITY_THRESHOLD = 0.38
-EMBEDDING_MODEL_NAME = "antelopev2_v3"
+# AdaFace IR-101's own p99 impostor threshold, derived from the human-labeled
+# clean eval set (eval/impostor_scope_eval.py) — NOT glintr100's 0.38, the two
+# backbones' cosine distributions aren't comparable.
+FACE_SIMILARITY_THRESHOLD = 0.28
+EMBEDDING_MODEL_NAME = "adaface_ir101_webface12m"
 ENROLLMENT_MIN_DET_SCORE = 0.50
+# Unchanged from glintr100: eval/build_gallery.py mirrors production enrollment
+# and kept this same value (0.50) when building the AdaFace gallery used in
+# every comparison run, so no evidence it needs to move for the new backbone.
 ENROLLMENT_OUTLIER_SIM_THRESHOLD = 0.50
 MAX_STORED_EMBEDDINGS = 128
 ENROLLMENT_SAMPLE_INTERVAL_S = 1.0   # sample one frame every 1 second for enrollment
 MAX_ENROLLMENT_FRAMES = 30           # cap to avoid overly long videos
+# Unchanged from glintr100 (see ENROLLMENT_OUTLIER_SIM_THRESHOLD note above) —
+# eval/build_gallery.py's _ANCHOR_SIM_THRESH stayed at 0.35 for AdaFace too.
 ANCHOR_CONSISTENCY_THRESHOLD = 0.35
 # Degradation scales applied during enrollment to bridge the gap between
 # close-up enrollment faces and small distant classroom faces.
@@ -51,7 +61,15 @@ def _normalize(vector: np.ndarray) -> np.ndarray:
     return vector / norm
 
 
-def _degrade_and_embed(fa: FaceAnalysis, frame: np.ndarray, bbox: tuple, scale: float) -> np.ndarray | None:
+def _align_and_embed(adaface: AdaFaceWrapper, frame: np.ndarray, kps: np.ndarray) -> np.ndarray:
+    """Align a face via InsightFace's norm_crop (same alignment AdaFace was
+    trained/calibrated against) and embed it with the AdaFace backbone."""
+    aligned = face_align.norm_crop(frame, landmark=np.asarray(kps, dtype=np.float32), image_size=112)
+    emb, _feat_norm = adaface.embed_aligned(aligned)
+    return _normalize(emb)
+
+
+def _degrade_and_embed(fa: FaceAnalysis, adaface: AdaFaceWrapper, frame: np.ndarray, bbox: tuple, scale: float) -> np.ndarray | None:
     """Shrink a face crop to `scale` fraction then bicubic back to original size,
     then re-run recognition on that blurry crop.  Simulates how a distant classroom
     face looks to the recognition model, so the enrollment gallery contains embeddings
@@ -68,13 +86,13 @@ def _degrade_and_embed(fa: FaceAnalysis, frame: np.ndarray, bbox: tuple, scale:
     degraded = cv2.resize(crop, (small_w, small_h), interpolation=cv2.INTER_AREA)
     degraded = cv2.resize(degraded, (cw, ch), interpolation=cv2.INTER_CUBIC)
 
-    # Paste degraded crop back into a copy of the frame and re-run fa.get
+    # Paste degraded crop back into a copy of the frame and re-run detection
     frame_copy = frame.copy()
     frame_copy[y1:y2, x1:x2] = degraded
     faces = fa.get(frame_copy)
 
     # Pick the face closest to the original bbox
-    best_emb, best_iou = None, 0.0
+    best_kps, best_iou = None, 0.0
     for f in faces:
         fx1, fy1, fx2, fy2 = [int(v) for v in f.bbox]
         ix1, iy1 = max(x1, fx1), max(y1, fy1)
@@ -82,14 +100,12 @@ def _degrade_and_embed(fa: FaceAnalysis, frame: np.ndarray, bbox: tuple, scale:
         inter = max(0, ix2 - ix1) * max(0, iy2 - iy1)
         union = (x2-x1)*(y2-y1) + (fx2-fx1)*(fy2-fy1) - inter
         iou = inter / union if union > 0 else 0.0
-        if iou > best_iou:
-            e = getattr(f, "normed_embedding", None)
-            if e is None:
-                e = getattr(f, "embedding", None)
-            if e is not None:
-                best_emb = _normalize(np.asarray(e, dtype=np.float32).flatten().copy())
-                best_iou = iou
-    return best_emb
+        if iou > best_iou and getattr(f, "kps", None) is not None:
+            best_kps = f.kps
+            best_iou = iou
+    if best_kps is None:
+        return None
+    return _align_and_embed(adaface, frame_copy, best_kps)
 
 
 
@@ -116,10 +132,11 @@ class AttendanceService:
         ensure_attendance_dirs()
         self.face_analysis = FaceAnalysis(
             name="antelopev2",
-            allowed_modules=["detection", "recognition"],
+            allowed_modules=["detection"],
             providers=["CPUExecutionProvider"],
         )
         self.face_analysis.prepare(ctx_id=0, det_size=(1280, 1280), det_thresh=0.5)
+        self.adaface = AdaFaceWrapper.load(ADAFACE_CKPT_PATH)
         self._migrate_legacy_embeddings_if_needed()
 
     def _read_store(self) -> dict:
@@ -202,15 +219,13 @@ class AttendanceService:
         faces = self.face_analysis.get(frame)
         samples: list[FaceSample] = []
         for face in faces:
-            embedding = getattr(face, "normed_embedding", None)
-            if embedding is None:
-                embedding = getattr(face, "embedding", None)
-            if embedding is None:
+            kps = getattr(face, "kps", None)
+            if kps is None:
                 continue
             bbox = tuple(int(value) for value in face.bbox)
             samples.append(
                 FaceSample(
-                    embedding=_normalize(np.asarray(embedding, dtype=np.float32)),
+                    embedding=_align_and_embed(self.adaface, frame, kps),
                     bbox=bbox,
                     score=float(getattr(face, "det_score", 0.0)),
                 )
@@ -252,7 +267,7 @@ class AttendanceService:
                 results.append((primary.embedding, float(primary.score)))
                 # Also enroll degraded versions to match distant classroom conditions
                 for scale in ENROLLMENT_DEGRADATION_SCALES:
-                    deg = _degrade_and_embed(self.face_analysis, frame, primary.bbox, scale)
+                    deg = _degrade_and_embed(self.face_analysis, self.adaface, frame, primary.bbox, scale)
                     if deg is not None:
                         results.append((deg, float(primary.score) * 0.9))
             else:
@@ -263,7 +278,7 @@ class AttendanceService:
                     results.append((best.embedding, float(best.score)))
                     # Also enroll degraded versions
                     for scale in ENROLLMENT_DEGRADATION_SCALES:
-                        deg = _degrade_and_embed(self.face_analysis, frame, best.bbox, scale)
+                        deg = _degrade_and_embed(self.face_analysis, self.adaface, frame, best.bbox, scale)
                         if deg is not None:
                             results.append((deg, float(best.score) * 0.9))
 

download_models.py

@@ -62,6 +62,19 @@ def _download(url: str, dest: Path, retries: int = 5) -> None:
                 sys.exit(1)
 
 
+def _download_adaface() -> bool:
+    from utils.adaface_backbone import DEFAULT_CKPT_PATH, download_model
+
+    if DEFAULT_CKPT_PATH.exists() and not _is_lfs_pointer(DEFAULT_CKPT_PATH):
+        print(f"  {DEFAULT_CKPT_PATH.name} already present, skipping.")
+        return False
+    print(f"  downloading {DEFAULT_CKPT_PATH.name} from HuggingFace (minchul/cvlface_adaface_ir101_webface12m) …")
+    download_model(DEFAULT_CKPT_PATH)
+    size_mb = DEFAULT_CKPT_PATH.stat().st_size / 1e6
+    print(f"  {DEFAULT_CKPT_PATH.name} — {size_mb:.1f} MB  ✓")
+    return True
+
+
 def main() -> None:
     any_downloaded = False
     for rel, url, _ in MODELS:
@@ -72,6 +85,8 @@ def main() -> None:
         _download(url, dest)
         any_downloaded = True
 
+    any_downloaded = _download_adaface() or any_downloaded
+
     if not any_downloaded:
         print("All model weights are already in place.")
     else:

requirements.txt

@@ -6,6 +6,7 @@ Pillow
 opencv-python-headless
 ultralytics
 insightface
+huggingface_hub
 torch
 torchvision
 imageio-ffmpeg

utils/adaface_backbone.py

@@ -0,0 +1,197 @@
+"""
+AdaFace IR-101 (CVLFace, minchul/cvlface_adaface_ir101_webface12m) recognition backbone.
+
+Replaces glintr100/antelopev2 as the production embedding model. InsightFace's
+`FaceAnalysis` is still used for detection (bbox + 5-pt landmarks); this module
+only handles alignment + embedding.
+
+Preprocessing:
+  - color_space: RGB → convert BGR→RGB before normalising
+  - normalisation: (pixel/255 - 0.5) / 0.5, applied after the BGR→RGB flip
+  - alignment: insightface.utils.face_align.norm_crop, 112x112 (same as glintr100)
+
+Output: (512-d L2-normalised embedding, pre-BN feature norm — quality proxy).
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import NamedTuple
+
+import cv2
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from torch.nn import (BatchNorm1d, BatchNorm2d, Conv2d, Dropout, Flatten,
+                      Linear, MaxPool2d, PReLU, Sequential)
+
+
+# ---------------------------------------------------------------------------
+# Building blocks (CVLFace IR-101 architecture)
+# ---------------------------------------------------------------------------
+
+class BasicBlockIR(nn.Module):
+    def __init__(self, in_channel: int, depth: int, stride: int):
+        super().__init__()
+        if in_channel == depth:
+            self.shortcut_layer = MaxPool2d(1, stride)
+        else:
+            self.shortcut_layer = Sequential(
+                Conv2d(in_channel, depth, (1, 1), stride, bias=False),
+                BatchNorm2d(depth),
+            )
+        self.res_layer = Sequential(
+            BatchNorm2d(in_channel),
+            Conv2d(in_channel, depth, (3, 3), (1, 1), 1, bias=False),
+            BatchNorm2d(depth),
+            PReLU(depth),
+            Conv2d(depth, depth, (3, 3), stride, 1, bias=False),
+            BatchNorm2d(depth),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.res_layer(x) + self.shortcut_layer(x)
+
+
+class _BlockSpec(NamedTuple):
+    in_channel: int
+    depth: int
+    stride: int
+
+
+def _get_blocks_ir101() -> list[list[_BlockSpec]]:
+    """Block specs for IR-101: unit counts [3, 13, 30, 3]."""
+    def _block(in_c: int, depth: int, n: int) -> list[_BlockSpec]:
+        return [_BlockSpec(in_c, depth, 2)] + [_BlockSpec(depth, depth, 1)] * (n - 1)
+
+    return [
+        _block(64,  64,  3),
+        _block(64,  128, 13),
+        _block(128, 256, 30),
+        _block(256, 512, 3),
+    ]
+
+
+class Backbone(nn.Module):
+    """CVLFace Backbone (InsightFace-style IR). Forward returns (emb, feat_norm)."""
+
+    def __init__(self, blocks_spec: list[list[_BlockSpec]], output_dim: int = 512,
+                 dropout: float = 0.4):
+        super().__init__()
+        self.input_layer = Sequential(
+            Conv2d(3, 64, (3, 3), 1, 1, bias=False),
+            BatchNorm2d(64),
+            PReLU(64),
+        )
+        units = [BasicBlockIR(b.in_channel, b.depth, b.stride)
+                 for block in blocks_spec for b in block]
+        self.body = Sequential(*units)
+
+        self.output_layer = Sequential(
+            BatchNorm2d(512),
+            Dropout(p=dropout),
+            Flatten(),
+            Linear(512 * 7 * 7, output_dim, bias=True),
+            BatchNorm1d(output_dim, affine=False),
+        )
+
+    def forward(self, x: torch.Tensor):
+        x = self.input_layer(x)
+        x = self.body(x)
+        x = self.output_layer[0](x)
+        x = self.output_layer[1](x)
+        x = self.output_layer[2](x)
+        x = self.output_layer[3](x)
+        feat_norm = torch.norm(x, p=2, dim=1)
+        x = self.output_layer[4](x)
+        emb = F.normalize(x, p=2, dim=1)
+        return emb, feat_norm
+
+
+def ir101(output_dim: int = 512) -> Backbone:
+    return Backbone(_get_blocks_ir101(), output_dim=output_dim)
+
+
+def load_cvlface_checkpoint(model: Backbone, ckpt_path: str) -> None:
+    """Load CVLFace model.pt, stripping the 'net.' prefix from state-dict keys."""
+    raw = torch.load(ckpt_path, map_location="cpu", weights_only=False)
+    if isinstance(raw, dict) and not isinstance(raw, nn.Module):
+        sd = raw.get("state_dict", raw)
+    else:
+        sd = raw
+    if any(k.startswith("net.") for k in sd):
+        sd = {k[4:]: v for k, v in sd.items() if k.startswith("net.")}
+    model.load_state_dict(sd, strict=False)
+
+
+# ---------------------------------------------------------------------------
+# Preprocessing + wrapper
+# ---------------------------------------------------------------------------
+
+def _to_tensor(aligned_bgr: np.ndarray) -> torch.Tensor:
+    """112x112 BGR ndarray -> (1,3,112,112) float32 tensor, RGB, normalised to [-1,1]."""
+    arr = aligned_bgr[:, :, ::-1].copy()       # BGR -> RGB
+    arr = arr.astype(np.float32)
+    arr = (arr / 255.0 - 0.5) / 0.5            # -> [-1, 1]
+    arr = arr.transpose(2, 0, 1)[np.newaxis]   # (1, C, H, W)
+    return torch.from_numpy(arr)
+
+
+class AdaFaceWrapper:
+    def __init__(self, model: torch.nn.Module):
+        self._model = model
+        self._model.eval()
+
+    @classmethod
+    def load(cls, ckpt_path: str | Path) -> "AdaFaceWrapper":
+        ckpt_path = Path(ckpt_path)
+        if not ckpt_path.exists():
+            raise FileNotFoundError(
+                f"AdaFace checkpoint not found: {ckpt_path}\n"
+                "Run `python download_models.py` to fetch it."
+            )
+        model = ir101()
+        load_cvlface_checkpoint(model, str(ckpt_path))
+        model.eval()
+        return cls(model)
+
+    def embed_aligned(self, aligned_bgr_112: np.ndarray) -> tuple[np.ndarray, float]:
+        """
+        aligned_bgr_112 : 112x112 BGR ndarray (from insightface.utils.face_align.norm_crop)
+        Returns (emb (512,) float32 L2-normalised, feat_norm quality proxy).
+        """
+        if aligned_bgr_112 is None or aligned_bgr_112.size == 0:
+            raise ValueError("aligned_bgr_112 cannot be empty")
+        crop = aligned_bgr_112
+        if crop.shape[:2] != (112, 112):
+            crop = cv2.resize(crop, (112, 112), interpolation=cv2.INTER_LINEAR)
+        tensor = _to_tensor(crop)
+        with torch.inference_mode():
+            emb_t, norm_t = self._model(tensor)
+        emb = emb_t[0].numpy().astype(np.float32)
+        norm = float(norm_t[0].item())
+        return emb, norm
+
+
+# ---------------------------------------------------------------------------
+# Download helper — mirrors the YOLO-weight pattern in download_models.py
+# ---------------------------------------------------------------------------
+
+_HF_REPO = "minchul/cvlface_adaface_ir101_webface12m"
+_HF_FILE = "pretrained_model/model.pt"
+
+DEFAULT_CKPT_PATH = Path(__file__).resolve().parent.parent / "models" / "adaface" / "adaface_ir101_webface12m.pt"
+
+
+def download_model(dest: Path = DEFAULT_CKPT_PATH, force: bool = False) -> Path:
+    dest = Path(dest)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    if dest.exists() and not force:
+        return dest
+
+    from huggingface_hub import hf_hub_download
+    tmp = hf_hub_download(repo_id=_HF_REPO, filename=_HF_FILE, local_dir=str(dest.parent))
+    tmp_p = Path(tmp)
+    if tmp_p != dest:
+        tmp_p.rename(dest)
+    return dest