Source code for luna_handlers.app.handlers.extractor_handler

""" Extractor handler """
from sanic.response import HTTPResponse

from app.api_sdk_adaptors.extractor import APISDKExtractorAdaptor
from app.api_sdk_adaptors.orientation import handleImageOrientation
from app.handlers.base_handler import BaseHandler
from classes.query_schemas.extractor import ExtractorQueries
from classes.schemas.extractor import Extractor
from crutches_on_wheels.cow.errors.errors import Error, ErrorInfo
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.monitoring.points import monitorTime
from crutches_on_wheels.cow.utils.functions import currentDateTime
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.filters import FaceDetectionFilters, Filters
from sdk.sdk_loop.tasks.task import TaskEstimationParams, TaskParams


class ExtractorHandler(BaseHandler):
    """
    Extract attributes such as gender, age, ethnicity, descriptor from samples.

    Resource: "/{api_version}/extractor"
    """

    async def post(self) -> HTTPResponse:
        """
        Extract attributes from samples. See `spec_extractor`_.

        .. _spec_extractor:
            _static/api.html#operation/extractAttributes

        Returns:
            response with list of extracted attributes
        Raises:
            VLException(Error.NotSelectedAttributesForExtract, 400, isCriticalError=True): if extract_descriptor=0 and
                extract_basic_attributes=0
        """
        self.accountId = self.getAccountIdFromHeader()
        queries: ExtractorQueries = self.loadDataFromQuery(ExtractorQueries)

        if not queries.estimationTargets.sdkTargets:
            raise VLException(Error.NotSelectedAttributesForExtract, 400, isCriticalError=False)

        faceFilters = FaceDetectionFilters(
            gcThreshold=queries.scoreThreshold
            if LoopEstimations.faceDescriptor in queries.estimationTargets.sdkTargets
            else None,
        )
        params = TaskParams(
            targets=queries.estimationTargets.sdkTargets,
            filters=Filters(faceDetection=faceFilters),
            estimatorsParams=TaskEstimationParams(faceDescriptorVersion=self.config.defaultFaceDescriptorVersion),
            autoRotation=self.config.useAutoRotation,
            multifacePolicy=MultifacePolicy.notAllowed,
            aggregate=bool(queries.aggregateAttributes),
        )

        sampleIds = self.request.json
        self.loadDataFromJson({"samples": sampleIds}, Extractor)

        with monitorTime(self.request.dataForMonitoring, "load_face_samples_time"):
            inputData = await self._getImagesFromSamples(
                inputJson={"samples": sampleIds},
                imageType=ImageType.FACE_WARP,
                defaultDetectTime=currentDateTime(self.config.storageTime),
            )

        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)

        extractorAdaptor = APISDKExtractorAdaptor(
            estimationResponseTargets=queries.estimationTargets.responseTargets,
            accountId=self.accountId,
            facesClient=self.luna3Client.lunaFaces,
            ttl=queries.ttl,
        )
        result, monitoringData = await extractorAdaptor.buildResult(
            task=task.result, meta=[metaImage.meta for metaImage in inputData]
        )
        self.handleMonitoringData(monitoringData)
        return self.success(201, outputJson=result)