Source code for luna_handlers.app.handlers.sdk_handler

""" SDK estimator handler"""
from typing import Union, List

import msgpack
from lunavl.sdk.estimators.face_estimators.livenessv1 import LivenessPrediction
from lunavl.sdk.estimators.face_estimators.mask import MaskState
from sanic.response import HTTPResponse
from werkzeug.http import parse_accept_header

from app.api_sdk_adaptors.sdk_adaptor import APISDKAdaptor
from app.global_vars.enums import ImageType
from app.handlers.base_handler import BaseHandlerWithMultipart
from app.handlers.custom_query_getters import (
    multifacePolicyGetter,
    int0180Getter,
    maskStatesValidator,
    livenessStatesValidator,
)
from classes.multipart_processing import SDKMultipartProcessor
from classes.schemas.sdk import SDKInputEstimationsModel
from configs.config import LIVENESS_V2_QUALITY_THRESHOLD
from crutches_on_wheels.enums.attributes import Liveness
from crutches_on_wheels.monitoring.points import monitorTime
from crutches_on_wheels.web.query_getters import int01Getter, float01Getter, boolFrom01Getter
from sdk.sdk_loop.enums import MultifacePolicy
from sdk.sdk_loop.estimation_targets import (
    SDKFaceEstimationTargets,
    SDKEstimationTargets,
    SDKHumanEstimationTargets,
    SDKLivenessEstimationPolicy,
)
from sdk.sdk_loop.sdk_task import SDKTask, SDKTaskFilters, SDKDetectableImage, FaceWarp


