Merge pull request #3 from agendrix/pp/switch-cli-to-api

Switch CLI to API usage

deepface/api/src/modules/core/routes.py

@@ -1,6 +1,8 @@
 from flask import Blueprint, request
+
 from deepface import DeepFace
 from deepface.api.src.modules.core import service
+from deepface.commons.agendrix.constants import DEFAULT_DETECTOR_BACKEND, DEFAULT_DISTANCE_METRIC, DEFAULT_MODEL
 from deepface.commons.logger import Logger
 
 logger = Logger()
@@ -13,30 +15,25 @@ def home():
     return f"<h1>Welcome to DeepFace API v{DeepFace.__version__}!</h1>"
 
 
-@blueprint.route("/represent", methods=["POST"])
-def represent():
+@blueprint.route("/detect", methods=["POST"])
+def detect():
     input_args = request.get_json()
 
     if input_args is None:
         return {"message": "empty input set passed"}
 
-    img_path = input_args.get("img") or input_args.get("img_path")
+    img_path = input_args.get("img_path")
     if img_path is None:
         return {"message": "you must pass img_path input"}
 
-    obj = service.represent(
+    output = service.detect(
         img_path=img_path,
-        model_name=input_args.get("model_name", "VGG-Face"),
-        detector_backend=input_args.get("detector_backend", "opencv"),
-        enforce_detection=input_args.get("enforce_detection", True),
-        align=input_args.get("align", True),
-        anti_spoofing=input_args.get("anti_spoofing", False),
-        max_faces=input_args.get("max_faces"),
+        model=input_args.get("model", DEFAULT_MODEL),
+        detector_backend=input_args.get("detector_backend", DEFAULT_DETECTOR_BACKEND),
+        redis_key=input_args.get("redis_key"),
     )
 
-    logger.debug(obj)
-
-    return obj
+    return output
 
 
 @blueprint.route("/verify", methods=["POST"])
@@ -46,51 +43,23 @@ def verify():
     if input_args is None:
         return {"message": "empty input set passed"}
 
-    img1_path = input_args.get("img1") or input_args.get("img1_path")
-    img2_path = input_args.get("img2") or input_args.get("img2_path")
+    img1_path = input_args.get("img1_path")
+    img2_path = input_args.get("img2_path")
 
     if img1_path is None:
         return {"message": "you must pass img1_path input"}
 
     if img2_path is None:
         return {"message": "you must pass img2_path input"}
 
-    verification = service.verify(
+    output = service.verify(
         img1_path=img1_path,
         img2_path=img2_path,
-        model_name=input_args.get("model_name", "VGG-Face"),
-        detector_backend=input_args.get("detector_backend", "opencv"),
-        distance_metric=input_args.get("distance_metric", "cosine"),
-        align=input_args.get("align", True),
-        enforce_detection=input_args.get("enforce_detection", True),
-        anti_spoofing=input_args.get("anti_spoofing", False),
+        model=input_args.get("model", DEFAULT_MODEL),
+        detector_backend=input_args.get("detector_backend", DEFAULT_DETECTOR_BACKEND),
+        distance_metric=input_args.get("distance_metric", DEFAULT_DISTANCE_METRIC),
+        threshold=input_args.get("threshold", 0.3),
+        redis_key=input_args.get("redis_key"),
     )
 
-    logger.debug(verification)
-
-    return verification
-
-
-@blueprint.route("/analyze", methods=["POST"])
-def analyze():
-    input_args = request.get_json()
-
-    if input_args is None:
-        return {"message": "empty input set passed"}
-
-    img_path = input_args.get("img") or input_args.get("img_path")
-    if img_path is None:
-        return {"message": "you must pass img_path input"}
-
-    demographies = service.analyze(
-        img_path=img_path,
-        actions=input_args.get("actions", ["age", "gender", "emotion", "race"]),
-        detector_backend=input_args.get("detector_backend", "opencv"),
-        enforce_detection=input_args.get("enforce_detection", True),
-        align=input_args.get("align", True),
-        anti_spoofing=input_args.get("anti_spoofing", False),
-    )
-
-    logger.debug(demographies)
-
-    return demographies
+    return output

deepface/api/src/modules/core/service.py

@@ -1,9 +1,14 @@
 # built-in dependencies
+import json
+import logging
 import traceback
 from typing import Optional
 
 # project dependencies
 from deepface import DeepFace
+from deepface.commons.agendrix.image_processing import get_faces_embeddings
+from deepface.commons.agendrix.redis import initialize_redis
+from deepface.commons.image_utils import load_image
 
 # pylint: disable=broad-except
 
@@ -35,33 +40,6 @@ def represent(
         return {"error": f"Exception while representing: {str(err)} - {tb_str}"}, 400
 
 
-def verify(
-    img1_path: str,
-    img2_path: str,
-    model_name: str,
-    detector_backend: str,
-    distance_metric: str,
-    enforce_detection: bool,
-    align: bool,
-    anti_spoofing: bool,
-):
-    try:
-        obj = DeepFace.verify(
-            img1_path=img1_path,
-            img2_path=img2_path,
-            model_name=model_name,
-            detector_backend=detector_backend,
-            distance_metric=distance_metric,
-            align=align,
-            enforce_detection=enforce_detection,
-            anti_spoofing=anti_spoofing,
-        )
-        return obj
-    except Exception as err:
-        tb_str = traceback.format_exc()
-        return {"error": f"Exception while verifying: {str(err)} - {tb_str}"}, 400
-
-
 def analyze(
     img_path: str,
     actions: list,
@@ -86,3 +64,75 @@ def analyze(
     except Exception as err:
         tb_str = traceback.format_exc()
         return {"error": f"Exception while analyzing: {str(err)} - {tb_str}"}, 400
+
+
+def detect(
+    img_path: str,
+    model: str,
+    detector_backend: str,
+    redis_key: Optional[str],
+):
+    try:
+        img, _ = load_image(img_path)
+        embeddings = get_faces_embeddings(img, model, detector_backend)
+        faces_count = len(embeddings)
+
+        raw_output = {"faces_count": faces_count}
+        logging.info(json.dumps(raw_output | {"redis_key": redis_key}))
+
+        if redis_key is not None:
+            r = initialize_redis()
+            r.setex(redis_key, 12 * 60 * 60, json.dumps({"faces_count": faces_count}))  # 12h expiration
+
+        return raw_output
+
+    except Exception as err:
+        tb_str = traceback.format_exc()
+        return {"error": f"Exception while analyzing: {str(err)} - {tb_str}"}, 400
+
+
+def verify(
+    img1_path: str,
+    img2_path: str,
+    model: str,
+    detector_backend: str,
+    distance_metric: str,
+    threshold: float,
+    redis_key: Optional[str],
+):
+    try:
+        img1, _ = load_image(img1_path)
+        img2, _ = load_image(img2_path)
+
+        embeddings_1 = get_faces_embeddings(img1, model, detector_backend)
+        embeddings_2 = get_faces_embeddings(img2, model, detector_backend)
+        faces_count = len(embeddings_2)
+
+        result = DeepFace.verify(
+            img1_path=embeddings_1,
+            img2_path=embeddings_2,
+            model_name=model,
+            detector_backend=detector_backend,
+            distance_metric=distance_metric,
+            enforce_detection=False,
+            silent=True,
+            threshold=threshold,
+            anti_spoofing=False,
+        )
+
+        raw_output = {
+            "matches": result["verified"],
+            "distance": result["distance"],
+            "faces_count": faces_count,
+        }
+        logging.info(json.dumps(raw_output | {"redis_key": redis_key}))
+
+        if redis_key is not None:
+            r = initialize_redis()
+            r.setex(redis_key, 12 * 60 * 60, json.dumps(raw_output))  # 12h expiration
+
+        return raw_output
+
+    except Exception as err:
+        tb_str = traceback.format_exc()
+        return {"error": f"Exception while analyzing: {str(err)} - {tb_str}"}, 400

deepface/cli/detect.py

@@ -36,15 +36,13 @@ def main():
     embeddings = get_faces_embeddings(img, args.model, args.detector_backend)
     faces_count = len(embeddings)
 
-    output = {
-        "faces_count": faces_count,
-    }
+    json_output = json.dumps({"faces_count": faces_count})
 
-    logging.info(json.dumps(output))
+    logging.info(json_output)
 
     if args.redis_key is not None:
         r = initialize_redis()
-        r.setex(args.redis_key, 12 * 60 * 60, json.dumps(output))  # 12h expiration
+        r.setex(args.redis_key, 12 * 60 * 60, json_output)  # 12h expiration
 
 
 if __name__ == "__main__":

deepface/cli/verify.py

@@ -55,17 +55,19 @@ def main():
         anti_spoofing=False,
     )
 
-    output = {
-        "matches": result["verified"],
-        "distance": result["distance"],
-        "faces_count": faces_count,
-    }
+    json_output = json.dumps(
+        {
+            "matches": result["verified"],
+            "distance": result["distance"],
+            "faces_count": faces_count,
+        }
+    )
 
-    logging.info(json.dumps(output))
+    logging.info(json_output)
 
     if args.redis_key is not None:
         r = initialize_redis()
-        r.setex(args.redis_key, 12 * 60 * 60, json.dumps(output))  # 12h expiration
+        r.setex(args.redis_key, 12 * 60 * 60, json_output)  # 12h expiration
 
 
 if __name__ == "__main__":

deepface/commons/agendrix/argparser.py

@@ -1,20 +1,20 @@
 from argparse import ArgumentParser
 
-from deepface.commons.agendrix.constants import DETECTOR_BACKENDS, DISTANCE_METRICS, MODELS
+from deepface.commons.agendrix.constants import DEFAULT_DETECTOR_BACKEND, DEFAULT_DISTANCE_METRIC, DEFAULT_MODEL, DETECTOR_BACKENDS, DISTANCE_METRICS, MODELS
 
 
 def add_model_arg(parser: ArgumentParser) -> ArgumentParser:
-    parser.add_argument("--model", type=str, default="Facenet512", choices=MODELS, help="Face recognition model")
+    parser.add_argument("--model", type=str, default=DEFAULT_MODEL, choices=MODELS, help="Face recognition model")
     return parser
 
 
 def add_detector_backend_arg(parser: ArgumentParser) -> ArgumentParser:
-    parser.add_argument("--detector_backend", type=str, default="ssd", choices=DETECTOR_BACKENDS, help="Face detector and alignment backend")
+    parser.add_argument("--detector_backend", type=str, default=DEFAULT_DETECTOR_BACKEND, choices=DETECTOR_BACKENDS, help="Face detector and alignment backend")
     return parser
 
 
 def add_distance_metric_arg(parser: ArgumentParser) -> ArgumentParser:
-    parser.add_argument("--distance_metric", type=str, default="cosine", choices=DISTANCE_METRICS, help="Type of distance metric to use")
+    parser.add_argument("--distance_metric", type=str, default=DEFAULT_DISTANCE_METRIC, choices=DISTANCE_METRICS, help="Type of distance metric to use")
     return parser
 
 

deepface/commons/agendrix/constants.py

@@ -1,5 +1,7 @@
 # MODELS = ['VGG-Face', 'Facenet', 'Facenet512', 'OpenFace', 'DeepFace', 'DeepID', 'Dlib', 'ArcFace', 'SFace', 'GhostFaceNet']
 MODELS = ["VGG-Face", "Facenet", "Facenet512", "OpenFace"]
+DEFAULT_MODEL = "Facenet512"
+
 # DETECTOR_BACKENDS = ['opencv', 'retinaface', 'mtcnn', 'ssd', 'dlib', 'mediapipe', 'yolov8', 'centerface', 'skip']
 DETECTOR_BACKENDS = [
     "opencv",
@@ -11,5 +13,8 @@
     "yolov8",
     "centerface",
 ]
+DEFAULT_DETECTOR_BACKEND = "ssd"
+
 # DISTANCE_METRICS = ['cosine', 'euclidean', 'euclidean_l2']
 DISTANCE_METRICS = ["cosine"]
+DEFAULT_DISTANCE_METRIC = "cosine"

deepface/commons/agendrix/redis.py

@@ -4,4 +4,4 @@
 
 
 def initialize_redis():
-    return Redis.from_url(os.getenv("REDIS_URL"))
+    return Redis.from_url(os.getenv("REDIS_URL", ""))

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "deepface-agendrix"
-version = "0.1.3"
+version = "0.1.4"
 description = "Face recognition library for Python, customized for Agendrix usage"
 authors = [
     {name = "Philippe Poulin", email = "[email protected]"},