api.py

import os
import cv2
import csv
import time
import torch
import pickle
import random
import numpy as np
import requests
from collections import deque, defaultdict
from datetime import datetime
from insightface.app import FaceAnalysis
from train_arcface import SimpleClassifier
import mediapipe as mp

# 설정값
CONFIRM_FRAMES = 5
DUPLICATE_INTERVAL_SEC = 10
LIVENESS_DURATION = 3
device = 'cuda' if torch.cuda.is_available() else 'cpu'

# 로그 파일
os.makedirs('logs', exist_ok=True)
log_file = 'logs/entry_log.csv'

# 레이블 매핑
with open('label_to_id.pkl', 'rb') as f: label2id = pickle.load(f)
id2label = {v: k for k, v in label2id.items()}

# 분류기 로드
classifier = SimpleClassifier(512, len(label2id)).to(device)
ckpt = torch.load('classifier.pth', map_location=device)
classifier.load_state_dict(ckpt['model'])
classifier.eval()

# 얼굴 감지기
provider = 'CUDAExecutionProvider' if torch.cuda.is_available() else 'CPUExecutionProvider'
app = FaceAnalysis(name='buffalo_l', providers=[provider])
app.prepare(ctx_id=0, det_size=(864, 576), det_thresh=0.5)

# FaceMesh 초기화
mp_face_mesh = mp.solutions.face_mesh
face_mesh = mp_face_mesh.FaceMesh(static_image_mode=False, max_num_faces=1, refine_landmarks=True)

# 지시문 리스트
instructions = ["Please open your mouth", "Turn your head to the left", "Turn your head to the right"]
current_instruction = None
instruction_start_time = None
instruction_active_until = None
waiting_for_instruction = True
liveness_passed_flag = defaultdict(bool)

# 출입 상태 기록
user_status = defaultdict(lambda: 'EXIT')

# 상태 저장
buffer = deque(maxlen=CONFIRM_FRAMES)
last_logged = defaultdict(lambda: datetime.min)

def get_embedding(img):
    faces = app.get(img)
    if not faces: return None
    e = faces[0].normed_embedding
    return e / np.linalg.norm(e)

def log_entry(name, score):
    now = datetime.now()
    with open(log_file, 'a', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow([now.strftime('%Y-%m-%d %H:%M:%S'), name, f"{score:.4f}"])
    print(f"[LOG] {name} access logged (score={score:.4f})")

def send_access_log(identifier, access_type, similarity):
    url = "http://13.125.140.66:8080/api/access/face"
    payload = {
        "identifier": identifier,
        "accessType": access_type,
        "similarity": round(similarity, 4)
    }
    try:
        response = requests.post(url, json=payload, timeout=3)
        if response.status_code == 200:
            print(f"✅ 서버 전송 완료: {payload}")
        else:
            print(f"⚠️ 서버 응답 오류: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"❌ 서버 전송 실패: {e}")

def get_face_landmarks(image):
    img_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    result = face_mesh.process(img_rgb)
    if not result.multi_face_landmarks:
        return None
    h, w, _ = image.shape
    return [(int(p.x * w), int(p.y * h)) for p in result.multi_face_landmarks[0].landmark]

def check_instruction_performed(instruction, landmarks):
    if not landmarks: return False
    if instruction == "Please open your mouth":
        return abs(landmarks[14][1] - landmarks[13][1]) > 15
    if instruction == "Turn your head to the right":
        cx = (landmarks[33][0] + landmarks[263][0]) // 2
        return landmarks[1][0] < cx - 15
    if instruction == "Turn your head to the left":
        cx = (landmarks[33][0] + landmarks[263][0]) // 2
        return landmarks[1][0] > cx + 15
    return False

def liveness_check(frame, name):
    global current_instruction, instruction_start_time, instruction_active_until, waiting_for_instruction
    now = time.time()
    landmarks = get_face_landmarks(frame)

    if waiting_for_instruction:
        current_instruction = random.choice(instructions)
        instruction_start_time = now
        instruction_active_until = now + LIVENESS_DURATION
        waiting_for_instruction = False
        print(f"[{name}] Instruction: {current_instruction}")

    if now <= instruction_active_until:
        if check_instruction_performed(current_instruction, landmarks):
            print(f"[{name}] Liveness success")
            liveness_passed_flag[name] = True
            waiting_for_instruction = True
    else:
        print(f"[{name}] Liveness failed")
        liveness_passed_flag[name] = False
        waiting_for_instruction = True

# 웹캠 실행
cap = cv2.VideoCapture(0)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 864)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 576)

print("▶ Face recognition with liveness + API started (press Q to quit)")

while True:
    ret, frame = cap.read()
    if not ret: break
    rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

    emb = get_embedding(rgb)
    name = "No Face"
    display_text = ""
    color = (0, 0, 255)

    if emb is not None:
        inp = torch.tensor(emb, dtype=torch.float32).unsqueeze(0).to(device)
        out = classifier(inp)
        probs = torch.softmax(out, 1)
        conf, pred = probs.max(1)
        conf = conf.item(); pid = pred.item()

        if conf > 0.7:
            name = id2label[pid]
            #name = "21011805"  # ← 임시 학번 고정
            buffer.append(name)

            if buffer.count(name) >= CONFIRM_FRAMES:
                liveness_check(frame, name)

                if liveness_passed_flag[name]:
                    now = datetime.now()
                    if (now - last_logged[name]).total_seconds() >= DUPLICATE_INTERVAL_SEC:
                        log_entry(name, conf)
                        last_logged[name] = now
                        liveness_passed_flag[name] = False

                        # ENTRY/EXIT 스위칭 및 서버 전송
                        prev_status = user_status[name]
                        new_status = "ENTRY" if prev_status == "EXIT" else "EXIT"
                        user_status[name] = new_status
                        send_access_log(name, new_status, conf)

                    display_text = f"{name} ✅ ACCESS GRANTED"
                    color = (0, 255, 0)
                else:
                    display_text = f"{name} ❌ LIVENESS FAILED"
                    color = (0, 0, 255)
            else:
                display_text = f"{name} ⏳ RECOGNIZING..."
        else:
            buffer.append("Unknown")
            name = f"Unknown ({conf:.2f})"
            display_text = name
    else:
        buffer.append("No Face")

    cv2.putText(frame, display_text, (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, color, 2)
    if current_instruction:
        cv2.putText(frame, f"[Instruction] {current_instruction}", (50, 100), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 0), 2)

    cv2.imshow('Face Recognition with Liveness + API', frame)
    if cv2.waitKey(1) & 0xFF == ord('q'): break

cap.release()
cv2.destroyAllWindows()