Source code for luna_handlers.app.handlers.handler_events_handler

""" Receive events on handler """

from datetime import datetime
from typing import List, Union, Tuple, Type
from sanic.response import HTTPResponse

from app.global_vars.enums import HandlerTarget, ImageType
from app.handlers.mixins import HandlerCacheMixin
from vlutils.helpers import convertTimeToString
from vlutils.jobs.async_runner import AsyncRunner
from app.handlers.base_handler import BaseHandlerWithMultipart
from app.handlers.custom_query_getters import (
    locationGetter,
    tagsValidator,
    sourceValidator,
    externalIdValidator,
    userDataValidator,
)
from classes.event import Event, Location, GeoPosition, eventAsDict
from classes.multipart_processing import HandlersMultipartProcessor
from classes.schemas.base_schema import BaseSchema
from classes.schemas.handler import CachedHandlerModel, HandlerInputEstimationsModel
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,
    timeFilterGetter,
    uuidGetter,
    floatInRangeGetterFactory,
    trackIdGetter,
)
from sdk.sdk_loop.sdk_task import SDKDetectableImage, FaceWarp

# longitude range
LONGITUDE_BOUNDS = (-180, 180)
# latitude range
LATITUDE_BOUNDS = (-90, 90)


class HandlerEventsHandler(BaseHandlerWithMultipart, HandlerCacheMixin):
    """
    Receiver of events.

    Resource: "/{api_version}/handlers/{handlerId}/events"
    """

    asyncRunner: AsyncRunner = None

    async def getDataFromMultipart(
        self, imageType: ImageType = ImageType.rawImage
    ) -> Tuple[Union[List[SDKDetectableImage], List[FaceWarp]], Union[dict, None]]:
        """
        Get data from multipart request
        Args:
            imageType: image type
        Returns:
            list of Images or list warps and optionally dict with policies (for multipart request with policies)
        """
        dataFromRequest = await HandlersMultipartProcessor().getData(self.request)
        estimationDataFromMultiPart = self._getDataFromMultipart(
            dataFromRequest.images, imageType, allowRawDescriptors=True
        )
        return estimationDataFromMultiPart, dataFromRequest.policies

    async def getInputEstimationData(
        self,
        request,
        validationModel: Type[BaseSchema],
        imageType: ImageType = ImageType.rawImage,
        autoOrient: bool = True,
        allowRawDescriptors: bool = True,
    ) -> Tuple[Union[List[SDKDetectableImage], List[FaceWarp]], Union[dict, None]]:
        """
        Get images from request body to detect faces and, optionally, policies for a dynamic handler.
        Args:
            request: request
            imageType: imageType
            validationModel: validation model
            autoOrient: whether to use orientation from exif or not
            allowRawDescriptors: whether raw descriptor mimetypes allowed or not
        Returns:
            list of Images or list warps and, optionally, policies for a dynamic handler.
        """
        if self.request.content_type.startswith("multipart/form-data"):
            images, policies = await self.getDataFromMultipart(imageType)
        else:
            images, policies = (
                await self.getDataFromRequest(
                    request, validationModel, imageType, allowRawDescriptors=allowRawDescriptors
                ),
                None,
            )

        if autoOrient:
            for image in images:
                if image.error:
                    continue

                if isinstance(image, SDKDetectableImage):
                    image.exifTranspose()

        return images, policies

    async def loadHandlerAndInputImages(
        self, handlerId
    ) -> tuple[CachedHandlerModel, Union[List[SDKDetectableImage], List[FaceWarp]], datetime]:
        """
        Prepare handler and input images for processing.

        If handler is `static`, policies are loaded from the database.
        If handler is `dynamic`, policies are required to be in the input request (multipart).

        Args:
            handlerId: handler id

        Returns:
            tuple with handler object, input images, version of cached handler or None
        """
        noCache = self.getQueryParam("no_cache", boolFrom01Getter)
        imageType = self.getQueryParam("image_type", lambda x: ImageType(int(x)), default=ImageType.rawImage)
        autoOrient = self.getQueryParam("use_exif_info", boolFrom01Getter, default=True)

        handlerJson, handlerVersion = await self.getHandlerWrapper(handlerId, self.accountId, noCache)

        if not handlerJson[HandlerTarget.isDynamic.value]:
            with monitorTime(self.request.dataForMonitoring, "load_images_for_processing_time"):
                inputData, policies = await self.getInputEstimationData(
                    self.request,
                    imageType=imageType,
                    validationModel=HandlerInputEstimationsModel,
                    autoOrient=autoOrient,
                )
            if policies is not None:
                raise VLException(
                    Error.BadMultipartInput.format("Not available to apply new policies to a non-dynamic handler"),
                    403,
                    False,
                )
        else:
            if "multipart/form-data" not in self.request.headers.get("Content-Type", ""):
                raise VLException(Error.BadMultipartInput.format("Required multipart in request"), 400, False)

            with monitorTime(self.request.dataForMonitoring, "load_images_for_processing_time"):
                inputData, policies = await self.getInputEstimationData(
                    self.request,
                    imageType=imageType,
                    validationModel=HandlerInputEstimationsModel,
                    autoOrient=autoOrient,
                )

            if policies is None:
                raise VLException(Error.InvalidInputJson.format("Required policies for dynamic handler."), 400, False)
            handlerJson[HandlerTarget.policies.value] = policies

        handler = self.loadDataFromJson(handlerJson, CachedHandlerModel)
        await handler.policies.checkListsAvailability(self.luna3Client, accountId=handler.accountId)
        self.checkPolicyLicensing(handler.policies)

        if not self.config.additionalServicesUsage.lunaEvents and handler.policies.storagePolicy.eventPolicy.storeEvent:
            raise VLException(Error.LunaEventsIsDisabled, 403, isCriticalError=False)

        return handler, inputData, handlerVersion

    def loadLocation(self) -> Location:
        """
        Load location data from the query parameters.

        Returns:
            location
        """
        latitude = self.getQueryParam("latitude", validator=floatInRangeGetterFactory(*LATITUDE_BOUNDS))
        longitude = self.getQueryParam("longitude", validator=floatInRangeGetterFactory(*LONGITUDE_BOUNDS))

        try:
            geoPosition = GeoPosition(latitude=latitude, longitude=longitude)
        except ValueError:
            raise VLException(Error.BadQueryParams.format("Geo position params is not properly specified"), 400, False)
        location = Location(
            street=self.getQueryParam("street", validator=locationGetter),
            houseNumber=self.getQueryParam("house_number", validator=locationGetter),
            district=self.getQueryParam("district", validator=locationGetter),
            area=self.getQueryParam("area", validator=locationGetter),
            city=self.getQueryParam("city", validator=locationGetter),
            geoPosition=geoPosition,
        )
        return location

    def parseTimeFromHeader(self, name: str, defaultTime: datetime) -> datetime:
        """
        Parse datetime from header by name.

        Args:
            name: header name
            defaultTime: default time

        Returns:
            user datetime if correct header or default datetime
        """
        timeFromHeader = self.request.headers.get(name)
        if timeFromHeader is not None:
            try:
                return timeFilterGetter(timeFromHeader)
            except ValueError:
                self.logger.warning(f"Incorrect '{name}' header: '{timeFromHeader}', " f"replaced with '{defaultTime}'")
        return defaultTime

    def getEventTimes(self) -> tuple[str, str]:
        """
        Get event times from header.

        Supported headers:

            Luna-Event-Time - event creation time
            Luna-Event-End-Time - event end time

        Returns:
            tuple with create and end event times
        """
        defaultEventTime = datetime.utcnow()
        inUTC = self.config.storageTime == "UTC"

        createEventTime = convertTimeToString(
            self.parseTimeFromHeader(name="Luna-Event-Time", defaultTime=defaultEventTime), inUTC=inUTC
        )
        endEventTime = convertTimeToString(
            self.parseTimeFromHeader(name="Luna-Event-End-Time", defaultTime=defaultEventTime), inUTC=inUTC
        )

        return createEventTime, endEventTime

    async def post(self, handlerId: str) -> HTTPResponse:
        """
        Receive events handler. See `spec receive_events`_.

        .. _`spec receive_events`:
            _static/api.html#operation/receiveEvents

        Args:
            handlerId: handler id

        Returns:
            Response with events and detections

        Raises:
            VLException(Error.BadMultipartInput, 400, isCriticalError=False): if failed to read multipart
        """
        self.addDataForMonitoring(tags={"handler_id": handlerId})

        source = self.getQueryParam("source", sourceValidator)
        externalId = self.getQueryParam("external_id", externalIdValidator, default="")
        userData = self.getQueryParam("user_data", userDataValidator, default="")
        tags = self.getQueryParam("tags", tagsValidator)
        self.accountId = self.getQueryParam("account_id", uuidGetter, require=True)
        aggregate = self.getQueryParam("aggregate_attributes", int01Getter)
        trackId = self.getQueryParam("track_id", trackIdGetter)

        createEventTime, endEventTime = self.getEventTimes()
        location = self.loadLocation()

        handler, inputData, handlerVersion = await self.loadHandlerAndInputImages(handlerId=handlerId)

        events: List[Event]
        imagesReply, events, filteredDetections, dataForMonitoring = await handler.policies.execute(
            accountId=self.accountId,
            inputData=inputData,
            luna3Client=self.luna3Client,
            aggregate=aggregate,
            userData=userData,
            externalId=externalId,
            sdkLoop=self.sdkLoop,
            source=source,
            userDefinedTags=tags,
            location=location,
            logger=self.logger,
            createEventTime=createEventTime,
            endEventTime=endEventTime,
            handlerId=handlerId,
            plugins=self.request.app.ctx.pluginManager,
            redisContext=self.redisContext,
            facesBucket=self.config.faceSamplesStorage.bucket,
            bodiesBucket=self.config.bodySamplesStorage.bucket,
            originBucket=self.config.imageOriginStorage.bucket,
            lunaEventsUsage=self.config.additionalServicesUsage.lunaEvents,
            lunaSenderUsage=self.config.additionalServicesUsage.lunaSender,
            trackId=trackId,
        )

        if handler.policies.detectPolicy.estimateLiveness.estimate:
            self.countLivenessEstimationsPerformed(sum(map(lambda x: x["status"], imagesReply)))

        self.handleMonitoringData(dataForMonitoring)

        # Provide info about caching to clients
        extraHeaders = {"Cache-Handler-Version": handlerVersion} if handlerVersion else None

        return self.success(
            201,
            outputJson={
                "images": imagesReply,
                "events": [eventAsDict(event) for event in events],
                "filtered_detections": filteredDetections,
            },
            extraHeaders=extraHeaders,
        )