From adfb99cd51328e67d153f7923083229736177108 Mon Sep 17 00:00:00 2001 From: acostapazo Date: Fri, 22 Dec 2023 11:32:06 +0100 Subject: [PATCH] feat(face): adding new feature to call face api (#96) * feat(face): adding new feature to call face api * chore: add headers as optional parameter to face selfie * chore: add metadata to SelfieResult * chore: add headers as optional parameter to face document --------- Co-authored-by: Miguel Lorenzo --- .mypy.ini | 6 +- alice/config.py | 6 +- alice/face/__init__.py | 0 alice/face/face.py | 120 +++++++++++++++++++++++ alice/face/face_models.py | 179 ++++++++++++++++++++++++++++++++++ alice/public_api.py | 5 +- examples/face.py | 45 +++++++++ requirements/requirements.txt | 3 +- 8 files changed, 360 insertions(+), 4 deletions(-) create mode 100644 alice/face/__init__.py create mode 100644 alice/face/face.py create mode 100644 alice/face/face_models.py create mode 100644 examples/face.py diff --git a/.mypy.ini b/.mypy.ini index ea3a867..0453acd 100644 --- a/.mypy.ini +++ b/.mypy.ini @@ -22,4 +22,8 @@ strict_equality = true [pydantic-mypy] init_forbid_extra = True init_typed = True -warn_required_dynamic_aliases = True \ No newline at end of file +warn_required_dynamic_aliases = True + + +[mypy-requests_toolbelt.*] +ignore_missing_imports = True \ No newline at end of file diff --git a/alice/config.py b/alice/config.py index 707fa0f..4409d66 100644 --- a/alice/config.py +++ b/alice/config.py @@ -12,7 +12,7 @@ class Config(BaseSettings): api_key: Union[str, None] = Field(default=None) environment: Union[Environment, None] = Field( - default=Environment.PRODUCTION, + default=Environment.PRODUCTION # , validation_alias="ALICE_ENVIRONMENT" ) verbose: bool = Field(default=False) session: Union[Session, None] = Field(default=None) @@ -23,6 +23,7 @@ class Config(BaseSettings): onboarding_url: Union[str, None] = Field( default="https://apis.alicebiometrics.com/onboarding" ) + face_url: Union[str, None] = Field(default="https://apis.alicebiometrics.com/face") sandbox_url: Union[str, None] = Field( default="https://apis.alicebiometrics.com/onboarding/sandbox", description="This path is only used for trials", @@ -47,6 +48,9 @@ def validate_urls(self) -> "Config": self.onboarding_url = ( f"https://apis.{self.environment.value}.alicebiometrics.com/onboarding" ) + self.face_url = ( + f"https://apis.{self.environment.value}.alicebiometrics.com/face" + ) self.sandbox_url = f"https://apis.{self.environment.value}.alicebiometrics.com/onboarding/sandbox" return self diff --git a/alice/face/__init__.py b/alice/face/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/alice/face/face.py b/alice/face/face.py new file mode 100644 index 0000000..c6f43d8 --- /dev/null +++ b/alice/face/face.py @@ -0,0 +1,120 @@ +from typing import Dict, Union + +from meiga import Failure, Result, Success +from requests import Response, Session + +from alice.config import Config +from alice.face.face_models import DocumentResult, FaceError, MatchResult, SelfieResult + + +class Face: + @staticmethod + def from_config(config: Config) -> "Face": + if config.session: + session = config.session + else: + session = Session() + return Face( + api_key=config.api_key, # type: ignore + url=config.face_url, # type: ignore + timeout=config.timeout, + send_agent=config.send_agent, + verbose=config.verbose, + session=session, + ) + + def __init__( + self, + api_key: str, + url: str, + session: Session, + timeout: Union[float, None] = None, + send_agent: bool = True, + verbose: bool = False, + ): + self.api_key = api_key + self.url = url + self.timeout = timeout + self.send_agent = send_agent + self.verbose = verbose + self.session = session + + def healthcheck(self) -> Response: + response = self.session.get( + url=self.url + "/healthcheck", headers={"apikey": self.api_key} + ) + return response + + def selfie( + self, + media: bytes, + extract_face_pad: bool = True, + extract_face_profile: bool = True, + headers: Union[Dict[str, str], None] = None, + ) -> Result[SelfieResult, FaceError]: + if headers is None: + headers = {} + + response = self.session.post( + url=f"{self.url}/selfie", + headers={"apikey": self.api_key} | headers, + files={"media": media}, + data={ + "extract_face_pad": extract_face_pad, + "extract_face_profile": extract_face_profile, + }, + ) + if response.status_code == 200: + return Success(SelfieResult.from_response(response)) + else: + return Failure(FaceError.from_response(response)) + + def document( + self, + image: bytes, + headers: Union[Dict[str, str], None] = None, + ) -> Result[DocumentResult, FaceError]: + if headers is None: + headers = {} + + response = self.session.post( + url=f"{self.url}/document", + headers={"apikey": self.api_key} | headers, + files={"image": image}, + ) + + if response.status_code == 200: + return Success(DocumentResult.from_response(response)) + else: + return Failure(FaceError.from_response(response)) + + def match_profiles( + self, + face_profile_probe: bytes, + face_profile_target: bytes, + ) -> Result[MatchResult, FaceError]: + response = self.session.post( + url=f"{self.url}/match/profiles", + headers={"apikey": self.api_key}, + files={ + "face_profile_probe": face_profile_probe, + "face_profile_target": face_profile_target, + }, + ) + if response.status_code == 200: + return Success(MatchResult(score=response.json().get("match_score"))) + else: + return Failure(FaceError.from_response(response)) + + def match_media( + self, selfie_media: bytes, document_media: bytes + ) -> Result[MatchResult, FaceError]: + response = self.session.post( + url=f"{self.url}/match/media", + headers={"apikey": self.api_key}, + files={"selfie_media": selfie_media, "document_media": document_media}, + ) + if response.status_code == 200: + return Success(MatchResult(score=response.json().get("match_score"))) + else: + return Failure(FaceError.from_response(response)) diff --git a/alice/face/face_models.py b/alice/face/face_models.py new file mode 100644 index 0000000..dc2ef63 --- /dev/null +++ b/alice/face/face_models.py @@ -0,0 +1,179 @@ +import json +from typing import Any, Dict, Union + +from meiga import Error +from pydantic import BaseModel, Field +from requests import Response +from requests_toolbelt import MultipartDecoder + + +class BoundingBox(BaseModel): + x: Union[int, float] = Field(ge=0) + y: Union[int, float] = Field(ge=0) + width: Union[int, float] = Field(gt=0) + height: Union[int, float] = Field(gt=0) + aspect_ratio: Union[float, None] = None + rotation_angle: Union[float, None] = None + + +def liveness_response(response_encoded: bytes) -> Union[float, None]: + if response_encoded == b"": + return None + pad_score = float(response_encoded.decode()) + return pad_score + + +def metadata_response(response_encoded: bytes) -> Union[Dict[str, Any], None]: + if response_encoded == b"": + return None + return json.loads(response_encoded.decode()) # type: ignore + + +def bytes_response(response_encoded: bytes) -> Union[bytes, None]: + if response_encoded == b"": + return None + return response_encoded + + +def face_bounding_box_response(response_encoded: bytes) -> BoundingBox: + face_bounding_box = json.loads(response_encoded.decode()) + return BoundingBox( + x=face_bounding_box["x"], + y=face_bounding_box["y"], + width=face_bounding_box["x2"] - face_bounding_box["x"], + height=face_bounding_box["y2"] - face_bounding_box["y"], + ) + + +def number_of_faces_response(response_encoded: bytes) -> int: + number_of_faces = int(response_encoded.decode()) + return number_of_faces + + +response_factory = { + "liveness_score": liveness_response, + "metadata": metadata_response, + "face_avatar": bytes_response, + "face_selfie": bytes_response, + "face_profile": bytes_response, + "face_bounding_box": face_bounding_box_response, + "number_of_faces": number_of_faces_response, +} + + +def decode_multipart_response(response: Response) -> Dict[str, Any]: + decoder = MultipartDecoder.from_response(response) + response_dict = {} + for part in decoder.parts: + header_name = part.headers._store.get(b"content-disposition")[1].decode() + header_name_part = header_name.split(";")[1] + separation_idx = header_name_part.index("=") + field_name = header_name_part[separation_idx + 2 : -1] # noqa + if field_name in response_factory: + response_dict[field_name] = response_factory[field_name](part.content) + + return response_dict + + +class SelfieResult(BaseModel): + face_profile: Union[bytes, None] = Field( + None, + description="Binary with the face profile extracted from given media.", + examples=[b"binary data"], + ) + liveness_score: Union[float, None] = Field( + None, + description="Liveness score to determine the input as a genuine attempt (>=50) or attack (<50).", + ge=0.0, + le=100.0, + examples=[90], + ) + number_of_faces: Union[int, None] = Field( + None, description="Number of faces detected.", examples=[1] + ) + metadata: Union[Dict[str, Any], None] = Field( + None, description="Dictionary with some optional metadata" + ) + + def __repr__(self) -> str: + face_profile = ( + f"face_profile(length)={len(self.face_profile)}" + if self.face_profile + else "face_profile=None" + ) + return f"[SelfieResult {face_profile} liveness_score={self.liveness_score} number_of_faces={self.number_of_faces} metadata={self.metadata}]" + + def __str__(self) -> str: + return self.__repr__() + + @staticmethod + def from_response(response: Response) -> "SelfieResult": + multipart_response_dict = decode_multipart_response(response) + face_profile = multipart_response_dict.get("face_profile") + liveness_score = multipart_response_dict.get("liveness_score") + number_of_faces = multipart_response_dict.get("number_of_faces") + metadata = multipart_response_dict.get("metadata") + + return SelfieResult( + face_profile=face_profile, + liveness_score=liveness_score, + number_of_faces=number_of_faces, + metadata=metadata, + ) + + def save_face_profile(self, filename: str) -> None: + if self.face_profile is None: + raise FileNotFoundError("Retrieved face profile is none") + with open(filename, "wb") as file: + file.write(self.face_profile) + + +class DocumentResult(BaseModel): + face_profile: Union[bytes, None] = Field( + None, + description="Binary with the face profile extracted from given image.", + examples=[b"binary data"], + ) + + @staticmethod + def from_response(response: Response) -> "DocumentResult": + multipart_response_dict = decode_multipart_response(response) + face_profile = multipart_response_dict.get("face_profile") + # face_bounding_box = multipart_response_dict.get("face_bounding_box") # TODO REVIEW + return DocumentResult(face_profile=face_profile) + + def save_face_profile(self, filename: str) -> None: + if self.face_profile is None: + raise FileNotFoundError("Retrieved face profile is none") + with open(filename, "wb") as file: + file.write(self.face_profile) + + +class MatchResult(BaseModel): + score: Union[float, None] = Field( + None, + description="Matching score to determine the input as a genuine attempt (>=50) or attack (<50).", + ge=0.0, + le=100.0, + examples=[90], + ) + + +class FaceError(Error): + def __init__(self, message: str, status_code: int, url: str): + self.message = message + self.status_code = status_code + self.url = url + + @staticmethod + def from_response(response: Response) -> "FaceError": + return FaceError( + message=str(response.content), + status_code=response.status_code, + url=response.url, + ) + + def __repr__(self) -> str: + return ( + f"{self.__class__.__name__}: {self.status_code} {self.message} | {self.url}" + ) diff --git a/alice/public_api.py b/alice/public_api.py index b6eb772..6597947 100644 --- a/alice/public_api.py +++ b/alice/public_api.py @@ -3,6 +3,8 @@ """Public API of Alice Onboarding Python SDK""" from typing import List +from alice.face.face import Face +from alice.face.face_models import DocumentResult, FaceError, SelfieResult from alice.onboarding.enums.environment import Environment # Modules @@ -65,6 +67,7 @@ "OtherTrustedDocumentReport", "Environment", ] +face = ["Face", "SelfieResult", "DocumentResult", "FaceError", "BoundingBox"] # Errors @@ -73,4 +76,4 @@ errors = ["OnboardingError", "SandboxError"] -__all__ = modules + classes + errors +__all__ = modules + classes + errors + face diff --git a/examples/face.py b/examples/face.py new file mode 100644 index 0000000..b9209ad --- /dev/null +++ b/examples/face.py @@ -0,0 +1,45 @@ +import os +from typing import Optional + +from alice import Config, Face + +RESOURCES_PATH = f"{os.path.dirname(os.path.abspath(__file__))}/../resources" + + +def face_example(api_key: str, verbose: Optional[bool] = False) -> None: + config = Config(api_key=api_key, verbose=verbose) + face = Face.from_config(config) + + selfie_media_data = given_any_selfie_image_media_data() + document_front_media_data = given_any_document_front_media_data() + + selfie_result = face.selfie(selfie_media_data).unwrap_or_raise() + + selfie_result.save_face_profile("face_profile_v5.bin") + + document_result = face.document(document_front_media_data).unwrap_or_raise() + + match_result = face.match_profiles( + selfie_result.face_profile, document_result.face_profile + ).unwrap_or_raise() + print(match_result) + + +def given_any_selfie_image_media_data() -> bytes: + with open(f"{RESOURCES_PATH}/selfie.png", "rb") as f: + return f.read() + + +def given_any_document_front_media_data() -> bytes: + with open(f"{RESOURCES_PATH}/idcard_esp_front_example.png", "rb") as f: + return f.read() + + +if __name__ == "__main__": + api_key = os.environ.get("ONBOARDING_API_KEY") + if api_key is None: + raise AssertionError( + "Please configure your ONBOARDING_API_KEY to run the example" + ) + print("Running face example...") + face_example(api_key=api_key, verbose=True) diff --git a/requirements/requirements.txt b/requirements/requirements.txt index 8030fac..7391ced 100644 --- a/requirements/requirements.txt +++ b/requirements/requirements.txt @@ -2,4 +2,5 @@ pyjwt>=2.3.0,<3 pydantic>=1.8.2,<3 pydantic-settings<3 requests>=2.26.0,<3 -meiga>=1.9.1,<2 \ No newline at end of file +meiga>=1.9.1,<2 +requests_toolbelt<2 \ No newline at end of file