refactor: 移除 PIL 依赖并重构推理逻辑为 NumPy
Browse files- 用 NumPy 替代 PIL 实现图像填充与裁剪操作,减少外部依赖
- 优化推理流程,将 Albumentations 和模型推理逻辑调整为以 NumPy 数组处理
- 使用 OpenCV 替代 PIL
app.py
CHANGED
|
@@ -1,4 +1,3 @@
|
|
| 1 |
-
import io
|
| 2 |
import json
|
| 3 |
from contextlib import asynccontextmanager
|
| 4 |
from pathlib import Path
|
|
@@ -7,7 +6,6 @@ import albumentations as A
|
|
| 7 |
import cv2
|
| 8 |
import numpy as np
|
| 9 |
import torch
|
| 10 |
-
from PIL import Image
|
| 11 |
from anyio.to_thread import run_sync
|
| 12 |
from fastapi import FastAPI, Request, UploadFile, File
|
| 13 |
from fastapi.responses import Response
|
|
@@ -95,66 +93,64 @@ async def predict(request: Request, file: UploadFile = File(...)):
|
|
| 95 |
new_w = (orig_w // TRAIN_SIZE + (1 if orig_w % TRAIN_SIZE != 0 else 0)) * TRAIN_SIZE
|
| 96 |
new_h = (orig_h // TRAIN_SIZE + (1 if orig_h % TRAIN_SIZE != 0 else 0)) * TRAIN_SIZE
|
| 97 |
|
| 98 |
-
# 2. Padding 原圖
|
| 99 |
-
padded_img =
|
| 100 |
-
|
|
|
|
| 101 |
|
| 102 |
-
|
|
|
|
| 103 |
|
| 104 |
# 3. 循環切割
|
| 105 |
for y in range(0, new_h, TRAIN_SIZE):
|
| 106 |
for x in range(0, new_w, TRAIN_SIZE):
|
| 107 |
-
|
| 108 |
-
patch = padded_img
|
| 109 |
|
| 110 |
-
#
|
| 111 |
-
|
| 112 |
-
|
| 113 |
-
input_tensor = transformed["image"].unsqueeze(0).to(device) # [1, C, H, W]
|
| 114 |
-
# -----------------------------------
|
| 115 |
|
| 116 |
output = model(input_tensor)
|
| 117 |
|
| 118 |
pred_mask = (output > 0.5).float().squeeze().cpu().numpy()
|
| 119 |
pred_mask = (pred_mask * 255).astype(np.uint8)
|
| 120 |
|
| 121 |
-
|
| 122 |
-
result_mask
|
| 123 |
|
| 124 |
-
|
|
|
|
| 125 |
|
| 126 |
# ==============================
|
| 127 |
# 情況 B: 圖片小於等於 512
|
| 128 |
# ==============================
|
| 129 |
else:
|
| 130 |
-
|
| 131 |
-
|
|
|
|
| 132 |
|
| 133 |
-
|
| 134 |
-
|
| 135 |
-
|
| 136 |
-
# --- 修改部分: Albumentations 處理 ---
|
| 137 |
-
patch_np = np.array(padded_img) # 轉換為 numpy
|
| 138 |
-
transformed = preprocess_fn(image=patch_np)
|
| 139 |
input_tensor = transformed["image"].unsqueeze(0).to(device)
|
| 140 |
-
# -----------------------------------
|
| 141 |
|
| 142 |
output = model(input_tensor)
|
| 143 |
|
| 144 |
pred_mask = (output > 0.5).float().squeeze().cpu().numpy()
|
| 145 |
pred_mask = (pred_mask * 255).astype(np.uint8)
|
| 146 |
|
| 147 |
-
|
|
|
|
| 148 |
|
| 149 |
return final_image
|
| 150 |
|
| 151 |
# 執行推理
|
| 152 |
result_image = await run_sync(_inference_logic)
|
| 153 |
|
| 154 |
-
# 返回圖片流
|
| 155 |
-
|
| 156 |
-
|
| 157 |
-
return Response(content=
|
| 158 |
|
| 159 |
|
| 160 |
@app.get("/")
|
|
|
|
|
|
|
| 1 |
import json
|
| 2 |
from contextlib import asynccontextmanager
|
| 3 |
from pathlib import Path
|
|
|
|
| 6 |
import cv2
|
| 7 |
import numpy as np
|
| 8 |
import torch
|
|
|
|
| 9 |
from anyio.to_thread import run_sync
|
| 10 |
from fastapi import FastAPI, Request, UploadFile, File
|
| 11 |
from fastapi.responses import Response
|
|
|
|
| 93 |
new_w = (orig_w // TRAIN_SIZE + (1 if orig_w % TRAIN_SIZE != 0 else 0)) * TRAIN_SIZE
|
| 94 |
new_h = (orig_h // TRAIN_SIZE + (1 if orig_h % TRAIN_SIZE != 0 else 0)) * TRAIN_SIZE
|
| 95 |
|
| 96 |
+
# 2. Padding 原圖 (使用 NumPy 創建黑色畫布)
|
| 97 |
+
padded_img = np.zeros((new_h, new_w, 3), dtype=np.uint8)
|
| 98 |
+
# 將原圖貼到左上角 (NumPy 切片賦值)
|
| 99 |
+
padded_img[:orig_h, :orig_w, :] = original_image
|
| 100 |
|
| 101 |
+
# 創建結果掩碼畫布 (單通道)
|
| 102 |
+
result_mask = np.zeros((new_h, new_w), dtype=np.uint8)
|
| 103 |
|
| 104 |
# 3. 循環切割
|
| 105 |
for y in range(0, new_h, TRAIN_SIZE):
|
| 106 |
for x in range(0, new_w, TRAIN_SIZE):
|
| 107 |
+
# NumPy 切片代替 crop
|
| 108 |
+
patch = padded_img[y:y + TRAIN_SIZE, x:x + TRAIN_SIZE]
|
| 109 |
|
| 110 |
+
# Albumentations 處理 (patch 已經是 numpy array)
|
| 111 |
+
transformed = preprocess_fn(image=patch)
|
| 112 |
+
input_tensor = transformed["image"].unsqueeze(0).to(device)
|
|
|
|
|
|
|
| 113 |
|
| 114 |
output = model(input_tensor)
|
| 115 |
|
| 116 |
pred_mask = (output > 0.5).float().squeeze().cpu().numpy()
|
| 117 |
pred_mask = (pred_mask * 255).astype(np.uint8)
|
| 118 |
|
| 119 |
+
# 將預測結果貼回大圖
|
| 120 |
+
result_mask[y:y + TRAIN_SIZE, x:x + TRAIN_SIZE] = pred_mask
|
| 121 |
|
| 122 |
+
# 裁剪回原始尺寸
|
| 123 |
+
final_image = result_mask[:orig_h, :orig_w]
|
| 124 |
|
| 125 |
# ==============================
|
| 126 |
# 情況 B: 圖片小於等於 512
|
| 127 |
# ==============================
|
| 128 |
else:
|
| 129 |
+
# 創建黑色畫布
|
| 130 |
+
padded_img = np.zeros((TRAIN_SIZE, TRAIN_SIZE, 3), dtype=np.uint8)
|
| 131 |
+
padded_img[:orig_h, :orig_w, :] = original_image
|
| 132 |
|
| 133 |
+
# Albumentations 處理
|
| 134 |
+
transformed = preprocess_fn(image=padded_img)
|
|
|
|
|
|
|
|
|
|
|
|
|
| 135 |
input_tensor = transformed["image"].unsqueeze(0).to(device)
|
|
|
|
| 136 |
|
| 137 |
output = model(input_tensor)
|
| 138 |
|
| 139 |
pred_mask = (output > 0.5).float().squeeze().cpu().numpy()
|
| 140 |
pred_mask = (pred_mask * 255).astype(np.uint8)
|
| 141 |
|
| 142 |
+
# 裁剪回原始尺寸
|
| 143 |
+
final_image = pred_mask[:orig_h, :orig_w]
|
| 144 |
|
| 145 |
return final_image
|
| 146 |
|
| 147 |
# 執行推理
|
| 148 |
result_image = await run_sync(_inference_logic)
|
| 149 |
|
| 150 |
+
# 返回圖片流 (使用 cv2.imencode)
|
| 151 |
+
# result_image 是单通道灰度图,可以直接编码为 PNG
|
| 152 |
+
success, encoded_image = cv2.imencode(".png", result_image)
|
| 153 |
+
return Response(content=encoded_image.tobytes(), media_type="image/png")
|
| 154 |
|
| 155 |
|
| 156 |
@app.get("/")
|