from typing import List, Optional, Union
from luna3.common.http_objs import MatchPolicy
from luna3.python_matcher.match_objects import Candidates, FaceFilters, Reference
from sanic.response import HTTPResponse
from app.handlers.custom_query_getters import listUUID4sGetter, uuid4Getter
from app.handlers.handlers_mixin import HandlersMixin
from app.handlers.matcher_basic_handler import MatcherBasicHandler
from app.mixins import CachedHandlerMixin, InputDataMixin
from classes.enums import HandlerType, ListType
from classes.handler_defaults import ImageProcessingParams, SearchHandlerPolicies
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
from crutches_on_wheels.web.query_getters import boolFrom01Getter, int01Getter
[docs]class MatchingVerifyHandler(MatcherBasicHandler):
"""
Verify handler
Resource: "/{api_version}/matching/verify"
"""
[docs] async def post(self) -> HTTPResponse:
"""
Verify a descriptor by a person, see `spec_matching_verify`_.
.. _`spec_matching_verify`:
_static/api.html#operation/matchingVerify
"""
descriptorId = self.getQueryParam("descriptor_id", require=True, validator=uuid4Getter)
personId = self.getQueryParam("person_id", require=True, validator=uuid4Getter)
personDescriptors = await self.getPersonDescriptors(personId=personId)
if not personDescriptors:
return self.generateEmptyMatchResponse()
reference = Reference(referenceType="face", referenceId=descriptorId)
candidates = Candidates(filters=FaceFilters(faceIds=personDescriptors), targets=["face_id", "similarity"])
matchResult = (
await self.lunaApiClient.match(references=[reference], candidates=[candidates], raiseError=True)
).json
return self.success(
statusCode=201, outputJson={"candidates": await self.generateVerifyMatchResult(matchResult, personId)}
)
[docs]class MatchHandler(MatcherBasicHandler):
"""
Match handler
Resource: "/{api_version}/matching/match"
"""
[docs] async def post(self) -> HTTPResponse:
"""
Match a descriptor or person by descriptors, see `spec_matching_match`_.
.. _`spec_matching_match`:
_static/api.html#operation/matchingMatch
Raises:
VLException(Error.RequiredQueryParameterNotFound, 400, isCriticalError=False)
if required query parameter not found
"""
descriptorId = self.getQueryParam("descriptor_id", validator=uuid4Getter)
personId = self.getQueryParam("person_id", validator=uuid4Getter)
listId = self.getQueryParam("list_id", validator=uuid4Getter)
descriptorIds = self.getQueryParam("descriptor_ids", validator=listUUID4sGetter)
limit = self.getQueryParam("limit", validator=self.candidateLimitGetter, default=3)
if listId is None and descriptorIds is None:
raise VLException(
Error.RequiredQueryParameterNotFound.format(["list_id", "descriptor_ids"]),
statusCode=400,
isCriticalError=False,
)
if personId is None and descriptorId is None:
raise VLException(
Error.RequiredQueryParameterNotFound.format(["person_id", "descriptor_id"]),
statusCode=400,
isCriticalError=False,
)
if personId is not None:
descriptorsForMatching = await self.getPersonDescriptors(personId)
if not descriptorsForMatching:
return self.generateEmptyMatchResponse()
else:
descriptorsForMatching = [descriptorId]
references = [Reference(referenceType="face", referenceId=descriptor) for descriptor in descriptorsForMatching]
if listId:
candidates = await self.getListAsCandidates(listId, limit)
else:
candidates = await self.generateCandidatesFromDescriptors(descriptorIds, limit)
matchResult = (
await self.lunaApiClient.match(references=references, candidates=[candidates], raiseError=True)
).json
return self.success(
statusCode=201, outputJson={"candidates": self.generateMatchByDescriptorsResult(matchResult)}
)
[docs]class MatchingIdentifyHandler(MatcherBasicHandler):
"""
Identify handler
Resource: "/{api_version}/matching/identify"
"""
[docs] async def post(self) -> HTTPResponse:
"""
Match a descriptor or person by persons, see `spec_matching_identify`_.
.. _`spec_matching_identify`:
_static/api.html#operation/matchingIdentify
Raises:
VLException(Error.RequiredQueryParameterNotFound, 400, isCriticalError=False)
if required query parameter not found
"""
descriptorId = self.getQueryParam("descriptor_id", validator=uuid4Getter)
personId = self.getQueryParam("person_id", validator=uuid4Getter)
listId = self.getQueryParam("list_id", validator=uuid4Getter)
personIds = self.getQueryParam("person_ids", validator=listUUID4sGetter)
limit = self.getQueryParam("limit", validator=self.candidateLimitGetter, default=3)
if personId is None and descriptorId is None:
raise VLException(
Error.RequiredQueryParameterNotFound.format(["person_id", "descriptor_id"]),
statusCode=400,
isCriticalError=False,
)
if listId is None and personIds is None:
raise VLException(
Error.RequiredQueryParameterNotFound.format(["list_id", "person_ids"]),
statusCode=400,
isCriticalError=False,
)
if personId is not None:
descriptorsForMatching = await self.getPersonDescriptors(personId)
if not descriptorsForMatching:
return self.generateEmptyMatchResponse()
else:
descriptorsForMatching = [descriptorId]
references = [Reference(referenceType="face", referenceId=descriptor) for descriptor in descriptorsForMatching]
if listId is not None:
candidates = await self.getListAsCandidates(listId, limit, ListType.persons)
else:
candidates = await self.getPersonsAsCandidates(personIds, limit)
if candidates is None:
return self.generateEmptyMatchResponse()
matchResult = (
await self.lunaApiClient.match(references=references, candidates=[candidates], raiseError=True)
).json
return self.success(
statusCode=201, outputJson={"candidates": await self.generateMatchByPersonsResult(matchResult)}
)
[docs]class MatchingSearchHandler(CachedHandlerMixin, MatcherBasicHandler, HandlersMixin, InputDataMixin):
"""
Search handler
Resource: "/{api_version}/matching/search"
"""
[docs] async def getMatchByPersons(self, listId: Optional[str] = None, personIds: Optional[List[str]] = None) -> bool:
"""
Get match by persons
Args:
listId: list id
personIds: person ids
Returns:
True if match by persons else False
Raises:
VLException(Error.ListNotFound, 400, isCriticalError=False) if list not found
"""
if listId:
try:
listType = await self.getCachedListType(listId)
except VLException as e:
if e.error == Error.ListNotFound:
e.statusCode = 400
raise e
raise
if listType == ListType.persons:
return True
return False
if personIds is not None:
return True
return False
[docs] @staticmethod
def getHandlerPolicies(
imageProcessingParams: ImageProcessingParams,
matchPolicy: MatchPolicy,
limit: int,
listId: Optional[str] = None,
personIds: Optional[List[str]] = None,
descriptorIds: Optional[List[str]] = None,
) -> SearchHandlerPolicies:
"""
Get policies as dict
Args:
imageProcessingParams: prepared image processing params
matchPolicy: prepared match policy
limit: matching limit
listId: list id
personIds: person ids
descriptorIds: descriptor ids
Returns:
search handler policies
"""
return SearchHandlerPolicies(
matchPolicy=matchPolicy,
limit=limit,
listId=listId,
personIds=personIds,
descriptorIds=descriptorIds,
**imageProcessingParams.asDictSnakeCaseKeys(),
)
[docs] async def getMatchPolicy(
self,
limit: int,
listId: Optional[str] = None,
personIds: Optional[List[str]] = None,
descriptorIds: Optional[List[str]] = None,
) -> Union[MatchPolicy, None]:
"""
Get match policy depends on input parameters
Args:
limit: limit
listId: list id
personIds: person ids
descriptorIds: descriptor ids
Returns:
match policy or None if got only person ids and persons have not descriptors
"""
if listId is not None:
return await self.generateMatchPolicyFromList(listId, limit)
if personIds is not None:
return await self.generateMatchPolicyFromPersons(personIds, limit)
return await self.generateMatchPolicyFromDescriptors(descriptorIds, limit)
[docs] async def post(self) -> HTTPResponse:
"""
Search a face on image by persons or descriptors, see `spec_matching_search`_.
.. _`spec_matching_search`:
_static/api.html#operation/matchingSearch
Raises:
VLException(Error.RequiredQueryParameterNotFound, isCriticalError=400, False)
if required query parameter not found
VLException(Error.UnsupportedQueryParam, 400, isCriticalError=False)
if unsupported query parameter found
"""
if not self.getQueryParam("extract_descriptor", validator=int01Getter, default=1):
raise VLException(
error=Error.UnsupportedQueryParam.format("extract_descriptor"), statusCode=400, isCriticalError=False
)
personIds = self.getQueryParam("person_ids", validator=listUUID4sGetter)
limit = self.getQueryParam("limit", validator=self.candidateLimitGetter, default=3)
descriptorIds = self.getQueryParam("descriptor_ids", validator=listUUID4sGetter)
listId = self.getQueryParam("list_id", validator=uuid4Getter)
noCache = self.getQueryParam("no_cache", boolFrom01Getter, default=False)
if listId is None and personIds is None and descriptorIds is None:
raise VLException(
Error.RequiredQueryParameterNotFound.format(["list_id", "person_ids", "descriptor_ids"]),
statusCode=400,
isCriticalError=False,
)
matchPolicy = await self.getMatchPolicy(limit, listId, personIds, descriptorIds)
if matchPolicy is None:
return self.generateEmptyMatchResponse()
data = await self.getRawDataContainer()
handler = await self.getHandler(HandlerType.search, noCache)
handlerId = handler["handler_id"]
imageProcessingParams = self.getImageProcessingParamsFromQuery()
handlerPolicies = self.getHandlerPolicies(
imageProcessingParams, matchPolicy, limit, listId, personIds, descriptorIds
)
events, exif = await self.emitEvents(
handlerType=HandlerType.search,
handlerId=handlerId,
inputData=data,
handlerPolicies=handlerPolicies,
warpedImage=imageProcessingParams.warpedImage,
)
unpreparedResult = await self.getFacesFromEstimation(
eventsFromReply=events, imageProcessingParams=imageProcessingParams, data=data
)
if len(unpreparedResult["faces"]) > 1:
error = Error.ManyFaces
error.detail = unpreparedResult
raise VLException(error=error, isCriticalError=False, statusCode=400)
result = {"face": unpreparedResult["faces"][0]}
if exif is not None:
result["exif"] = exif
resultCandidates = events[0]["matches"][0]["candidates"]
matchByPersons = await self.getMatchByPersons(listId, personIds)
if matchByPersons:
result["candidates"] = await self.generateMatchByPersonsResult(resultCandidates, aggregateMatches=False)
else:
result["candidates"] = self.generateMatchByDescriptorsResult(resultCandidates, aggregateMatches=False)
return self.success(statusCode=201, outputJson=result, extraHeaders=self.getExtraHeaders())
