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