[docs]class SDKHandler(BaseHandlerWithMultipart): """ SDK estimator handler Resource: "/{api_version}/sdk" """
[docs] async def getDataFromMultipart( self, imageType: ImageType = ImageType.rawImage ) -> Union[List[SDKDetectableImage], List[FaceWarp]]: """Description see :func:`~BaseHandlerWithMultipart.getDataFromMultipart`.""" dataFromRequest = await SDKMultipartProcessor().getData(self.request) estimationDataFromMultiPart = self._getDataFromMultipart(dataFromRequest.images, imageType) return estimationDataFromMultiPart
[docs] async def post(self) -> HTTPResponse: """ SDK estimations handler. See `spec sdk`_. .. _`spec sdk`: _static/api.html#operation/sdk Returns: Response with estimations """ detectFace = self.getQueryParam("detect_face", int01Getter, default=0) detectBody = self.getQueryParam("detect_body", int01Getter, default=0) estimateHeadPose = self.getQueryParam("estimate_head_pose", int01Getter, default=0) estimateLandmarks68 = self.getQueryParam("estimate_landmarks68", int01Getter, default=0) estimateLandmarks5 = self.getQueryParam("estimate_landmarks5", int01Getter, default=0) estimateQuality = self.getQueryParam("estimate_quality", int01Getter, default=0) estimateGaze = self.getQueryParam("estimate_gaze", int01Getter, default=0) estimateEyesAttributes = self.getQueryParam("estimate_eyes_attributes", int01Getter, default=0) estimateMouthAttributes = self.getQueryParam("estimate_mouth_attributes", int01Getter, default=0) estimateEmotions = self.getQueryParam("estimate_emotions", int01Getter, default=0) estimateMask = self.getQueryParam("estimate_mask", int01Getter, default=0) estimateGlasses = self.getQueryParam("estimate_glasses", int01Getter, default=0) estimateLiveness = self.getQueryParam("estimate_liveness", int01Getter, default=0) multifacePolicy = self.getQueryParam("multiface_policy", multifacePolicyGetter, default=MultifacePolicy.allowed) extractFaceDescriptor = self.getQueryParam("estimate_face_descriptor", int01Getter, default=0) extractBasicAttributes = self.getQueryParam("estimate_basic_attributes", int01Getter, default=0) extractHumanDescriptor = self.getQueryParam("estimate_body_descriptor", int01Getter, default=0) aggregate = self.getQueryParam("aggregate_attributes", int01Getter, default=0) imageType = self.getQueryParam("image_type", lambda x: ImageType(int(x)), default=ImageType.rawImage) estimateHumanWarp = self.getQueryParam("estimate_body_warp", int01Getter, default=0) estimateFaceWarp = self.getQueryParam("estimate_face_warp", int01Getter, default=0) pitchThreshold = self.getQueryParam("pitch_threshold", int0180Getter, default=None) rollThreshold = self.getQueryParam("roll_threshold", int0180Getter, default=None) yawThreshold = self.getQueryParam("yaw_threshold", int0180Getter, default=None) scoreThreshold = self.getQueryParam("score_threshold", float01Getter, default=None) maskStates = self.getQueryParam("mask_states", maskStatesValidator, default=None) if maskStates: maskStates = [MaskState(int(x)) for x in maskStates] livenessStates = self.getQueryParam("liveness_states", livenessStatesValidator, default=None) if livenessStates: livenessStates = [LivenessPrediction(Liveness(int(x)).name) for x in livenessStates] autoOrient = self.getQueryParam("use_exif_info", boolFrom01Getter, default=True) filters = SDKTaskFilters( yawThreshold=yawThreshold, pitchThreshold=pitchThreshold, rollThreshold=rollThreshold, garbageScoreThreshold=scoreThreshold, maskStates=maskStates, livenessStates=livenessStates, ) with monitorTime(self.request.dataForMonitoring, "load_images_for_processing_time"): inputData = await self.getInputEstimationData( self.request, imageType=imageType, validationModel=SDKInputEstimationsModel, autoOrient=autoOrient ) faceTargets = SDKFaceEstimationTargets( estimateQuality=estimateQuality, estimateMouthAttributes=estimateMouthAttributes, estimateAGS=0, estimateGaze=estimateGaze, estimateEyesAttributes=estimateEyesAttributes, estimateEmotions=estimateEmotions, estimateHeadPose=estimateHeadPose, estimateFaceDescriptor=extractFaceDescriptor, estimateBasicAttributes=extractBasicAttributes, estimateMask=estimateMask, estimateGlasses=estimateGlasses, estimateLiveness=SDKLivenessEstimationPolicy( estimate=estimateLiveness, qualityThreshold=LIVENESS_V2_QUALITY_THRESHOLD ), ) self.checkLivenessEstimationLicensing(faceTargets.estimateLiveness.estimate) humanTargets = SDKHumanEstimationTargets(estimateHumanDescriptor=extractHumanDescriptor) estimateFace = int( any((estimateLandmarks68, estimateLandmarks5, detectFace, estimateFaceWarp, not faceTargets.isEmpty())) ) if scoreThreshold is not None: faceTargets.estimateFaceDescriptor = 1 if filters.needFilterByAngles(): faceTargets.estimateHeadPose = 1 estimateHuman = int(detectBody or estimateHumanWarp or not humanTargets.isEmpty()) toEstimate = SDKEstimationTargets( estimateHuman=estimateHuman, faceEstimationTargets=faceTargets, estimateFace=estimateFace, humanEstimationTargets=humanTargets, ) task = SDKTask( toEstimate, data=inputData, filters=filters, multifacePolicy=multifacePolicy, aggregateAttributes=aggregate ) detector = APISDKAdaptor(accountId="", logger=self.logger, sdkLoop=self.sdkLoop) acceptHeader = self.request.headers.get("Accept", "application/json") responseContentType = parse_accept_header(acceptHeader).best_match( ("application/msgpack", "application/json"), "application/json" ) encodeToBase64 = True if responseContentType == "application/msgpack": encodeToBase64 = False resTask, monitoringData = await detector.estimate( task, estimateLandmarks68, estimateHumanWarp=estimateHumanWarp, estimateFaceWarp=estimateFaceWarp, estimateLandmarks5=estimateLandmarks5, estimateHeadPose=estimateHeadPose, estimateFaceDescriptor=extractFaceDescriptor, encodeBytesToBase64=encodeToBase64, ) if faceTargets.estimateLiveness.estimate: self.countLivenessEstimationsPerformed(sum(map(lambda x: not x.error, task.images))) self.handleMonitoringData(monitoringData) if responseContentType == "application/msgpack": body = msgpack.packb(resTask, use_bin_type=True) return self.success(200, body=body, contentType="application/msgpack") return self.success(200, outputJson=resTask)