# built-in dependencies import os from typing import Any, Dict, IO, List, Union, Optional, cast import uuid import time import math import tempfile # 3rd party dependencies import pandas as pd import numpy as np from numpy.typing import NDArray # project dependencies from deepface.modules.database.types import Database from deepface.modules.database.inventory import database_inventory from deepface.modules.representation import represent from deepface.modules.verification import ( find_angular_distance, find_cosine_distance, find_euclidean_distance, l2_normalize as find_l2_normalize, find_threshold, find_confidence, ) from deepface.commons.logger import Logger logger = Logger() # pylint: disable=too-many-positional-arguments, no-else-return def register( img: Union[str, NDArray[Any], IO[bytes], List[str], List[NDArray[Any]], List[IO[bytes]]], img_name: Optional[str] = None, model_name: str = "VGG-Face", detector_backend: str = "opencv", enforce_detection: bool = True, align: bool = True, l2_normalize: bool = False, expand_percentage: int = 0, normalization: str = "base", anti_spoofing: bool = False, database_type: str = "postgres", connection_details: Optional[Union[Dict[str, Any], str]] = None, connection: Any = None, ) -> Dict[str, Any]: """ Register identities to database for face recognition Args: img (str or np.ndarray or IO[bytes] or list): The exact path to the image, a numpy array in BGR format, a file object that supports at least `.read` and is opened in binary mode, or a base64 encoded image. If a list is provided, each element should be a string or numpy array representing an image, and the function will process images in batch. img_name (optional str): image name to store in db, if not provided then we will try to extract it from given img. model_name (str): Model for face recognition. Options: VGG-Face, Facenet, Facenet512, OpenFace, DeepFace, DeepID, Dlib, ArcFace, SFace and GhostFaceNet (default is VGG-Face). detector_backend (string): face detector backend. Options: 'opencv', 'retinaface', 'mtcnn', 'ssd', 'dlib', 'mediapipe', 'yolov8n', 'yolov8m', 'yolov8l', 'yolov11n', 'yolov11s', 'yolov11m', 'yolov11l', 'yolov12n', 'yolov12s', 'yolov12m', 'yolov12l', 'centerface' or 'skip' (default is opencv). enforce_detection (boolean): If no face is detected in an image, raise an exception. Set to False to avoid the exception for low-resolution images (default is True). align (bool): Flag to enable face alignment (default is True). l2_normalize (bool): Flag to enable L2 normalization (unit vector normalization) expand_percentage (int): expand detected facial area with a percentage (default is 0). normalization (string): Normalize the input image before feeding it to the model. Options: base, raw, Facenet, Facenet2018, VGGFace, VGGFace2, ArcFace (default is base). anti_spoofing (boolean): Flag to enable anti spoofing (default is False). database_type (str): Type of database to register identities. Options: 'postgres', 'mongo', 'weaviate', 'neo4j', 'pgvector', 'pinecone' (default is 'postgres'). connection_details (dict or str): Connection details for the database. connection (Any): Existing database connection object. If provided, this connection will be used instead of creating a new one. Note: Instead of providing `connection` or `connection_details`, database connection information can be supplied via environment variables: - DEEPFACE_POSTGRES_URI - DEEPFACE_MONGO_URI - DEEPFACE_WEAVIATE_URI - DEEPFACE_NEO4J_URI - DEEPFACE_PINECONE_API_KEY Returns: result (dict): A dictionary containing registration results with following keys. - inserted (int): Number of embeddings successfully registered to the database. """ db_client = __connect_database( database_type=database_type, connection_details=connection_details, connection=connection, ) results = __get_embeddings( img=img, model_name=model_name, detector_backend=detector_backend, enforce_detection=enforce_detection, align=align, anti_spoofing=anti_spoofing, expand_percentage=expand_percentage, normalization=normalization, l2_normalize=l2_normalize, return_face=True, ) embedding_records: List[Dict[str, Any]] = [] for result in results: img_identifier = img_name or ( img if isinstance(img, str) and img.endswith((".jpg", ".jpeg", ".png")) else str(uuid.uuid4()) ) embedding_record = { "id": None, "img_name": img_identifier, "face": result["face"], "model_name": model_name, "detector_backend": detector_backend, "embedding": result["embedding"], "aligned": align, "l2_normalized": l2_normalize, } embedding_records.append(embedding_record) inserted = db_client.insert_embeddings(embedding_records, batch_size=100) logger.debug(f"Successfully registered {inserted} embeddings to the database.") # Close the database connection if it was created internally if connection is None: db_client.close() return {"inserted": inserted} def search( img: Union[str, NDArray[Any], IO[bytes], List[str], List[NDArray[Any]], List[IO[bytes]]], model_name: str = "VGG-Face", detector_backend: str = "opencv", distance_metric: str = "cosine", enforce_detection: bool = True, align: bool = True, l2_normalize: bool = False, expand_percentage: int = 0, normalization: str = "base", anti_spoofing: bool = False, similarity_search: bool = False, k: Optional[int] = None, database_type: str = "postgres", connection_details: Optional[Union[Dict[str, Any], str]] = None, connection: Any = None, search_method: str = "exact", ) -> List[pd.DataFrame]: """ Search for identities in database for face recognition. This is a stateless facial recognition function. Use find function to do it in a stateful way. Args: img (str or np.ndarray or IO[bytes] or list): The exact path to the image, a numpy array in BGR format, a file object that supports at least `.read` and is opened in binary mode, or a base64 encoded image. If a list is provided, each element should be a string or numpy array representing an image, and the function will process images in batch. model_name (str): Model for face recognition. Options: VGG-Face, Facenet, Facenet512, OpenFace, DeepFace, DeepID, Dlib, ArcFace, SFace and GhostFaceNet (default is VGG-Face). detector_backend (string): face detector backend. Options: 'opencv', 'retinaface', 'mtcnn', 'ssd', 'dlib', 'mediapipe', 'yolov8n', 'yolov8m', 'yolov8l', 'yolov11n', 'yolov11s', 'yolov11m', 'yolov11l', 'yolov12n', 'yolov12s', 'yolov12m', 'yolov12l', 'centerface' or 'skip' (default is opencv). distance_metric (string): Metric for measuring similarity. Options: 'cosine', 'euclidean', 'angular' (default is cosine). enforce_detection (boolean): If no face is detected in an image, raise an exception. Set to False to avoid the exception for low-resolution images (default is True). align (bool): Flag to enable face alignment (default is True). l2_normalize (bool): Flag to enable L2 normalization (unit vector normalization) expand_percentage (int): expand detected facial area with a percentage (default is 0). normalization (string): Normalize the input image before feeding it to the model. Options: base, raw, Facenet, Facenet2018, VGGFace, VGGFace2, ArcFace (default is base). anti_spoofing (boolean): Flag to enable anti spoofing (default is False). similarity_search (boolean): If False, performs identity verification and returns images of the same person. If True, performs similarity search and returns visually similar faces (e.g., celebrity or parental look-alikes). Default is False. k (int): Number of top similar faces to retrieve from the database for each detected face. If not specified, all faces within the threshold will be returned (default is None). search_method (str): Method to use for searching identities. Options: 'exact', 'ann'. To use ann search, you must run build_index function first to create the index. database_type (str): Type of database to search identities. Options: 'postgres', 'mongo', 'weaviate', 'neo4j', 'pgvector', 'pinecone' (default is 'postgres'). connection_details (dict or str): Connection details for the database. connection (Any): Existing database connection object. If provided, this connection will be used instead of creating a new one. Note: Instead of providing `connection` or `connection_details`, database connection information can be supplied via environment variables: - DEEPFACE_POSTGRES_URI - DEEPFACE_MONGO_URI - DEEPFACE_WEAVIATE_URI - DEEPFACE_NEO4J_URI - DEEPFACE_PINECONE_API_KEY Returns: results (List[pd.DataFrame]): A list of pandas dataframes or a list of dicts. Each dataframe or dict corresponds to the identity information for an individual detected in the source image. The DataFrame columns or dict keys include: - id: ID of the detected individual. - img_name: Name of the image file in the database. - model_name: Name of the model used for recognition. - aligned: Whether face alignment was performed. - l2_normalized: Whether L2 normalization was applied. - search_method: Method used for searching identities: exact or ann. - target_x, target_y, target_w, target_h: Bounding box coordinates of the target face in the database. Notice that source image's face coordinates are not included in the result here. - threshold: threshold to determine a pair whether same person or different persons - confidence: Confidence score indicating the likelihood that the images represent the same person. The score is between 0 and 100, where higher values indicate greater confidence in the verification result. - distance_metric: Distance metric used for similarity measurement. Distance metric will be ignored for ann search, and set to cosine if l2_normalize is True, euclidean if l2_normalize is False. - distance: Similarity score between the faces based on the specified model and distance metric """ dfs: List[pd.DataFrame] = [] # adjust distance metric if search_method == "ann": # ann does cosine for l2 normalized vectors, euclidean for non-l2 normalized vectors new_distance_metric = "cosine" if l2_normalize is True else "euclidean" if new_distance_metric != distance_metric: logger.warn( f"Overwriting distance_metric to '{new_distance_metric}' since " f"{'vectors are L2-norm' if l2_normalize else 'vectors are not L2-norm'}." ) distance_metric = new_distance_metric elif search_method != "exact": if l2_normalize is True and distance_metric == "euclidean": logger.warn( "Overwriting distance_metric to 'euclidean_l2' since vectors are L2 normalized." ) distance_metric = "euclidean_l2" threshold = find_threshold(model_name=model_name, distance_metric=distance_metric) db_client = __connect_database( database_type=database_type, connection_details=connection_details, connection=connection, ) results = __get_embeddings( img=img, model_name=model_name, detector_backend=detector_backend, enforce_detection=enforce_detection, align=align, anti_spoofing=anti_spoofing, expand_percentage=expand_percentage, normalization=normalization, l2_normalize=l2_normalize, return_face=False, ) is_vector_db = database_inventory[database_type]["is_vector_db"] if search_method == "ann" and is_vector_db is False: try: import faiss except ImportError as e: raise ValueError( "faiss is not installed. Please install faiss to use approximate nearest neighbour." ) from e embeddings_index_bytes = db_client.get_embeddings_index( model_name=model_name, detector_backend=detector_backend, aligned=align, l2_normalized=l2_normalize, ) embeddings_index_buffer = np.frombuffer(embeddings_index_bytes, dtype=np.uint8) embeddings_index = faiss.deserialize_index(embeddings_index_buffer) logger.info("Loaded embeddings index from database.") for result in results: query_vector = np.array(result["embedding"], dtype="float32").reshape(1, -1) distances, indices = embeddings_index.search(query_vector, k or 20) instances = [] for i, index in enumerate(indices[0]): distance = ( math.sqrt(distances[0][i]) if distance_metric == "euclidean" else distances[0][i] / 2 ) verified = bool(distance <= threshold) instance = { "id": index, # "img_name": "N/A", # need to fetch from DB if required "model_name": model_name, "detector_backend": detector_backend, "aligned": align, "l2_normalized": l2_normalize, "search_method": search_method, "target_x": result.get("facial_area", {}).get("x", None), "target_y": result.get("facial_area", {}).get("y", None), "target_w": result.get("facial_area", {}).get("w", None), "target_h": result.get("facial_area", {}).get("h", None), "threshold": threshold, "distance_metric": distance_metric, "distance": distance, "confidence": find_confidence( distance=distance, model_name=model_name, distance_metric=distance_metric, verified=verified, ), } if similarity_search is False and verified: instances.append(instance) if len(instances) == 0: continue df = pd.DataFrame(instances) df = df.sort_values(by="distance", ascending=True).reset_index(drop=True) if k is not None and k > 0: df = df.nsmallest(k, "distance") # we should query DB to get img_name for each id id_mappings = db_client.search_by_id(ids=df["id"].tolist()) ids_df = pd.DataFrame(id_mappings, columns=["id", "img_name"]) df = df.merge(ids_df, on="id", how="left") del ids_df dfs.append(df) return dfs elif search_method == "ann" and is_vector_db is True: for result in results: target_vector: List[float] = result["embedding"] neighbours = db_client.search_by_vector( vector=target_vector, model_name=model_name, detector_backend=detector_backend, aligned=align, l2_normalized=l2_normalize, limit=k or 20, ) if not neighbours: raise ValueError( "No embeddings found in the database for the criteria " f"{model_name=}, {detector_backend=}, {align=}, {l2_normalize=}." "You must call register some embeddings to the database before using search." ) instances = [] for neighbour in neighbours: distance = neighbour["distance"] verified = bool(distance <= threshold) instance = { "id": neighbour["id"], "img_name": neighbour["img_name"], "model_name": model_name, "detector_backend": detector_backend, "aligned": align, "l2_normalized": l2_normalize, "search_method": search_method, "target_x": result.get("facial_area", {}).get("x", None), "target_y": result.get("facial_area", {}).get("y", None), "target_w": result.get("facial_area", {}).get("w", None), "target_h": result.get("facial_area", {}).get("h", None), "threshold": threshold, "distance_metric": distance_metric, "distance": distance, "confidence": find_confidence( distance=distance, model_name=model_name, distance_metric=distance_metric, verified=verified, ), } if similarity_search is False and verified: instances.append(instance) if len(instances) > 0: df = pd.DataFrame(instances) df = df.sort_values(by="distance", ascending=True).reset_index(drop=True) if k is not None and k > 0: df = df.nsmallest(k, "distance") dfs.append(df) return dfs elif search_method == "exact": source_embeddings = db_client.fetch_all_embeddings( model_name=model_name, detector_backend=detector_backend, aligned=align, l2_normalized=l2_normalize, ) if not source_embeddings: raise ValueError( "No embeddings found in the database for the criteria " f"{model_name=}, {detector_backend=}, {align=}, {l2_normalize=}." "You must call register some embeddings to the database before using search." ) for result in results: target_embedding = cast(List[float], result["embedding"]) df = pd.DataFrame(source_embeddings) df["target_embedding"] = [target_embedding for _ in range(len(df))] df["search_method"] = search_method df["target_x"] = result.get("facial_area", {}).get("x", None) df["target_y"] = result.get("facial_area", {}).get("y", None) df["target_w"] = result.get("facial_area", {}).get("w", None) df["target_h"] = result.get("facial_area", {}).get("h", None) df["threshold"] = threshold df["distance_metric"] = distance_metric if distance_metric == "cosine": df["distance"] = df.apply( lambda row: find_cosine_distance(row["embedding"], row["target_embedding"]), axis=1, ) elif distance_metric == "euclidean": df["distance"] = df.apply( lambda row: find_euclidean_distance(row["embedding"], row["target_embedding"]), axis=1, ) elif distance_metric == "angular": df["distance"] = df.apply( lambda row: find_angular_distance(row["embedding"], row["target_embedding"]), axis=1, ) elif distance_metric == "euclidean_l2": df["distance"] = df.apply( lambda row: find_euclidean_distance( find_l2_normalize(row["embedding"]), find_l2_normalize(row["target_embedding"]), ), axis=1, ) else: raise ValueError(f"Unsupported distance metric: {distance_metric}") df["confidence"] = df.apply( lambda row: find_confidence( distance=row["distance"], model_name=model_name, distance_metric=distance_metric, verified=bool(row["distance"] <= threshold), ), axis=1, ) df = df.drop(columns=["embedding", "target_embedding"]) if similarity_search is False: df = df[df["distance"] <= threshold] if k is not None and k > 0: df = df.nsmallest(k, "distance") df = df.sort_values(by="distance", ascending=True).reset_index(drop=True) dfs.append(df) return dfs else: raise ValueError(f"Unsupported search method: {search_method}") def build_index( model_name: str = "VGG-Face", detector_backend: str = "opencv", align: bool = True, l2_normalize: bool = False, database_type: str = "postgres", connection_details: Optional[Union[Dict[str, Any], str]] = None, connection: Any = None, batch_size: int = 1000, max_neighbors_per_node: int = 32, ) -> None: """ Build index for faster search in the database. You should set search_method to 'ann' in the search function to use the built index. Args: model_name (str): Model for face recognition. Options: VGG-Face, Facenet, Facenet512, OpenFace, DeepFace, DeepID, Dlib, ArcFace, SFace and GhostFaceNet (default is VGG-Face). detector_backend (string): face detector backend. Options: 'opencv', 'retinaface', 'mtcnn', 'ssd', 'dlib', 'mediapipe', 'yolov8n', 'yolov8m', 'yolov8l', 'yolov11n', 'yolov11s', 'yolov11m', 'yolov11l', 'yolov12n', 'yolov12s', 'yolov12m', 'yolov12l', 'centerface' or 'skip' (default is opencv). align (bool): Flag to enable face alignment (default is True). l2_normalize (bool): Flag to enable L2 normalization (unit vector normalization) max_neighbors_per_node (int): Maximum number of neighbors per node in the index (default is 32). database_type (str): Type of database to build index. Options: 'postgres', 'mongo', 'weaviate', 'neo4j', 'pgvector', 'pinecone' (default is 'postgres'). connection (Any): Existing database connection object. If provided, this connection will be used instead of creating a new one. connection_details (dict or str): Connection details for the database. Note: Instead of providing `connection` or `connection_details`, database connection information can be supplied via environment variables: - DEEPFACE_POSTGRES_URI - DEEPFACE_MONGO_URI - DEEPFACE_WEAVIATE_URI - DEEPFACE_NEO4J_URI - DEEPFACE_PINECONE_API_KEY """ if database_inventory.get(database_type) is None: raise ValueError(f"Unsupported database type: {database_type}") is_vector_db = database_inventory[database_type]["is_vector_db"] if is_vector_db is True: logger.info(f"{database_type} manages its own indexes. No need to build index manually.") return try: import faiss except ImportError as e: raise ValueError("faiss is not installed. Please install faiss to use build_index.") from e db_client = __connect_database( database_type=database_type, connection_details=connection_details, connection=connection, ) index = __get_index( db_client=db_client, model_name=model_name, detector_backend=detector_backend, align=align, l2_normalize=l2_normalize, ) if index is not None: indexed_indices = faiss.vector_to_array(index.id_map) indexed_embeddings = set(indexed_indices) logger.info(f"Found {len(indexed_embeddings)} embeddings already indexed in the database.") else: logger.info("No existing index found in the database. A new index will be created.") indexed_embeddings = set() tic = time.time() source_embeddings = db_client.fetch_all_embeddings( model_name=model_name, detector_backend=detector_backend, aligned=align, l2_normalized=l2_normalize, ) toc = time.time() if not source_embeddings: raise ValueError( "No embeddings found in the database for the criteria " f"{model_name=}, {detector_backend=}, {align=}, {l2_normalize=}." "You must call register some embeddings to the database before using build_index." ) logger.info( f"Fetched {len(source_embeddings)} embeddings from database in {toc - tic:.2f} seconds." ) unindexed_source_embeddings = [ item for item in source_embeddings if item["id"] not in indexed_embeddings ] if not unindexed_source_embeddings: logger.info("All embeddings are already indexed. No new embeddings to index.") return ids = [item["id"] for item in unindexed_source_embeddings] vectors = np.array([item["embedding"] for item in unindexed_source_embeddings], dtype="float32") embedding_dim_size = len(source_embeddings[0]["embedding"]) if index is None: base_index = faiss.IndexHNSWFlat(embedding_dim_size, max_neighbors_per_node) index = faiss.IndexIDMap(base_index) tic = time.time() for i in range(0, len(vectors), batch_size): batch_ids = np.array(ids[i : i + batch_size], dtype="int64") batch_vectors = vectors[i : i + batch_size] index.add_with_ids(batch_vectors, batch_ids) toc = time.time() logger.info(f"Added {len(vectors)} embeddings to index in {toc - tic:.2f} seconds.") index_path = os.path.join( tempfile.gettempdir(), f"{model_name}_{detector_backend}_{align}_{l2_normalize}.faiss" ) # now create index from scratch, then think how to load an index and add new vectors to it tic = time.time() faiss.write_index(index, index_path) toc = time.time() logger.info(f"Saved index to disk in {toc - tic:.2f} seconds") with open(index_path, "rb") as f: index_data = f.read() tic = time.time() db_client.upsert_embeddings_index( model_name=model_name, detector_backend=detector_backend, aligned=align, l2_normalized=l2_normalize, index_data=index_data, ) toc = time.time() logger.info(f"Upserted index to database in {toc - tic:.2f} seconds.") # clean up temp file if os.path.exists(index_path): os.remove(index_path) def __get_embeddings( img: Union[str, NDArray[Any], IO[bytes], List[str], List[NDArray[Any]], List[IO[bytes]]], model_name: str = "VGG-Face", detector_backend: str = "opencv", enforce_detection: bool = True, align: bool = True, l2_normalize: bool = False, expand_percentage: int = 0, normalization: str = "base", anti_spoofing: bool = False, return_face: bool = True, ) -> List[Dict[str, Any]]: """ Get embeddings for given image(s) Args: img (str or np.ndarray or IO[bytes] or list): The exact path to the image, a numpy array in BGR format, a file object that supports at least `.read` and is opened in binary mode, or a base64 encoded image. If a list is provided, each element should be a string or numpy array representing an image, and the function will process images in batch. model_name (str): Model for face recognition. Options: VGG-Face, Facenet, Facenet512, OpenFace, DeepFace, DeepID, Dlib, ArcFace, SFace and GhostFaceNet (default is VGG-Face). detector_backend (string): face detector backend. Options: 'opencv', 'retinaface', 'mtcnn', 'ssd', 'dlib', 'mediapipe', 'yolov8n', 'yolov8m', 'yolov8l', 'yolov11n', 'yolov11s', 'yolov11m', 'yolov11l', 'yolov12n', 'yolov12s', 'yolov12m', 'yolov12l', 'centerface' or 'skip' (default is opencv). enforce_detection (boolean): If no face is detected in an image, raise an exception. Set to False to avoid the exception for low-resolution images (default is True). align (bool): Flag to enable face alignment (default is True). l2_normalize (bool): Flag to enable L2 normalization (unit vector normalization) expand_percentage (int): expand detected facial area with a percentage (default is 0). normalization (string): Normalize the input image before feeding it to the model. Options: base, raw, Facenet, Facenet2018, VGGFace, VGGFace2, ArcFace (default is base). anti_spoofing (boolean): Flag to enable anti spoofing (default is False). return_face (bool): Whether to return the aligned face along with the embedding (default is True). Returns: results (List[Dict]): A list of dictionaries containing embeddings and optionally aligned face images for each detected face in the input image(s). """ results = represent( img_path=img, model_name=model_name, detector_backend=detector_backend, enforce_detection=enforce_detection, align=align, anti_spoofing=anti_spoofing, expand_percentage=expand_percentage, normalization=normalization, l2_normalize=l2_normalize, return_face=return_face, ) if len(results) == 0: raise ValueError("No embeddings were detected in the provided image(s).") flat_results: List[Dict[str, Any]] = [] for result in results: if isinstance(result, dict): flat_results.append(result) elif isinstance(result, list): flat_results.extend(result) return flat_results def __connect_database( database_type: str = "postgres", connection_details: Optional[Union[Dict[str, Any], str]] = None, connection: Any = None, ) -> Database: """ Connect to the specified database type Args: database_type (str): Type of database to connect. Options: 'postgres', 'mongo', 'weaviate', 'neo4j', 'pgvector', 'pinecone' (default is 'postgres'). connection_details (dict or str): Connection details for the database. connection (Any): Existing database connection object. If provided, this connection will be used instead of creating a new one. Note: Instead of providing `connection` or `connection_details`, database connection information can be supplied via environment variables: - DEEPFACE_POSTGRES_URI - DEEPFACE_MONGO_URI - DEEPFACE_WEAVIATE_URI - DEEPFACE_NEO4J_URI - DEEPFACE_PINECONE_API_KEY Returns: db_client (Database): An instance of the connected database client. """ if database_inventory.get(database_type) is None: raise ValueError(f"Unsupported database type: {database_type}") client_class = database_inventory[database_type]["client"] return client_class(connection_details=connection_details, connection=connection) def __get_index( db_client: Database, model_name: str, detector_backend: str, align: bool, l2_normalize: bool, ) -> Any: """ Retrieve the embeddings index from the database Args: db_client (Database): An instance of the connected database client. model_name (str): Model for face recognition. detector_backend (string): face detector backend. align (bool): Flag to enable face alignment. l2_normalize (bool): Flag to enable L2 normalization (unit vector normalization) Returns: embeddings_index (Any): The deserialized embeddings index object, or None if not found. """ import faiss try: embeddings_index_bytes = db_client.get_embeddings_index( model_name=model_name, detector_backend=detector_backend, aligned=align, l2_normalized=l2_normalize, ) embeddings_index_buffer = np.frombuffer(embeddings_index_bytes, dtype=np.uint8) embeddings_index = faiss.deserialize_index(embeddings_index_buffer) return embeddings_index except ValueError: return None