"""医学影像预处理:优先 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}, }