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