Source code for luna_api.app.handlers.samples_handler

""" Save samples to storage """
import io
from abc import abstractmethod
from typing import Tuple

from luna3.common.luna_response import LunaResponse
from PIL import Image
from sanic.response import HTTPResponse

from app.handlers.base_handler import (
    BaseBodySampleProxyHandler,
    BaseFaceSampleProxyHandler,
    BaseProxyHandler,
    ProxyRequest,
)
from app.version import VERSION
from configs.configs.configs.settings.classes import ImageStoreAddressSettings
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException

IMG_CONTENT_TYPES = (
    "image/jpeg",
    "image/png",
    "image/bmp",
    "image/x-portable-pixmap",
    "image/tiff",
)


[docs]def checkWarpedImage(byteImg: bytes, expectedSize: Tuple[int, int]) -> None: """ Check image is warp (checking size and type). Args: byteImg: images expectedSize: expected image size Raises: VLException(Error.BadContentType, 400, isCriticalError=False): if mime type is not 'JPEG' VLException(Error.BadWarpImageSize, 400, isCriticalError=False): if image size is not valid """ allowedFormats = ("JPEG", "PNG", "BMP", "TIFF", "PPM") try: img = Image.open(io.BytesIO(byteImg)) except OSError: raise VLException(Error.ConvertImageError.format("image"), 400, isCriticalError=False) if img.format not in allowedFormats: raise VLException(Error.BadContentType, 400, isCriticalError=False) if img.size != expectedSize: error = Error.BadWarpImage.format(f"Image size is not equal to {expectedSize[0]}x{expectedSize[1]}") raise VLException(error, 400, isCriticalError=False)
class _BaseSamplesProxyHandler(BaseProxyHandler): """ Base handler for saving samples in the storage. """ allowedMethods = {"POST"} serviceAddress: ImageStoreAddressSettings serviceTimeouts: ImageStoreAddressSettings @abstractmethod def checkWarpedImage(self, warp: bytes) -> None: """ Check, whether an input image is a warp. Args: warp: a image """ @abstractmethod def prepareLocationHeader(self, sampleId: str) -> str: """ Prepare sample Location header. Args: sampleId: sample id """ def prepareUrl(self) -> str: """ Prepare url to the LIS service. Returns: A str, url """ return f"{self.serviceUrl}/buckets/{self.serviceAddress.bucket}/images" async def prepareRequestPost(self): """ Build request object for the LIS service. Returns: ProxyRequest """ contentType = self.request.headers.get("Content-Type") if contentType not in IMG_CONTENT_TYPES: raise VLException(Error.BadContentType, 400, isCriticalError=False) warp = self.request.body self.checkWarpedImage(warp) return ProxyRequest(warp, self.prepareHeaders(), self.prepareQuery()) async def postProcessingPost(self, response: LunaResponse) -> HTTPResponse: """ Post process LIS response. Args: response: response Returns: response in api format """ sampleId = response.json["image_id"] location = self.prepareLocationHeader(sampleId=sampleId) self.respHeaders["Location"] = location return self.success(201, outputJson={"sample_id": sampleId, "url": location})
[docs]class FaceSamplesProxyHandler(_BaseSamplesProxyHandler, BaseFaceSampleProxyHandler): """ Face sample proxy handler. See `spec face samples <_static/api.html#operation/saveFaceSample>`_. Resource: "/{api_version}/samples/faces" """
[docs] def checkTokenPermissions(self) -> None: """ Description see :func:`~BaseRequestHandler.checkTokenPermissions`. """ self.checkTokenPermissionsDefault(objectName="faceSample")
[docs] def prepareLocationHeader(self, sampleId: str) -> str: """ Prepare relative url to the created image. Returns: A str, url """ return f"/{VERSION['Version']['api']}/samples/faces/{sampleId}"
[docs] def checkWarpedImage(self, warp: bytes) -> None: """ Check whether image is warped. Args: warp: warp to check Raises: VLException """ checkWarpedImage(warp, (250, 250))
[docs]class BodySamplesProxyHandler(_BaseSamplesProxyHandler, BaseBodySampleProxyHandler): """ Body sample proxy handler. See `spec body samples <_static/api.html#operation/saveBodySample>`_. Resource: "/{api_version}/samples/bodies" """
[docs] def checkTokenPermissions(self) -> None: """ Description see :func:`~BaseRequestHandler.checkTokenPermissions`. """ self.checkTokenPermissionsDefault(objectName="bodySample")
[docs] def prepareLocationHeader(self, sampleId: str) -> str: """ Prepare relative url to the created image. Returns: A str, url """ return f"/{VERSION['Version']['api']}/samples/bodies/{sampleId}"
[docs] def checkWarpedImage(self, warp: bytes) -> None: """ Check whether image is warped. Args: warp: warp to check Raises: VLException """ checkWarpedImage(warp, (128, 256))