FacialRecognitionPython/grpc_server.py

import grpc
from concurrent import futures
import time
import os
from deepface import DeepFace
import pandas as pd

import face_recognition_pb2
import face_recognition_pb2_grpc


EMPLOYEE_DB_PATH = "data/employees"


class FaceRecognitionServicer(face_recognition_pb2_grpc.FaceRecognitionServiceServicer):

    def Recognize(self, request, context):
        print("[INFO] Received request...")

        # Save temp image
        temp_path = "temp_upload.jpg"
        with open(temp_path, "wb") as f:
            f.write(request.image)

        try:
            print("[INFO] Running DeepFace search...")

            result = DeepFace.find(
                img_path=temp_path,
                db_path=EMPLOYEE_DB_PATH,
                enforce_detection=False
            )

            # result is a list of pandas DataFrames for each model;
            # we only use first result
            if isinstance(result, list):
                result = result[0]

            if result is None or len(result) == 0:
                print("[INFO] No match found.")
                return face_recognition_pb2.FaceResponse(
                    name="Unknown",
                    confidence=0.0
                )

            # Pick best match
            best_row = result.iloc[0]
            matched_image_path = best_row["identity"]
            name = os.path.splitext(os.path.basename(matched_image_path))[0]

            # DeepFace gives "distance" values, convert roughly to confidence
            distance = best_row.get("VGG-Face_cosine", 0.3)
            confidence = float(max(0.0, 1.0 - distance))

            print(f"[INFO] Match: {name}, confidence: {confidence:.2f}")

            return face_recognition_pb2.FaceResponse(
                name=name,
                confidence=confidence
            )

        except Exception as e:
            print("[ERROR] DeepFace failed:", e)
            return face_recognition_pb2.FaceResponse(
                name="Error",
                confidence=0.0
            )

        finally:
            if os.path.exists(temp_path):
                os.remove(temp_path)


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    face_recognition_pb2_grpc.add_FaceRecognitionServiceServicer_to_server(
        FaceRecognitionServicer(), server
    )

    server.add_insecure_port("[::]:50051")
    server.start()
    print("[INFO] gRPC Face Recognition server running on port 50051")
    server.wait_for_termination()


if __name__ == "__main__":
    serve()