Spaces:
Sleeping
Sleeping
Update main.py
Browse files
main.py
CHANGED
|
@@ -7,7 +7,7 @@ import torch
|
|
| 7 |
from dataclasses import dataclass
|
| 8 |
from typing import Optional
|
| 9 |
from scipy.ndimage import median_filter
|
| 10 |
-
from scipy.optimize import curve_fit
|
| 11 |
from ultralytics import YOLO
|
| 12 |
from fastapi import FastAPI, UploadFile, File
|
| 13 |
from fastapi.middleware.cors import CORSMiddleware
|
|
@@ -25,7 +25,7 @@ MAX_IMAGE_SIZE = 2048
|
|
| 25 |
CHECKER_WIDTH_CM = 6.3
|
| 26 |
|
| 27 |
# ==============================================================================
|
| 28 |
-
# --- COLOR CALIBRATION
|
| 29 |
# ==============================================================================
|
| 30 |
def to_linear_srgb(u8_bgr):
|
| 31 |
rgb = cv2.cvtColor(u8_bgr, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
|
|
@@ -39,10 +39,18 @@ def to_srgb_u8(lin_rgb):
|
|
| 39 |
|
| 40 |
def detect_checker_corners(img_bgr):
|
| 41 |
det = cv2.mcc.CCheckerDetector_create()
|
| 42 |
-
if
|
| 43 |
-
|
| 44 |
-
|
| 45 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 46 |
|
| 47 |
def warp_checker(img, corners, out_w=600, out_h=400):
|
| 48 |
dst = np.float32([[0, out_h-1],[0, 0],[out_w-1, 0],[out_w-1, out_h-1]])
|
|
@@ -67,28 +75,31 @@ def compute_deltaE_00(lin_src, lin_ref):
|
|
| 67 |
return color.deltaE_ciede2000(color.rgb2lab(srgb_src.reshape(1, -1, 3)), color.rgb2lab(srgb_ref.reshape(1, -1, 3))).flatten()
|
| 68 |
|
| 69 |
def apply_color_pipeline(target_bgr, ref24, tgt24):
|
| 70 |
-
"""
|
| 71 |
-
|
| 72 |
-
|
| 73 |
-
|
| 74 |
-
|
| 75 |
-
|
| 76 |
-
|
| 77 |
-
|
| 78 |
-
|
| 79 |
-
|
| 80 |
-
|
| 81 |
-
|
| 82 |
-
|
| 83 |
-
|
| 84 |
-
|
| 85 |
-
|
| 86 |
-
|
| 87 |
-
|
| 88 |
-
|
| 89 |
-
|
| 90 |
-
|
| 91 |
|
|
|
|
|
|
|
|
|
|
| 92 |
return to_srgb_u8(np.clip(corrected_lin, 0, 1))
|
| 93 |
|
| 94 |
# ==============================================================================
|
|
@@ -199,7 +210,7 @@ class WatermelonProcessor:
|
|
| 199 |
|
| 200 |
ys_extrap = np.linspace(0, h, 500)
|
| 201 |
xs_extrap = parabola((ys_extrap - y_mean)/y_span, *popt_mid)
|
| 202 |
-
pts_rot = np.vstack([xs_extrap, ys_extrap, np.ones_like(
|
| 203 |
else:
|
| 204 |
ys_extrap = np.linspace(0, h, 500)
|
| 205 |
pts_rot = np.vstack([np.full_like(ys_extrap, cx), ys_extrap, np.ones_like(ys_extrap)])
|
|
@@ -232,8 +243,7 @@ class WatermelonProcessor:
|
|
| 232 |
print(f"Calibration skipped for {source_name}: {e}")
|
| 233 |
|
| 234 |
results = self.model(image, conf=0.25, retina_masks=True, verbose=False)
|
| 235 |
-
rind_mask = np.zeros((h, w), dtype=np.uint8)
|
| 236 |
-
flesh_contours = []
|
| 237 |
|
| 238 |
if results[0].masks is None:
|
| 239 |
return ProcessResult(success=False, message="No masks detected.")
|
|
@@ -241,18 +251,14 @@ class WatermelonProcessor:
|
|
| 241 |
for mask_data, cls in zip(results[0].masks.xy, results[0].boxes.cls):
|
| 242 |
contour = np.array(mask_data, dtype=np.int32)
|
| 243 |
c_id = int(cls)
|
| 244 |
-
if c_id == 0:
|
| 245 |
-
|
| 246 |
-
elif c_id == 1:
|
| 247 |
-
flesh_contours.append(contour)
|
| 248 |
-
elif c_id == 2:
|
| 249 |
-
flesh_contours.append(contour)
|
| 250 |
|
| 251 |
flesh_contours.sort(key=lambda cnt: cv2.moments(cnt)['m10'] / (cv2.moments(cnt)['m00'] + 1e-5))
|
| 252 |
flesh_l_m, flesh_r_m = np.zeros((h, w), dtype=np.uint8), np.zeros((h, w), dtype=np.uint8)
|
| 253 |
|
| 254 |
if len(flesh_contours) >= 2:
|
| 255 |
-
cv2.drawContours(flesh_l_m,
|
| 256 |
cv2.drawContours(flesh_r_m,[flesh_contours[1]], -1, 255, -1)
|
| 257 |
elif len(flesh_contours) == 1:
|
| 258 |
cv2.drawContours(flesh_l_m,[flesh_contours[0]], -1, 255, -1)
|
|
@@ -290,22 +296,7 @@ class WatermelonProcessor:
|
|
| 290 |
perimeter_val = float(perimeter_px * cm_per_px * orig_scale)
|
| 291 |
|
| 292 |
midline = self.get_dual_mask_midline(flesh_l_m, flesh_r_m, rind_cnt, fit_pts, cx, cy)
|
| 293 |
-
|
| 294 |
-
output = image.copy().astype(np.float32)
|
| 295 |
-
alpha = 0.42
|
| 296 |
-
output[..., 0] = np.where(rind_mask > 0, output[..., 0] * (1 - alpha) + 0.0 * alpha, output[..., 0])
|
| 297 |
-
output[..., 1] = np.where(rind_mask > 0, output[..., 1] * (1 - alpha) + 170.0 * alpha, output[..., 1])
|
| 298 |
-
output[..., 2] = np.where(rind_mask > 0, output[..., 2] * (1 - alpha) + 0.0 * alpha, output[..., 2])
|
| 299 |
-
|
| 300 |
-
output[..., 0] = np.where(flesh_l_m > 0, output[..., 0] * (1 - alpha) + 255.0 * alpha, output[..., 0])
|
| 301 |
-
output[..., 1] = np.where(flesh_l_m > 0, output[..., 1] * (1 - alpha) + 0.0 * alpha, output[..., 1])
|
| 302 |
-
output[..., 2] = np.where(flesh_l_m > 0, output[..., 2] * (1 - alpha) + 0.0 * alpha, output[..., 2])
|
| 303 |
-
|
| 304 |
-
output[..., 0] = np.where(flesh_r_m > 0, output[..., 0] * (1 - alpha) + 0.0 * alpha, output[..., 0])
|
| 305 |
-
output[..., 1] = np.where(flesh_r_m > 0, output[..., 1] * (1 - alpha) + 0.0 * alpha, output[..., 1])
|
| 306 |
-
output[..., 2] = np.where(flesh_r_m > 0, output[..., 2] * (1 - alpha) + 255.0 * alpha, output[..., 2])
|
| 307 |
-
|
| 308 |
-
output = np.clip(output, 0, 255).astype(np.uint8)
|
| 309 |
|
| 310 |
if checker_corners is not None:
|
| 311 |
cv2.polylines(output, [np.int32(checker_corners)], True, (0, 165, 255), 4)
|
|
@@ -323,13 +314,14 @@ class WatermelonProcessor:
|
|
| 323 |
cv2.circle(output, p1, 6, (255, 0, 255), -1)
|
| 324 |
cv2.line(output, p1, p2, (255, 0, 255), 2)
|
| 325 |
|
| 326 |
-
_, buffer = cv2.imencode('.jpg', output,
|
|
|
|
| 327 |
|
| 328 |
return ProcessResult(
|
| 329 |
success=True, message="Success", r2_score=float(r2),
|
| 330 |
width_val=width_val, height_val=height_val, perimeter_val=perimeter_val,
|
| 331 |
delta_e_initial=dE_initial, delta_e_final=dE_final,
|
| 332 |
-
image_base64=
|
| 333 |
)
|
| 334 |
|
| 335 |
|
|
|
|
| 7 |
from dataclasses import dataclass
|
| 8 |
from typing import Optional
|
| 9 |
from scipy.ndimage import median_filter
|
| 10 |
+
from scipy.optimize import curve_fit, minimize
|
| 11 |
from ultralytics import YOLO
|
| 12 |
from fastapi import FastAPI, UploadFile, File
|
| 13 |
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
| 25 |
CHECKER_WIDTH_CM = 6.3
|
| 26 |
|
| 27 |
# ==============================================================================
|
| 28 |
+
# --- DIRECT DELTA-E COLOR CALIBRATION ---
|
| 29 |
# ==============================================================================
|
| 30 |
def to_linear_srgb(u8_bgr):
|
| 31 |
rgb = cv2.cvtColor(u8_bgr, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
|
|
|
|
| 39 |
|
| 40 |
def detect_checker_corners(img_bgr):
|
| 41 |
det = cv2.mcc.CCheckerDetector_create()
|
| 42 |
+
if det.process(img_bgr, cv2.mcc.MCC24):
|
| 43 |
+
cc = det.getListColorChecker()[0]
|
| 44 |
+
return np.array(cc.getBox() if hasattr(cc, "getBox") else cc.getCorners(), dtype=np.float32)
|
| 45 |
+
|
| 46 |
+
# Fallback: Sometimes OpenCV fails on high-res noise. Scale by 50% and try again.
|
| 47 |
+
img_small = cv2.resize(img_bgr, (0,0), fx=0.5, fy=0.5)
|
| 48 |
+
if det.process(img_small, cv2.mcc.MCC24):
|
| 49 |
+
cc = det.getListColorChecker()[0]
|
| 50 |
+
pts = np.array(cc.getBox() if hasattr(cc, "getBox") else cc.getCorners(), dtype=np.float32)
|
| 51 |
+
return pts * 2.0 # Scale corners back up
|
| 52 |
+
|
| 53 |
+
raise RuntimeError("ColorChecker not found")
|
| 54 |
|
| 55 |
def warp_checker(img, corners, out_w=600, out_h=400):
|
| 56 |
dst = np.float32([[0, out_h-1],[0, 0],[out_w-1, 0],[out_w-1, out_h-1]])
|
|
|
|
| 75 |
return color.deltaE_ciede2000(color.rgb2lab(srgb_src.reshape(1, -1, 3)), color.rgb2lab(srgb_ref.reshape(1, -1, 3))).flatten()
|
| 76 |
|
| 77 |
def apply_color_pipeline(target_bgr, ref24, tgt24):
|
| 78 |
+
"""Safely scales White Balance, then uses Powell's method to minimize Delta E."""
|
| 79 |
+
tgt_lin = to_linear_srgb(target_bgr)
|
| 80 |
+
|
| 81 |
+
# 1. White Balance (Von Kries scaling using 6 neutrals)
|
| 82 |
+
gains = np.median(ref24[18:24], axis=0) / np.maximum(np.median(tgt24[18:24], axis=0), 1e-6)
|
| 83 |
+
tgt_lin_wb = tgt_lin * gains.reshape(1, 1, 3)
|
| 84 |
+
tgt24_wb = tgt24 * gains
|
| 85 |
+
|
| 86 |
+
# 2. Objective Function: Explicitly optimize the matrix for the lowest Delta E score
|
| 87 |
+
def objective(W_flat):
|
| 88 |
+
W = W_flat.reshape(3, 3)
|
| 89 |
+
pred_lin = np.clip(tgt24_wb @ W, 0, 1)
|
| 90 |
+
return np.mean(compute_deltaE_00(pred_lin, ref24))
|
| 91 |
+
|
| 92 |
+
# Get a fast starting point using standard Ridge Regression
|
| 93 |
+
X, Y = tgt24_wb, ref24
|
| 94 |
+
W_init = np.linalg.inv(X.T @ X + 0.05 * np.eye(3)) @ X.T @ Y
|
| 95 |
+
|
| 96 |
+
# Optimize matrix to minimize CIEDE2000 natively
|
| 97 |
+
res = minimize(objective, W_init.flatten(), method='Powell')
|
| 98 |
+
W_opt = res.x.reshape(3, 3)
|
| 99 |
|
| 100 |
+
# 3. Apply to full image securely
|
| 101 |
+
corrected_lin = (tgt_lin_wb.reshape(-1, 3) @ W_opt).reshape(tgt_lin_wb.shape)
|
| 102 |
+
|
| 103 |
return to_srgb_u8(np.clip(corrected_lin, 0, 1))
|
| 104 |
|
| 105 |
# ==============================================================================
|
|
|
|
| 210 |
|
| 211 |
ys_extrap = np.linspace(0, h, 500)
|
| 212 |
xs_extrap = parabola((ys_extrap - y_mean)/y_span, *popt_mid)
|
| 213 |
+
pts_rot = np.vstack([xs_extrap, ys_extrap, np.ones_like(xs_extrap)])
|
| 214 |
else:
|
| 215 |
ys_extrap = np.linspace(0, h, 500)
|
| 216 |
pts_rot = np.vstack([np.full_like(ys_extrap, cx), ys_extrap, np.ones_like(ys_extrap)])
|
|
|
|
| 243 |
print(f"Calibration skipped for {source_name}: {e}")
|
| 244 |
|
| 245 |
results = self.model(image, conf=0.25, retina_masks=True, verbose=False)
|
| 246 |
+
rind_mask, flesh_contours = np.zeros((h, w), dtype=np.uint8), []
|
|
|
|
| 247 |
|
| 248 |
if results[0].masks is None:
|
| 249 |
return ProcessResult(success=False, message="No masks detected.")
|
|
|
|
| 251 |
for mask_data, cls in zip(results[0].masks.xy, results[0].boxes.cls):
|
| 252 |
contour = np.array(mask_data, dtype=np.int32)
|
| 253 |
c_id = int(cls)
|
| 254 |
+
if c_id == 0: cv2.drawContours(rind_mask, [contour], -1, 255, -1)
|
| 255 |
+
elif c_id == 1: flesh_contours.append(contour)
|
|
|
|
|
|
|
|
|
|
|
|
|
| 256 |
|
| 257 |
flesh_contours.sort(key=lambda cnt: cv2.moments(cnt)['m10'] / (cv2.moments(cnt)['m00'] + 1e-5))
|
| 258 |
flesh_l_m, flesh_r_m = np.zeros((h, w), dtype=np.uint8), np.zeros((h, w), dtype=np.uint8)
|
| 259 |
|
| 260 |
if len(flesh_contours) >= 2:
|
| 261 |
+
cv2.drawContours(flesh_l_m,[flesh_contours[0]], -1, 255, -1)
|
| 262 |
cv2.drawContours(flesh_r_m,[flesh_contours[1]], -1, 255, -1)
|
| 263 |
elif len(flesh_contours) == 1:
|
| 264 |
cv2.drawContours(flesh_l_m,[flesh_contours[0]], -1, 255, -1)
|
|
|
|
| 296 |
perimeter_val = float(perimeter_px * cm_per_px * orig_scale)
|
| 297 |
|
| 298 |
midline = self.get_dual_mask_midline(flesh_l_m, flesh_r_m, rind_cnt, fit_pts, cx, cy)
|
| 299 |
+
output = blend_mask_overlays(image, rind_mask, flesh_combined)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 300 |
|
| 301 |
if checker_corners is not None:
|
| 302 |
cv2.polylines(output, [np.int32(checker_corners)], True, (0, 165, 255), 4)
|
|
|
|
| 314 |
cv2.circle(output, p1, 6, (255, 0, 255), -1)
|
| 315 |
cv2.line(output, p1, p2, (255, 0, 255), 2)
|
| 316 |
|
| 317 |
+
_, buffer = cv2.imencode('.jpg', output,[cv2.IMWRITE_JPEG_QUALITY, 85])
|
| 318 |
+
img_base64 = base64.b64encode(buffer).decode('utf-8')
|
| 319 |
|
| 320 |
return ProcessResult(
|
| 321 |
success=True, message="Success", r2_score=float(r2),
|
| 322 |
width_val=width_val, height_val=height_val, perimeter_val=perimeter_val,
|
| 323 |
delta_e_initial=dE_initial, delta_e_final=dE_final,
|
| 324 |
+
image_base64=img_base64, filename=source_name
|
| 325 |
)
|
| 326 |
|
| 327 |
|