Spaces:
Running
Running
File size: 1,943 Bytes
40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 40aaca9 0b7ba62 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
import os
import boto3
from botocore.client import Config
import uuid
from PIL import Image
import io
def upload_mask(image, prefix="mask"):
"""
Upload the segmentation mask image to DigitalOcean Spaces
Args:
image: PIL Image object
prefix: filename prefix
Returns:
URL of the uploaded file
"""
try:
# Get credentials from environment variables
do_key = os.environ.get('DO_SPACES_KEY')
do_secret = os.environ.get('DO_SPACES_SECRET')
do_region = os.environ.get('DO_SPACES_REGION')
do_bucket = os.environ.get('DO_SPACES_BUCKET')
# Validate that credentials exist
if not all([do_key, do_secret, do_region, do_bucket]):
raise ValueError("Missing DigitalOcean Spaces credentials")
# Create S3 client
session = boto3.session.Session()
client = session.client('s3',
region_name=do_region,
endpoint_url=f'https://{do_region}.digitaloceanspaces.com',
aws_access_key_id=do_key,
aws_secret_access_key=do_secret)
# Generate a unique filename
filename = f"{prefix}_{uuid.uuid4().hex}.png"
# Convert the image to a byte stream
img_byte_arr = io.BytesIO()
image.save(img_byte_arr, format='PNG')
img_byte_arr.seek(0)
# Upload to Spaces
client.upload_fileobj(
img_byte_arr,
do_bucket,
filename,
ExtraArgs={'ACL': 'public-read', 'ContentType': 'image/png'}
)
# Return the public URL
url = f'https://{do_bucket}.{do_region}.digitaloceanspaces.com/{filename}'
return url
except Exception as e:
print(f"Upload failed: {str(e)}")
return f"Upload error: {str(e)}"
|