Source code for luna_handlers.app.handlers.iso_handler

""" ISO estimator handler"""
from typing import List

import msgpack
from sanic.response import HTTPResponse

from app.api_sdk_adaptors.base import ISO_FACE_ESTIMATION_TARGETS
from app.api_sdk_adaptors.iso_adaptor import APISDKCheckISOAdaptor
from app.api_sdk_adaptors.orientation import handleImageOrientation
from app.handlers.base_handler import BaseHandlerWithMultipart
from app.handlers.custom_query_getters import multifacePolicyGetter
from classes.image_meta import InputImageData
from classes.multipart_processing import ISOMultipartProcessor
from classes.schemas.iso import ISOInputEstimationsModel
from crutches_on_wheels.errors.errors import Error, ErrorInfo
from crutches_on_wheels.errors.exception import VLException
from crutches_on_wheels.monitoring.points import monitorTime
from crutches_on_wheels.web.query_getters import boolFrom01Getter, int01Getter
from sdk.sdk_loop.enums import LoopEstimations, MultifacePolicy
from sdk.sdk_loop.models.image import ImageType
from sdk.sdk_loop.task import HandlersTask
from sdk.sdk_loop.tasks.task import TaskParams


class ISOHandler(BaseHandlerWithMultipart):
    """
    ISO estimator handler

    Resource: "/{api_version}/iso"
    """

    async def getDataFromMultipart(self, imageType: ImageType = ImageType.IMAGE) -> List[InputImageData]:
        """Description see :func:`~BaseHandlerWithMultipart.getDataFromMultipart`."""
        dataFromRequest = await ISOMultipartProcessor().getData(self.request)
        estimationDataFromMultiPart = self._getDataFromMultipart(dataFromRequest.images, imageType)
        return estimationDataFromMultiPart

    def checkIsoLicensing(self):
        """
        Check that an ISO license is available.

        Raises:
            VLException(Error.LicenseProblem, 403): if some checks failed
        """
        if not self.app.ctx.licenseChecker.checkExpirationTime():
            raise VLException(Error.LicenseProblem.format("License expired"), 403, isCriticalError=False)

        if (
            not self.app.ctx.licenseChecker.licenseState.iso
            or not self.app.ctx.licenseChecker.licenseState.iso.isAvailable
        ):
            raise VLException(
                Error.LicenseProblem.format("ISO license feature is disabled."),
                statusCode=403,
                isCriticalError=False,
            )

    async def post(self) -> HTTPResponse:
        """
        ISO estimations handler. See spec `check iso`_.

        .. _`check iso`:
            _static/api.html#operation/checkISO

        Returns:
            Response with estimations
        """
        self.checkIsoLicensing()

        targets = ISO_FACE_ESTIMATION_TARGETS.copy()
        if self.getQueryParam("extract_exif", boolFrom01Getter, default=False):
            targets.add(LoopEstimations.exif)

        multifacePolicy = self.getQueryParam(
            "multiface_policy", multifacePolicyGetter, default=MultifacePolicy.notAllowed
        )
        params = TaskParams(
            targets=targets,
            multifacePolicy=multifacePolicy,
            useExifInfo=False,
            autoRotation=self.config.useAutoRotation,
        )

        imageType = self.getQueryParam("image_type", lambda x: ImageType(int01Getter(x)), default=ImageType.IMAGE)
        with monitorTime(self.request.dataForMonitoring, "load_images_for_processing_time"):
            inputData = await self.getInputEstimationData(
                self.request, imageType=imageType, validationModel=ISOInputEstimationsModel
            )

        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)

        isoAdaptor = APISDKCheckISOAdaptor(estimationTargets=targets)
        result, monitoringData = await isoAdaptor.buildResult(
            task.result, meta=[metaImage.meta for metaImage in inputData]
        )
        self.handleMonitoringData(monitoringData)

        if self.getResponseContentType() == "application/msgpack":
            body = msgpack.packb(result, use_bin_type=True)
            return self.success(200, body=body, contentType="application/msgpack")
        return self.success(200, outputJson=result)