"""S3 Image methods.""" # Based on: https://gist.github.com/ghandic/a48f450f3c011f44d42eea16a0c7014d from io import BytesIO import logging import os import boto3 from PIL import Image class S3ImageInvalidExtension(Exception): """Error for invalid extension.""" pass class S3ImageUploadFailed(Exception): """Error for failed upload.""" pass def transform_image(image_name, image_object): """Transform a Pillow image to match ripper spec.""" # if the image is not square if image_object.width != image_object.height: # Check how much longer the long side is long_edge = max(image_object.width, image_object.height) print('Correcting non-square image to {}x{}px.'.format( long_edge, long_edge)) # Reshape image_object = image_object.resize((long_edge, long_edge)) # Convert Mode if image_object.mode != 'RGB': print('Converting {} to RGB.'.format(image_name)) image_object = image_object.convert('RGB') # Upscale dimensions if image_object.width < 3000: print('Upscaling image {} to 3000x3000px.'.format(image_name)) image_object = image_object.resize((3000, 3000)) return image_object def get_image_from_s3(bucket, key): """Get an image from an S3 source as a Pillow object.""" # Get S3 object s3_client = boto3.client('s3') file_byte_string = s3_client.get_object(Bucket=bucket, Key=key)[ 'Body'].read() print('Converting object to Image') return Image.open(BytesIO(file_byte_string)) def save_image_to_s3(img, bucket, key, dpi=None): """Save a Pillow image object to an S3 target from memory.""" if not dpi: dpi = (72, 72) print('Using dpi: {}'.format(dpi)) file_name = os.path.split(key)[1] # Transform image print('Transforming {}'.format(file_name)) img = transform_image(file_name, img) buffer = BytesIO() ext = __get_safe_ext(file_name) print('Extension is: {}'.format(ext)) try: print('Saving: {} as {}'.format(file_name, key)) img.save(buffer, ext, dpi=dpi) except Exception as e: logging.exception('Save failed!') print('Error - {} ({}):, {}'.format(type(e), e.args, str(e))) buffer.seek(0) # Rewind buffer for saving to S3 print('Puting {} to {}'.format(key, bucket)) s3_client = boto3.client('s3') sent_data = s3_client.put_object( Body=buffer, Key=key, Bucket=bucket, ContentType='image/tiff') if sent_data['ResponseMetadata']['HTTPStatusCode'] != 200: raise S3ImageUploadFailed( 'Failed to upload image {} to bucket {}'.format(key, bucket)) def __get_safe_ext(key): ext = os.path.splitext(key)[-1].strip('.').upper() if ext in ['JPG', 'JPEG']: return 'JPEG' elif ext in ['PNG']: return 'PNG' elif ext in ['TIF', 'TIFF']: return 'TIFF' else: raise S3ImageInvalidExtension('Extension is invalid')