Deploy files from GitHub repository with LFS

.gitignore

@@ -1,2 +1,5 @@
 ./face_db
 myenv/
+facerecog/
+test_verify-image.py
+test_verify.py

main.py

@@ -2,7 +2,7 @@ import os
 import base64
 import asyncio
 import json
-from typing import List, Optional, Iterable
+from typing import List, Dict, Optional, Iterable
 
 import numpy as np
 import cv2
@@ -12,246 +12,464 @@ from starlette.concurrency import run_in_threadpool
 from starlette.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
+# Import Uniface (Single Library)
 from uniface import RetinaFace, ArcFace
 
-# =========================
-# CONFIG
-# =========================
-DETECTION_THRESHOLD = 0.70
-SIM_THRESHOLD = 0.40
+# Konfigurasi Threshold
+# Stage 1: Detection Confidence (Pengganti Liveness Check sementara)
+# Wajah dengan confidence di bawah ini dianggap tidak valid/buruk
+DETECTION_THRESHOLD = 0.70 
 
-FACE_DB_ROOT = "face_db"
-LIVE_DB_ROOT = os.path.join(FACE_DB_ROOT, "_live")
+# 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()
 
-# =========================
-# INIT MODELS
-# =========================
+# --- INITIALIZATION (UNIFACE ONLY) ---
 detector = None
 recognizer = None
 
 try:
-    print("[Uniface] Initializing RetinaFace...")
-    detector = RetinaFace()
-    print("[Uniface] Initializing ArcFace...")
+    print("[Uniface] Initializing RetinaFace (Stage 1)...")
+    detector = RetinaFace() 
+    print("[Uniface] Initializing ArcFace (Stage 2)...")
     recognizer = ArcFace()
 except Exception as e:
-    print("[Uniface] Init error:", e)
+    print(f"[Uniface] Error initializing models: {e}")
 
-# =========================
-# FASTAPI
-# =========================
-app = FastAPI(
-    title="Face Verification API (Uniface Fixed)",
-    version="2.3.0",
-)
 
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# =========================
-# SCHEMAS
-# =========================
 class EnrollRequest(BaseModel):
-    employee_id: str = "1"
-    images: List[str]
+    employee_id: str = Field(default="1")
+    images: List[str] = Field(..., description="Base64 images")
 
 class VerifyImageRequest(BaseModel):
-    employee_id: str = "1"
-    image: str
-    threshold: Optional[float] = None
+    employee_id: str = Field(default="1")
+    image: str = Field(..., description="Base64 image")
+    threshold: Optional[float] = Field(default=None)
 
 class ClearRequest(BaseModel):
-    employee_id: str = "1"
+    employee_id: str = Field(default="1")
+
 
-# =========================
-# HELPERS
-# =========================
 def _strip_b64(s: str) -> str:
-    if "," in s and s.lower().startswith("data:"):
+    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):
+def _b64_to_bgr(b64: str) -> Optional[np.ndarray]:
     try:
         raw = base64.b64decode(_strip_b64(b64), validate=False)
-        return cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
+        arr = np.frombuffer(raw, np.uint8)
+        return cv2.imdecode(arr, cv2.IMREAD_COLOR)
     except Exception:
         return None
 
-def _bytes_to_bgr(data: bytes):
+def _bytes_to_bgr(data: bytes) -> Optional[np.ndarray]:
     try:
-        return cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_COLOR)
+        arr = np.frombuffer(data, np.uint8)
+        return cv2.imdecode(arr, cv2.IMREAD_COLOR)
     except Exception:
         return None
 
-def _frame_path(employee_id: str):
-    safe = employee_id.replace("/", "_")
+def _frame_path_for(employee_id: str) -> str:
+    safe = (employee_id or "live").replace("/", "_")
     return os.path.join(LIVE_DB_ROOT, f"{safe}.jpg")
 
-def _compute_embedding(img, landmarks):
+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 recognizer is None:
+        return None
     try:
