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.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

# constant face estimation targets for ISO handler
ISO_FACE_ESTIMATION_TARGETS = {
    LoopEstimations.faceLandmarks68,
    LoopEstimations.faceWarpQuality,
    LoopEstimations.mouthAttributes,
    LoopEstimations.gaze,
    LoopEstimations.eyes,
    LoopEstimations.headPose,
    LoopEstimations.glasses,
    LoopEstimations.imageColorType,
    LoopEstimations.headwear,
    LoopEstimations.faceNaturalLight,
    LoopEstimations.fisheye,
    LoopEstimations.eyebrowExpression,
    LoopEstimations.redEyes,
    # faceDetectionBackground,
}


[docs]class ISOHandler(BaseHandlerWithMultipart): """ ISO estimator handler Resource: "/{api_version}/iso" """
[docs] 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
[docs] 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, )
[docs] 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)