""" Handler for verify resource"""
from luna3.common.luna_response import LunaResponse
from luna3.python_matcher.match_objects import Candidates, FaceFilters, Reference
from sanic.response import HTTPResponse
from app.handlers.base_handler import APIProxyBaseHandler
from app.handlers.handlers_mixin import HandlersMixin
from app.handlers.matcher_basic_handler import MatcherBasicHandler
from app.handlers.schemas import PATCH_HANDLER_VERIFY_SCHEMA
from app.mixins import CachedHandlerMixin
from classes.enums import HandlerType
from classes.handler_defaults import VerifyHandlerPolicies
from crutches_on_wheels.cow.errors.errors import ErrorInfo
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.web import query_getters as queryGetters
[docs]class HandlerVerify(CachedHandlerMixin, HandlersMixin, MatcherBasicHandler):
"""
Handler for verifying that photo contains given person
Resource: "/{api_version}/verify"
"""
[docs] async def post(self) -> HTTPResponse:
"""
Verify person on provided image, see `spec_handlers_verify`_.
.. _`spec_handlers_verify`:
_static/api.html#operation/verifyFace
"""
noCache = self.getQueryParam("no_cache", queryGetters.boolFrom01Getter, default=False)
personId = self.getQueryParam("person_id", queryGetters.uuidGetter, require=True)
warpedImage = self.getQueryParam("warped_image", queryGetters.int01Getter, default=0)
handler = await self.getHandler(HandlerType.verify, noCache)
handlerId, policiesDict = handler["handler_id"], handler["policies"]
handlerPolicies = VerifyHandlerPolicies(**policiesDict)
event = await self.emitHandlerEvent(
handlerType=HandlerType.verify,
handlerId=handlerId,
handlerPolicies=handlerPolicies,
warpedImage=warpedImage,
)
face = event.face
exif = event.exif
reference = Reference(referenceType="face", referenceId=face["id"])
verifyThreshold = policiesDict["verify_threshold"]
result = {"candidates": [], "face": face, "status": False}
if exif is not None:
result["exif"] = exif
personDescriptors = await self.getPersonDescriptors(personId=personId)
if not personDescriptors:
return self.success(statusCode=201, outputJson=result, extraHeaders=self.getExtraHeaders())
candidates = Candidates(filters=FaceFilters(faceIds=personDescriptors), targets=["face_id", "similarity"])
matchResult = (
await self.lunaApiClient.match(references=[reference], candidates=[candidates], raiseError=True)
).json
candidates = await self.generateVerifyMatchResult(matchResult, personId)
status = candidates and candidates[0]["similarity"] > verifyThreshold
result["candidates"] = candidates
result["status"] = status
return self.success(statusCode=201, outputJson=result, extraHeaders=self.getExtraHeaders())
[docs] async def get(self) -> HTTPResponse:
"""
Get handler info. See `spec_get_handler_verify`_.
.. _spec_get_handler_verify:
_static/api.html#operation/getHandlerVerify
"""
handler = await self.getHandler(handlerType=HandlerType.verify, noCache=True)
return self.success(200, outputJson=handler)
[docs] async def patch(self) -> HTTPResponse:
"""
Patch handler. See `spec_patch_handler_verify`_.
.. _spec_patch_handler_verify:
_static/api.html#operation/patchHandlerVerify
"""
reqJson = self.request.json
self.validateJson(reqJson, PATCH_HANDLER_VERIFY_SCHEMA, useJsonSchema=False)
actualHandlerVersion = await self.patchHandler(handlerType=HandlerType.verify, policies=reqJson)
return self.success(200, outputJson={"version": actualHandlerVersion})
[docs]class HandlerVerifyRaw(CachedHandlerMixin, HandlersMixin, APIProxyBaseHandler):
"""
Handler for raw verifying reference descriptors vs candidates descriptors
Resource: "/{api_version}/verify/raw"
"""
allowedMethods = ("POST",)
[docs] async def postProcessingPost(self, response: LunaResponse) -> HTTPResponse:
"""
Add a verification status to match results
Args:
response: response
Returns:
match response in api format
"""
responseJson = response.json
noCache = self.getQueryParam("no_cache", queryGetters.boolFrom01Getter, default=False)
handler = await self.getHandler(HandlerType.verify, noCache)
verifyThreshold = handler["policies"]["verify_threshold"]
for referenceMatches in responseJson["matches"]:
for match in referenceMatches["matches"]:
match["status"] = match["similarity"] > verifyThreshold
return self.success(200, outputJson=responseJson, extraHeaders=response.headers)
[docs] async def postProcessingFailedRequest(self, response: LunaResponse):
"""
Post processing failed response from the api
Args:
response: response
Raises:
VLException(ErrorInfo): re raise error
"""
raise VLException(ErrorInfo.fromDict(response.json), response.statusCode, False)
[docs] def prepareUrl(self) -> str:
"""
Prepare url for raw matching
Returns:
raw matching url
"""
return f"{self.lunaApiClient.baseUri}/matcher/raw"
