Source code for luna_handlers.app.handlers.detector_handler

""" Detector Handler

Module realize detector handler.
"""
from typing import Union, List, Awaitable

from sanic.response import HTTPResponse

from app.api_sdk_adaptors.detector import APISDKDetectorAdaptor
from app.global_vars.enums import ImageType
from app.handlers.base_handler import BaseHandlerWithMultipart
from app.handlers.custom_query_getters import int0180Getter, multifacePolicyGetter
from classes.multipart_processing import DetectorMultipartProcessor
from classes.schemas.detector import Detector
from classes.schemas.storage_policy import saveSamples
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.web.query_getters import uuidGetter, int01Getter, boolFrom01Getter
from crutches_on_wheels.monitoring.points import monitorTime, DataForMonitoring
from img_utils.utils import getExif
from sdk.sdk_loop.enums import MultifacePolicy
from sdk.sdk_loop.estimation_targets import SDKEstimationTargets, SDKFaceEstimationTargets
from sdk.sdk_loop.sdk_task import SDKTask, SDKTaskFilters, SDKDetectableImage, FaceWarp


[docs]class DetectorHandler(BaseHandlerWithMultipart): """ Handler for detecting faces on images. Resource: "/{api_version}/detector" """
[docs] async def getDataFromMultipart( self, imageType: ImageType = ImageType.rawImage ) -> Union[List[SDKDetectableImage], List[FaceWarp]]: """Description see :func:`~BaseHandlerWithMultipart.getDataFromMultipart`.""" dataFromRequest = await DetectorMultipartProcessor().getData(self.request) estimationDataFromMultiPart = self._getDataFromMultipart(dataFromRequest.images, imageType) return estimationDataFromMultiPart
def _getImagesFromSamples( self, inputJson: dict, imageType: Union[ImageType, None], defaultDetectTime: str ) -> Awaitable[Union[List[SDKDetectableImage], List[FaceWarp]]]: """ Stub unknown image type for face samples. """ sampleImageType = imageType if imageType is None: sampleImageType = ImageType.faceWarp return super()._getImagesFromSamples( inputJson=inputJson, imageType=sampleImageType, defaultDetectTime=defaultDetectTime )
[docs] async def post(self) -> HTTPResponse: """ Detect faces on images. See `spec_detector`_. .. _spec_detector: _static/api.html#operation/detectFaces Returns: response with succeeded processed images and failed processed images """ self.request.dataForMonitoring: DataForMonitoring estimateHeadPose = self.getQueryParam("estimate_head_pose", int01Getter, default=False) detectLandmarks68 = self.getQueryParam("detect_landmarks68", int01Getter, default=False) extractExif = self.getQueryParam("extract_exif", int01Getter, default=False) estimateQuality = self.getQueryParam("estimate_quality", int01Getter, default=False) estimateGaze = self.getQueryParam("estimate_gaze", int01Getter, default=False) estimateEyesAttributes = self.getQueryParam("estimate_eyes_attributes", int01Getter, default=False) estimateMouthAttributes = self.getQueryParam("estimate_mouth_attributes", int01Getter, default=False) estimateEmotions = self.getQueryParam("estimate_emotions", int01Getter, default=False) estimateMask = self.getQueryParam("estimate_mask", int01Getter, default=False) pitchThreshold = self.getQueryParam("pitch_threshold", int0180Getter, default=None) rollThreshold = self.getQueryParam("roll_threshold", int0180Getter, default=None) yawThreshold = self.getQueryParam("yaw_threshold", int0180Getter, default=None) warpedImage = self.getQueryParam("warped_image", int01Getter) imageType = {1: ImageType.faceWarp, 0: ImageType.rawImage, None: None}[warpedImage] self.accountId = self.getQueryParam("account_id", uuidGetter, require=True) multifacePolicy = self.getQueryParam("multiface_policy", multifacePolicyGetter, default=MultifacePolicy.allowed) autoOrient = self.getQueryParam("use_exif_info", boolFrom01Getter, default=True) with monitorTime(self.request.dataForMonitoring, "download_images_time"): inputData = await self.getInputEstimationData( self.request, imageType=imageType, validationModel=Detector, autoOrient=autoOrient ) faceTargets = SDKFaceEstimationTargets( estimateQuality=estimateQuality, estimateMouthAttributes=estimateMouthAttributes, estimateAGS=0, estimateGaze=estimateGaze, estimateEyesAttributes=estimateEyesAttributes, estimateEmotions=estimateEmotions, estimateMask=estimateMask, estimateHeadPose=estimateHeadPose, ) toEstimate = SDKEstimationTargets(estimateHuman=0, faceEstimationTargets=faceTargets) filters = SDKTaskFilters(yawThreshold=yawThreshold, pitchThreshold=pitchThreshold, rollThreshold=rollThreshold) task = SDKTask(toEstimate, data=inputData, filters=filters, multifacePolicy=multifacePolicy) detector = APISDKDetectorAdaptor(accountId=self.accountId, logger=self.logger, sdkLoop=self.sdkLoop,) replyImages, monitoringData, warpsToSave = await detector.detect(task, detectLandmarks68) for detectImage in replyImages: for imageDetection in detectImage.detections: if imageDetection.face is not None: imageDetection.face.url = ( f"{self.luna3Client.lunaFaceSamplesStore.baseUri}/buckets/" f"{self.config.faceSamplesStorage.bucket}/images/" f"{imageDetection.face.sdkEstimation.warp.sampleId}" ) with monitorTime(monitoringData.request, "save_warps_time"): await saveSamples( warpsToSave=warpsToSave, bucket=self.config.faceSamplesStorage.bucket, accountId=self.accountId, storeApiClient=self.luna3Client.lunaFaceSamplesStore, ) self.handleMonitoringData(monitoringData) prepareResponseData = {image.id: image.asDict() for image in replyImages} if extractExif: for sdkImage, replyImage in zip(inputData, replyImages): if sdkImage.error is None and replyImage.error == Error.Success: prepareResponseData[sdkImage.id]["exif"] = getExif(sdkImage) return self.success(201, outputJson={"images": list(prepareResponseData.values())})