Source code for luna_handlers.app.handlers.detector_handler

""" Detector Handler

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

from sanic.response import HTTPResponse

from app.api_sdk_adaptors.detector import APISDKDetectorAdaptor
from app.api_sdk_adaptors.orientation import handleImageOrientation
from app.handlers.base_handler import BaseHandlerWithMultipart
from classes.image_meta import InputImageData
from classes.multipart_processing import DetectorMultipartProcessor
from classes.query_schemas.detector import DetectorQueries
from classes.schemas.detector import Detector
from crutches_on_wheels.cow.errors.errors import ErrorInfo
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.monitoring.points import monitorTime
from sdk.sdk_loop.models.image import ImageType
from sdk.sdk_loop.task import HandlersTask
from sdk.sdk_loop.tasks.filters import FaceDetectionFilters, Filters
from sdk.sdk_loop.tasks.task import TaskParams


class DetectorHandler(BaseHandlerWithMultipart):
    """
    Handler for detecting faces on images.

    Resource: "/{api_version}/detector"
    """

    async def getDataFromMultipart(self, imageType: ImageType = ImageType.IMAGE) -> List[InputImageData]:
        """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[List[InputImageData]]:
        """
        Stub unknown image type for face samples.
        """
        sampleImageType = imageType
        if imageType is None:
            sampleImageType = ImageType.FACE_WARP
        return super()._getImagesFromSamples(
            inputJson=inputJson, imageType=sampleImageType, defaultDetectTime=defaultDetectTime
        )

    def _getImagesFromUrls(
        self, inputJson: dict, imageType: Union[ImageType, None], defaultDetectTime: str
    ) -> Awaitable[List[InputImageData]]:
        """
        Stub unknown image type for urls.
        """
        urlImageType = imageType
        if imageType is None:
            urlImageType = ImageType.IMAGE
        return super()._getImagesFromUrls(
            inputJson=inputJson, imageType=urlImageType, defaultDetectTime=defaultDetectTime
        )

    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.accountId = self.getAccountIdFromHeader()
        queries: DetectorQueries = self.loadDataFromQuery(DetectorQueries)

        if queries.estimateFaceQuality:
            self.assertLicence(iso=True)

        params = TaskParams(
            targets=queries.estimationTargets.sdkTargets,
            filters=Filters(
                faceDetection=FaceDetectionFilters(
                    rollThreshold=queries.rollThreshold,
                    yawThreshold=queries.yawThreshold,
                    pitchThreshold=queries.pitchThreshold,
                )
            ),
            multifacePolicy=queries.multifacePolicy,
            useExifInfo=queries.useExifInfo,
            autoRotation=self.config.useAutoRotation,
        )

        with monitorTime(self.request.dataForMonitoring, "download_images_time"):
            inputData = await self.getInputEstimationData(
                self.request, imageType=queries.warpedImage, validationModel=Detector
            )

        task = HandlersTask(data=[metaImage.image for metaImage in inputData], params=params)
        await task.execute()
        if task.result.error:
            raise VLException(ErrorInfo.fromDict(task.result.error.asDict()), 400, isCriticalError=False)
        if self.config.useAutoRotation:
            handleImageOrientation(task.result.images)

        detectorAdaptor = APISDKDetectorAdaptor(
            responseEstimationTargets=queries.estimationTargets.responseTargets,
            accountId=self.accountId,
            sampleStore=self.luna3Client.lunaFaceSamplesStore,
            sampleBucket=self.config.faceSamplesStorage.bucket,
            estimateFaceQuality=queries.estimateFaceQuality,
        )
        result, monitoringData = await detectorAdaptor.buildResult(
            task.result, meta=[metaImage.meta for metaImage in inputData]
        )
        self.handleMonitoringData(monitoringData)
        return self.success(201, outputJson=result)