Source code for luna_python_matcher.app_proxy.handlers.base_handler

""" Proxy Base handler

Module realize base class for all handlers.
"""

from typing import Optional, Iterable

from luna3.client import Client
from vlutils.descriptors.data import DescriptorType

from app_common.handlers.base_matcher_handler import CommonBaseMatcherHandler
from app_proxy.matcher.base_struct import (
    MatchResult,
    MatchUnitType,
    Reference,
    Candidate,
    MatchRequest,
    Descriptor,
    ErrorMatchResult,
    FaceExternalIdReference,
    EventExternalIdReference,
    EventTrackIdReference,
)
from classes.descriptor_validator import DescriptorValidator, InvalidDescriptor
from classes.matching_storage import MatchingStorage
from classes.enums import ReferenceType
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException


[docs]class BaseMatcherHandler(CommonBaseMatcherHandler): """ Proxy Base handler for other handlers. Attributes: luna3Client (luna3.client.Client): luna3 client """ def __init__(self, request): super().__init__(request) self.luna3Client: Optional[Client] = self.app.ctx.luna3Session.getClient(self.requestId)
[docs] async def getMatchRequests( self, inputJson: dict, accountId: Optional[str] = None, matchDescriptorType: DescriptorType = DescriptorType.face, ) -> MatchingStorage: """ Get match candidates and match references from the request. Args: inputJson: input json accountId: account id matchDescriptorType: match descriptor type - face or body Returns: tuple with match requests list and loading errors list """ storage = MatchingStorage() candidates: tuple[Candidate] = tuple(self._getCandidates(inputJson["candidates"], accountId)) references: Iterable[tuple[Reference, Optional[ErrorMatchResult]]] = self._getReferences( rawReferences=inputJson["references"], accountId=accountId, referenceDescriptorType=matchDescriptorType ) async for reference, error in references: for candidate in candidates: matchRequest = MatchRequest( reference=reference, candidate=candidate, accountId=accountId, matchDescriptorType=matchDescriptorType, ) if error: storage.addMatchResult(MatchResult(matchRequest=matchRequest, results=error)) else: storage.addMatchRequest(matchRequest) return storage
[docs] def getDescriptor(self, inputReference: dict) -> Descriptor: """ Override getBinaryReference behaviour in part of returned type. Args: inputReference: a dict from the request Returns: Descriptor obj """ binaryDescriptor, version = self._getBinaryReference(inputReference) return Descriptor(descriptor=binaryDescriptor, version=version)
async def _getReferences( self, rawReferences: list[dict], referenceDescriptorType: DescriptorType, accountId: Optional[str] = None ) -> list[tuple[Reference, Optional[ErrorMatchResult]]]: """ Get references from request. Args: rawReferences: list with references from json referenceDescriptorType: reference descriptor type - face or body accountId: account id from request Yields: tuples of (Reference, ErrorMatchResult) Raises: VLException(Error.LunaEventsIsDisabled, 403, isCriticalError=False) if one of references is `event`, but `luna-events` is disabled """ if referenceDescriptorType == DescriptorType.face: validate = DescriptorValidator(self.config.defaultFaceDescriptorVersion) else: validate = DescriptorValidator(self.config.defaultHumanDescriptorVersion) for inputReference in rawReferences: referenceId, referenceStrType = inputReference["id"], inputReference["type"] if not self.configLunaEventsUsage and referenceStrType == ReferenceType.event.value: raise VLException(Error.LunaEventsIsDisabled, 403, isCriticalError=False) if referenceStrType in self.ENTITY_REFERENCE: yield Reference(type=MatchUnitType[referenceStrType], id=referenceId, descriptor=None), None elif referenceStrType in self.BINARY_REFERENCE: descriptor = self.getDescriptor(inputReference) reference = Reference(type=MatchUnitType.descriptor, descriptor=descriptor, id=referenceId) try: validate(descriptor.descriptor, descriptor.version) except InvalidDescriptor as exc: errorInfo = exc.error error = ErrorMatchResult(errorInfo.errorCode, errorInfo.description, errorInfo.detail) yield reference, error else: yield reference, None elif referenceStrType == ReferenceType.faceExternal.value: faces = await self.facesDBContext.getFaces( externalId=referenceId, accountId=accountId, targets=["face_id"] ) if not faces: errorInfo = Error.NoOneFaceFoundByExternalId.format(referenceId) yield FaceExternalIdReference( type=MatchUnitType.face, id=referenceId ), ErrorMatchResult(errorInfo.errorCode, errorInfo.description, errorInfo.detail) for face in faces: yield FaceExternalIdReference( type=MatchUnitType.face, id=face["face_id"], externalObjectId=referenceId ), None elif referenceStrType == ReferenceType.eventExternal.value: eventIds = await self.eventsDBContext.getEventIdsByExternalFields( externalId=referenceId, accountId=accountId ) if not eventIds: errorInfo = Error.NoOneEventFoundByExternalId.format(referenceId) yield EventExternalIdReference( type=MatchUnitType.event, id=referenceId ), ErrorMatchResult(errorInfo.errorCode, errorInfo.description, errorInfo.detail) for eventId in eventIds: yield EventExternalIdReference( type=MatchUnitType.event, id=eventId, externalObjectId=referenceId ), None elif referenceStrType == ReferenceType.eventTrack.value: eventIds = await self.eventsDBContext.getEventIdsByExternalFields( trackId=referenceId, accountId=accountId ) if not eventIds: errorInfo = Error.NoOneEventFoundByTrackId.format(referenceId) yield EventTrackIdReference( type=MatchUnitType.event, id=referenceId ), ErrorMatchResult(errorInfo.errorCode, errorInfo.description, errorInfo.detail) for eventId in eventIds: yield EventTrackIdReference( type=MatchUnitType.event, id=eventId, externalObjectId=referenceId ), None else: raise ValueError(f'Unknown reference type "{referenceStrType}"')