Source code for luna_backport3.app.handlers.handler_identify

""" Handler for identify resource"""

from sanic.response import HTTPResponse

from app.handlers.handlers_mixin import HandlersMixin
from app.handlers.matcher_basic_handler import MatcherBasicHandler
from app.handlers.schemas import PATCH_HANDLER_IDENTIFY_SCHEMA
from app.mixins import CachedHandlerMixin
from classes.enums import HandlerType, ListType
from classes.handler_defaults import IdentifyHandlerPolicies
from crutches_on_wheels.web.query_getters import boolFrom01Getter, int01Getter


class HandlerIdentify(CachedHandlerMixin, HandlersMixin, MatcherBasicHandler):
    """
    Handler for identifying persons on provided image.

    Resource: "/{api_version}/identify"
    """

    async def post(self) -> HTTPResponse:
        """
        Identify person on provided image, see `spec_handlers_identify`_.

        .. _`spec_handlers_identify`:
            _static/api.html#operation/identifyFace
        """

        noCache = self.getQueryParam("no_cache", boolFrom01Getter, default=False)
        warpedImage = self.getQueryParam("warped_image", int01Getter, default=0)

        handler = await self.getHandler(HandlerType.identify, noCache)
        handlerId, policiesDict = handler["handler_id"], handler["policies"]
        handlerPolicies = IdentifyHandlerPolicies(**policiesDict)

        event = await self.emitHandlerEvent(
            handlerType=HandlerType.identify,
            handlerId=handlerId,
            handlerPolicies=handlerPolicies,
            warpedImage=warpedImage,
        )
        face = event.face
        exif = event.exif
        faceCandidates = event.event["matches"][0]["candidates"]

        faceCandidatesIds = [match["face"]["face_id"] for match in faceCandidates]
        descriptorToPersonMap = await self.dbContext.getPersonsByDescriptors(descriptorIds=faceCandidatesIds)

        listType = await self.getCachedListType(policiesDict["list_id"])

        if listType == ListType.persons:
            candidates = self.makeCandidatesList(descriptorToPersonMap, faceCandidates)
        else:
            candidates = [
                {"id": match["face"]["face_id"], "similarity": match["similarity"]} for match in faceCandidates
            ]

        result = {"candidates": candidates, "face": face}

        if exif is not None:
            result["exif"] = exif

        return self.success(201, outputJson=result, extraHeaders=self.getExtraHeaders())

    async def get(self) -> HTTPResponse:
        """
        Get handler info. See `spec_get_handler_identify`_.

        .. _spec_get_handler_identify:
            _static/api.html#operation/getHandlerIdentify
        """
        handler = await self.getHandler(handlerType=HandlerType.identify, noCache=True)

        return self.success(200, outputJson=handler)

    async def patch(self) -> HTTPResponse:
        """
        Patch handler. See `spec_patch_handler_identify`_.

        .. _spec_patch_handler_identify:
            _static/api.html#operation/patchHandlerIdentify
        """
        reqJson = self.request.json
        self.validateJson(reqJson, PATCH_HANDLER_IDENTIFY_SCHEMA, useJsonSchema=False)
        if "limit" in reqJson:
            reqJson["limit"] = min(self.config.maxCandidateCount, reqJson["limit"])
        actualHandlerVersion = await self.patchHandler(handlerType=HandlerType.identify, policies=reqJson)

        return self.success(200, outputJson={"version": actualHandlerVersion})