Source code for luna_backport3.app.handlers.matcher_handler

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())