""" 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.schemas.extractor import Extractor
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.utils.functions import currentDateTime
from crutches_on_wheels.web.query_getters import float01Getter, int01Getter, uuidGetter
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
[docs]class ExtractorHandler(BaseHandler):
"""
Extract attributes such as gender, age, ethnicity, descriptor from samples.
Resource: "/{api_version}/extractor"
"""
[docs] 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.getQueryParam("account_id", uuidGetter, require=True)
targets = set()
if self.getQueryParam("extract_descriptor", int01Getter, default=1):
targets.add(LoopEstimations.faceDescriptor)
if self.getQueryParam("extract_basic_attributes", int01Getter, default=0):
targets.add(LoopEstimations.basicAttributes)
if not targets:
raise VLException(Error.NotSelectedAttributesForExtract, 400, isCriticalError=False)
faceFilters = FaceDetectionFilters(
gcThreshold=self.getQueryParam("score_threshold", float01Getter)
if LoopEstimations.faceDescriptor in targets
else None,
)
params = TaskParams(
targets=targets,
filters=Filters(faceDetection=faceFilters),
estimatorsParams=TaskEstimationParams(
faceDescriptorVersion=self.config.defaultFaceDescriptorVersion,
),
autoRotation=self.config.useAutoRotation,
multifacePolicy=MultifacePolicy.notAllowed,
aggregate=self.getQueryParam("aggregate_attributes", int01Getter, default=0),
)
ttl = self.getQueryParam("ttl", lambda ttl: max(min(int(ttl), 86400), 1), default=300)
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(
estimationTargets=targets,
accountId=self.accountId,
facesClient=self.luna3Client.lunaFaces,
ttl=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)
