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