-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
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 <[email protected]>
- Loading branch information
1 parent
e971051
commit adfb99c
Showing
8 changed files
with
360 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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}" | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.