You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
352 lines
16 KiB
352 lines
16 KiB
import gradio as gr
|
|
import numpy as np
|
|
from typing import List, Optional
|
|
import asyncio
|
|
from PIL import Image
|
|
import cv2
|
|
|
|
from src.application.use_cases import (
|
|
RegisterFaceUseCase,
|
|
VerifyFaceUseCase,
|
|
IdentifyFaceUseCase
|
|
)
|
|
from src.domain.entities import OperationType
|
|
|
|
def create_gradio_interface(
|
|
register_use_case: RegisterFaceUseCase,
|
|
verify_use_case: VerifyFaceUseCase,
|
|
identify_use_case: IdentifyFaceUseCase
|
|
):
|
|
async def process_registration(user_name: str, images):
|
|
if not user_name:
|
|
return "Error: Please provide a user name"
|
|
|
|
if not images:
|
|
return "Error: Please upload at least one image"
|
|
|
|
np_images = []
|
|
for img_file in images:
|
|
if hasattr(img_file, 'name'):
|
|
img = Image.open(img_file.name)
|
|
np_images.append(np.array(img))
|
|
else:
|
|
np_images.append(np.array(img_file))
|
|
|
|
success, message = await register_use_case.execute(user_name, np_images)
|
|
|
|
return message
|
|
|
|
async def process_recognition(operation_type: str, user_name: Optional[str], image: Image.Image):
|
|
if not image:
|
|
return "Error: Please upload an image"
|
|
|
|
np_image = np.array(image)
|
|
|
|
if operation_type == OperationType.VERIFICATION.value:
|
|
if not user_name:
|
|
return "Error: Please provide a user name for verification"
|
|
|
|
result = await verify_use_case.execute(user_name, np_image)
|
|
|
|
if result.is_verified:
|
|
return f"✅ Verified as {user_name}\nConfidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s"
|
|
else:
|
|
return f"❌ Not verified as {user_name}\nConfidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s"
|
|
|
|
else:
|
|
result = await identify_use_case.execute(np_image)
|
|
|
|
if result.is_identified:
|
|
candidates_str = "\n".join([f" - {name}: {conf:.2%}" for name, conf in result.candidates[:5]])
|
|
return f"✅ Identified as: {result.user_id}\nConfidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s\n\nTop candidates:\n{candidates_str}"
|
|
else:
|
|
candidates_str = "\n".join([f" - {name}: {conf:.2%}" for name, conf in result.candidates[:5]])
|
|
return f"❌ Person not identified\nBest match confidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s\n\nTop candidates:\n{candidates_str}"
|
|
|
|
def sync_process_registration(user_name, images):
|
|
return asyncio.run(process_registration(user_name, images))
|
|
|
|
def sync_process_recognition(operation_type, user_name, image):
|
|
return asyncio.run(process_recognition(operation_type, user_name, image))
|
|
|
|
# Webcam streaming functions
|
|
def process_webcam_frame(frame, operation_type, user_name):
|
|
if frame is None:
|
|
return frame
|
|
|
|
# Convert frame to PIL Image for processing
|
|
image = Image.fromarray(frame)
|
|
|
|
try:
|
|
if operation_type == OperationType.VERIFICATION.value and user_name:
|
|
result = asyncio.run(verify_use_case.execute(user_name, np.array(image)))
|
|
|
|
# Draw result on frame
|
|
cv2.putText(frame, f"Verification: {'✓' if result.is_verified else '✗'}",
|
|
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1,
|
|
(0, 255, 0) if result.is_verified else (0, 0, 255), 2)
|
|
cv2.putText(frame, f"Confidence: {result.confidence:.2%}",
|
|
(10, 70), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
|
|
|
elif operation_type == OperationType.IDENTIFICATION.value:
|
|
result = asyncio.run(identify_use_case.execute(np.array(image)))
|
|
|
|
# Draw result on frame
|
|
if result.is_identified:
|
|
cv2.putText(frame, f"Identified: {result.user_id}",
|
|
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
|
|
cv2.putText(frame, f"Confidence: {result.confidence:.2%}",
|
|
(10, 70), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
|
else:
|
|
cv2.putText(frame, "No match found",
|
|
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2)
|
|
|
|
except Exception as e:
|
|
cv2.putText(frame, f"Error: {str(e)[:50]}",
|
|
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)
|
|
|
|
return frame
|
|
|
|
# Webcam registration capture
|
|
captured_images = []
|
|
|
|
def capture_image_for_registration(frame):
|
|
if frame is not None:
|
|
captured_images.append(frame.copy())
|
|
return f"Captured {len(captured_images)} images"
|
|
return "No frame to capture"
|
|
|
|
def register_from_captures(user_name):
|
|
if not user_name:
|
|
return "Error: Please provide a user name"
|
|
|
|
if len(captured_images) == 0:
|
|
return "Error: No images captured. Please use webcam to capture images first."
|
|
|
|
try:
|
|
# Convert captured frames to numpy arrays
|
|
np_images = [np.array(img) for img in captured_images]
|
|
success, message = asyncio.run(register_use_case.execute(user_name, np_images))
|
|
|
|
# Clear captured images after registration
|
|
captured_images.clear()
|
|
return message
|
|
except Exception as e:
|
|
return f"Error during registration: {str(e)}"
|
|
|
|
def clear_captured_images():
|
|
captured_images.clear()
|
|
return "Cleared all captured images"
|
|
|
|
# CSS for better layout
|
|
css = """
|
|
.webcam-container {
|
|
max-width: 640px !important;
|
|
margin: 0 auto;
|
|
}
|
|
.capture-info {
|
|
text-align: center;
|
|
font-weight: bold;
|
|
color: #007bff;
|
|
}
|
|
"""
|
|
|
|
with gr.Blocks(title="Face Recognition System", css=css) as interface:
|
|
gr.Markdown("# Face Recognition System")
|
|
gr.Markdown("Advanced face recognition using RetinaFace detector and SFace recognizer")
|
|
|
|
with gr.Tabs():
|
|
# File Upload Registration Tab
|
|
with gr.TabItem("📁 File Registration"):
|
|
gr.Markdown("## Register New User - Upload Files")
|
|
gr.Markdown("Upload multiple images for better recognition accuracy")
|
|
|
|
with gr.Row():
|
|
reg_user_name = gr.Textbox(
|
|
label="User Name",
|
|
placeholder="Enter user name..."
|
|
)
|
|
reg_images = gr.File(
|
|
label="Upload Face Images",
|
|
file_count="multiple",
|
|
file_types=["image"]
|
|
)
|
|
|
|
reg_button = gr.Button("Register User", variant="primary")
|
|
reg_output = gr.Textbox(label="Registration Result", lines=3)
|
|
|
|
reg_button.click(
|
|
fn=sync_process_registration,
|
|
inputs=[reg_user_name, reg_images],
|
|
outputs=reg_output
|
|
)
|
|
|
|
# Webcam Registration Tab
|
|
with gr.TabItem("📷 Webcam Registration"):
|
|
gr.Markdown("## Register New User - Using Webcam")
|
|
gr.Markdown("Capture multiple images from webcam for better recognition accuracy")
|
|
|
|
with gr.Row():
|
|
with gr.Column():
|
|
webcam_reg_user_name = gr.Textbox(
|
|
label="User Name",
|
|
placeholder="Enter user name..."
|
|
)
|
|
webcam_reg_image = gr.Image(
|
|
label="Webcam Feed",
|
|
sources=["webcam"],
|
|
type="numpy",
|
|
elem_classes=["webcam-container"]
|
|
)
|
|
|
|
with gr.Row():
|
|
capture_btn = gr.Button("📸 Capture Image", variant="secondary")
|
|
clear_btn = gr.Button("🗑️ Clear Captures", variant="secondary")
|
|
register_webcam_btn = gr.Button("✅ Register User", variant="primary")
|
|
|
|
with gr.Column():
|
|
capture_status = gr.Textbox(
|
|
label="Capture Status",
|
|
value="Ready to capture images",
|
|
elem_classes=["capture-info"]
|
|
)
|
|
webcam_reg_output = gr.Textbox(label="Registration Result", lines=8)
|
|
|
|
capture_btn.click(
|
|
fn=capture_image_for_registration,
|
|
inputs=[webcam_reg_image],
|
|
outputs=[capture_status]
|
|
)
|
|
|
|
clear_btn.click(
|
|
fn=clear_captured_images,
|
|
outputs=[capture_status]
|
|
)
|
|
|
|
register_webcam_btn.click(
|
|
fn=register_from_captures,
|
|
inputs=[webcam_reg_user_name],
|
|
outputs=[webcam_reg_output]
|
|
)
|
|
|
|
# File Upload Recognition Tab
|
|
with gr.TabItem("📁 File Recognition"):
|
|
gr.Markdown("## Face Recognition - Upload Image")
|
|
gr.Markdown("Choose between verification (1:1) or identification (1:N)")
|
|
|
|
with gr.Row():
|
|
operation_type = gr.Radio(
|
|
choices=[OperationType.VERIFICATION.value, OperationType.IDENTIFICATION.value],
|
|
value=OperationType.VERIFICATION.value,
|
|
label="Operation Type"
|
|
)
|
|
rec_user_name = gr.Textbox(
|
|
label="User Name (for verification)",
|
|
placeholder="Enter user name for verification...",
|
|
visible=True
|
|
)
|
|
|
|
rec_image = gr.Image(
|
|
label="Upload Face Image",
|
|
type="pil"
|
|
)
|
|
|
|
rec_button = gr.Button("Process", variant="primary")
|
|
rec_output = gr.Textbox(label="Recognition Result", lines=8)
|
|
|
|
def update_user_name_visibility(operation):
|
|
return gr.update(visible=operation == OperationType.VERIFICATION.value)
|
|
|
|
operation_type.change(
|
|
fn=update_user_name_visibility,
|
|
inputs=operation_type,
|
|
outputs=rec_user_name
|
|
)
|
|
|
|
rec_button.click(
|
|
fn=sync_process_recognition,
|
|
inputs=[operation_type, rec_user_name, rec_image],
|
|
outputs=rec_output
|
|
)
|
|
|
|
# Webcam Recognition Tab
|
|
with gr.TabItem("📷 Live Recognition"):
|
|
gr.Markdown("## Live Face Recognition - Webcam Stream")
|
|
gr.Markdown("Real-time face verification or identification using webcam")
|
|
|
|
with gr.Row():
|
|
with gr.Column():
|
|
stream_operation_type = gr.Radio(
|
|
choices=[OperationType.VERIFICATION.value, OperationType.IDENTIFICATION.value],
|
|
value=OperationType.IDENTIFICATION.value,
|
|
label="Operation Type"
|
|
)
|
|
stream_user_name = gr.Textbox(
|
|
label="User Name (for verification)",
|
|
placeholder="Enter user name for verification...",
|
|
visible=False
|
|
)
|
|
|
|
webcam_stream = gr.Image(
|
|
label="Live Webcam Feed",
|
|
sources=["webcam"],
|
|
type="numpy",
|
|
elem_classes=["webcam-container"]
|
|
)
|
|
|
|
with gr.Column():
|
|
gr.Markdown("### Instructions:")
|
|
gr.Markdown("- For **Identification**: The system will try to identify any face in the camera")
|
|
gr.Markdown("- For **Verification**: Enter a username to verify if the face matches that user")
|
|
gr.Markdown("- Results are displayed directly on the video feed")
|
|
|
|
def update_stream_user_name_visibility(operation):
|
|
return gr.update(visible=operation == OperationType.VERIFICATION.value)
|
|
|
|
stream_operation_type.change(
|
|
fn=update_stream_user_name_visibility,
|
|
inputs=stream_operation_type,
|
|
outputs=stream_user_name
|
|
)
|
|
|
|
# Note: Streaming removed due to Gradio version compatibility
|
|
# Use the capture button approach instead
|
|
stream_button = gr.Button("🔄 Process Current Frame", variant="primary")
|
|
stream_output = gr.Textbox(label="Recognition Result", lines=4)
|
|
|
|
def process_current_frame(image, operation_type, user_name):
|
|
if image is None:
|
|
return "No image captured from webcam"
|
|
|
|
try:
|
|
pil_image = Image.fromarray(image)
|
|
np_image = np.array(pil_image)
|
|
|
|
if operation_type == OperationType.VERIFICATION.value and user_name:
|
|
result = asyncio.run(verify_use_case.execute(user_name, np_image))
|
|
if result.is_verified:
|
|
return f"✅ Verified as {user_name}\nConfidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s"
|
|
else:
|
|
return f"❌ Not verified as {user_name}\nConfidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s"
|
|
|
|
elif operation_type == OperationType.IDENTIFICATION.value:
|
|
result = asyncio.run(identify_use_case.execute(np_image))
|
|
if result.is_identified:
|
|
candidates_str = "\n".join([f" - {name}: {conf:.2%}" for name, conf in result.candidates[:5]])
|
|
return f"✅ Identified as: {result.user_id}\nConfidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s\n\nTop candidates:\n{candidates_str}"
|
|
else:
|
|
candidates_str = "\n".join([f" - {name}: {conf:.2%}" for name, conf in result.candidates[:5]])
|
|
return f"❌ Person not identified\nBest match confidence: {result.confidence:.2%}\nThreshold: {result.threshold:.2%}\nProcessing time: {result.processing_time:.3f}s\n\nTop candidates:\n{candidates_str}"
|
|
else:
|
|
return "Please select operation type and enter username for verification"
|
|
|
|
except Exception as e:
|
|
return f"Error processing image: {str(e)}"
|
|
|
|
stream_button.click(
|
|
fn=process_current_frame,
|
|
inputs=[webcam_stream, stream_operation_type, stream_user_name],
|
|
outputs=[stream_output]
|
|
)
|
|
|
|
return interface |