Source code for luna_handlers.app.handlers.iso_handler

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

import msgpack
from sanic.response import HTTPResponse
from werkzeug.http import parse_accept_header

from app.api_sdk_adaptors.iso_adaptor import APISDKCheckISOAdaptor
from app.global_vars.enums import ImageType
from app.handlers.base_handler import BaseHandlerWithMultipart
from classes.multipart_processing import ISOMultipartProcessor
from classes.schemas.iso import ISOInputEstimationsModel
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
from crutches_on_wheels.monitoring.points import monitorTime
from crutches_on_wheels.web.query_getters import int01Getter, boolFrom01Getter
from sdk.sdk_loop.enums import MultifacePolicy
from sdk.sdk_loop.estimation_targets import (
    SDKFaceEstimationTargets,
    SDKEstimationTargets,
)
from sdk.sdk_loop.sdk_task import SDKTask, SDKDetectableImage, FaceWarp


# constant face estimation targets for ISO handler
ISO_FACE_ESTIMATION_TARGETS = SDKFaceEstimationTargets(
    estimateQuality=1,
    estimateMouthAttributes=1,
    estimateGaze=1,
    estimateEyesAttributes=1,
    estimateHeadPose=1,
    estimateGlasses=1,
)


[docs]class ISOHandler(BaseHandlerWithMultipart): """ ISO estimator handler Resource: "/{api_version}/iso" """
[docs] async def getDataFromMultipart( self, imageType: ImageType = ImageType.rawImage ) -> Union[List[SDKDetectableImage], List[FaceWarp]]: """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 """ multifacePolicy = self.getQueryParam( "multiface_policy", lambda x: MultifacePolicy(int(x)), default=MultifacePolicy.notAllowed ) imageType = self.getQueryParam("image_type", lambda x: ImageType(int01Getter(x)), default=ImageType.rawImage) extractExif = self.getQueryParam("extract_exif", boolFrom01Getter, default=False) with monitorTime(self.request.dataForMonitoring, "load_images_for_processing_time"): inputData = await self.getInputEstimationData( self.request, imageType=imageType, validationModel=ISOInputEstimationsModel, autoOrient=False ) self.checkIsoLicensing() toEstimate = SDKEstimationTargets( estimateHuman=0, estimateFace=1, faceEstimationTargets=ISO_FACE_ESTIMATION_TARGETS, ) task = SDKTask(toEstimate, data=inputData, multifacePolicy=multifacePolicy) isoAdaptor = APISDKCheckISOAdaptor(accountId="", logger=self.logger, sdkLoop=self.sdkLoop) resTask, monitoringData = await isoAdaptor.estimate(task, extractExif=extractExif) self.handleMonitoringData(monitoringData) acceptHeader = self.request.headers.get("Accept", "application/json") responseContentType = parse_accept_header(acceptHeader).best_match( ("application/msgpack", "application/json"), default="application/json" ) if responseContentType == "application/msgpack": body = msgpack.packb(resTask, use_bin_type=True) return self.success(200, body=body, contentType="application/msgpack") return self.success(200, outputJson=resTask)