File size: 5,929 Bytes
fcc071b |
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 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
from fastapi import FastAPI, File, Form, UploadFile
from fastapi.responses import JSONResponse
from google.cloud import storage
import os
import io
import numpy as np
import cv2
from PIL import Image
import uuid
import base64
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = "./da-kalbe-63ee33c9cdbb.json"
bucket_name = "da-kalbe-ml-result-png"
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
app = FastAPI()
# Function to upload file to Google Cloud Storage
def upload_to_gcs(image: Image, filename: str):
"""Uploads an image to Google Cloud Storage."""
try:
blob = bucket.blob(filename)
image_buffer = io.BytesIO()
image.save(image_buffer, format='PNG')
image_buffer.seek(0)
blob.upload_from_file(image_buffer, content_type='image/png')
except Exception as e:
return {'error': f"An unexpected error occurred: {e}"}
def upload_folder_images(image_path, enhanced_image_path):
# Extract the base name of the uploaded image without the extension
folder_name = os.path.splitext(os.path.basename(image_path))[0]
# Create the folder in Cloud Storage
bucket.blob(folder_name + '/').upload_from_string('', content_type='application/x-www-form-urlencoded')
# Open the images
original_image = Image.open(image_path)
enhanced_image = Image.open(enhanced_image_path)
# Upload images to GCS
upload_to_gcs(original_image, folder_name + '/' + 'original_image.png')
upload_to_gcs(enhanced_image, folder_name + '/' + enhancement_type + '.png')
def calculate_mse(original_image, enhanced_image):
mse = np.mean((original_image - enhanced_image) ** 2)
return mse
def calculate_psnr(original_image, enhanced_image):
mse = calculate_mse(original_image, enhanced_image)
if mse == 0:
return float('inf')
max_pixel_value = 255.0
psnr = 20 * np.log10(max_pixel_value / np.sqrt(mse))
return psnr
def calculate_maxerr(original_image, enhanced_image):
maxerr = np.max((original_image - enhanced_image) ** 2)
return maxerr
def calculate_l2rat(original_image, enhanced_image):
l2norm_ratio = np.sum(original_image ** 2) / np.sum((original_image - enhanced_image) ** 2)
return l2norm_ratio
def process_image(original_image, enhancement_type, fix_monochrome=True):
if fix_monochrome and original_image.shape[-1] == 3:
original_image = cv2.cvtColor(original_image, cv2.COLOR_BGR2GRAY)
image = original_image - np.min(original_image)
image = image / np.max(original_image)
image = (image * 255).astype(np.uint8)
enhanced_image = enhance_image(image, enhancement_type)
mse = calculate_mse(original_image, enhanced_image)
psnr = calculate_psnr(original_image, enhanced_image)
maxerr = calculate_maxerr(original_image, enhanced_image)
l2rat = calculate_l2rat(original_image, enhanced_image)
return enhanced_image, mse, psnr, maxerr, l2rat
def apply_clahe(image):
clahe = cv2.createCLAHE(clipLimit=40.0, tileGridSize=(8, 8))
return clahe.apply(image)
def invert(image):
return cv2.bitwise_not(image)
def hp_filter(image, kernel=None):
if kernel is None:
kernel = np.array([[-1, -1, -1], [-1, 9, -1], [-1, -1, -1]])
return cv2.filter2D(image, -1, kernel)
def unsharp_mask(image, radius=5, amount=2):
def usm(image, radius, amount):
blurred = cv2.GaussianBlur(image, (0, 0), radius)
sharpened = cv2.addWeighted(image, 1.0 + amount, blurred, -amount, 0)
return sharpened
return usm(image, radius, amount)
def hist_eq(image):
return cv2.equalizeHist(image)
def enhance_image(image, enhancement_type):
if enhancement_type == "Invert":
return invert(image)
elif enhancement_type == "High Pass Filter":
return hp_filter(image)
elif enhancement_type == "Unsharp Masking":
return unsharp_mask(image)
elif enhancement_type == "Histogram Equalization":
return hist_eq(image)
elif enhancement_type == "CLAHE":
return apply_clahe(image)
else:
raise ValueError(f"Unknown enhancement type: {enhancement_type}")
@app.post("/process_image")
async def process_image_api(image: UploadFile = File(...), enhancement_type: str = Form(...)):
"""Processes an uploaded image and returns the enhanced image and metrics."""
if not image:
return JSONResponse(status_code=400, content={'error': 'No image file provided'})
allowed_extensions = {'png', 'jpg', 'jpeg'}
if '.' not in image.filename or image.filename.split('.')[-1].lower() not in allowed_extensions:
return JSONResponse(status_code=400, content={'error': 'Invalid image file'})
try:
# Open the image using Pillow
image_pil = Image.open(image.file).convert('RGB')
# Convert to NumPy array
image_np = np.array(image_pil)
# Apply image processing
enhanced_image, mse, psnr, maxerr, l2rat = process_image(image_np, enhancement_type)
# Convert processed image back to PIL format for saving
enhanced_image_pil = Image.fromarray(enhanced_image)
# Save to in-memory buffer
image_buffer = io.BytesIO()
enhanced_image_pil.save(image_buffer, format='PNG')
image_buffer.seek(0)
# Encode to base64
image_base64 = base64.b64encode(image_buffer.getvalue()).decode('utf-8')
response = {
'message': 'Image processed successfully!',
'processed_image': image_base64,
'mse': float(mse),
'psnr': float(psnr),
'maxerr': float(maxerr),
'l2rat': float(l2rat)
}
upload_folder_images(image, enhanced_image_pil)
return JSONResponse(status_code=200, content=response)
except Exception as e:
return JSONResponse(status_code=500, content={'error': f'Error processing image: {str(e)}'})
|