Source code for luna_python_matcher.app.handlers.matcher_face_handler

""" Matcher handler. """

from typing import Awaitable

from sanic.response import HTTPResponse
from vlutils.descriptors.data import DescriptorType

from app.handlers.base_handler import BaseMatcherHandler
from app_common.handlers.schemas import FaceMatch
from classes.enums import MatchSource
from classes.filters import EventFilters, FaceFilters
from classes.match_processing import MatchLPMProcessor
from crutches_on_wheels.web.query_getters import uuidGetter

CANDIDATE_FILTERS_MAP = {MatchSource.faces.value: FaceFilters, MatchSource.events.value: EventFilters}


class FaceMatcherHandler(BaseMatcherHandler):
    """
    Matcher handler allows to submit tasks to a service that searches for faces similar to a given reference by
    matching them.

    Resource: "/{api_version}/matcher"
    """

    async def post(self) -> HTTPResponse:
        """
        Match events, faces and attributes by faces. See `spec_matcher`_.

        .. _spec_matcher:
            _static/api.html#operation/matching

        Returns:
            response with matching results
        """
        inputJson: dict = self.request.json
        self.validateMatchJson(inputJson, fastjsonSchema=FaceMatch.schema)

        accountId = self.getQueryParam("account_id", uuidGetter, default=None)

        candidates, references = await self.getMatchStructuresFromRequest(
            inputJson=inputJson,
            accountId=accountId,
            descriptorVersion=self.config.defaultFaceDescriptorVersion,
            descriptorType=DescriptorType.face,
        )

        matching = MatchLPMProcessor(
            eventsDBContext=self.eventsDBContext,
            facesDBContext=self.facesDBContext,
            attributesDBContext=self.attributesDBContext,
            candidates=candidates,
            references=references,
            accountId=accountId,
            storageTime=self.config.storageTime,
            requestId=self.request.requestId,
        )
        matchResults = await matching.match()

        return self.success(200, outputJson=matchResults)


class UnwantedFaceMatcherHandler(FaceMatcherHandler):
    """Unwanted matcher handler."""

    def post(self) -> Awaitable[HTTPResponse]:
        """
        Print warning also.
        """
        self.logger.warning("Resource `/1/matcher` is deprecated. Use `/1/matcher/faces` instead.")
        return super().post()