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