""" 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}"')
