Update main.py

main.py

@@ -2,7 +2,7 @@ import os
 import base64
 import asyncio
 import json
-from typing import List, Dict, Optional, Iterable
+from typing import List, Optional, Iterable
 
 import numpy as np
 import cv2
@@ -12,445 +12,246 @@ 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
 
-# Konfigurasi Threshold
-# Stage 1: Detection Confidence (Pengganti Liveness Check sementara)
-# Wajah dengan confidence di bawah ini dianggap tidak valid/buruk
-DETECTION_THRESHOLD = 0.70 
-
-# 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=["*"],
-)
+# =========================
+# CONFIG
+# =========================
+DETECTION_THRESHOLD = 0.70
+SIM_THRESHOLD = 0.40
 
 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(FACE_DB_ROOT, exist_ok=True)
 os.makedirs(LIVE_DB_ROOT, exist_ok=True)
 
 processing_lock = asyncio.Lock()
 
-# --- INITIALIZATION (UNIFACE ONLY) ---
+# =========================
+# INIT MODELS
+# =========================
 detector = None
 recognizer = None
 
 try:
-    print("[Uniface] Initializing RetinaFace (Stage 1)...")
-    detector = RetinaFace() 
-    print("[Uniface] Initializing ArcFace (Stage 2)...")
+    print("[Uniface] Initializing RetinaFace...")
+    detector = RetinaFace()
+    print("[Uniface] Initializing ArcFace...")
     recognizer = ArcFace()
 except Exception as e:
-    print(f"[Uniface] Error initializing models: {e}")
+    print("[Uniface] Init error:", 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 = Field(default="1")
-    images: List[str] = Field(..., description="Base64 images")
+    employee_id: str = "1"
+    images: List[str]
 
 class VerifyImageRequest(BaseModel):
-    employee_id: str = Field(default="1")
-    image: str = Field(..., description="Base64 image")
-    threshold: Optional[float] = Field(default=None)
+    employee_id: str = "1"
+    image: str
+    threshold: Optional[float] = None
 
 class ClearRequest(BaseModel):
-    employee_id: str = Field(default="1")
-
+    employee_id: str = "1"
 
+# =========================
+# HELPERS
+# =========================
 def _strip_b64(s: str) -> str:
-    if isinstance(s, str) and "," in s and s.lstrip().lower().startswith("data:"):
+    if "," in s and s.lower().startswith("data:"):
         return s.split(",", 1)[1]
     return s
 
-def _b64_to_bgr(b64: str) -> Optional[np.ndarray]:
+def _b64_to_bgr(b64: str):
     try:
         raw = base64.b64decode(_strip_b64(b64), validate=False)
-        arr = np.frombuffer(raw, np.uint8)
-        return cv2.imdecode(arr, cv2.IMREAD_COLOR)
+        return cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
     except Exception:
         return None
 
-def _bytes_to_bgr(data: bytes) -> Optional[np.ndarray]:
+def _bytes_to_bgr(data: bytes):
     try:
-        arr = np.frombuffer(data, np.uint8)
-        return cv2.imdecode(arr, cv2.IMREAD_COLOR)
+        return cv2.imdecode(np.frombuffer(data, np.uint8), cv2.IMREAD_COLOR)
     except Exception:
         return None
 
-def _frame_path_for(employee_id: str) -> str:
-    safe = (employee_id or "live").replace("/", "_")
+def _frame_path(employee_id: str):
+    safe = employee_id.replace("/", "_")
     return os.path.join(LIVE_DB_ROOT, f"{safe}.jpg")
 
-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()
-    if "multipart/form-data" in ct:
-        form = await request.form()
-        for name in field_names:
-            file = form.get(name)
-            if isinstance(file, UploadFile):
-                data = await file.read()
-                return _bytes_to_bgr(data)
-        for name in field_names:
-            raw = form.get(name)
-            if isinstance(raw, (bytes, bytearray)):
-                return _bytes_to_bgr(bytes(raw))
-            if isinstance(raw, str):
-                img = _b64_to_bgr(raw)
-                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
+def _compute_embedding(img, landmarks):
     try:
-        # get_normalized_embedding requires the image and landmarks
-        embedding = recognizer.get_normalized_embedding(img_bgr, landmarks)
-        return embedding
+        return recognizer.get_normalized_embedding(img, landmarks)
     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: DETECTION & QUALITY CHECK ---
-    # Menggunakan RetinaFace untuk mendeteksi wajah
+    # -------- STAGE 1: DETECT --------
     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
-    # Jika confidence rendah, anggap sebagai "Fake" atau "Low Quality" dan tolak
-    if confidence < DETECTION_THRESHOLD:
+    face = faces[0]
+    bbox = list(map(float, face["bbox"]))
+    landmarks = face["landmarks"]
+    det_conf = float(face["confidence"])
+
+    if det_conf < DETECTION_THRESHOLD:
         return {
-            "bbox": [float(x1), float(y1), float(x2), float(y2)],
+            "bbox": bbox,
             "match_user": None,
             "confidence": 0.0,
             "authorized": False,
-            "det_score": float(confidence),
-            "fake": True, # Flagged as fake/bad quality
+            "det_score": det_conf,
+            "fake": True,
             "model_threshold": DETECTION_THRESHOLD,
             "raw_distance": 0.0,
-            "reason": "Low quality/confidence detection"
+            "reason": "low detection confidence",
         }
 
-    # --- STAGE 2: RECOGNITION (ArcFace) ---
-    # Jika lolos Stage 1, lanjut ke Recognition
-    probe_emb = await run_in_threadpool(_compute_embedding_uniface, img, landmarks)
+    # -------- STAGE 2: EMBEDDING --------
+    probe_emb = await run_in_threadpool(_compute_embedding, img, landmarks)
     if probe_emb is None:
         return None
 
-    best_score = -1.0
+    best_score = 0.0
     best_user = None
 
-    # Determine which users to check
+    # users to search
     search_dirs = []
     if target_id == "*":
-        for d in os.listdir(FACE_DB_ROOT):
-            if not d.startswith("_"):
-                search_dirs.append(d)
+        search_dirs = [d for d in os.listdir(FACE_DB_ROOT) if not d.startswith("_")]
     else:
-        if os.path.exists(os.path.join(FACE_DB_ROOT, target_id)):
-            search_dirs.append(target_id)
+        if os.path.isdir(os.path.join(FACE_DB_ROOT, target_id)):
+            search_dirs = [target_id]
 
-    # Search in DB
+    # -------- MATCHING --------
     for uid in search_dirs:
         user_dir = os.path.join(FACE_DB_ROOT, uid)
         for fname in os.listdir(user_dir):
-            if not fname.lower().endswith(('.jpg', '.png', '.jpeg')):
+            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
-
-                # Detect face in ref image to get landmarks
-                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:
-                    # Cosine similarity
-                    sim = np.dot(probe_emb, ref_emb)
-                    if sim > best_score:
-                        best_score = sim
-                        best_user = uid
-            except Exception:
+
+            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:
                 continue
 
-    authorized = (best_score > SIM_THRESHOLD)
-    
+            ref_emb = _compute_embedding(ref_img, ref_faces[0]["landmarks"])
+            if ref_emb is None:
+                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
+
     return {
-        "bbox": [float(x1), float(y1), float(x2), float(y2)],
+        "bbox": bbox,
         "match_user": best_user if authorized else None,
-        "confidence": round(float(best_score), 4),
+        "confidence": round(best_score, 4),
         "authorized": authorized,
-        "det_score": float(confidence),
-        "fake": False, # Passed Stage 1
+        "det_score": det_conf,
+        "fake": False,
         "model_threshold": SIM_THRESHOLD,
-        "raw_distance": float(best_score)
+        "raw_distance": round(best_score, 4),
     }
 
-
+# =========================
+# ENDPOINTS
+# =========================
 @app.get("/health")
 async def health():
-    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)"}
-
+    return {"status": "ok" if detector and recognizer else "model_error"}
 
 @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)