-        return recognizer.get_normalized_embedding(img, landmarks)
+        # get_normalized_embedding requires the image and landmarks
+        embedding = recognizer.get_normalized_embedding(img_bgr, landmarks)
+        return embedding
     except Exception:
         return None
 
-# =========================
-# PIPELINE (FIXED)
-# =========================
 async def _run_pipeline(img_path: str, target_id: str):
     if detector is None or recognizer is None:
         return None
 
+    # Read image
     img = cv2.imread(img_path)
     if img is None:
         return None
 
-    # -------- STAGE 1: DETECT --------
+    # --- 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
 
-    face = faces[0]
-    bbox = list(map(float, face["bbox"]))
-    landmarks = face["landmarks"]
-    det_conf = float(face["confidence"])
-
-    if det_conf < DETECTION_THRESHOLD:
+    # 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
+    # Jika confidence rendah, anggap sebagai "Fake" atau "Low Quality" dan tolak
+    if confidence < DETECTION_THRESHOLD:
+        
         return {
-            "bbox": bbox,
+            "bbox": [float(x1), float(y1), float(x2), float(y2)],
             "match_user": None,
             "confidence": 0.0,
             "authorized": False,
-            "det_score": det_conf,
-            "fake": True,
+            "det_score": float(confidence),
+            "fake": True, # Flagged as fake/bad quality
             "model_threshold": DETECTION_THRESHOLD,
             "raw_distance": 0.0,
-            "reason": "low detection confidence",
+            "reason": "Low quality/confidence detection"
         }
 
-    # -------- STAGE 2: EMBEDDING --------
-    probe_emb = await run_in_threadpool(_compute_embedding, img, landmarks)
+    # --- STAGE 2: RECOGNITION (ArcFace) ---
+    # Jika lolos Stage 1, lanjut ke Recognition
+    probe_emb = await run_in_threadpool(_compute_embedding_uniface, img, landmarks)
     if probe_emb is None:
         return None
 
-    best_score = 0.0
+    best_score = -1.0
     best_user = None
 
-    # users to search
+    # Determine which users to check
     search_dirs = []
     if target_id == "*":
-        search_dirs = [d for d in os.listdir(FACE_DB_ROOT) if not d.startswith("_")]
+        for d in os.listdir(FACE_DB_ROOT):
+            if not d.startswith("_"):
+                search_dirs.append(d)
     else:
-        if os.path.isdir(os.path.join(FACE_DB_ROOT, target_id)):
-            search_dirs = [target_id]
+        if os.path.exists(os.path.join(FACE_DB_ROOT, target_id)):
+            search_dirs.append(target_id)
 
-    # -------- MATCHING --------
+    # 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_img = cv2.imread(os.path.join(user_dir, fname))
-            if ref_img is None:
-                continue
-
-            # IMPORTANT FIX: lock + threadpool
-            async with processing_lock:
-                ref_faces = await run_in_threadpool(detector.detect, ref_img)
-
-            if not ref_faces:
+            if not fname.lower().endswith(('.jpg', '.png', '.jpeg')):
                 continue
-
-            ref_emb = _compute_embedding(ref_img, ref_faces[0]["landmarks"])
-            if ref_emb is None:
+            
+            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
 
-            sim = float(np.dot(probe_emb, ref_emb))
-            if sim > best_score:
-                best_score = sim
-                best_user = uid
-
-    authorized = best_user is not None and best_score >= SIM_THRESHOLD
-
+    authorized = bool(best_score > SIM_THRESHOLD)
+    
     return {
-        "bbox": bbox,
+        "bbox": [float(x1), float(y1), float(x2), float(y2)],
         "match_user": best_user if authorized else None,
-        "confidence": round(best_score, 4),
+        "confidence": round(float(best_score), 4),
         "authorized": authorized,
-        "det_score": det_conf,
-        "fake": False,
+        "det_score": float(confidence),
+        "fake": False, # Passed Stage 1
         "model_threshold": SIM_THRESHOLD,
-        "raw_distance": round(best_score, 4),
+        "raw_distance": float(best_score)
     }
 
