Spaces:
Sleeping
Sleeping
| import os | |
| import cv2 | |
| import numpy as np | |
| import base64 | |
| import gc | |
| import torch | |
| import time | |
| import traceback | |
| from dataclasses import dataclass | |
| from typing import Optional | |
| from scipy.ndimage import median_filter | |
| from scipy.optimize import curve_fit, minimize | |
| from ultralytics import YOLO | |
| from fastapi import FastAPI, UploadFile, File, Query, Form | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.concurrency import run_in_threadpool | |
| import uvicorn | |
| from skimage import color | |
| import hashlib | |
| import asyncio | |
| # OOM PREVENTION | |
| torch.set_num_threads(1) | |
| # --- CONFIGURATION --- | |
| MODEL_PATH = "best.pt" | |
| MAX_IMAGE_SIZE = 2048 | |
| CHECKER_WIDTH_CM = 6.3 | |
| MIN_RIND_FLESH_OVERLAP_RATIO = 0.10 | |
| MIN_FLESH_PIXELS_FOR_FALLBACK = 100 | |
| MAX_RIND_TO_FLESH_AREA_RATIO = 3.5 | |
| MAX_RIND_CENTER_OFFSET_RATIO = 0.60 | |
| # ============================================================================== | |
| # --- COLOR CALIBRATION LOGIC --- | |
| # ============================================================================== | |
| def to_linear_srgb(u8_bgr): | |
| rgb = cv2.cvtColor(u8_bgr, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0 | |
| a = 0.055 | |
| return np.where(rgb <= 0.04045, rgb / 12.92, ((rgb + a) / (1 + a)) ** 2.4) | |
| def to_srgb_u8(lin_rgb): | |
| a = 0.055 | |
| srgb = np.where(lin_rgb <= 0.0031308, 12.92 * lin_rgb, (1 + a) * np.power(np.maximum(lin_rgb, 0), 1/2.4) - a) | |
| return cv2.cvtColor((np.clip(srgb, 0, 1) * 255.0).astype(np.uint8), cv2.COLOR_RGB2BGR) | |
| def detect_checker_corners(img_bgr): | |
| det = cv2.mcc.CCheckerDetector_create() | |
| if det.process(img_bgr, cv2.mcc.MCC24): | |
| cc = det.getListColorChecker()[0] | |
| return np.array(cc.getBox() if hasattr(cc, "getBox") else cc.getCorners(), dtype=np.float32) | |
| img_small = cv2.resize(img_bgr, (0,0), fx=0.5, fy=0.5) | |
| if det.process(img_small, cv2.mcc.MCC24): | |
| cc = det.getListColorChecker()[0] | |
| pts = np.array(cc.getBox() if hasattr(cc, "getBox") else cc.getCorners(), dtype=np.float32) | |
| return pts * 2.0 | |
| raise RuntimeError("ColorChecker not found") | |
| def warp_checker(img, corners, out_w=600, out_h=400): | |
| dst = np.float32([[0, out_h-1],[0, 0],[out_w-1, 0],[out_w-1, out_h-1]]) | |
| H_mat = cv2.getPerspectiveTransform(corners, dst) | |
| return cv2.warpPerspective(img, H_mat, (out_w, out_h), flags=cv2.INTER_CUBIC) | |
| def sample_24_patches(warped, margin=12): | |
| H, W = warped.shape[:2] | |
| cell_w, cell_h = W / 6.0, H / 4.0 | |
| lin = to_linear_srgb(warped) | |
| means =[] | |
| for r in range(4): | |
| for c in range(6): | |
| x0, x1 = int(c * cell_w + margin), int((c+1) * cell_w - margin) | |
| y0, y1 = int(r * cell_h + margin), int((r+1) * cell_h - margin) | |
| means.append(np.median(lin[y0:y1, x0:x1].reshape(-1, 3), axis=0)) | |
| return np.stack(means, 0) | |
| def compute_deltaE_00(lin_src, lin_ref): | |
| srgb_src = to_srgb_u8(lin_src).astype(np.float32) / 255.0 | |
| srgb_ref = to_srgb_u8(lin_ref).astype(np.float32) / 255.0 | |
| return color.deltaE_ciede2000(color.rgb2lab(srgb_src.reshape(1, -1, 3)), color.rgb2lab(srgb_ref.reshape(1, -1, 3))).flatten() | |
| def apply_color_pipeline(target_bgr, ref24, tgt24): | |
| tgt_lin = to_linear_srgb(target_bgr) | |
| # 1. White Balance (Von Kries scaling) | |
| gains = np.median(ref24[18:24], axis=0) / np.maximum(np.median(tgt24[18:24], axis=0), 1e-6) | |
| tgt_lin_wb = tgt_lin * gains.reshape(1, 1, 3) | |
| tgt24_wb = tgt24 * gains | |
| # 2. 3x3 Matrix Solver | |
| def objective(W_flat): | |
| W = W_flat.reshape(3, 3) | |
| pred_lin = np.clip(tgt24_wb @ W, 0, 1) | |
| return np.mean(compute_deltaE_00(pred_lin, ref24)) | |
| X, Y = tgt24_wb, ref24 | |
| W_init = np.linalg.inv(X.T @ X + 0.05 * np.eye(3)) @ X.T @ Y | |
| res = minimize( | |
| objective, | |
| W_init.flatten(), | |
| method='Powell', | |
| options={"maxiter": 150, "xtol": 1e-4, "ftol": 1e-4}, | |
| ) | |
| W_opt = res.x.reshape(3, 3) | |
| # 3. Apply to full image | |
| corrected_lin = (tgt_lin_wb.reshape(-1, 3) @ W_opt).reshape(tgt_lin_wb.shape) | |
| return to_srgb_u8(np.clip(corrected_lin, 0, 1)) | |
| # ============================================================================== | |
| # --- CORE API & PROCESSOR --- | |
| # ============================================================================== | |
| class ProcessResult: | |
| success: bool | |
| message: str | |
| r2_rind: Optional[float] = None | |
| r2_flesh: Optional[float] = None | |
| raw_width: Optional[float] = None | |
| sm_width: Optional[float] = None | |
| raw_height: Optional[float] = None | |
| sm_height: Optional[float] = None | |
| raw_perimeter: Optional[float] = None | |
| sm_perimeter: Optional[float] = None | |
| raw_flesh_width: Optional[float] = None | |
| sm_flesh_width: Optional[float] = None | |
| raw_flesh_height: Optional[float] = None | |
| sm_flesh_height: Optional[float] = None | |
| raw_flesh_perimeter: Optional[float] = None | |
| sm_flesh_perimeter: Optional[float] = None | |
| raw_rind_thick: Optional[float] = None | |
| sm_rind_thick: Optional[float] = None | |
| raw_rind_ratio: Optional[float] = None | |
| sm_rind_ratio: Optional[float] = None | |
| raw_total_area: Optional[float] = None | |
| sm_total_area: Optional[float] = None | |
| raw_flesh_area: Optional[float] = None | |
| sm_flesh_area: Optional[float] = None | |
| raw_flesh_ratio: Optional[float] = None | |
| sm_flesh_ratio: Optional[float] = None | |
| raw_elongation: Optional[float] = None | |
| sm_elongation: Optional[float] = None | |
| raw_asym: Optional[float] = None | |
| sm_asym: Optional[float] = None | |
| raw_flesh_asym: Optional[float] = None | |
| sm_flesh_asym: Optional[float] = None | |
| raw_circ: Optional[float] = None | |
| sm_circ: Optional[float] = None | |
| midline_curvature: Optional[float] = None | |
| delta_e_initial: Optional[float] = None | |
| delta_e_final: Optional[float] = None | |
| image_raw_base64: Optional[str] = None | |
| image_sm_base64: Optional[str] = None | |
| filename: Optional[str] = None | |
| measurement_unit: Optional[str] = None | |
| area_unit: Optional[str] = None | |
| scale_source: Optional[str] = None | |
| color_checker_found: bool = False | |
| rind_source: Optional[str] = None | |
| rind_overlap_ratio: Optional[float] = None | |
| warnings: Optional[list] = None | |
| timings_ms: Optional[dict] = None | |
| processing_ms: Optional[int] = None | |
| class WatermelonProcessor: | |
| def __init__(self, model_path: str): | |
| self.model = YOLO(model_path) | |
| self.ref24 = None | |
| if os.path.exists("reference.png"): | |
| try: | |
| ref_img = cv2.imread("reference.png") | |
| ref_corners = detect_checker_corners(ref_img) | |
| self.ref24 = sample_24_patches(warp_checker(ref_img, ref_corners)) | |
| print("Reference ColorChecker patches loaded.") | |
| except Exception as e: | |
| print(f"Reference extraction failed: {e}") | |
| def watermelon_model(theta, Rx, Ry, c_a, d_top, w_top, d_bot, w_bot, phi, c_skew, c_bend): | |
| t = theta - phi | |
| ellipse = (Rx * Ry) / np.sqrt((Ry * np.cos(t)) ** 2 + (Rx * np.sin(t)) ** 2) | |
| asymmetry = 1 + c_a * np.cos(t) ** 3 | |
| divot_top = d_top * np.exp(w_top * (np.sin(t) - 1)) | |
| divot_bot = d_bot * np.exp(w_bot * (-np.sin(t) - 1)) | |
| return (ellipse * asymmetry) - divot_top - divot_bot + c_skew * np.sin(t) + c_bend * np.cos(t) * (np.sin(t) ** 2) | |
| def calculate_axis_metrics(cx, cy, phi, rind_mask, flesh_mask): | |
| """Instantly finds axes and rind thickness using fast OpenCV bitwise operations.""" | |
| h, w = rind_mask.shape | |
| def get_intersections(theta, mask): | |
| temp = np.zeros((h, w), dtype=np.uint8) | |
| L = max(h, w) | |
| # Draw a line spanning across the entire image | |
| p1 = (int(cx + L * np.cos(theta)), int(cy - L * np.sin(theta))) | |
| p2 = (int(cx - L * np.cos(theta)), int(cy + L * np.sin(theta))) | |
| cv2.line(temp, p1, p2, 255, 1) | |
| # Find where the line overlaps the mask | |
| overlap = cv2.bitwise_and(mask, temp) | |
| y_pts, x_pts = np.where(overlap > 0) | |
| if len(x_pts) == 0: | |
| return (int(cx), int(cy)), (int(cx), int(cy)), 0.0 | |
| # Project points to find the two extreme ends of the line segment | |
| dx, dy = x_pts - cx, y_pts - cy | |
| proj = dx * np.cos(theta) - dy * np.sin(theta) | |
| idx_max, idx_min = np.argmax(proj), np.argmin(proj) | |
| pt1 = (int(x_pts[idx_max]), int(y_pts[idx_max])) | |
| pt2 = (int(x_pts[idx_min]), int(y_pts[idx_min])) | |
| dist = float(np.hypot(pt1[0] - pt2[0], pt1[1] - pt2[1])) | |
| return pt1, pt2, dist | |
| # Height line (perpendicular to phi) | |
| pt_top, pt_bot, height_px = get_intersections(phi + np.pi/2, rind_mask) | |
| # Width line (parallel to phi) | |
| pt_right, pt_left, width_px = get_intersections(phi, rind_mask) | |
| # Flesh width along the exact same width line | |
| _, _, flesh_width_px = get_intersections(phi, flesh_mask) | |
| rind_thick_px = None | |
| if width_px > 0 and flesh_width_px > 0: | |
| rind_thick_px = float(max(0.0, (width_px - flesh_width_px) / 2.0)) | |
| return height_px, width_px, rind_thick_px, (pt_top, pt_bot), (pt_left, pt_right) | |
| def contour_centroid(contour): | |
| M = cv2.moments(contour) | |
| if M["m00"] == 0: | |
| return None | |
| return np.array([M["m10"] / M["m00"], M["m01"] / M["m00"]], dtype=np.float32) | |
| def mask_centroid(mask): | |
| M = cv2.moments(mask) | |
| if M["m00"] == 0: | |
| return None | |
| return np.array([M["m10"] / M["m00"], M["m01"] / M["m00"]], dtype=np.float32) | |
| def draw_single_contour(shape, contour): | |
| mask = np.zeros(shape, dtype=np.uint8) | |
| cv2.drawContours(mask, [contour.astype(np.int32)], -1, 255, -1) | |
| return mask | |
| def flesh_envelope_mask(flesh_combined): | |
| ys, xs = np.where(flesh_combined > 0) | |
| if len(xs) < MIN_FLESH_PIXELS_FOR_FALLBACK: | |
| return None | |
| pts = np.column_stack([xs, ys]).astype(np.int32) | |
| hull = cv2.convexHull(pts.reshape(-1, 1, 2)) | |
| envelope = np.zeros_like(flesh_combined) | |
| cv2.drawContours(envelope, [hull], -1, 255, -1) | |
| _, _, bw, bh = cv2.boundingRect(hull) | |
| pad = int(max(12, min(80, round(max(bw, bh) * 0.035)))) | |
| kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (pad * 2 + 1, pad * 2 + 1)) | |
| envelope = cv2.dilate(envelope, kernel, iterations=1) | |
| close_size = max(5, (pad // 2) * 2 + 1) | |
| close_kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (close_size, close_size)) | |
| return cv2.morphologyEx(envelope, cv2.MORPH_CLOSE, close_kernel) | |
| def choose_target_rind_mask(rind_mask, flesh_combined): | |
| warnings = [] | |
| flesh_area = cv2.countNonZero(flesh_combined) | |
| cnts, _ = cv2.findContours(rind_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE) | |
| if not cnts: | |
| envelope = WatermelonProcessor.flesh_envelope_mask(flesh_combined) | |
| if envelope is None: | |
| return rind_mask, "missing", None, ["No whole-watermelon mask and not enough flesh mask for fallback."] | |
| return envelope, "flesh_envelope", 1.0, ["No whole-watermelon mask; estimated perimeter from flesh masks."] | |
| scored = [] | |
| for cnt in cnts: | |
| temp = WatermelonProcessor.draw_single_contour(rind_mask.shape, cnt) | |
| overlap = cv2.countNonZero(cv2.bitwise_and(temp, flesh_combined)) | |
| ratio = overlap / max(flesh_area, 1) | |
| scored.append((ratio, cv2.contourArea(cnt), cnt, temp)) | |
| scored.sort(key=lambda item: (item[0], item[1]), reverse=True) | |
| best_ratio, best_area, best_cnt, best_mask = scored[0] | |
| if flesh_area == 0: | |
| warnings.append("No flesh masks detected; using largest whole-watermelon mask.") | |
| largest = max(cnts, key=cv2.contourArea) | |
| return WatermelonProcessor.draw_single_contour(rind_mask.shape, largest), "whole_mask_no_flesh", None, warnings | |
| flesh_center = WatermelonProcessor.mask_centroid(flesh_combined) | |
| rind_center = WatermelonProcessor.contour_centroid(best_cnt) | |
| ys, xs = np.where(flesh_combined > 0) | |
| flesh_extent = max(float(np.ptp(xs)) if len(xs) else 1.0, float(np.ptp(ys)) if len(ys) else 1.0, 1.0) | |
| center_offset_ratio = 0.0 | |
| if flesh_center is not None and rind_center is not None: | |
| center_offset_ratio = float(np.linalg.norm(flesh_center - rind_center) / flesh_extent) | |
| area_ratio = float(best_area / max(flesh_area, 1)) | |
| if best_ratio >= MIN_RIND_FLESH_OVERLAP_RATIO: | |
| if area_ratio <= MAX_RIND_TO_FLESH_AREA_RATIO and center_offset_ratio <= MAX_RIND_CENTER_OFFSET_RATIO: | |
| return best_mask, "whole_mask_overlap", float(best_ratio), warnings | |
| warnings.append("Whole-watermelon mask overlapped flesh but looked too large or off-center; using fallback.") | |
| if flesh_center is not None and rind_center is not None: | |
| shifted_cnt = best_cnt.astype(np.float32) + (flesh_center - rind_center).reshape(1, 1, 2) | |
| shifted_mask = WatermelonProcessor.draw_single_contour(rind_mask.shape, shifted_cnt) | |
| shifted_ratio = cv2.countNonZero(cv2.bitwise_and(shifted_mask, flesh_combined)) / max(flesh_area, 1) | |
| if shifted_ratio >= MIN_RIND_FLESH_OVERLAP_RATIO and area_ratio <= MAX_RIND_TO_FLESH_AREA_RATIO: | |
| if best_ratio >= MIN_RIND_FLESH_OVERLAP_RATIO: | |
| warnings.append("Whole-watermelon mask was suspicious; translated it to the flesh-mask centroid.") | |
| else: | |
| warnings.append("Whole-watermelon mask did not overlap flesh; translated it to the flesh-mask centroid.") | |
| return shifted_mask, "translated_whole_mask", float(shifted_ratio), warnings | |
| envelope = WatermelonProcessor.flesh_envelope_mask(flesh_combined) | |
| if envelope is not None: | |
| warnings.append("Whole-watermelon mask did not overlap flesh; estimated perimeter from flesh masks.") | |
| return envelope, "flesh_envelope", float(best_ratio), warnings | |
| warnings.append("Whole-watermelon mask did not overlap flesh and fallback was unavailable.") | |
| return best_mask, "low_overlap_whole_mask", float(best_ratio), warnings | |
| def contour_area_px(points): | |
| cnt = points.reshape(-1, 1, 2).astype(np.float32) | |
| return float(abs(cv2.contourArea(cnt))) | |
| def elongation_from_points(points): | |
| cnt = points.reshape(-1, 1, 2).astype(np.float32) | |
| if len(cnt) >= 5: | |
| _, (axis_a, axis_b), _ = cv2.fitEllipse(cnt) | |
| minor = max(min(axis_a, axis_b), 1e-6) | |
| return float(max(axis_a, axis_b) / minor) | |
| _, _, bw, bh = cv2.boundingRect(cnt.astype(np.int32)) | |
| return float(max(bw, bh) / max(min(bw, bh), 1)) | |
| def split_asymmetry(region_mask, midline, thickness=5): | |
| if midline is None or len(midline) < 2 or cv2.countNonZero(region_mask) == 0: return None | |
| split_mask = region_mask.copy() | |
| # FIX: Extrapolate the line 100 pixels in both directions to guarantee it completely bisects smoothed/expanded masks | |
| m = midline.astype(np.float32) | |
| p0, p1, pn, pn_1 = m[0], m[1], m[-1], m[-2] | |
| n0, n1 = np.linalg.norm(p0 - p1), np.linalg.norm(pn - pn_1) | |
| ext_start = p0 + (p0 - p1) / n0 * 100 if n0 > 1e-5 else p0 | |
| ext_end = pn + (pn - pn_1) / n1 * 100 if n1 > 1e-5 else pn | |
| ext_midline = np.vstack([ext_start, m, ext_end]).astype(np.int32) | |
| cv2.polylines(split_mask, [ext_midline], False, 0, thickness) | |
| n_labels, _, stats, _ = cv2.connectedComponentsWithStats((split_mask > 0).astype(np.uint8), connectivity=8) | |
| if n_labels <= 2: return None | |
| areas = sorted([int(stats[i, cv2.CC_STAT_AREA]) for i in range(1, n_labels)], reverse=True) | |
| if len(areas) < 2 or areas[0] + areas[1] == 0: return None | |
| return float(abs(areas[0] - areas[1]) / (areas[0] + areas[1])) | |
| def midline_curvature_score(midline): | |
| if midline is None or len(midline) < 3: | |
| return None | |
| diffs = np.diff(midline.astype(np.float32), axis=0) | |
| path_len = float(np.sum(np.linalg.norm(diffs, axis=1))) | |
| chord_len = float(np.linalg.norm(midline[-1] - midline[0])) | |
| if chord_len <= 1e-6: | |
| return None | |
| return float(max(0.0, (path_len / chord_len) - 1.0)) | |
| def get_polar_data(mask): | |
| """Universal polar extractor for either Rind or Flesh masks.""" | |
| cnts, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE) | |
| if not cnts: return None | |
| cnt = max(cnts, key=cv2.contourArea) | |
| M = cv2.moments(cnt) | |
| if M["m00"] == 0: return None | |
| cx, cy = M["m10"]/M["m00"], M["m01"]/M["m00"] | |
| pts = cnt.reshape(-1, 2) | |
| dx, dy = pts[:, 0] - cx, cy - pts[:, 1] | |
| r_vals, t_vals = np.sqrt(dx**2 + dy**2), np.arctan2(dy, dx) | |
| num_bins = 360 | |
| bins = np.linspace(-np.pi, np.pi, num_bins + 1) | |
| raw_r = np.full(num_bins, np.nan) | |
| for i in range(num_bins): | |
| b_mask = (t_vals >= bins[i]) & (t_vals < bins[i + 1]) | |
| if np.any(b_mask): raw_r[i] = np.max(r_vals[b_mask]) | |
| valid_idx = np.where(~np.isnan(raw_r))[0] | |
| if len(valid_idx) == 0: return None | |
| raw_r[np.isnan(raw_r)] = np.interp(np.where(np.isnan(raw_r))[0], valid_idx, raw_r[valid_idx], period=360) | |
| return (bins[:-1] + bins[1:])/2.0, median_filter(raw_r, size=7, mode="wrap"), (cx, cy), cnt | |
| def calculate_axis_metrics(cx, cy, phi, rind_mask, flesh_mask): | |
| """Instantly finds axes and rind thickness using fast OpenCV bitwise operations.""" | |
| h, w = rind_mask.shape | |
| def get_intersections(theta, mask): | |
| temp = np.zeros((h, w), dtype=np.uint8) | |
| L = max(h, w) | |
| p1 = (int(cx + L * np.cos(theta)), int(cy - L * np.sin(theta))) | |
| p2 = (int(cx - L * np.cos(theta)), int(cy + L * np.sin(theta))) | |
| cv2.line(temp, p1, p2, 255, 1) | |
| overlap = cv2.bitwise_and(mask, temp) | |
| y_pts, x_pts = np.where(overlap > 0) | |
| if len(x_pts) == 0: return (int(cx), int(cy)), (int(cx), int(cy)), 0.0 | |
| dx, dy = x_pts - cx, y_pts - cy | |
| proj = dx * np.cos(theta) - dy * np.sin(theta) | |
| idx_max, idx_min = np.argmax(proj), np.argmin(proj) | |
| pt1 = (int(x_pts[idx_max]), int(y_pts[idx_max])) | |
| pt2 = (int(x_pts[idx_min]), int(y_pts[idx_min])) | |
| dist = float(np.hypot(pt1[0] - pt2[0], pt1[1] - pt2[1])) | |
| return pt1, pt2, dist | |
| pt_top, pt_bot, height_px = get_intersections(phi + np.pi/2, rind_mask) | |
| pt_right, pt_left, width_px = get_intersections(phi, rind_mask) | |
| _, _, f_height_px = get_intersections(phi + np.pi/2, flesh_mask) | |
| _, _, f_width_px = get_intersections(phi, flesh_mask) | |
| rind_thick_px = None | |
| if width_px > 0 and f_width_px > 0: | |
| rind_thick_px = float(max(0.0, (width_px - f_width_px) / 2.0)) | |
| return height_px, width_px, rind_thick_px, (pt_top, pt_bot), (pt_left, pt_right), f_height_px, f_width_px | |
| def get_dual_mask_midline(f_left, f_right, rind_cnt, pred_cnt, cx, cy): | |
| h, w = f_left.shape | |
| _, (ma, Ma), angle = cv2.fitEllipse(rind_cnt) if len(rind_cnt) > 5 else (None, (0,0), 0) | |
| rot_angle = angle if ma < Ma else angle + 90 | |
| m_rot = cv2.getRotationMatrix2D((cx, cy), rot_angle, 1.0) | |
| m_inv = cv2.getRotationMatrix2D((cx, cy), -rot_angle, 1.0) | |
| l_rot, r_rot = cv2.warpAffine(f_left, m_rot, (w, h)), cv2.warpAffine(f_right, m_rot, (w, h)) | |
| l_idx, r_idx = np.where(l_rot > 0)[1], np.where(r_rot > 0)[1] | |
| if len(l_idx) > 0 and len(r_idx) > 0: | |
| if np.mean(l_idx) > np.mean(r_idx): | |
| l_rot, r_rot = r_rot, l_rot | |
| gap_points =[] | |
| y_l, y_r = np.where(l_rot > 0)[0], np.where(r_rot > 0)[0] | |
| if len(y_l) > 0 and len(y_r) > 0: | |
| for y in range(max(np.min(y_l), np.min(y_r)), min(np.max(y_l), np.max(y_r))): | |
| row_l, row_r = np.where(l_rot[y, :] > 0)[0], np.where(r_rot[y, :] > 0)[0] | |
| if len(row_l) > 0 and len(row_r) > 0: | |
| gap_points.append([y, (row_l[-1] + row_r[0]) / 2.0]) | |
| gap_points = np.array(gap_points) | |
| if len(gap_points) > 10: | |
| y_min_g, y_max_g = np.min(gap_points[:, 0]), np.max(gap_points[:, 0]) | |
| y_span, y_mean = max(y_max_g - y_min_g, 1), (y_max_g + y_min_g) / 2.0 | |
| def parabola(y_n, a, b, c): return a*(y_n**2) + b*y_n + c | |
| max_bend = w * 0.08 | |
| try: | |
| popt_mid, _ = curve_fit( | |
| parabola, | |
| (gap_points[:,0]-y_mean)/y_span, | |
| gap_points[:,1], | |
| bounds=([-max_bend, -np.inf, -np.inf],[max_bend, np.inf, np.inf]), | |
| max_nfev=1500, | |
| ) | |
| except: popt_mid = [0.0, 0.0, cx] | |
| ys_extrap = np.linspace(0, h, 500) | |
| xs_extrap = parabola((ys_extrap - y_mean)/y_span, *popt_mid) | |
| pts_rot = np.vstack([xs_extrap, ys_extrap, np.ones_like(ys_extrap)]) | |
| else: | |
| ys_extrap = np.linspace(0, h, 500) | |
| pts_rot = np.vstack([np.full_like(ys_extrap, cx), ys_extrap, np.ones_like(ys_extrap)]) | |
| pts_orig = (m_inv @ pts_rot).T | |
| pred_cnt_cv = pred_cnt.reshape(-1, 1, 2).astype(np.int32) | |
| return np.array([pt for pt in pts_orig if cv2.pointPolygonTest(pred_cnt_cv, (float(pt[0]), float(pt[1])), False) >= 0]) | |
| def process_image(self, image: np.ndarray, source_name: str, scale_ratio: float, include_image: bool = True) -> ProcessResult: | |
| timings = {} | |
| stage_t = time.perf_counter() | |
| def mark(stage_name): | |
| nonlocal stage_t | |
| now = time.perf_counter() | |
| timings[stage_name] = int(round((now - stage_t) * 1000)) | |
| stage_t = now | |
| def fail(message, **extra): | |
| return ProcessResult(success=False, message=message, filename=source_name, warnings=warnings or None, timings_ms=timings, **extra) | |
| warnings =[] | |
| if image is None: return ProcessResult(success=False, message="Could not decode image.", filename=source_name) | |
| h, w = image.shape[:2] | |
| # 1. CALIBRATION & SCALING | |
| dE_initial, dE_final, cm_per_px, checker_corners = None, None, None, None | |
| try: | |
| checker_corners = detect_checker_corners(image) | |
| top_width = np.linalg.norm(checker_corners[1] - checker_corners[2]) | |
| bot_width = np.linalg.norm(checker_corners[0] - checker_corners[3]) | |
| cm_per_px = CHECKER_WIDTH_CM / ((top_width + bot_width) / 2.0) | |
| if self.ref24 is not None: | |
| tgt_warped = warp_checker(image, checker_corners) | |
| tgt24 = sample_24_patches(tgt_warped) | |
| dE_initial = float(np.mean(compute_deltaE_00(tgt24, self.ref24))) | |
| image = apply_color_pipeline(image, self.ref24, tgt24) | |
| tgt_warped_corr = warp_checker(image, checker_corners) | |
| tgt24_corr = sample_24_patches(tgt_warped_corr) | |
| dE_final = float(np.mean(compute_deltaE_00(tgt24_corr, self.ref24))) | |
| except Exception as e: | |
| if cm_per_px is None: warnings.append("ColorChecker not found; dimensions in original-image pixels.") | |
| else: warnings.append("Color correction skipped after ColorChecker detection.") | |
| mark("calibration") | |
| # 2. YOLO INFERENCE | |
| results = self.model(image, conf=0.25, retina_masks=True, verbose=False) | |
| mark("yolo_inference") | |
| rind_mask = np.zeros((h, w), dtype=np.uint8) | |
| f_l_cnts, f_r_cnts = [], [] | |
| if results[0].masks is None: | |
| return fail("No masks detected.", measurement_unit="cm" if cm_per_px is not None else "px", scale_source="color_checker" if cm_per_px is not None else "original_pixels", color_checker_found=checker_corners is not None) | |
| for mask_data, cls in zip(results[0].masks.xy, results[0].boxes.cls): | |
| c = np.array(mask_data, dtype=np.int32) | |
| if int(cls) == 0: cv2.drawContours(rind_mask, [c], -1, 255, -1) | |
| elif int(cls) == 1: f_l_cnts.append(c) | |
| elif int(cls) == 2: f_r_cnts.append(c) | |
| if len(f_l_cnts) >= 2 and len(f_r_cnts) == 0: | |
| f_l_cnts.sort(key=lambda cnt: cv2.moments(cnt)["m10"] / (cv2.moments(cnt)["m00"] + 1e-5)) | |
| f_r_cnts.append(f_l_cnts.pop()) | |
| warnings.append("Only flesh_left was detected; split by x-position.") | |
| elif len(f_r_cnts) >= 2 and len(f_l_cnts) == 0: | |
| f_r_cnts.sort(key=lambda cnt: cv2.moments(cnt)["m10"] / (cv2.moments(cnt)["m00"] + 1e-5)) | |
| f_l_cnts.append(f_r_cnts.pop(0)) | |
| warnings.append("Only flesh_right was detected; split by x-position.") | |
| flesh_l_m, flesh_r_m = np.zeros((h, w), dtype=np.uint8), np.zeros((h, w), dtype=np.uint8) | |
| for c in f_l_cnts: cv2.drawContours(flesh_l_m, [c], -1, 255, -1) | |
| for c in f_r_cnts: cv2.drawContours(flesh_r_m, [c], -1, 255, -1) | |
| flesh_combined = cv2.bitwise_or(flesh_l_m, flesh_r_m) | |
| target_rind_mask, rind_source, rind_overlap_ratio, r_warn = self.choose_target_rind_mask(rind_mask, flesh_combined) | |
| warnings.extend(r_warn) | |
| mark("mask_parse") | |
| # 3. EXTRACTION (Rind First) | |
| rind_data = self.get_polar_data(target_rind_mask) | |
| if rind_data is None: | |
| return fail("No stable perimeter.", measurement_unit="cm" if cm_per_px else "px", scale_source="color_checker" if cm_per_px else "original_pixels", color_checker_found=checker_corners is not None, rind_source=rind_source, rind_overlap_ratio=rind_overlap_ratio) | |
| t_r, raw_r_r, (cx, cy), raw_rind_cnt = rind_data | |
| pts_r_raw = raw_rind_cnt.reshape(-1, 2).astype(np.float32) | |
| # --- SCAN-LINE FLESH GAP FILLING (Preserves V-shape) --- | |
| _, (ma, Ma), angle = cv2.fitEllipse(raw_rind_cnt) if len(raw_rind_cnt) > 5 else (None, (0,0), 0) | |
| rot_angle = angle if ma < Ma else angle + 90 | |
| M_rot = cv2.getRotationMatrix2D((cx, cy), rot_angle, 1.0) | |
| M_inv = cv2.getRotationMatrix2D((cx, cy), -rot_angle, 1.0) | |
| l_rot = cv2.warpAffine(flesh_l_m, M_rot, (w, h)) | |
| r_rot = cv2.warpAffine(flesh_r_m, M_rot, (w, h)) | |
| l_idx = np.where(l_rot > 0)[1] | |
| r_idx = np.where(r_rot > 0)[1] | |
| if len(l_idx) > 0 and len(r_idx) > 0 and np.mean(l_idx) > np.mean(r_idx): | |
| l_rot, r_rot = r_rot, l_rot | |
| flesh_closed_rot = cv2.bitwise_or(l_rot, r_rot) | |
| y_l, y_r = np.where(l_rot > 0)[0], np.where(r_rot > 0)[0] | |
| if len(y_l) > 0 and len(y_r) > 0: | |
| for y in range(max(np.min(y_l), np.min(y_r)), min(np.max(y_l), np.max(y_r))): | |
| row_l = np.where(l_rot[y, :] > 0)[0] | |
| row_r = np.where(r_rot[y, :] > 0)[0] | |
| if len(row_l) > 0 and len(row_r) > 0: | |
| x_start = row_l[-1] | |
| x_end = row_r[0] | |
| if x_start < x_end: | |
| flesh_closed_rot[y, x_start:x_end] = 255 | |
| flesh_closed = cv2.warpAffine(flesh_closed_rot, M_inv, (w, h)) | |
| _, flesh_closed = cv2.threshold(flesh_closed, 127, 255, cv2.THRESH_BINARY) | |
| # NOW we extract the flesh_data! | |
| flesh_data = self.get_polar_data(flesh_closed) | |
| raw_flesh_perim = None | |
| if flesh_data: | |
| _, _, _, raw_flesh_cnt = flesh_data | |
| raw_flesh_perim = float(cv2.arcLength(raw_flesh_cnt, True)) | |
| # 4. RAW FEATURES | |
| _, _, r_angle = cv2.fitEllipse(raw_rind_cnt) | |
| raw_phi = np.deg2rad(180 - r_angle) if r_angle > 90 else np.deg2rad(-r_angle) | |
| # Capture raw_f_h and raw_f_w | |
| raw_h, raw_w, raw_rt, raw_h_line, raw_w_line, raw_f_h, raw_f_w = self.calculate_axis_metrics(cx, cy, raw_phi, target_rind_mask, flesh_closed) | |
| # Midline is found using flesh_combined (with gap) and clipped to the raw rind contour | |
| midline = self.get_dual_mask_midline(flesh_l_m, flesh_r_m, raw_rind_cnt, pts_r_raw, cx, cy) | |
| raw_perim = float(cv2.arcLength(raw_rind_cnt, True)) | |
| raw_tot_a = self.contour_area_px(pts_r_raw) | |
| raw_f_a = float(cv2.countNonZero(flesh_combined)) | |
| raw_f_rat = float(raw_f_a / raw_tot_a) if raw_tot_a > 0 else None | |
| raw_elong = self.elongation_from_points(pts_r_raw) | |
| raw_circ = float((4.0 * np.pi * raw_tot_a) / (raw_perim ** 2)) if raw_perim > 0 and raw_tot_a > 0 else None | |
| raw_asym = self.split_asymmetry(target_rind_mask, midline) | |
| raw_f_asym = self.split_asymmetry(flesh_combined, midline, thickness=3) | |
| midline_curve = self.midline_curvature_score(midline) | |
| # 5. SMOOTHED FEATURES | |
| sm_w, sm_h, sm_perim, sm_rt, sm_tot_a, sm_f_a, sm_f_rat = None, None, None, None, None, None, None | |
| sm_elong, sm_circ, sm_asym, sm_f_asym, r2_rind, r2_flesh = None, None, None, None, None, None | |
| sm_rind_cnt, sm_flesh_cnt = None, None | |
| sm_h_line, sm_w_line = None, None | |
| try: | |
| # Fit Rind | |
| scale_r = np.mean(raw_r_r) | |
| popt_r, _ = curve_fit(self.watermelon_model, t_r, raw_r_r/scale_r, | |
| p0=[1.0, 1.1, 0.0, 0.05, 3.0, 0.05, 3.0, 0.0, 0.0, 0.0], | |
| bounds=([0.5, 0.5, -0.4, 0.0, 0.1, 0.0, 0.1, -1.5, -0.2, -0.2], [2.0, 2.0, 0.4, 0.5, 50.0, 0.5, 50.0, 1.5, 0.2, 0.2]), max_nfev=3000) | |
| d_r = np.sum((raw_r_r/scale_r - 1)**2) | |
| r2_rind = 1 - (np.sum((raw_r_r/scale_r - self.watermelon_model(t_r, *popt_r))**2) / d_r) if d_r != 0 else None | |
| # R² Warning Flag | |
| if r2_rind is not None and r2_rind < 0.85: | |
| warnings.append(f"R² below 0.85 ({r2_rind:.2f}). Fruit may be damaged or irregular.") | |
| t_fit = np.linspace(-np.pi, np.pi, 500) | |
| fit_r = self.watermelon_model(t_fit, *popt_r) * scale_r | |
| sm_rind_pts = np.array([[r*np.cos(t)+cx, cy-r*np.sin(t)] for t, r in zip(t_fit, fit_r)], dtype=np.float32) | |
| sm_rind_cnt = sm_rind_pts.reshape(-1, 1, 2).astype(np.int32) | |
| sm_rind_mask = np.zeros_like(target_rind_mask) | |
| cv2.fillPoly(sm_rind_mask, [sm_rind_cnt], 255) | |
| sm_perim = float(np.sum(np.linalg.norm(np.diff(sm_rind_pts, axis=0), axis=1)) + np.linalg.norm(sm_rind_pts[-1]-sm_rind_pts[0])) | |
| sm_tot_a = self.contour_area_px(sm_rind_pts) | |
| sm_elong = self.elongation_from_points(sm_rind_pts) | |
| sm_circ = float((4.0 * np.pi * sm_tot_a) / (sm_perim ** 2)) if sm_perim > 0 and sm_tot_a > 0 else None | |
| sm_asym = self.split_asymmetry(sm_rind_mask, midline) | |
| sm_phi = popt_r[7] | |
| # Fit Flesh | |
| if flesh_data: | |
| t_f, raw_r_f, (fcx, fcy), _ = flesh_data | |
| scale_f = np.mean(raw_r_f) | |
| popt_f, _ = curve_fit(self.watermelon_model, t_f, raw_r_f/scale_f, | |
| p0=[1.0, 1.1, 0.0, 0.05, 3.0, 0.05, 3.0, 0.0, 0.0, 0.0], | |
| bounds=([0.5, 0.5, -0.4, 0.0, 0.1, 0.0, 0.1, -1.5, -0.2, -0.2], [2.0, 2.0, 0.4, 0.5, 50.0, 0.5, 50.0, 1.5, 0.2, 0.2]), max_nfev=3000) | |
| d_f = np.sum((raw_r_f/scale_f - 1)**2) | |
| r2_flesh = 1 - (np.sum((raw_r_f/scale_f - self.watermelon_model(t_f, *popt_f))**2) / d_f) if d_f != 0 else None | |
| fit_f = self.watermelon_model(t_fit, *popt_f) * scale_f | |
| sm_flesh_pts = np.array([[r*np.cos(t)+fcx, fcy-r*np.sin(t)] for t, r in zip(t_fit, fit_f)], dtype=np.float32) | |
| sm_flesh_cnt = sm_flesh_pts.reshape(-1, 1, 2).astype(np.int32) | |
| sm_flesh_mask = np.zeros_like(target_rind_mask) | |
| cv2.fillPoly(sm_flesh_mask, [sm_flesh_cnt], 255) | |
| sm_f_a = self.contour_area_px(sm_flesh_pts) | |
| sm_flesh_asym = self.split_asymmetry(sm_flesh_mask, midline, thickness=3) | |
| sm_flesh_perim = float(np.sum(np.linalg.norm(np.diff(sm_flesh_pts, axis=0), axis=1)) + np.linalg.norm(sm_flesh_pts[-1]-sm_flesh_pts[0])) | |
| else: | |
| sm_flesh_mask = np.zeros_like(target_rind_mask) | |
| sm_flesh_perim = None | |
| # Smooth axes (using the filled smooth masks) | |
| sm_h, sm_w, sm_rt, sm_h_line, sm_w_line, sm_f_h, sm_f_w = self.calculate_axis_metrics(cx, cy, sm_phi, sm_rind_mask, sm_flesh_mask) | |
| if sm_tot_a > 0 and sm_f_a is not None: | |
| sm_f_rat = float(sm_f_a / sm_tot_a) | |
| except Exception as exc: | |
| warnings.append(f"Smoothing fit failed: {exc}") | |
| mark("fit") | |
| # 6. APPLY SCALES | |
| sc_src = "color_checker" if cm_per_px else "original_pixels" | |
| m_unit = "cm" if cm_per_px else "px" | |
| a_unit = "cm2" if cm_per_px else "px2" | |
| scaler = cm_per_px if cm_per_px else (1.0 / scale_ratio) | |
| a_scaler = scaler ** 2 | |
| def s(v): return float(v * scaler) if v is not None else None | |
| def a(v): return float(v * a_scaler) if v is not None else None | |
| def rt_rat(thick, w): return float((thick * 2.0) / w) if thick is not None and w and w > 0 else None | |
| res = ProcessResult( | |
| success=True, message="Success", filename=source_name, measurement_unit=m_unit, area_unit=a_unit, | |
| scale_source=sc_src, color_checker_found=bool(cm_per_px), | |
| rind_source=rind_source, rind_overlap_ratio=rind_overlap_ratio, warnings=warnings or None, | |
| r2_rind=r2_rind, r2_flesh=r2_flesh, midline_curvature=midline_curve, | |
| delta_e_initial=dE_initial, delta_e_final=dE_final, timings_ms=timings, | |
| raw_width=s(raw_w), sm_width=s(sm_w), raw_height=s(raw_h), sm_height=s(sm_h), | |
| raw_perimeter=s(raw_perim), sm_perimeter=s(sm_perim), | |
| raw_flesh_width=s(raw_f_w), sm_flesh_width=s(sm_f_w), | |
| raw_flesh_height=s(raw_f_h), sm_flesh_height=s(sm_f_h), | |
| raw_flesh_perimeter=s(raw_flesh_perim), sm_flesh_perimeter=s(sm_flesh_perim), | |
| raw_rind_thick=s(raw_rt), sm_rind_thick=s(sm_rt), | |
| raw_rind_ratio=rt_rat(raw_rt, raw_w), sm_rind_ratio=rt_rat(sm_rt, sm_w), | |
| raw_total_area=a(raw_tot_a), sm_total_area=a(sm_tot_a), | |
| raw_flesh_area=a(raw_f_a), sm_flesh_area=a(sm_f_a), | |
| raw_flesh_ratio=raw_f_rat, sm_flesh_ratio=sm_f_rat, | |
| raw_elongation=raw_elong, sm_elongation=sm_elong, | |
| raw_asym=raw_asym, sm_asym=sm_asym, raw_flesh_asym=raw_f_asym, sm_flesh_asym=sm_flesh_asym, | |
| raw_circ=raw_circ, sm_circ=sm_circ | |
| ) | |
| # 7. DRAW PREVIEWS | |
| if include_image: | |
| def encode_img(canvas): | |
| _, b = cv2.imencode(".jpg", canvas, [cv2.IMWRITE_JPEG_QUALITY, 85]) | |
| return base64.b64encode(b).decode("utf-8") | |
| def draw_base(r_m, f_m): | |
| out = image.copy().astype(np.float32) | |
| alpha = 0.18 # Transparency | |
| # --- THE FIX: Only paint the rind where there is NO flesh --- | |
| rind_only = (r_m > 0) & (f_m == 0) | |
| # Rind fill: Green | |
| out[..., 0] = np.where(rind_only, out[..., 0]*(1-alpha) + 0, out[..., 0]) | |
| out[..., 1] = np.where(rind_only, out[..., 1]*(1-alpha) + 170, out[..., 1]) | |
| out[..., 2] = np.where(rind_only, out[..., 2]*(1-alpha) + 0, out[..., 2]) | |
| # Flesh fill: Red | |
| out[..., 0] = np.where(f_m > 0, out[..., 0]*(1-alpha) + 0, out[..., 0]) | |
| out[..., 1] = np.where(f_m > 0, out[..., 1]*(1-alpha) + 0, out[..., 1]) | |
| out[..., 2] = np.where(f_m > 0, out[..., 2]*(1-alpha) + 255, out[..., 2]) | |
| out = np.clip(out, 0, 255).astype(np.uint8) | |
| if checker_corners is not None: | |
| cv2.polylines(out, [np.int32(checker_corners)], True, (0, 165, 255), 4) | |
| if len(midline) > 1: | |
| cv2.polylines(out, [midline.astype(np.int32)], False, (0, 255, 255), 3) | |
| pt1, pt2 = tuple(midline[0].astype(int)), tuple(midline[-1].astype(int)) | |
| for pt in (pt1, pt2): | |
| cv2.circle(out, pt, 8, (0,0,0), 2) | |
| cv2.circle(out, pt, 6, (255,255,255), -1) | |
| return out | |
| # RAW | |
| out_raw = draw_base(target_rind_mask, flesh_closed) | |
| cv2.line(out_raw, raw_h_line[0], raw_h_line[1], (255, 100, 255), 2) | |
| cv2.line(out_raw, raw_w_line[0], raw_w_line[1], (255, 255, 100), 2) | |
| f_cnts_raw, _ = cv2.findContours(flesh_closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) | |
| if f_cnts_raw: | |
| cv2.polylines(out_raw, [max(f_cnts_raw, key=cv2.contourArea)], True, (0, 0, 255), 2) # Red, 2px | |
| cv2.polylines(out_raw, [raw_rind_cnt], True, (0, 200, 0), 2) # Dark Green, 2px | |
| res.image_raw_base64 = encode_img(out_raw) | |
| # SMOOTH | |
| if sm_rind_cnt is not None: | |
| # Use raw fills for the smoothed preview (decoupled visual) | |
| out_sm = draw_base(target_rind_mask, flesh_closed) | |
| cv2.line(out_sm, sm_h_line[0], sm_h_line[1], (255, 100, 255), 2) | |
| cv2.line(out_sm, sm_w_line[0], sm_w_line[1], (255, 255, 100), 2) | |
| if sm_flesh_cnt is not None: | |
| cv2.polylines(out_sm, [sm_flesh_cnt], True, (0, 0, 255), 2) # Red, 2px | |
| cv2.polylines(out_sm, [sm_rind_cnt], True, (0, 200, 0), 2) # Dark Green, 2px | |
| res.image_sm_base64 = encode_img(out_sm) | |
| mark("render") | |
| return res | |
| app = FastAPI() | |
| app.add_middleware( | |
| CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], | |
| ) | |
| processor = WatermelonProcessor(MODEL_PATH) | |
| # --- CONCURRENCY & QUEUE MANAGEMENT --- | |
| dev_lock = asyncio.Lock() | |
| gen_lock = asyncio.Lock() | |
| dev_queue_count = 0 | |
| gen_queue_count = 0 | |
| def read_root(): return {"status": "Watermelon API is awake and running!"} | |
| def get_queue_status(): | |
| return {"dev_queue": dev_queue_count, "gen_queue": gen_queue_count} | |
| async def process_single( | |
| file: UploadFile = File(...), | |
| include_image: bool = Query(True), | |
| password: str = Form(""), | |
| username: str = Form("") | |
| ): | |
| global dev_queue_count, gen_queue_count | |
| request_t = time.perf_counter() | |
| expected_hash = "9139eb3676d5dfafced7613f044d86d9e7c84f40a04c83ddce062878621315d0" | |
| if hashlib.sha256(password.encode('utf-8')).hexdigest() != expected_hash: | |
| return ProcessResult(success=False, message="Unauthorized.", filename=file.filename).__dict__ | |
| contents, img = None, None | |
| try: | |
| contents = await file.read() | |
| if not contents: return ProcessResult(success=False, message="Empty.", filename=file.filename).__dict__ | |
| img = cv2.imdecode(np.frombuffer(contents, np.uint8), cv2.IMREAD_COLOR) | |
| if img is None: return ProcessResult(success=False, message="Decode error.", filename=file.filename).__dict__ | |
| scale_ratio = 1.0 | |
| h, w = img.shape[:2] | |
| if max(h, w) > MAX_IMAGE_SIZE: | |
| scale_ratio = MAX_IMAGE_SIZE / float(max(h, w)) | |
| img = cv2.resize(img, (int(w * scale_ratio), int(h * scale_ratio)), interpolation=cv2.INTER_AREA) | |
| # --- CPU CORE ROUTING & QUEUE LOGIC --- | |
| is_dev = (username.strip().lower() == 'devtest') | |
| # run_in_threadpool prevents OpenCV/YOLO from freezing the API so /queue_status can still answer | |
| if is_dev: | |
| dev_queue_count += 1 | |
| try: | |
| async with dev_lock: | |
| res = await run_in_threadpool(processor.process_image, img, file.filename, scale_ratio, include_image) | |
| finally: | |
| dev_queue_count -= 1 | |
| else: | |
| gen_queue_count += 1 | |
| try: | |
| async with gen_lock: | |
| res = await run_in_threadpool(processor.process_image, img, file.filename, scale_ratio, include_image) | |
| finally: | |
| gen_queue_count -= 1 | |
| res.processing_ms = int(round((time.perf_counter() - request_t) * 1000)) | |
| return res.__dict__ | |
| except Exception as exc: | |
| traceback.print_exc() | |
| return ProcessResult(success=False, message=str(exc), filename=file.filename).__dict__ | |
| finally: | |
| del img, contents | |
| gc.collect() |