Files
Shixun/ai-service/app/services/monai_preprocess.py
T
2026-07-31 12:27:41 +08:00

79 lines
2.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""医学影像预处理:优先 MONAI,失败则用 OpenCV/Pillow 降级。"""
from __future__ import annotations
import logging
from typing import Any
import cv2
import numpy as np
logger = logging.getLogger(__name__)
_MONAI_OK = False
try:
import monai # noqa: F401
from monai.transforms import Compose, ScaleIntensity, Resize
_MONAI_OK = True
except Exception: # pragma: no cover
_MONAI_OK = False
logger.info("MONAI 未安装,使用 OpenCV 预处理管线")
def monai_available() -> bool:
return _MONAI_OK
def load_image_bgr(image_bytes: bytes) -> np.ndarray:
arr = np.frombuffer(image_bytes, dtype=np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if img is None:
raise ValueError("无法解码影像文件,请上传常见图片格式(jpg/png 等)")
return img
def preprocess(image_bgr: np.ndarray, target_size: int = 640) -> dict[str, Any]:
"""
返回:
- image_bgr: 原始 BGR
- image_rgb: RGB
- tensor_like: 归一化后的 float32 CHW(MONAI 或 numpy 模拟)
- meta: 尺寸信息
"""
h, w = image_bgr.shape[:2]
rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB)
if _MONAI_OK:
try:
# 灰度/三通道统一为 CHW float,再经 MONAI ScaleIntensity + Resize
chw = np.transpose(rgb.astype(np.float32) / 255.0, (2, 0, 1))
transforms = Compose(
[
ScaleIntensity(minv=0.0, maxv=1.0),
Resize(spatial_size=(target_size, target_size), mode="bilinear"),
]
)
tensor = transforms(chw)
if hasattr(tensor, "numpy"):
tensor = tensor.numpy()
return {
"image_bgr": image_bgr,
"image_rgb": rgb,
"tensor_like": np.asarray(tensor),
"backend": "monai",
"meta": {"orig_h": h, "orig_w": w, "target": target_size},
}
except Exception as e: # pragma: no cover
logger.warning("MONAI 预处理失败,降级 OpenCV: %s", e)
# OpenCV 降级:resize + normalize
resized = cv2.resize(rgb, (target_size, target_size), interpolation=cv2.INTER_LINEAR)
tensor = np.transpose(resized.astype(np.float32) / 255.0, (2, 0, 1))
return {
"image_bgr": image_bgr,
"image_rgb": rgb,
"tensor_like": tensor,
"backend": "opencv",
"meta": {"orig_h": h, "orig_w": w, "target": target_size},
}