+async def enroll(payload: EnrollRequest):
+    path = os.path.join(FACE_DB_ROOT, payload.employee_id)
+    os.makedirs(path, 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)
+        cv2.imwrite(os.path.join(path, f"face_{count}.jpg"), 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 face_verify(
+async def verify(
     request: Request,
-    employee_id: str = Query(default="1"),
-    threshold: float = Query(default=SIM_THRESHOLD),
+    employee_id: str = Query("1"),
+    threshold: float = Query(SIM_THRESHOLD),
 ):
-    employee_id = employee_id.strip() or "1"
-    
     img = await _decode_image_from_request(request, ("frame", "image"))
     if img is None:
-        raise HTTPException(status_code=400, detail="No image provided")
-
-    frame_path = _frame_path_for(employee_id)
-    cv2.imwrite(frame_path, img)
-
-    try:
-        det = await _run_pipeline(frame_path, employee_id)
-        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
+        raise HTTPException(400, "no image")
 
-    frame_path = _frame_path_for(employee_id)
+    frame_path = _frame_path(employee_id)
     cv2.imwrite(frame_path, img)
 
     det = await _run_pipeline(frame_path, employee_id)
     if det is None:
         return {
             "employee_id": employee_id,
-            "threshold": thr,
-            "detections": [],
-            "count": 0,
             "authorized": False,
-            "reason": "no face found",
+            "detections": [],
         }
+
     return {
         "employee_id": employee_id,
-        "threshold": thr,
+        "authorized": det["authorized"],
         "detections": [det],
-        "count": 1,
-        "authorized": bool(det["authorized"]),
     }
 
-
 @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)
+async def clear(payload: ClearRequest):
+    path = os.path.join(FACE_DB_ROOT, payload.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}
+    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}