first commit

.gitignore

@@ -0,0 +1,15 @@
+./face_db
+myenv/
+facerecog/
+test_verify-image.py
+test_verify.py
+test/*
+face_db/*
+*.onnx
+*.onnx.*
+*.bin
+*.jpg
+*.jpeg
+*.png
+*.webp
+*.onnx

Dockerfile

@@ -0,0 +1,48 @@
+# ---------- CPU BASE ----------
+# Menggunakan image Python official yang ringan (slim) berbasis Debian
+# Tidak perlu image NVIDIA/CUDA karena kita hanya menggunakan CPU
+FROM python:3.10-slim
+
+ENV DEBIAN_FRONTEND=noninteractive \
+    PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1
+
+WORKDIR /app
+
+# System deps
+# Kita tetap butuh library untuk OpenCV (libglib, libsm6, dll)
+# dan build-essential untuk berjaga-jaga jika ada library python yang perlu dicompile
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential gcc g++ make \
+    libglib2.0-0 libsm6 libxrender1 libxext6 \
+ && rm -rf /var/lib/apt/lists/*
+# Source - https://stackoverflow.com/a
+# Posted by Tushar Kolhe, modified by community. See post 'Timeline' for change history
+# Retrieved 2025-12-05, License - CC BY-SA 4.0
+
+RUN apt-get update && apt-get install ffmpeg libsm6 libxext6  -y
+
+# Python deps
+COPY requirements.txt /app/requirements.txt
+# Upgrade pip dan install requirements
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r /app/requirements.txt
+
+# App
+COPY main.py /app/main.py
+COPY liveness.py /app/liveness.py
+COPY helper.py /app/helper.py
+COPY 2.7_80x80_MiniFASNetV2.onnx /app/2.7_80x80_MiniFASNetV2.onnx
+
+# Runtime env
+ENV FACE_DET_SIZE=480,480
+# PENTING: Set provider hanya ke CPU
+ENV ORT_PROVIDERS=CPUExecutionProvider
+
+# Hugging Face Spaces secara default biasanya mengharapkan port 7860.
+# Jika config Space Anda mengharuskan 8000, biarkan 8000.
+# Namun standard Docker Space adalah 7860.
+EXPOSE 7860
+
+# Workers bisa dinaikkan sedikit karena di CPU tidak ada limit context locking seperti di GPU
+CMD ["uvicorn", "main:app", "--host=0.0.0.0", "--port=7860", "--workers=2"]

README.md

@@ -0,0 +1,220 @@
+---
+title: Face Recognition Sss Beta
+emoji: 📊
+colorFrom: red
+colorTo: yellow
+sdk: docker
+pinned: false
+---
+
+# Face Verification API (FastAPI + InsightFace)
+
+FastAPI service for face enroll & verification using InsightFace (`buffalo_l`).  
+Supports:
+- Enroll via **JSON (base64)** or **multipart files**
+- Verify via **single HTTP POST** (multipart or JSON) and **WebSocket stream** (binary or JSON)
+- Clear per-employee embeddings
+- Swagger UI
+
+---
+
+## Endpoints
+
+- `POST /face/enroll` — JSON base64 enroll
+- `POST /face/enroll-files` — multipart file enroll
+- `POST /face/verify` — verify one frame (multipart `frame`/`image` or JSON data URL)
+- `WEBSOCKET /face/verify` — stream frames (binary bytes or JSON data URL)
+- `POST /face/verify-image` — JSON base64 verify
+- `POST /face/clear` — clear embeddings
+- `GET /health` — health check
+
+Swagger UI: [http://localhost:8000/docs](http://localhost:8000/docs)  
+ReDoc: [http://localhost:8000/redoc](http://localhost:8000/redoc)
+
+---
+
+## GPU Setup
+
+Use if you want speed and have an NVIDIA GPU.
+
+**Image requirements:**
+- `onnxruntime-gpu` in requirements.txt
+- Base image like `nvidia/cuda:12.4.1-cudnn-runtime-ubuntu22.04`
+
+### 1. Install NVIDIA driver on host
+
+```bash
+nvidia-smi   # should print your GPU details
+```
+
+### 2. Install NVIDIA Container Toolkit (Ubuntu example)
+
+```bash
+distribution=$(. /etc/os-release;echo $ID$VERSION_ID)
+curl -fsSL https://nvidia.github.io/libnvidia-container/gpgkey \
+ | sudo gpg --dearmor -o /usr/share/keyrings/nvidia-container-toolkit-keyring.gpg
+curl -s -L https://nvidia.github.io/libnvidia-container/$distribution/libnvidia-container.list \
+ | sed 's#deb https://#deb [signed-by=/usr/share/keyrings/nvidia-container-toolkit-keyring.gpg] https://#g' \
+ | sudo tee /etc/apt/sources.list.d/nvidia-container-toolkit.list
+sudo apt-get update
+sudo apt-get install -y nvidia-container-toolkit
+sudo nvidia-ctk runtime configure --runtime=docker
+sudo systemctl restart docker
+```
+
+### 3. Test GPU passthrough
+
+```bash
+sudo docker run --rm --gpus all nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi
+```
+
+### 4. Build & Run (GPU)
+
+```bash
+sudo docker build -t face-api-gpu .
+sudo docker run --rm -it --gpus all -p 8000:8000 \
+  -e ORT_PROVIDERS=CUDAExecutionProvider,CPUExecutionProvider \
+  -e FACE_SIM_THRESHOLD=0.75 \
+  -v $HOME/.insightface:/root/.insightface \
+  face-api-gpu
+```
+
+Open: [http://localhost:8000/docs](http://localhost:8000/docs)
+
+---
+
+## Troubleshooting
+
+- **Error:** could not select device driver "" with capabilities: [[gpu]]
+  - NVIDIA Container Toolkit not set up → follow the GPU Setup steps.
+- **Old Docker (<19.03):**
+  ```bash
+  sudo docker run --rm -it --runtime=nvidia -e NVIDIA_VISIBLE_DEVICES=all ...
+  ```
+- **Warning:** The NVIDIA Driver was not detected
+  - Install the host driver (`nvidia-smi` must work), then restart Docker.
+- **Container exits during startup**
+  - Try CPU first:
+    ```bash
+    sudo docker run --rm -it -p 8000:8000 \
+      -e ORT_PROVIDERS=CPUExecutionProvider \
+      -v $HOME/.insightface:/root/.insightface \
+      face-api-gpu
+    ```
+- **Permissions error on Docker socket**
+  - Add user to docker group:
+    ```bash
+    sudo usermod -aG docker $USER
+    # log out/in or reboot
+    ```
+  - Or use `sudo docker ...`
+
+
+---
+
+
+## Open Swagger
+```bash
+http://localhost:8000/docs/
+```
+
+---
+
+## Curl Examples
+
+### Enroll (files)
+
+```bash
+curl -X POST "http://localhost:8000/face/enroll-files" \
+  -F "employee_id=1" \
+  -F "files=@/path/to/3x4.jpg" \
+  -F "files=@/path/to/another.jpg"
+```
+
+### Enroll (base64 JSON)
+
+```bash
+IMG_B64="$(base64 -w0 /path/to/3x4.jpg)"
+curl -X POST "http://localhost:8000/face/enroll" \
+  -H "Content-Type: application/json" \
+  -d "{\"employee_id\":\"1\",\"images\":[\"data:image/jpeg;base64,$IMG_B64\"]}"
+```
+
+### Verify (single, multipart file)
+
+```bash
+curl -X POST "http://localhost:8000/face/verify?employee_id=1&threshold=0.75" \
+  -F "frame=@/path/to/capture.jpg"
+```
+
+### Verify (single, JSON data URL)
+
+```bash
+IMG_B64="$(base64 -w0 /path/to/capture.jpg)"
+curl -X POST "http://localhost:8000/face/verify?employee_id=1&threshold=0.75" \
+  -H "Content-Type: application/json" \
+  -d "{\"frame\":\"data:image/jpeg;base64,$IMG_B64\"}"
+```
+
+### Verify (image-only JSON)
+
+```bash
+IMG_B64="$(base64 -w0 /path/to/capture.jpg)"
+curl -X POST "http://localhost:8000/face/verify-image" \
+  -H "Content-Type: application/json" \
+  -d "{\"employee_id\":\"1\",\"image\":\"data:image/jpeg;base64,$IMG_B64\",\"threshold\":0.75}"
+```
+
+### Clear embeddings
+
+```bash
+curl -X POST "http://localhost:8000/face/clear" \
+  -H "Content-Type: application/json" \
+  -d '{"employee_id":"1"}'
+```
+---
+
+## Dev (local, no Docker)
+
+```bash
+pip install -r requirements.txt
+export ORT_PROVIDERS=CPUExecutionProvider
+uvicorn main:app --host 0.0.0.0 --port 8000 --reload
+```
+
+---
+
+## IF CPU-only
+
+### 1. Dockerfile (CPU)
+
+```dockerfile
+FROM python:3.11-slim
+ENV DEBIAN_FRONTEND=noninteractive PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libglib2.0-0 libsm6 libxrender1 libxext6 build-essential gcc g++ make \
+  && rm -rf /var/lib/apt/lists/*
+WORKDIR /app
+COPY requirements.txt /app/requirements.txt
+# requirements.txt should use "onnxruntime" (CPU), not "onnxruntime-gpu"
+RUN pip install --no-cache-dir -r /app/requirements.txt
+COPY main.py /app/main.py
+ENV ORT_PROVIDERS=CPUExecutionProvider FACE_DET_SIZE=480,480
+EXPOSE 8000
+CMD ["uvicorn","main:app","--host=0.0.0.0","--port=8000","--workers=1"]
+```
+
+#### Build & Run (CPU)
+
+```bash
+sudo docker build -t face-api-cpu .
+sudo docker run --rm -it -p 8000:8000 \
+  -e ORT_PROVIDERS=CPUExecutionProvider \
+  -v $HOME/.insightface:/root/.insightface \
+  face-api-cpu
+```
+
+Open: [http://localhost:8000/docs](http://localhost:8000/docs)
+
+
+---

helper.py

@@ -0,0 +1,63 @@
+from typing import Optional
+import os
+import base64
+import cv2
+import numpy as np
+
+FACE_DB_ROOT = "face_db"
+os.makedirs(FACE_DB_ROOT, exist_ok=True)
+
+LIVE_DB_ROOT = os.path.join(FACE_DB_ROOT, "_live")
+os.makedirs(LIVE_DB_ROOT, exist_ok=True)
+
+def _strip_b64(s: str) -> str:
+    if isinstance(s, str) and "," in s and s.lstrip().lower().startswith("data:"):
+        return s.split(",", 1)[1]
+    return s
+
+def _b64_to_bgr(b64: str) -> Optional[np.ndarray]:
+    try:
+        raw = base64.b64decode(_strip_b64(b64), validate=False)
+        arr = np.frombuffer(raw, np.uint8)
+        return cv2.imdecode(arr, cv2.IMREAD_COLOR)
+    except Exception:
+        return None
+
+def _bytes_to_bgr(data: bytes) -> Optional[np.ndarray]:
+    try:
+        arr = np.frombuffer(data, np.uint8)
+        return cv2.imdecode(arr, cv2.IMREAD_COLOR)
+    except Exception:
+        return None
+
+def _frame_path_for(employee_id: str) -> str:
+    safe = (employee_id or "live").replace("/", "_")
+    return os.path.join(LIVE_DB_ROOT, f"{safe}.jpg")
+
+class MotionVerifier:
+    def __init__(self, required_frames=5, threshold=1.5):
+        self.required_frames = required_frames
+        self.threshold = threshold
+        self.landmarks_buffer = []
+        
+    def add_landmarks(self, landmarks):
+        self.landmarks_buffer.append(landmarks)
+        if len(self.landmarks_buffer) > self.required_frames:
+            self.landmarks_buffer.pop(0)
+            
+    def is_ready(self):
+        return len(self.landmarks_buffer) >= self.required_frames
+
+    def check_motion(self):
+        if not self.is_ready():
+            return False, "buffering"
+        data = np.array(self.landmarks_buffer)        
+        std_dev = np.std(data, axis=0) 
+        avg_movement = np.mean(std_dev)
+        print(f"[Motion] Avg Movement: {avg_movement:.4f} px")
+        if avg_movement < self.threshold:
+            return False, "static_spoof_detected"
+        return True, "passed"
+
+    def reset(self):
+        self.landmarks_buffer = []

liveness.py

@@ -0,0 +1,59 @@
+import cv2
+import numpy as np
+import onnxruntime as ort
+
+class LivenessDetector:
+    def __init__(self, model_path, device="cpu"):
+        providers = ['CUDAExecutionProvider', 'CPUExecutionProvider'] if device == "cuda" else ['CPUExecutionProvider']
+        self.session = ort.InferenceSession(model_path, providers=providers)
+        self.input_name = self.session.get_inputs()[0].name
+        print(f"[INFO] Model loaded dari {model_path} pada device {device}")
+        
+    def _get_kernel(self, height, width):
+        kernel_size = (height + width) / 2
+        scale = 2.7 
+        return int(kernel_size * scale)
+
+    def predict(self, img, bbox):
+        """
+        img: gambar full BGR
+        bbox: boundix box [x1, y1, x2, y2] hasil dari RetinaFace()
+        """
+        h_img, w_img, _ = img.shape
+        x1, y1, x2, y2 = map(int, bbox)
+        
+        box_h = y2 - y1
+        box_w = x2 - x1
+        
+        center_x = x1 + box_w // 2
+        center_y = y1 + box_h // 2
+        
+        side = self._get_kernel(box_h, box_w)
+        
+        new_x1 = max(0, center_x - side // 2)
+        new_y1 = max(0, center_y - side // 2)
+        new_x2 = min(w_img, center_x + side // 2)
+        new_y2 = min(h_img, center_y + side // 2)
+        
+        cropped = img[new_y1:new_y2, new_x1:new_x2]
+        
+        if cropped.size == 0:
+            return False, 0.0
+
+        # prep yaitu resize ke 80x80
+        resized = cv2.resize(cropped, (80, 80))
+        
+        # Convert ke Float32 & Transpose (HWC -> CHW)
+        blob = resized.astype(np.float32)
+        blob = np.transpose(blob, (2, 0, 1)) 
+        blob = np.expand_dims(blob, 0)       
+        
+        result = self.session.run(None, {self.input_name: blob})
+        
+        probs = self._softmax(result[0][0])
+        
+        return probs
+
+    def _softmax(self, x):
+        e_x = np.exp(x - np.max(x))
+        return e_x / e_x.sum()

main.py

@@ -0,0 +1,449 @@
+import os
+import base64
+import asyncio
+import json
+from typing import List, Optional, Iterable
+
+# tool buat liveness check
+from liveness import LivenessDetector
+
+import numpy as np
+import cv2
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, UploadFile, File, Form, HTTPException, Request
+from pydantic import BaseModel, Field
+from starlette.concurrency import run_in_threadpool
+from starlette.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from helper import _b64_to_bgr, _bytes_to_bgr, _frame_path_for
+from helper import MotionVerifier
+
+# Import Uniface (Single Library)
+from uniface import RetinaFace, ArcFace
+
+# Konfigurasi Threshold
+# Stage 1: Detection Confidence (Pengganti Liveness Check sementara)
+# Wajah dengan confidence di bawah ini dianggap tidak valid/buruk
+DETECTION_THRESHOLD = 0.7 
+
+# Stage 2: Recognition Similarity (Cosine Similarity)
+SIM_THRESHOLD = 0.40 
+
+app = FastAPI(
+    title="Face Verification API (Uniface Pure)",
+    version="2.2.0",
+    description="2-Stage Verification: High-Confidence Detection -> Uniface Recognition",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+FACE_DB_ROOT = "face_db"
+os.makedirs(FACE_DB_ROOT, exist_ok=True)
+
+LIVE_DB_ROOT = os.path.join(FACE_DB_ROOT, "_live")
+os.makedirs(LIVE_DB_ROOT, exist_ok=True)
+
+processing_lock = asyncio.Lock()
+
+# --- INITIALIZATION (UNIFACE and Minivision) ---
+models = {
+    "detector": None,
+    "recognizer": None,
+    "liveness": None,
+}
+
+try:
+    print("[Uniface] Initializing RetinaFace (Stage 1)...")
+    models["detector"] = RetinaFace() 
+    print("[Minivision] Initializing Liveness Detector...")
+    models["liveness"] = LivenessDetector("2.7_80x80_MiniFASNetV2.onnx")
+    print("[Uniface] Initializing ArcFace (Stage 2)...")
+    models["recognizer"] = ArcFace()
+    
+except Exception as e:
+    print(f"[Uniface] Error initializing models: {e}")
+
+
+class EnrollRequest(BaseModel):
+    employee_id: str = Field(default="1")
+    images: List[str] = Field(..., description="Base64 images")
+
+class VerifyImageRequest(BaseModel):
+    employee_id: str = Field(default="1")
+    image: str = Field(..., description="Base64 image")
+    threshold: Optional[float] = Field(default=None)
+
+class ClearRequest(BaseModel):
+    employee_id: str = Field(default="1")
+
+
+
+
+async def _decode_image_from_request(request: Request, field_names: Iterable[str] = ("frame", "image")) -> Optional[np.ndarray]:
+    ct = (request.headers.get("content-type") or "").lower()
+    print(f"--- DEBUG START ---")
+    print(f"1. Content-Type yang diterima: {ct}")
+    if "multipart/form-data" in ct:
+        form = await request.form()
+        for name in field_names:
+            file = form.get(name)
+            if not file:
+                continue
+            
+            if hasattr(file, "read"):
+                await file.seek(0)
+                data = await file.read()
+                img = _bytes_to_bgr(data)
+                if img is not None:
+                    return img
+            
+            elif isinstance(file, bytes):
+                img = _bytes_to_bgr(file)
+                if img is not None:
+                    return img
+                
+                # Coba decode
+            elif isinstance(file, str):
+                img = _b64_to_bgr(file)
+                if img is not None:
+                    return img
+            
+        return None
+    if "application/json" in ct or "text/json" in ct:
+        try:
+            obj = await request.json()
+        except Exception:
+            return None
+        for name in field_names:
+            val = obj.get(name)
+            if isinstance(val, str):
+                img = _b64_to_bgr(val)
+                if img is not None:
+                    return img
+        return None
+    return None
+
+def _compute_embedding_uniface(img_bgr, landmarks):
+    """Helper to get embedding using Uniface ArcFace"""
+    if models["recognizer"] is None:
+        return None
+    try:
+        # get_normalized_embedding requires the image and landmarks
+        embedding = models["recognizer"].get_normalized_embedding(img_bgr, landmarks)
+        return embedding
+    except Exception:
+        return None
+
+async def _run_pipeline(img: np.ndarray, target_id: str):
+    
+    detector = models["detector"]
+    liveness = models["liveness"]
+    
+    if detector is None or liveness is None:
+        raise RuntimeError("Models not initialized")
+    # --- STAGE 1: DETECTION & QUALITY CHECK ---
+    # Menggunakan RetinaFace untuk mendeteksi wajah
+    async with processing_lock:
+        faces = await run_in_threadpool(detector.detect, img)
+    
+    if not faces:
+        return None
+
+    # Ambil wajah dengan confidence tertinggi atau area terbesar
+    target_face = faces[0]
+    bbox = target_face['bbox'] # [x1, y1, x2, y2]
+    landmarks = target_face['landmarks']
+    confidence = target_face['confidence']
+    
+    x1, y1, x2, y2 = map(int, bbox)
+
+    # Filter Stage 1: Check Confidence
+    if confidence < DETECTION_THRESHOLD:
+        
+        return {
+            "bbox": [float(x1), float(y1), float(x2), float(y2)],
+            "match_user": None,
+            "confidence": 0.0,
+            "authorized": False,
+            "det_score": float(confidence),
+            "fake": True, # flag sebagai fake
+            "model_threshold": DETECTION_THRESHOLD,
+            "raw_distance": 0.0,
+            "reason": "Low quality/confidence detection"
+        }
+        
+    # STAGE 1.5 : LIVENESS CHECK (MINIFASNETV2)
+    if liveness is not None:
+        PROBS_REAL_IDX = 1  
+        probs = liveness.predict(img, bbox)
+        max_probs = np.argmax(probs)
+        live_score = probs[PROBS_REAL_IDX]
+        
+        if max_probs != PROBS_REAL_IDX:
+            return {
+                "bbox": [float(x1), float(y1), float(x2), float(y2)],
+                "match_user": None,
+                "confidence": 0.0,
+                "authorized": False,
+                "det_score": float(confidence),
+                "fake": True, 
+                "model_threshold": DETECTION_THRESHOLD,
+                "raw_distance": 0.0,
+                "reason": "Liveness check failed",
+                "live_score": float(live_score)
+            }
+    
+    # --- STAGE 2: RECOGNITION (ArcFace) ---
+    # Jika lolos Stage 1.5, lanjut ke Recognition
+    probe_emb = await run_in_threadpool(_compute_embedding_uniface, img, landmarks)
+    
+    if probe_emb is None:
+        return None
+
+    best_score = -1.0
+    best_user = None
+
+    # Determine which users to check
+    search_dirs = []
+    if target_id == "*":
+        for d in os.listdir(FACE_DB_ROOT):
+            if not d.startswith("_"):
+                search_dirs.append(d)
+    else:
+        if os.path.exists(os.path.join(FACE_DB_ROOT, target_id)):
+            search_dirs.append(target_id)
+
+    # Search in DB
+    for uid in search_dirs:
+        user_dir = os.path.join(FACE_DB_ROOT, uid)
+        files = os.listdir(user_dir)
+        
+        if not files:
+            print(f"[DEBUG] Folder {uid} kosong")
+        
+        for fname in os.listdir(user_dir):
+            if not fname.lower().endswith(('.jpg', '.png', '.jpeg')):
+                continue
+            
+            ref_path = os.path.join(user_dir, fname)
+            try:
+                # Load ref image
+                # Note: In production, embeddings should be cached in memory/database
+                ref_img = cv2.imread(ref_path)
+                if ref_img is None: continue
+                
+                ref_faces = detector.detect(ref_img)
+                
+                if not ref_faces:
+                    continue
+                
+                # Get embedding
+                ref_emb = _compute_embedding_uniface(ref_img, ref_faces[0]['landmarks'])
+                if ref_emb is not None:
+                    probe_flat = probe_emb.flatten()
+                    ref_flat = ref_emb.flatten()
+                    # Cosine similarity
+                    sim = np.dot(probe_flat, ref_flat)
+                    if sim > best_score:
+                        best_score = sim
+                        best_user = uid
+            except Exception:
+                continue
+
+    authorized = bool(best_score > SIM_THRESHOLD)
+    
+    return {
+        "bbox": [float(x1), float(y1), float(x2), float(y2)],
+        "match_user": best_user if authorized else None,
+        "confidence": round(float(best_score), 4),
+        "authorized": authorized,
+        "det_score": float(confidence),
+        "fake": False, # passed all stages
+        "live_score": float(live_score),
+        "model_threshold": SIM_THRESHOLD,
+        "raw_distance": float(best_score)
+    }
+
+
+@app.get("/health")
+async def health():
+    status = "ok"
+    if models["detector"] is None or models["recognizer"] is None or models["liveness"] is None:
+        status = "models_loading_or_failed"
+    return {"status": status, "system": "Uniface Pure (Stage1:Detect, Stage2:Recognize)"}
+
+
+@app.post("/face/enroll")
+async def face_enroll(payload: EnrollRequest):
+    if not payload.images:
+        raise HTTPException(status_code=400, detail="No images provided")
+    employee_id = payload.employee_id.strip() or "1"
+    save_dir = os.path.join(FACE_DB_ROOT, employee_id)
+    os.makedirs(save_dir, exist_ok=True)
+    count = 0
+    for s in payload.images:
+        img = _b64_to_bgr(s)
+        if img is None:
+            continue
+        out_path = os.path.join(save_dir, f"face_{count}.jpg")
+        cv2.imwrite(out_path, img)
+        count += 1
+    return {"employee_id": employee_id, "added": count, "total": len(os.listdir(save_dir))}
+
+
+@app.post("/face/enroll-files")
+async def face_enroll_files(
+    employee_id: str = Form(default="1"),
+    files: List[UploadFile] = File(default=[]),
+):
+    employee_id = employee_id.strip() or "1"
+    if not files:
+        raise HTTPException(status_code=400, detail="No files uploaded")
+    save_dir = os.path.join(FACE_DB_ROOT, employee_id)
+    os.makedirs(save_dir, exist_ok=True)
+    count = 0
+    for f in files:
+        try:
+            data = await f.read()
+            img = cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_COLOR)
+            if img is None:
+                continue
+            out_path = os.path.join(save_dir, f"face_{count}.jpg")
+            cv2.imwrite(out_path, img)
+            count += 1
+        except Exception:
+            continue
+    return {"employee_id": employee_id, "added": count, "total": len(os.listdir(save_dir))}
+
+@app.websocket("/face/verify")
+async def face_verify_ws(
+    websocket: WebSocket,
+    employee_id: str = Query(default="1"),
+    threshold: float = Query(default=SIM_THRESHOLD),
+):
+    await websocket.accept()
+    employee_id = employee_id.strip() or "1"
+    
+    motion_verifier = MotionVerifier(required_frames=5)
+
+    try:
+        while True:
+            try:
+                msg = await websocket.receive()
+            except WebSocketDisconnect:
+                break
+
+            img = None
+            if "bytes" in msg and msg["bytes"] is not None:
+                img = _bytes_to_bgr(msg["bytes"])
+            elif "text" in msg and msg["text"] is not None:
+                try:
+                    obj = json.loads(msg["text"])
+                    s = obj.get("frame") or obj.get("image")
+                    if isinstance(s, str):
+                        img = _b64_to_bgr(s)
+                except Exception:
+                    img = None
+
+            if img is None:
+                try:
+                    await websocket.send_json({"error": "no/invalid frame"})
+                except (RuntimeError, WebSocketDisconnect):
+                    break
+                continue
+
+            if models["detector"] is None:
+                await websocket.send_json({"error": "Model not loaded"})
+                break
+                
+            faces = await run_in_threadpool(models["detector"].detect, img)
+            
+            if not faces:
+                motion_verifier.reset()
+                await websocket.send_json({
+                    "status": "searching",
+                    "message": "Wajah tidak ditemukan, posisikan wajah di tengah..."
+                })
+                continue
+            
+            current_landmarks = faces[0]['landmarks']
+            motion_verifier.add_landmarks(current_landmarks)
+            
+            if not motion_verifier.is_ready():
+                progress = len(motion_verifier.landmarks_buffer)
+                await websocket.send_json({
+                    "status": "analyzing_motion",
+                    "progress": f"{progress}/{motion_verifier.required_frames}",
+                    "message": "Tahan posisi wajah, menganalisa gerakan..."
+                })
+                continue
+            
+            passed_motion, reason = motion_verifier.check_motion()
+            
+            if not passed_motion:
+                motion_verifier.reset()
+                await websocket.send_json({
+                    "status": "failed",
+                    "authorized": False,
+                    "reason": reason,
+                    "motion_check": False
+                })
+                continue
+            
+            try:
+                det = await _run_pipeline(img, employee_id)
+            except Exception as e:
+                try:
+                    await websocket.send_json({"error": f"{type(e).__name__}: {e}"})
+                except (RuntimeError, WebSocketDisconnect):
+                    break
+                continue
+
+            if det is None:
+                payload = {
+                    "employee_id": employee_id,
+                    "threshold": threshold,
+                    "detections": [],
+                    "count": 0,
+                    "authorized": False,
+                    "reason": "no face found",
+                }
+            else:
+                payload = {
+                    "employee_id": employee_id,
+                    "threshold": threshold,
+                    "detections": [det],
+                    "count": 1,
+                    "authorized": bool(det["authorized"]),
+                }
+
+            try:
+                await websocket.send_json(payload)
+            except (RuntimeError, WebSocketDisconnect):
+                break
+    finally:
+        return
+
+@app.post("/face/clear")
+async def face_clear(payload: ClearRequest):
+    employee_id = payload.employee_id.strip() or "1"
+    folder = os.path.join(FACE_DB_ROOT, employee_id)
+    removed = 0
+    if os.path.isdir(folder):
+        for f in os.listdir(folder):
+            try:
+                os.remove(os.path.join(folder, f))
+                removed += 1
+            except Exception:
+                pass
+        try:
+            os.rmdir(folder)
+        except Exception:
+            pass
+    return {"employee_id": employee_id, "removed": removed}

packages.txt

@@ -0,0 +1,3 @@
+libgl1
+libgl1-mesa-glx
+libglib2.0-0

requirements.txt

@@ -0,0 +1,12 @@
+fastapi
+uvicorn[standard]
+numpy
+onnxruntime-gpu
+onnxruntime
+pydantic
+starlette
+python-multipart
+Pillow
+typing_extensions
+uniface
+opencv-python