wuhp commited on
Commit
4166933
·
verified ·
1 Parent(s): 2bb9b3a

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +3 -982
app.py CHANGED
@@ -1,985 +1,6 @@
1
- import os
2
- import sys
3
- import math
4
- import requests
5
- import numpy as np
6
- import cv2
7
- import torch
8
- import pickle
9
- import logging
10
- from PIL import Image
11
- from typing import Optional, Dict, List, Tuple
12
- from dataclasses import dataclass, field
13
- from collections import Counter
14
-
15
  import gradio as gr
 
16
 
17
- from ultralytics import YOLO
18
- from facenet_pytorch import InceptionResnetV1
19
- from torchvision import transforms
20
- from deep_sort_realtime.deepsort_tracker import DeepSort
21
-
22
- import mediapipe as mp
23
-
24
- logging.basicConfig(
25
- level=logging.INFO,
26
- format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
27
- handlers=[logging.FileHandler('face_pipeline.log'), logging.StreamHandler()],
28
- )
29
- logger = logging.getLogger(__name__)
30
-
31
- logging.getLogger('torch').setLevel(logging.ERROR)
32
- logging.getLogger('mediapipe').setLevel(logging.ERROR)
33
- logging.getLogger('deep_sort_realtime').setLevel(logging.ERROR)
34
-
35
- DEFAULT_MODEL_URL = "https://github.com/wuhplaptop/face-11-n/blob/main/face2.pt?raw=true"
36
- DEFAULT_DB_PATH = os.path.expanduser("~/.face_pipeline/known_faces.pkl")
37
- MODEL_DIR = os.path.expanduser("~/.face_pipeline/models")
38
- CONFIG_PATH = os.path.expanduser("~/.face_pipeline/config.pkl")
39
-
40
- LEFT_EYE_IDX = [33, 160, 158, 133, 153, 144]
41
- RIGHT_EYE_IDX = [263, 387, 385, 362, 380, 373]
42
-
43
- mp_drawing = mp.solutions.drawing_utils
44
- mp_face_mesh = mp.solutions.face_mesh
45
- mp_hands = mp.solutions.hands
46
-
47
- @dataclass
48
- class PipelineConfig:
49
- detector: Dict = field(default_factory=dict)
50
- tracker: Dict = field(default_factory=dict)
51
- recognition: Dict = field(default_factory=dict)
52
- anti_spoof: Dict = field(default_factory=dict)
53
- blink: Dict = field(default_factory=dict)
54
- face_mesh_options: Dict = field(default_factory=dict)
55
- hand: Dict = field(default_factory=dict)
56
- eye_color: Dict = field(default_factory=dict)
57
- enabled_components: Dict = field(default_factory=dict)
58
-
59
- detection_conf_thres: float = 0.4
60
- recognition_conf_thres: float = 0.85
61
-
62
- bbox_color: Tuple[int, int, int] = (0, 255, 0)
63
- spoofed_bbox_color: Tuple[int, int, int] = (0, 0, 255)
64
- unknown_bbox_color: Tuple[int, int, int] = (0, 0, 255)
65
- eye_outline_color: Tuple[int, int, int] = (255, 255, 0)
66
- blink_text_color: Tuple[int, int, int] = (0, 0, 255)
67
- hand_landmark_color: Tuple[int, int, int] = (255, 210, 77)
68
- hand_connection_color: Tuple[int, int, int] = (204, 102, 0)
69
- hand_text_color: Tuple[int, int, int] = (255, 255, 255)
70
- mesh_color: Tuple[int, int, int] = (100, 255, 100)
71
- contour_color: Tuple[int, int, int] = (200, 200, 0)
72
- iris_color: Tuple[int, int, int] = (255, 0, 255)
73
- eye_color_text_color: Tuple[int, int, int] = (255, 255, 255)
74
-
75
- def __post_init__(self):
76
- self.detector = self.detector or {
77
- 'model_path': os.path.join(MODEL_DIR, "face2.pt"),
78
- 'device': 'cuda' if torch.cuda.is_available() else 'cpu',
79
- }
80
- self.tracker = self.tracker or {'max_age': 30}
81
- self.recognition = self.recognition or {'enable': True}
82
- self.anti_spoof = self.anti_spoof or {'enable': True, 'lap_thresh': 80.0}
83
- self.blink = self.blink or {'enable': True, 'ear_thresh': 0.25}
84
- self.face_mesh_options = self.face_mesh_options or {
85
- 'enable': False,
86
- 'tesselation': False,
87
- 'contours': False,
88
- 'irises': False,
89
- }
90
- self.hand = self.hand or {
91
- 'enable': True,
92
- 'min_detection_confidence': 0.5,
93
- 'min_tracking_confidence': 0.5,
94
- }
95
- self.eye_color = self.eye_color or {'enable': False}
96
- self.enabled_components = self.enabled_components or {
97
- 'detection': True,
98
- 'tracking': True,
99
- 'anti_spoof': True,
100
- 'recognition': True,
101
- 'blink': True,
102
- 'face_mesh': False,
103
- 'hand': True,
104
- 'eye_color': False,
105
- }
106
-
107
- def save(self, path: str):
108
- try:
109
- os.makedirs(os.path.dirname(path), exist_ok=True)
110
- with open(path, 'wb') as f:
111
- pickle.dump(self.__dict__, f)
112
- logger.info(f"Saved config to {path}")
113
- except Exception as e:
114
- logger.error(f"Config save failed: {str(e)}")
115
- raise RuntimeError(f"Config save failed: {str(e)}") from e
116
-
117
- @classmethod
118
- def load(cls, path: str) -> 'PipelineConfig':
119
- try:
120
- if os.path.exists(path):
121
- with open(path, 'rb') as f:
122
- data = pickle.load(f)
123
- return cls(**data)
124
- return cls()
125
- except Exception as e:
126
- logger.error(f"Config load failed: {str(e)}")
127
- return cls()
128
-
129
- class FaceDatabase:
130
- def __init__(self, db_path: str = DEFAULT_DB_PATH):
131
- self.db_path = db_path
132
- self.embeddings: Dict[str, List[np.ndarray]] = {}
133
- self._load()
134
-
135
- def _load(self):
136
- try:
137
- if os.path.exists(self.db_path):
138
- with open(self.db_path, 'rb') as f:
139
- self.embeddings = pickle.load(f)
140
- logger.info(f"Loaded database from {self.db_path}")
141
- except Exception as e:
142
- logger.error(f"Database load failed: {str(e)}")
143
- self.embeddings = {}
144
-
145
- def save(self):
146
- try:
147
- os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
148
- with open(self.db_path, 'wb') as f:
149
- pickle.dump(self.embeddings, f)
150
- logger.info(f"Saved database to {self.db_path}")
151
- except Exception as e:
152
- logger.error(f"Database save failed: {str(e)}")
153
- raise RuntimeError(f"Database save failed: {str(e)}") from e
154
-
155
- def add_embedding(self, label: str, embedding: np.ndarray):
156
- try:
157
- if not isinstance(embedding, np.ndarray) or embedding.ndim != 1:
158
- raise ValueError("Invalid embedding format")
159
- if label not in self.embeddings:
160
- self.embeddings[label] = []
161
- self.embeddings[label].append(embedding)
162
- logger.debug(f"Added embedding for {label}")
163
- except Exception as e:
164
- logger.error(f"Add embedding failed: {str(e)}")
165
- raise
166
-
167
- def remove_label(self, label: str):
168
- try:
169
- if label in self.embeddings:
170
- del self.embeddings[label]
171
- logger.info(f"Removed {label}")
172
- else:
173
- logger.warning(f"Label {label} not found")
174
- except Exception as e:
175
- logger.error(f"Remove label failed: {str(e)}")
176
- raise
177
-
178
- def list_labels(self) -> List[str]:
179
- return list(self.embeddings.keys())
180
-
181
- def get_embeddings_by_label(self, label: str) -> Optional[List[np.ndarray]]:
182
- return self.embeddings.get(label)
183
-
184
- def search_by_image(self, query_embedding: np.ndarray, threshold: float = 0.7) -> List[Tuple[str, float]]:
185
- results = []
186
- for lbl, embs in self.embeddings.items():
187
- for db_emb in embs:
188
- similarity = FacePipeline.cosine_similarity(query_embedding, db_emb)
189
- if similarity >= threshold:
190
- results.append((lbl, similarity))
191
- return sorted(results, key=lambda x: x[1], reverse=True)
192
-
193
- class YOLOFaceDetector:
194
- def __init__(self, model_path: str, device: str = 'cpu'):
195
- self.model = None
196
- self.device = device
197
- try:
198
- if not os.path.exists(model_path):
199
- logger.info(f"Model not found at {model_path}. Downloading from GitHub...")
200
- resp = requests.get(DEFAULT_MODEL_URL)
201
- resp.raise_for_status()
202
- os.makedirs(os.path.dirname(model_path), exist_ok=True)
203
- with open(model_path, 'wb') as f:
204
- f.write(resp.content)
205
- logger.info(f"Downloaded YOLO model to {model_path}")
206
-
207
- self.model = YOLO(model_path)
208
- self.model.to(device)
209
- logger.info(f"Loaded YOLO model from {model_path}")
210
- except Exception as e:
211
- logger.error(f"YOLO init failed: {str(e)}")
212
- raise
213
-
214
- def detect(self, image: np.ndarray, conf_thres: float) -> List[Tuple[int, int, int, int, float, int]]:
215
- try:
216
- results = self.model.predict(
217
- source=image, conf=conf_thres, verbose=False, device=self.device
218
- )
219
- detections = []
220
- for result in results:
221
- for box in result.boxes:
222
- x1, y1, x2, y2 = box.xyxy[0].cpu().numpy()
223
- conf = float(box.conf[0].cpu().numpy())
224
- cls = int(box.cls[0].cpu().numpy()) if box.cls is not None else 0
225
- detections.append((int(x1), int(y1), int(x2), int(y2), conf, cls))
226
- logger.debug(f"Detected {len(detections)} faces.")
227
- return detections
228
- except Exception as e:
229
- logger.error(f"Detection error: {str(e)}")
230
- return []
231
-
232
- class FaceTracker:
233
- def __init__(self, max_age: int = 30):
234
- self.tracker = DeepSort(max_age=max_age, embedder='mobilenet')
235
-
236
- def update(self, detections: List[Tuple], frame: np.ndarray):
237
- try:
238
- ds_detections = [
239
- ([x1, y1, x2 - x1, y2 - y1], conf, cls)
240
- for (x1, y1, x2, y2, conf, cls) in detections
241
- ]
242
- tracks = self.tracker.update_tracks(ds_detections, frame=frame)
243
- logger.debug(f"Updated tracker with {len(tracks)} tracks.")
244
- return tracks
245
- except Exception as e:
246
- logger.error(f"Tracking error: {str(e)}")
247
- return []
248
-
249
- class FaceNetEmbedder:
250
- def __init__(self, device: str = 'cpu'):
251
- self.device = device
252
- self.model = InceptionResnetV1(pretrained='vggface2').eval().to(device)
253
- self.transform = transforms.Compose([
254
- transforms.Resize((160, 160)),
255
- transforms.ToTensor(),
256
- transforms.Normalize([0.5]*3, [0.5]*3),
257
- ])
258
-
259
- def get_embedding(self, face_bgr: np.ndarray) -> Optional[np.ndarray]:
260
- try:
261
- face_rgb = cv2.cvtColor(face_bgr, cv2.COLOR_BGR2RGB)
262
- pil_img = Image.fromarray(face_rgb).convert('RGB')
263
- tens = self.transform(pil_img).unsqueeze(0).to(self.device)
264
- with torch.no_grad():
265
- embedding = self.model(tens)[0].cpu().numpy()
266
- logger.debug(f"Generated embedding sample: {embedding[:5]}...")
267
- return embedding
268
- except Exception as e:
269
- logger.error(f"Embedding failed: {str(e)}")
270
- return None
271
-
272
- def detect_blink(face_roi: np.ndarray, threshold: float = 0.25) -> Tuple[bool, float, float, np.ndarray, np.ndarray]:
273
- """
274
- Returns:
275
- (blink_bool, left_ear, right_ear, left_eye_points, right_eye_points).
276
- """
277
- try:
278
- face_mesh_proc = mp_face_mesh.FaceMesh(
279
- static_image_mode=True,
280
- max_num_faces=1,
281
- refine_landmarks=True,
282
- min_detection_confidence=0.5
283
- )
284
- result = face_mesh_proc.process(cv2.cvtColor(face_roi, cv2.COLOR_BGR2RGB))
285
- face_mesh_proc.close()
286
-
287
- if not result.multi_face_landmarks:
288
- return False, 0.0, 0.0, None, None
289
-
290
- landmarks = result.multi_face_landmarks[0].landmark
291
- h, w = face_roi.shape[:2]
292
-
293
- def eye_aspect_ratio(indices):
294
- pts = [(landmarks[i].x * w, landmarks[i].y * h) for i in indices]
295
- vertical = np.linalg.norm(np.array(pts[1]) - np.array(pts[5])) + \
296
- np.linalg.norm(np.array(pts[2]) - np.array(pts[4]))
297
- horizontal = np.linalg.norm(np.array(pts[0]) - np.array(pts[3]))
298
- return vertical / (2.0 * horizontal + 1e-6)
299
-
300
- left_ear = eye_aspect_ratio(LEFT_EYE_IDX)
301
- right_ear = eye_aspect_ratio(RIGHT_EYE_IDX)
302
-
303
- blink = (left_ear < threshold) and (right_ear < threshold)
304
-
305
- left_eye_pts = np.array([(int(landmarks[i].x * w), int(landmarks[i].y * h)) for i in LEFT_EYE_IDX])
306
- right_eye_pts = np.array([(int(landmarks[i].x * w), int(landmarks[i].y * h)) for i in RIGHT_EYE_IDX])
307
-
308
- return blink, left_ear, right_ear, left_eye_pts, right_eye_pts
309
-
310
- except Exception as e:
311
- logger.error(f"Blink detection error: {str(e)}")
312
- return False, 0.0, 0.0, None, None
313
-
314
- def process_face_mesh(face_roi: np.ndarray):
315
- try:
316
- fm_proc = mp_face_mesh.FaceMesh(
317
- static_image_mode=True,
318
- max_num_faces=1,
319
- refine_landmarks=True,
320
- min_detection_confidence=0.5
321
- )
322
- result = fm_proc.process(cv2.cvtColor(face_roi, cv2.COLOR_BGR2RGB))
323
- fm_proc.close()
324
- if result.multi_face_landmarks:
325
- return result.multi_face_landmarks[0]
326
- return None
327
- except Exception as e:
328
- logger.error(f"Face mesh error: {str(e)}")
329
- return None
330
-
331
- def draw_face_mesh(image: np.ndarray, face_landmarks, config: Dict, pipeline_config: PipelineConfig):
332
- mesh_color_bgr = pipeline_config.mesh_color[::-1]
333
- contour_color_bgr = pipeline_config.contour_color[::-1]
334
- iris_color_bgr = pipeline_config.iris_color[::-1]
335
-
336
- if config.get('tesselation'):
337
- mp_drawing.draw_landmarks(
338
- image,
339
- face_landmarks,
340
- mp_face_mesh.FACEMESH_TESSELATION,
341
- landmark_drawing_spec=mp_drawing.DrawingSpec(color=mesh_color_bgr, thickness=1, circle_radius=1),
342
- connection_drawing_spec=mp_drawing.DrawingSpec(color=mesh_color_bgr, thickness=1),
343
- )
344
- if config.get('contours'):
345
- mp_drawing.draw_landmarks(
346
- image,
347
- face_landmarks,
348
- mp_face_mesh.FACEMESH_CONTOURS,
349
- landmark_drawing_spec=None,
350
- connection_drawing_spec=mp_drawing.DrawingSpec(color=contour_color_bgr, thickness=2)
351
- )
352
- if config.get('irises'):
353
- mp_drawing.draw_landmarks(
354
- image,
355
- face_landmarks,
356
- mp_face_mesh.FACEMESH_IRISES,
357
- landmark_drawing_spec=None,
358
- connection_drawing_spec=mp_drawing.DrawingSpec(color=iris_color_bgr, thickness=2)
359
- )
360
-
361
- EYE_COLOR_RANGES = {
362
- "amber": (255, 191, 0),
363
- "blue": (0, 0, 255),
364
- "brown": (139, 69, 19),
365
- "green": (0, 128, 0),
366
- "gray": (128, 128, 128),
367
- "hazel": (102, 51, 0),
368
- }
369
-
370
- def classify_eye_color(rgb_color: Tuple[int,int,int]) -> str:
371
- if rgb_color is None:
372
- return "Unknown"
373
- min_dist = float('inf')
374
- best = "Unknown"
375
- for color_name, ref_rgb in EYE_COLOR_RANGES.items():
376
- dist = math.sqrt(sum([(a-b)**2 for a,b in zip(rgb_color, ref_rgb)]))
377
- if dist < min_dist:
378
- min_dist = dist
379
- best = color_name
380
- return best
381
-
382
- def get_dominant_color(image_roi, k=3):
383
- if image_roi.size == 0:
384
- return None
385
- pixels = np.float32(image_roi.reshape(-1, 3))
386
- criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 100, 0.1)
387
- _, labels, palette = cv2.kmeans(pixels, k, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS)
388
- _, counts = np.unique(labels, return_counts=True)
389
- dom_color = tuple(palette[np.argmax(counts)].astype(int).tolist())
390
- return dom_color
391
-
392
- def detect_eye_color(face_roi: np.ndarray, face_landmarks) -> Optional[str]:
393
- if face_landmarks is None:
394
- return None
395
- h, w = face_roi.shape[:2]
396
- iris_inds = set()
397
- for conn in mp_face_mesh.FACEMESH_IRISES:
398
- iris_inds.update(conn)
399
-
400
- iris_points = []
401
- for idx in iris_inds:
402
- lm = face_landmarks.landmark[idx]
403
- iris_points.append((int(lm.x * w), int(lm.y * h)))
404
- if not iris_points:
405
- return None
406
-
407
- min_x = min(pt[0] for pt in iris_points)
408
- max_x = max(pt[0] for pt in iris_points)
409
- min_y = min(pt[1] for pt in iris_points)
410
- max_y = max(pt[1] for pt in iris_points)
411
-
412
- pad = 5
413
- x1 = max(0, min_x - pad)
414
- y1 = max(0, min_y - pad)
415
- x2 = min(w, max_x + pad)
416
- y2 = min(h, max_y + pad)
417
-
418
- eye_roi = face_roi[y1:y2, x1:x2]
419
-
420
- eye_roi_resize = cv2.resize(eye_roi, (40, 40), interpolation=cv2.INTER_AREA)
421
-
422
- if eye_roi_resize.size == 0:
423
- return None
424
-
425
- dom_rgb = get_dominant_color(eye_roi_resize)
426
- if dom_rgb is not None:
427
- return classify_eye_color(dom_rgb)
428
- return None
429
-
430
- class HandTracker:
431
- def __init__(self, min_detection_confidence=0.5, min_tracking_confidence=0.5):
432
- self.hands = mp_hands.Hands(
433
- static_image_mode=True,
434
- max_num_hands=2,
435
- min_detection_confidence=min_detection_confidence,
436
- min_tracking_confidence=min_tracking_confidence,
437
- )
438
- logger.info("Initialized Mediapipe HandTracking")
439
-
440
- def detect_hands(self, image: np.ndarray):
441
- try:
442
- img_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
443
- results = self.hands.process(img_rgb)
444
- return results.multi_hand_landmarks, results.multi_handedness
445
- except Exception as e:
446
- logger.error(f"Hand detection error: {str(e)}")
447
- return None, None
448
-
449
- def draw_hands(self, image: np.ndarray, hand_landmarks, handedness, config):
450
- if not hand_landmarks:
451
- return image
452
-
453
- mpdraw = mp_drawing
454
- for i, hlms in enumerate(hand_landmarks):
455
-
456
- hl_color = config.hand_landmark_color[::-1]
457
- hc_color = config.hand_connection_color[::-1]
458
- mpdraw.draw_landmarks(
459
- image,
460
- hlms,
461
- mp_hands.HAND_CONNECTIONS,
462
- mpdraw.DrawingSpec(color=hl_color, thickness=2, circle_radius=4),
463
- mpdraw.DrawingSpec(color=hc_color, thickness=2, circle_radius=2),
464
- )
465
- if handedness and i < len(handedness):
466
- label = handedness[i].classification[0].label
467
- score = handedness[i].classification[0].score
468
- text = f"{label}: {score:.2f}"
469
-
470
- wrist_lm = hlms.landmark[mp_hands.HandLandmark.WRIST]
471
- h, w, _ = image.shape
472
- cx, cy = int(wrist_lm.x * w), int(wrist_lm.y * h)
473
- ht_color = config.hand_text_color[::-1]
474
- cv2.putText(image, text, (cx, cy - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, ht_color, 2)
475
- return image
476
-
477
- class FacePipeline:
478
- def __init__(self, config: PipelineConfig):
479
- self.config = config
480
- self.detector = None
481
- self.tracker = None
482
- self.facenet = None
483
- self.db = None
484
- self.hand_tracker = None
485
- self._initialized = False
486
-
487
- def initialize(self):
488
- try:
489
-
490
- self.detector = YOLOFaceDetector(
491
- model_path=self.config.detector['model_path'],
492
- device=self.config.detector['device']
493
- )
494
-
495
- self.tracker = FaceTracker(max_age=self.config.tracker['max_age'])
496
-
497
- self.facenet = FaceNetEmbedder(device=self.config.detector['device'])
498
-
499
- self.db = FaceDatabase()
500
-
501
- if self.config.hand['enable']:
502
- self.hand_tracker = HandTracker(
503
- min_detection_confidence=self.config.hand['min_detection_confidence'],
504
- min_tracking_confidence=self.config.hand['min_tracking_confidence']
505
- )
506
-
507
- self._initialized = True
508
- logger.info("FacePipeline initialized successfully.")
509
- except Exception as e:
510
- logger.error(f"Initialization failed: {str(e)}")
511
- self._initialized = False
512
- raise
513
-
514
- def process_frame(self, frame: np.ndarray) -> Tuple[np.ndarray, List[Dict]]:
515
- """
516
- Main pipeline processing: detection, tracking, hand detection, face mesh, blink detection, etc.
517
- Returns annotated_frame, detection_results.
518
- """
519
- if not self._initialized:
520
- logger.error("Pipeline not initialized.")
521
- return frame, []
522
-
523
- try:
524
-
525
- detections = self.detector.detect(frame, self.config.detection_conf_thres)
526
- tracked_objs = self.tracker.update(detections, frame)
527
- annotated = frame.copy()
528
- results = []
529
-
530
- hand_landmarks_list = None
531
- handedness_list = None
532
- if self.config.hand['enable'] and self.hand_tracker:
533
- hand_landmarks_list, handedness_list = self.hand_tracker.detect_hands(annotated)
534
- annotated = self.hand_tracker.draw_hands(
535
- annotated, hand_landmarks_list, handedness_list, self.config
536
- )
537
-
538
- for obj in tracked_objs:
539
- if not obj.is_confirmed():
540
- continue
541
-
542
- track_id = obj.track_id
543
- bbox = obj.to_tlbr().astype(int)
544
- x1, y1, x2, y2 = bbox
545
- conf = getattr(obj, 'score', 1.0)
546
- cls = getattr(obj, 'class_id', 0)
547
-
548
- face_roi = frame[y1:y2, x1:x2]
549
- if face_roi.size == 0:
550
- logger.warning(f"Empty face ROI for track={track_id}")
551
- continue
552
-
553
- is_spoofed = False
554
- if self.config.anti_spoof.get('enable', True):
555
- is_spoofed = not self.is_real_face(face_roi)
556
- if is_spoofed:
557
- cls = 1
558
-
559
- if is_spoofed:
560
- box_color_bgr = self.config.spoofed_bbox_color[::-1]
561
- name = "Spoofed"
562
- similarity = 0.0
563
- else:
564
-
565
- emb = self.facenet.get_embedding(face_roi)
566
- if emb is not None and self.config.recognition.get('enable', True):
567
- name, similarity = self.recognize_face(emb, self.config.recognition_conf_thres)
568
- else:
569
- name = "Unknown"
570
- similarity = 0.0
571
-
572
- box_color_rgb = (self.config.bbox_color if name != "Unknown"
573
- else self.config.unknown_bbox_color)
574
- box_color_bgr = box_color_rgb[::-1]
575
-
576
- label_text = name
577
- cv2.rectangle(annotated, (x1, y1), (x2, y2), box_color_bgr, 2)
578
- cv2.putText(annotated, label_text, (x1, y1 - 10),
579
- cv2.FONT_HERSHEY_SIMPLEX, 0.5, box_color_bgr, 2)
580
-
581
- blink = False
582
- if self.config.blink.get('enable', False):
583
- blink, left_ear, right_ear, left_eye_pts, right_eye_pts = detect_blink(
584
- face_roi, threshold=self.config.blink.get('ear_thresh', 0.25)
585
- )
586
- if left_eye_pts is not None and right_eye_pts is not None:
587
-
588
- le_g = left_eye_pts + np.array([x1, y1])
589
- re_g = right_eye_pts + np.array([x1, y1])
590
-
591
- eye_outline_bgr = self.config.eye_outline_color[::-1]
592
- cv2.polylines(annotated, [le_g], True, eye_outline_bgr, 1)
593
- cv2.polylines(annotated, [re_g], True, eye_outline_bgr, 1)
594
- if blink:
595
- blink_msg_color = self.config.blink_text_color[::-1]
596
- cv2.putText(annotated, "Blink Detected",
597
- (x1, y2 + 20),
598
- cv2.FONT_HERSHEY_SIMPLEX, 0.5,
599
- blink_msg_color, 2)
600
-
601
- face_mesh_landmarks = None
602
- eye_color_name = None
603
- if (self.config.face_mesh_options.get('enable') or
604
- self.config.eye_color.get('enable')):
605
- face_mesh_landmarks = process_face_mesh(face_roi)
606
- if face_mesh_landmarks:
607
-
608
- if self.config.face_mesh_options.get('enable', False):
609
- draw_face_mesh(
610
- annotated[y1:y2, x1:x2],
611
- face_mesh_landmarks,
612
- self.config.face_mesh_options,
613
- self.config
614
- )
615
-
616
- if self.config.eye_color.get('enable', False):
617
- color_found = detect_eye_color(face_roi, face_mesh_landmarks)
618
- if color_found:
619
- eye_color_name = color_found
620
- text_col_bgr = self.config.eye_color_text_color[::-1]
621
- cv2.putText(
622
- annotated, f"Eye Color: {eye_color_name}",
623
- (x1, y2 + 40),
624
- cv2.FONT_HERSHEY_SIMPLEX, 0.5,
625
- text_col_bgr, 2
626
- )
627
-
628
- detection_info = {
629
- "track_id": track_id,
630
- "bbox": (x1, y1, x2, y2),
631
- "confidence": float(conf),
632
- "class_id": cls,
633
- "name": name,
634
- "similarity": similarity,
635
- "blink": blink if self.config.blink.get('enable') else None,
636
- "face_mesh": bool(face_mesh_landmarks) if self.config.face_mesh_options.get('enable') else False,
637
- "hands_detected": bool(hand_landmarks_list),
638
- "hand_count": len(hand_landmarks_list) if hand_landmarks_list else 0,
639
- "eye_color": eye_color_name if self.config.eye_color.get('enable') else None
640
- }
641
- results.append(detection_info)
642
-
643
- return annotated, results
644
-
645
- except Exception as e:
646
- logger.error(f"Frame process error: {str(e)}")
647
- return frame, []
648
-
649
- def is_real_face(self, face_roi: np.ndarray) -> bool:
650
- try:
651
- gray = cv2.cvtColor(face_roi, cv2.COLOR_BGR2GRAY)
652
- lapv = cv2.Laplacian(gray, cv2.CV_64F).var()
653
- return lapv > self.config.anti_spoof.get('lap_thresh', 80.0)
654
- except Exception as e:
655
- logger.error(f"Anti-spoof error: {str(e)}")
656
- return False
657
-
658
- def recognize_face(self, embedding: np.ndarray, threshold: float) -> Tuple[str, float]:
659
- try:
660
- best_name = "Unknown"
661
- best_sim = 0.0
662
- for lbl, embs in self.db.embeddings.items():
663
- for db_emb in embs:
664
- sim = FacePipeline.cosine_similarity(embedding, db_emb)
665
- if sim > best_sim:
666
- best_sim = sim
667
- best_name = lbl
668
- if best_sim < threshold:
669
- best_name = "Unknown"
670
- return best_name, best_sim
671
- except Exception as e:
672
- logger.error(f"Recognition error: {str(e)}")
673
- return ("Unknown", 0.0)
674
-
675
- @staticmethod
676
- def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
677
- return float(np.dot(a, b) / ((np.linalg.norm(a)*np.linalg.norm(b)) + 1e-6))
678
-
679
- pipeline = None
680
- def load_pipeline() -> FacePipeline:
681
- global pipeline
682
- if pipeline is None:
683
- cfg = PipelineConfig.load(CONFIG_PATH)
684
- pipeline = FacePipeline(cfg)
685
- pipeline.initialize()
686
- return pipeline
687
-
688
- def hex_to_bgr(hexstr: str) -> Tuple[int,int,int]:
689
- if not hexstr.startswith('#'):
690
- hexstr = '#' + hexstr
691
- h = hexstr.lstrip('#')
692
- if len(h) != 6:
693
- return (255, 0, 0)
694
- r = int(h[0:2], 16)
695
- g = int(h[2:4], 16)
696
- b = int(h[4:6], 16)
697
- return (b,g,r)
698
-
699
- def bgr_to_hex(bgr: Tuple[int,int,int]) -> str:
700
- b,g,r = bgr
701
- return f"#{r:02x}{g:02x}{b:02x}"
702
-
703
- def update_config(
704
-
705
- enable_recognition, enable_antispoof, enable_blink, enable_hand, enable_eyecolor, enable_facemesh,
706
- show_tesselation, show_contours, show_irises,
707
-
708
- detection_conf, recognition_thresh, antispoof_thresh, blink_thresh, hand_det_conf, hand_track_conf,
709
-
710
- bbox_hex, spoofed_hex, unknown_hex, eye_hex, blink_hex,
711
- hand_landmark_hex, hand_connect_hex, hand_text_hex,
712
- mesh_hex, contour_hex, iris_hex, eye_color_text_hex
713
- ):
714
- pl = load_pipeline()
715
- cfg = pl.config
716
-
717
- cfg.recognition['enable'] = enable_recognition
718
- cfg.anti_spoof['enable'] = enable_antispoof
719
- cfg.blink['enable'] = enable_blink
720
- cfg.hand['enable'] = enable_hand
721
- cfg.eye_color['enable'] = enable_eyecolor
722
- cfg.face_mesh_options['enable'] = enable_facemesh
723
-
724
- cfg.face_mesh_options['tesselation'] = show_tesselation
725
- cfg.face_mesh_options['contours'] = show_contours
726
- cfg.face_mesh_options['irises'] = show_irises
727
-
728
- cfg.detection_conf_thres = detection_conf
729
- cfg.recognition_conf_thres = recognition_thresh
730
- cfg.anti_spoof['lap_thresh'] = antispoof_thresh
731
- cfg.blink['ear_thresh'] = blink_thresh
732
- cfg.hand['min_detection_confidence'] = hand_det_conf
733
- cfg.hand['min_tracking_confidence'] = hand_track_conf
734
-
735
- cfg.bbox_color = hex_to_bgr(bbox_hex)[::-1]
736
- cfg.spoofed_bbox_color = hex_to_bgr(spoofed_hex)[::-1]
737
- cfg.unknown_bbox_color = hex_to_bgr(unknown_hex)[::-1]
738
- cfg.eye_outline_color = hex_to_bgr(eye_hex)[::-1]
739
- cfg.blink_text_color = hex_to_bgr(blink_hex)[::-1]
740
- cfg.hand_landmark_color = hex_to_bgr(hand_landmark_hex)[::-1]
741
- cfg.hand_connection_color = hex_to_bgr(hand_connect_hex)[::-1]
742
- cfg.hand_text_color = hex_to_bgr(hand_text_hex)[::-1]
743
- cfg.mesh_color = hex_to_bgr(mesh_hex)[::-1]
744
- cfg.contour_color = hex_to_bgr(contour_hex)[::-1]
745
- cfg.iris_color = hex_to_bgr(iris_hex)[::-1]
746
- cfg.eye_color_text_color = hex_to_bgr(eye_color_text_hex)[::-1]
747
-
748
- cfg.save(CONFIG_PATH)
749
- return "Configuration saved successfully!"
750
-
751
- def enroll_user(label_name: str, filepaths: List[str]) -> str:
752
- """Enrolls a user by name using multiple image file paths."""
753
- pl = load_pipeline()
754
- if not label_name:
755
- return "Please provide a user name."
756
- if not filepaths or len(filepaths) == 0:
757
- return "No images provided."
758
-
759
- enrolled_count = 0
760
- for path in filepaths:
761
- if not os.path.isfile(path):
762
- continue
763
- img_bgr = cv2.imread(path)
764
- if img_bgr is None:
765
- continue
766
-
767
- dets = pl.detector.detect(img_bgr, pl.config.detection_conf_thres)
768
- for x1, y1, x2, y2, conf, cls in dets:
769
- roi = img_bgr[y1:y2, x1:x2]
770
- if roi.size == 0:
771
- continue
772
- emb = pl.facenet.get_embedding(roi)
773
- if emb is not None:
774
- pl.db.add_embedding(label_name, emb)
775
- enrolled_count += 1
776
-
777
- if enrolled_count > 0:
778
- pl.db.save()
779
- return f"Enrolled '{label_name}' with {enrolled_count} face(s)!"
780
- else:
781
- return "No faces detected in provided images."
782
-
783
- def search_by_name(name: str) -> str:
784
- pl = load_pipeline()
785
- if not name:
786
- return "No name entered."
787
- embs = pl.db.get_embeddings_by_label(name)
788
- if embs:
789
- return f"'{name}' found with {len(embs)} embedding(s)."
790
- else:
791
- return f"No embeddings found for '{name}'."
792
-
793
- def search_by_image(img: np.ndarray) -> str:
794
- pl = load_pipeline()
795
- if img is None:
796
- return "No image uploaded."
797
- img_bgr = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
798
- dets = pl.detector.detect(img_bgr, pl.config.detection_conf_thres)
799
- if not dets:
800
- return "No faces detected in the uploaded image."
801
- x1, y1, x2, y2, conf, cls = dets[0]
802
- roi = img_bgr[y1:y2, x1:x2]
803
- if roi.size == 0:
804
- return "Empty face ROI in the uploaded image."
805
-
806
- emb = pl.facenet.get_embedding(roi)
807
- if emb is None:
808
- return "Could not generate embedding from face."
809
- results = pl.db.search_by_image(emb, pl.config.recognition_conf_thres)
810
- if not results:
811
- return "No matches in the database under current threshold."
812
- lines = [f"- {lbl} (sim={sim:.3f})" for lbl, sim in results]
813
- return "Search results:\n" + "\n".join(lines)
814
-
815
- def remove_user(label: str) -> str:
816
- pl = load_pipeline()
817
- if not label:
818
- return "No user label selected."
819
- pl.db.remove_label(label)
820
- pl.db.save()
821
- return f"User '{label}' removed."
822
-
823
- def list_users() -> str:
824
- pl = load_pipeline()
825
- labels = pl.db.list_labels()
826
- if labels:
827
- return "Enrolled users:\n" + ", ".join(labels)
828
- return "No users enrolled."
829
-
830
- def process_test_image(img: np.ndarray) -> Tuple[np.ndarray, str]:
831
- """Single-image test: run pipeline and return annotated image + JSON results."""
832
- if img is None:
833
- return None, "No image uploaded."
834
-
835
- pl = load_pipeline()
836
- bgr = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
837
- processed, detections = pl.process_frame(bgr)
838
- result_rgb = cv2.cvtColor(processed, cv2.COLOR_BGR2RGB)
839
- return result_rgb, str(detections)
840
-
841
- def build_app():
842
- with gr.Blocks() as demo:
843
- gr.Markdown("# Complete Face Recognition System (Single-Image) with Mediapipe")
844
-
845
- with gr.Tab("Image Test"):
846
- gr.Markdown("Upload a single image to detect faces, run blink detection, face mesh, hand tracking, etc.")
847
- test_in = gr.Image(type="numpy", label="Upload Image")
848
- test_out = gr.Image()
849
- test_info = gr.Textbox(label="Detections")
850
- process_btn = gr.Button("Process Image")
851
-
852
- process_btn.click(
853
- fn=process_test_image,
854
- inputs=test_in,
855
- outputs=[test_out, test_info],
856
- )
857
-
858
- with gr.Tab("Configuration"):
859
- gr.Markdown("Adjust toggles, thresholds, and colors. Click Save to persist changes.")
860
-
861
- with gr.Row():
862
- enable_recognition = gr.Checkbox(label="Enable Recognition", value=True)
863
- enable_antispoof = gr.Checkbox(label="Enable Anti-Spoof", value=True)
864
- enable_blink = gr.Checkbox(label="Enable Blink Detection", value=True)
865
- enable_hand = gr.Checkbox(label="Enable Hand Tracking", value=True)
866
- enable_eyecolor = gr.Checkbox(label="Enable Eye Color Detection", value=False)
867
- enable_facemesh = gr.Checkbox(label="Enable Face Mesh", value=False)
868
-
869
- gr.Markdown("**Face Mesh Options**")
870
- with gr.Row():
871
- show_tesselation = gr.Checkbox(label="Tesselation", value=False)
872
- show_contours = gr.Checkbox(label="Contours", value=False)
873
- show_irises = gr.Checkbox(label="Irises", value=False)
874
-
875
- gr.Markdown("**Thresholds**")
876
- detection_conf = gr.Slider(0, 1, 0.4, step=0.01, label="Detection Confidence")
877
- recognition_thresh = gr.Slider(0.5, 1.0, 0.85, step=0.01, label="Recognition Threshold")
878
- antispoof_thresh = gr.Slider(0, 200, 80, step=1, label="Anti-Spoof Threshold")
879
- blink_thresh = gr.Slider(0, 0.5, 0.25, step=0.01, label="Blink EAR Threshold")
880
- hand_det_conf = gr.Slider(0, 1, 0.5, step=0.01, label="Hand Detection Confidence")
881
- hand_track_conf = gr.Slider(0, 1, 0.5, step=0.01, label="Hand Tracking Confidence")
882
-
883
- gr.Markdown("**Color Options (Hex)**")
884
- bbox_hex = gr.Textbox(label="Box Color (Recognized)", value="#00ff00")
885
- spoofed_hex = gr.Textbox(label="Box Color (Spoofed)", value="#ff0000")
886
- unknown_hex = gr.Textbox(label="Box Color (Unknown)", value="#ff0000")
887
- eye_hex = gr.Textbox(label="Eye Outline Color", value="#ffff00")
888
- blink_hex = gr.Textbox(label="Blink Text Color", value="#0000ff")
889
-
890
- hand_landmark_hex = gr.Textbox(label="Hand Landmark Color", value="#ffd24d")
891
- hand_connect_hex = gr.Textbox(label="Hand Connection Color", value="#cc6600")
892
- hand_text_hex = gr.Textbox(label="Hand Text Color", value="#ffffff")
893
-
894
- mesh_hex = gr.Textbox(label="Mesh Color", value="#64ff64")
895
- contour_hex = gr.Textbox(label="Contour Color", value="#c8c800")
896
- iris_hex = gr.Textbox(label="Iris Color", value="#ff00ff")
897
- eye_color_text_hex = gr.Textbox(label="Eye Color Text Color", value="#ffffff")
898
-
899
- save_btn = gr.Button("Save Configuration")
900
- save_msg = gr.Textbox(label="", interactive=False)
901
-
902
- save_btn.click(
903
- fn=update_config,
904
- inputs=[
905
- enable_recognition, enable_antispoof, enable_blink, enable_hand, enable_eyecolor, enable_facemesh,
906
- show_tesselation, show_contours, show_irises,
907
- detection_conf, recognition_thresh, antispoof_thresh, blink_thresh, hand_det_conf, hand_track_conf,
908
- bbox_hex, spoofed_hex, unknown_hex, eye_hex, blink_hex,
909
- hand_landmark_hex, hand_connect_hex, hand_text_hex,
910
- mesh_hex, contour_hex, iris_hex, eye_color_text_hex
911
- ],
912
- outputs=[save_msg]
913
- )
914
-
915
- with gr.Tab("Database Management"):
916
- gr.Markdown("Enroll multiple images per user, search by name or image, remove users, list all users.")
917
-
918
- with gr.Accordion("User Enrollment", open=False):
919
- enroll_name = gr.Textbox(label="User Name")
920
- enroll_paths = gr.File(file_count="multiple", type="filepath", label="Upload Multiple Images")
921
- enroll_btn = gr.Button("Enroll User")
922
- enroll_result = gr.Textbox()
923
-
924
- enroll_btn.click(
925
- fn=enroll_user,
926
- inputs=[enroll_name, enroll_paths],
927
- outputs=[enroll_result]
928
- )
929
-
930
- with gr.Accordion("User Search", open=False):
931
- search_mode = gr.Radio(["Name", "Image"], label="Search By", value="Name")
932
- search_name_box = gr.Dropdown(label="Select User", choices=[], value=None, visible=True)
933
- search_image_box = gr.Image(label="Upload Search Image", type="numpy", visible=False)
934
- search_btn = gr.Button("Search")
935
- search_out = gr.Textbox()
936
-
937
- def toggle_search(mode):
938
- if mode == "Name":
939
- return gr.update(visible=True), gr.update(visible=False)
940
- else:
941
- return gr.update(visible=False), gr.update(visible=True)
942
-
943
- search_mode.change(
944
- fn=toggle_search,
945
- inputs=[search_mode],
946
- outputs=[search_name_box, search_image_box]
947
- )
948
-
949
- def do_search(mode, uname, img):
950
- if mode == "Name":
951
- return search_by_name(uname)
952
- else:
953
- return search_by_image(img)
954
-
955
- search_btn.click(
956
- fn=do_search,
957
- inputs=[search_mode, search_name_box, search_image_box],
958
- outputs=[search_out]
959
- )
960
-
961
- with gr.Accordion("User Management Tools", open=False):
962
- list_btn = gr.Button("List Enrolled Users")
963
- list_out = gr.Textbox()
964
- list_btn.click(fn=lambda: list_users(), inputs=[], outputs=[list_out])
965
-
966
- def refresh_choices():
967
- pl = load_pipeline()
968
- return gr.update(choices=pl.db.list_labels())
969
-
970
- refresh_btn = gr.Button("Refresh User List")
971
- refresh_btn.click(fn=refresh_choices, inputs=[], outputs=[search_name_box])
972
-
973
- remove_box = gr.Dropdown(label="Select User to Remove", choices=[])
974
- remove_btn = gr.Button("Remove")
975
- remove_out = gr.Textbox()
976
-
977
- remove_btn.click(fn=remove_user, inputs=[remove_box], outputs=[remove_out])
978
- refresh_btn.click(fn=refresh_choices, inputs=[], outputs=[remove_box])
979
-
980
- return demo
981
-
982
  if __name__ == "__main__":
983
- app = build_app()
984
-
985
- app.queue().launch(server_name="0.0.0.0", server_port=7860)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
  import gradio as gr
2
+ from face_pipeline import FacePipeline
3
 
4
+ pipeline = FacePipeline()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
5
  if __name__ == "__main__":
6
+ iface.launch()