-# =========================
-# ENDPOINTS
-# =========================
+
 @app.get("/health")
 async def health():
-    return {"status": "ok" if detector and recognizer else "model_error"}
+    status = "ok"
+    if detector is None or recognizer is None:
+        status = "models_loading_or_failed"
+    return {"status": status, "system": "Uniface Pure (Stage1:Detect, Stage2:Recognize)"}
 
-@app.post("/face/enroll")
-async def enroll(payload: EnrollRequest):
-    path = os.path.join(FACE_DB_ROOT, payload.employee_id)
-    os.makedirs(path, exist_ok=True)
 
+@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
-        cv2.imwrite(os.path.join(path, f"face_{count}.jpg"), img)
+        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))}
 
-    return {"employee_id": payload.employee_id, "added": count}
 
 @app.post("/face/verify")
-async def verify(
+async def face_verify(
     request: Request,
-    employee_id: str = Query("1"),
-    threshold: float = Query(SIM_THRESHOLD),
+    employee_id: str = Query(default="1"),
+    threshold: float = Query(default=SIM_THRESHOLD),
 ):
+    employee_id = employee_id.strip() or "1"
+    
     img = await _decode_image_from_request(request, ("frame", "image"))
     if img is None:
-        raise HTTPException(400, "no image")
+        raise HTTPException(status_code=400, detail="No image provided")
 
-    frame_path = _frame_path(employee_id)
+    frame_path = _frame_path_for(employee_id)
+    cv2.imwrite(frame_path, img)
+
+    try:
+        det = await _run_pipeline(frame_path, employee_id)
+        print(f"[DEBUG] Pipeline result: {det}")
+        if det is None:
+            return {
+                "employee_id": employee_id,
+                "threshold": threshold,
+                "detections": [],
+                "count": 0,
+                "authorized": False,
+                "reason": "no face found",
+            }
+        
+        return {
+            "employee_id": employee_id,
+            "threshold": threshold,
+            "detections": [det],
+            "count": 1,
+            "authorized": bool(det["authorized"]),
+        }
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        return JSONResponse(status_code=500, content={"error": f"{type(e).__name__}: {e}"})
+
+
+@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"
+
+    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
+
+            frame_path = _frame_path_for(employee_id)
+            cv2.imwrite(frame_path, img)
+
+            try:
+                det = await _run_pipeline(frame_path, 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/verify-image")
+async def face_verify_image(payload: VerifyImageRequest):
+    img = _b64_to_bgr(payload.image)
+    if img is None:
+        raise HTTPException(status_code=400, detail="Invalid or missing image")
+    employee_id = payload.employee_id.strip() or "1"
+    thr = payload.threshold if payload.threshold is not None else SIM_THRESHOLD
+
+    frame_path = _frame_path_for(employee_id)
     cv2.imwrite(frame_path, img)
 
     det = await _run_pipeline(frame_path, employee_id)
     if det is None:
         return {
             "employee_id": employee_id,
-            "authorized": False,
+            "threshold": thr,
             "detections": [],
+            "count": 0,
+            "authorized": False,
+            "reason": "no face found",
         }
-
     return {
         "employee_id": employee_id,
-        "authorized": det["authorized"],
+        "threshold": thr,
         "detections": [det],
+        "count": 1,
+        "authorized": bool(det["authorized"]),
     }
 
+
 @app.post("/face/clear")
-async def clear(payload: ClearRequest):
-    path = os.path.join(FACE_DB_ROOT, payload.employee_id)
+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(path):
-        for f in os.listdir(path):
-            os.remove(os.path.join(path, f))
-            removed += 1
-        os.rmdir(path)
-    return {"employee_id": payload.employee_id, "removed": removed}
+    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}