ilaandClaude Sonnet 4.6
37a4c13799
- 新增 mingxi-capture（PyQt5桌面程序）目录骨架 - 新增 mingxi-vision（FastAPI推理服务）目录骨架 - 新增 mingxi-backend（Django业务后端）目录骨架 - 新增 mingxi-frontend（Vue3管理后台）目录骨架 - 迁入 docs/（技术设计文档和PCB调研资料） - 补充 .gitignore（模型文件、Node.js依赖、运行时数据库） Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
827 lines
26 KiB
Markdown
827 lines
26 KiB
Markdown
# mingxi-vision 技术架构与设计分析
|
||
|
||
> 版本：v1.0 · 日期：2026-05-24
|
||
> 硬件约束：联想 Y7000P · RTX 4060 Laptop 8GB · Python 3.8 · Windows 11
|
||
> 上游依赖：`yolo_classification_system/yolo8/inference.py`（迁移基础）
|
||
|
||
---
|
||
|
||
## 1. 定位与职责边界
|
||
|
||
`mingxi-vision` 是整个明析平台的**推理微服务**，职责单一：
|
||
|
||
```
|
||
接收图像 → 前处理 → 模型推理 → 后处理 → 返回结构化缺陷JSON
|
||
```
|
||
|
||
它**不**做的事：
|
||
- 不持久化数据（无数据库依赖）
|
||
- 不做鉴权（内网服务，由 backend 代理）
|
||
- 不管理产线/批次业务（backend 的职责）
|
||
- 不直接控制相机（capture 的职责）
|
||
|
||
---
|
||
|
||
## 2. 框架选型：FastAPI
|
||
|
||
### 为什么不用 Django
|
||
|
||
现有系统用 Django，但推理服务有不同的特征：
|
||
|
||
| 维度 | Django（现有） | FastAPI（选用） |
|
||
|------|--------------|----------------|
|
||
| 定位 | 全功能Web框架 | 轻量API框架 |
|
||
| 启动时间 | ~3-5s | ~0.5s |
|
||
| ORM/DB | 必须配置 | 无需数据库 |
|
||
| 异步支持 | 有限（Django 4+ 部分支持） | 原生 async/await |
|
||
| 自动文档 | 需要额外配置 | 内置 /docs (Swagger) |
|
||
| 推理场景适配 | 过重 | 刚好合适 |
|
||
|
||
### Python 3.8 的约束
|
||
|
||
FastAPI 在 Python 3.8 下完全可用，需注意：
|
||
- 类型注解用 `Optional[X]` 而非 `X | None`（3.10+ 语法）
|
||
- `from __future__ import annotations` 可缓解部分问题
|
||
- Pydantic v1（随 FastAPI 早期版本）在 3.8 稳定
|
||
|
||
```
|
||
# requirements.txt 核心版本锁定
|
||
fastapi==0.104.1 # 3.8 兼容的最后稳定版系列
|
||
uvicorn[standard]==0.24.0
|
||
pydantic==1.10.13 # v1，Python 3.8 最稳定
|
||
ultralytics==8.0.235 # 训练阶段用
|
||
onnxruntime-gpu==1.16.3 # 推理部署阶段用
|
||
opencv-python-headless==4.8.1.78
|
||
numpy==1.24.4 # 3.8 + torch 2.0 兼容版本
|
||
python-multipart==0.0.6 # FastAPI 文件上传必须
|
||
```
|
||
|
||
---
|
||
|
||
## 3. 双运行时策略
|
||
|
||
这是 mingxi-vision 最核心的设计决策。
|
||
|
||
### 问题
|
||
|
||
- **训练阶段**：用 `ultralytics` 的 `.pt` 格式，便于迭代和验证
|
||
- **部署阶段**：用 `.onnx` 格式，跨平台、无需安装 PyTorch、性能更稳定
|
||
- 两套格式的推理 API 有差异，需要统一抽象
|
||
|
||
### 解决方案：运行时适配器（Adapter 模式）
|
||
|
||
```
|
||
┌─────────────────────────────┐
|
||
│ InferenceEngine │ ← 统一接口
|
||
│ detect(image) → [Defect] │
|
||
└──────────┬──────────────────┘
|
||
│ 根据配置选择
|
||
┌───────────────┴────────────────┐
|
||
▼ ▼
|
||
UltralyticsAdapter OnnxRuntimeAdapter
|
||
(.pt 文件，训练验证用) (.onnx 文件，生产部署用)
|
||
ultralytics.YOLO onnxruntime.InferenceSession
|
||
```
|
||
|
||
### 代码设计
|
||
|
||
```python
|
||
# engine/base.py
|
||
from abc import ABC, abstractmethod
|
||
from dataclasses import dataclass
|
||
from typing import List
|
||
import numpy as np
|
||
|
||
@dataclass
|
||
class DefectBox:
|
||
class_id: int
|
||
class_name: str
|
||
class_name_zh: str
|
||
confidence: float
|
||
severity: str # fatal / major / minor / rework / none
|
||
box_xyxy: List[float] # [x1, y1, x2, y2]，像素坐标
|
||
|
||
|
||
@dataclass
|
||
class DetectResult:
|
||
defects: List[DefectBox]
|
||
duration_ms: float
|
||
image_width: int
|
||
image_height: int
|
||
model_version: str
|
||
|
||
@property
|
||
def defect_count(self) -> int:
|
||
return len(self.defects)
|
||
|
||
@property
|
||
def max_severity(self) -> str:
|
||
order = ["fatal", "major", "minor", "rework", "none"]
|
||
found = {d.severity for d in self.defects}
|
||
for s in order:
|
||
if s in found:
|
||
return s
|
||
return "none"
|
||
|
||
@property
|
||
def avg_confidence(self):
|
||
if not self.defects:
|
||
return None
|
||
return round(sum(d.confidence for d in self.defects) / len(self.defects), 4)
|
||
|
||
|
||
class BaseInferenceEngine(ABC):
|
||
@abstractmethod
|
||
def detect(self, image: np.ndarray, conf: float) -> DetectResult:
|
||
...
|
||
|
||
@abstractmethod
|
||
def warmup(self) -> None:
|
||
"""启动时预热，避免第一次推理延迟"""
|
||
...
|
||
```
|
||
|
||
```python
|
||
# engine/ultralytics_adapter.py
|
||
import time
|
||
import numpy as np
|
||
from .base import BaseInferenceEngine, DefectBox, DetectResult
|
||
from .defect_classes import DEFECT_CLASSES
|
||
|
||
class UltralyticsAdapter(BaseInferenceEngine):
|
||
"""开发/训练验证阶段使用"""
|
||
|
||
def __init__(self, weights_path: str):
|
||
from ultralytics import YOLO
|
||
self._model = YOLO(weights_path)
|
||
self._weights_path = weights_path
|
||
|
||
def warmup(self):
|
||
dummy = np.zeros((640, 640, 3), dtype=np.uint8)
|
||
self._model.predict(source=dummy, verbose=False, conf=0.01)
|
||
|
||
def detect(self, image: np.ndarray, conf: float = 0.45) -> DetectResult:
|
||
h, w = image.shape[:2]
|
||
t0 = time.perf_counter()
|
||
results = self._model.predict(source=image, verbose=False, conf=conf)
|
||
duration_ms = (time.perf_counter() - t0) * 1000
|
||
|
||
defects = []
|
||
result = results[0]
|
||
if result.boxes is not None and len(result.boxes):
|
||
xyxy = result.boxes.xyxy.cpu().numpy()
|
||
confs = result.boxes.conf.cpu().numpy()
|
||
clsids = result.boxes.cls.cpu().numpy()
|
||
for i in range(len(xyxy)):
|
||
cid = int(clsids[i])
|
||
meta = DEFECT_CLASSES.get(cid, {
|
||
"name": str(cid), "zh": str(cid), "severity": "minor"
|
||
})
|
||
defects.append(DefectBox(
|
||
class_id=cid,
|
||
class_name=meta["name"],
|
||
class_name_zh=meta["zh"],
|
||
confidence=round(float(confs[i]), 4),
|
||
severity=meta["severity"],
|
||
box_xyxy=[round(float(x), 1) for x in xyxy[i]],
|
||
))
|
||
|
||
return DetectResult(
|
||
defects=defects,
|
||
duration_ms=round(duration_ms, 1),
|
||
image_width=w,
|
||
image_height=h,
|
||
model_version=str(self._weights_path),
|
||
)
|
||
```
|
||
|
||
```python
|
||
# engine/onnx_adapter.py
|
||
import time
|
||
import cv2
|
||
import numpy as np
|
||
import onnxruntime as ort
|
||
from .base import BaseInferenceEngine, DefectBox, DetectResult
|
||
from .defect_classes import DEFECT_CLASSES
|
||
|
||
class OnnxRuntimeAdapter(BaseInferenceEngine):
|
||
"""生产部署阶段使用，无 PyTorch 依赖"""
|
||
|
||
def __init__(self, model_path: str, device: str = "cuda"):
|
||
providers = (
|
||
["CUDAExecutionProvider", "CPUExecutionProvider"]
|
||
if device == "cuda"
|
||
else ["CPUExecutionProvider"]
|
||
)
|
||
self._session = ort.InferenceSession(model_path, providers=providers)
|
||
self._input_name = self._session.get_inputs()[0].name
|
||
self._input_shape = self._session.get_inputs()[0].shape # [1,3,640,640]
|
||
self._imgsz = self._input_shape[2] # 通常 640
|
||
self._model_path = model_path
|
||
|
||
def warmup(self):
|
||
dummy = np.zeros(
|
||
(1, 3, self._imgsz, self._imgsz), dtype=np.float32
|
||
)
|
||
self._session.run(None, {self._input_name: dummy})
|
||
|
||
def _preprocess(self, image: np.ndarray):
|
||
"""BGR → RGB → letterbox → NCHW float32 [0,1]"""
|
||
img = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
|
||
img, ratio, (dw, dh) = self._letterbox(img, self._imgsz)
|
||
img = img.astype(np.float32) / 255.0
|
||
img = np.transpose(img, (2, 0, 1)) # HWC → CHW
|
||
img = np.expand_dims(img, 0) # CHW → NCHW
|
||
return img, ratio, dw, dh
|
||
|
||
@staticmethod
|
||
def _letterbox(img, new_size=640):
|
||
h, w = img.shape[:2]
|
||
ratio = min(new_size / h, new_size / w)
|
||
nh, nw = int(h * ratio), int(w * ratio)
|
||
img = cv2.resize(img, (nw, nh), interpolation=cv2.INTER_LINEAR)
|
||
dw = (new_size - nw) / 2
|
||
dh = (new_size - nh) / 2
|
||
top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1))
|
||
left, right = int(round(dw - 0.1)), int(round(dw + 0.1))
|
||
img = cv2.copyMakeBorder(img, top, bottom, left, right,
|
||
cv2.BORDER_CONSTANT, value=(114, 114, 114))
|
||
return img, ratio, dw, dh
|
||
|
||
def _postprocess(self, outputs, orig_h, orig_w, ratio, dw, dh, conf_thres):
|
||
"""YOLOv8 ONNX 输出解码：[1, 84, 8400] → DefectBox列表"""
|
||
pred = outputs[0][0] # [84, 8400]
|
||
pred = pred.T # [8400, 84]
|
||
boxes = pred[:, :4] # cx,cy,w,h
|
||
scores = pred[:, 4:] # [8400, num_classes]
|
||
|
||
class_ids = np.argmax(scores, axis=1)
|
||
confidences = scores[np.arange(len(scores)), class_ids]
|
||
|
||
mask = confidences > conf_thres
|
||
boxes, class_ids, confidences = (
|
||
boxes[mask], class_ids[mask], confidences[mask]
|
||
)
|
||
if len(boxes) == 0:
|
||
return []
|
||
|
||
# cx,cy,w,h → x1,y1,x2,y2
|
||
x1 = boxes[:, 0] - boxes[:, 2] / 2
|
||
y1 = boxes[:, 1] - boxes[:, 3] / 2
|
||
x2 = boxes[:, 0] + boxes[:, 2] / 2
|
||
y2 = boxes[:, 1] + boxes[:, 3] / 2
|
||
|
||
# 去除 letterbox padding，还原到原始坐标
|
||
x1 = np.clip((x1 - dw) / ratio, 0, orig_w)
|
||
y1 = np.clip((y1 - dh) / ratio, 0, orig_h)
|
||
x2 = np.clip((x2 - dw) / ratio, 0, orig_w)
|
||
y2 = np.clip((y2 - dh) / ratio, 0, orig_h)
|
||
|
||
# NMS
|
||
nms_ids = cv2.dnn.NMSBoxes(
|
||
np.stack([x1, y1, x2 - x1, y2 - y1], axis=1).tolist(),
|
||
confidences.tolist(), conf_thres, iou_threshold=0.45
|
||
)
|
||
if len(nms_ids) == 0:
|
||
return []
|
||
|
||
defects = []
|
||
for idx in nms_ids.flatten():
|
||
cid = int(class_ids[idx])
|
||
meta = DEFECT_CLASSES.get(cid, {
|
||
"name": str(cid), "zh": str(cid), "severity": "minor"
|
||
})
|
||
defects.append(DefectBox(
|
||
class_id=cid,
|
||
class_name=meta["name"],
|
||
class_name_zh=meta["zh"],
|
||
confidence=round(float(confidences[idx]), 4),
|
||
severity=meta["severity"],
|
||
box_xyxy=[
|
||
round(float(x1[idx]), 1), round(float(y1[idx]), 1),
|
||
round(float(x2[idx]), 1), round(float(y2[idx]), 1),
|
||
],
|
||
))
|
||
return defects
|
||
|
||
def detect(self, image: np.ndarray, conf: float = 0.45) -> DetectResult:
|
||
orig_h, orig_w = image.shape[:2]
|
||
inp, ratio, dw, dh = self._preprocess(image)
|
||
|
||
t0 = time.perf_counter()
|
||
outputs = self._session.run(None, {self._input_name: inp})
|
||
duration_ms = (time.perf_counter() - t0) * 1000
|
||
|
||
defects = self._postprocess(outputs, orig_h, orig_w, ratio, dw, dh, conf)
|
||
|
||
return DetectResult(
|
||
defects=defects,
|
||
duration_ms=round(duration_ms, 1),
|
||
image_width=orig_w,
|
||
image_height=orig_h,
|
||
model_version=str(self._model_path),
|
||
)
|
||
```
|
||
|
||
---
|
||
|
||
## 4. 模型管理：单例 + 启动加载
|
||
|
||
### 问题根源
|
||
|
||
现有 `inference.py` 的 `_MODEL_CACHE` 是进程级字典，首次请求时才加载模型（懒加载）。对推理服务来说，这会导致第一个请求有数秒延迟，在演示现场是灾难性的。
|
||
|
||
### 解决方案：FastAPI 生命周期钩子
|
||
|
||
```python
|
||
# engine/loader.py
|
||
from typing import Optional
|
||
from .base import BaseInferenceEngine
|
||
from config import settings
|
||
|
||
_engine: Optional[BaseInferenceEngine] = None
|
||
|
||
|
||
def get_engine() -> BaseInferenceEngine:
|
||
if _engine is None:
|
||
raise RuntimeError("推理引擎未初始化，请检查启动日志")
|
||
return _engine
|
||
|
||
|
||
def init_engine() -> BaseInferenceEngine:
|
||
global _engine
|
||
if settings.runtime == "onnx":
|
||
from .onnx_adapter import OnnxRuntimeAdapter
|
||
_engine = OnnxRuntimeAdapter(
|
||
model_path=settings.model_path,
|
||
device=settings.device,
|
||
)
|
||
else:
|
||
from .ultralytics_adapter import UltralyticsAdapter
|
||
_engine = UltralyticsAdapter(weights_path=settings.model_path)
|
||
|
||
_engine.warmup() # 预热，消除第一次推理的延迟
|
||
return _engine
|
||
```
|
||
|
||
```python
|
||
# main.py
|
||
from contextlib import asynccontextmanager
|
||
from fastapi import FastAPI
|
||
from engine.loader import init_engine
|
||
from api.detect import router as detect_router
|
||
from api.health import router as health_router
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
# 启动时加载模型
|
||
engine = init_engine()
|
||
print(f"[mingxi-vision] 引擎就绪: {engine.__class__.__name__}")
|
||
yield
|
||
# 关闭时释放资源（ONNX session会自动GC）
|
||
|
||
|
||
app = FastAPI(
|
||
title="明析推理服务",
|
||
description="PCB缺陷检测推理接口",
|
||
version="1.0.0",
|
||
lifespan=lifespan,
|
||
)
|
||
|
||
app.include_router(detect_router, prefix="/api")
|
||
app.include_router(health_router, prefix="/api")
|
||
```
|
||
|
||
---
|
||
|
||
## 5. 图像接收与前处理
|
||
|
||
### 接收来源
|
||
|
||
mingxi-vision 接受两种图像来源，用同一个端点处理：
|
||
|
||
```python
|
||
# api/detect.py
|
||
import cv2
|
||
import numpy as np
|
||
from fastapi import APIRouter, File, Form, UploadFile, HTTPException
|
||
from fastapi.responses import JSONResponse
|
||
from engine.loader import get_engine
|
||
from engine.annotator import draw_boxes
|
||
from schema import DetectResponse
|
||
import base64, uuid
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.post("/detect", response_model=DetectResponse)
|
||
async def detect(
|
||
image: UploadFile = File(...),
|
||
conf: float = Form(default=0.45, ge=0.01, le=0.99),
|
||
line_id: str = Form(default=""),
|
||
batch_id: str = Form(default=""),
|
||
return_annotated: bool = Form(default=False),
|
||
):
|
||
# 1. 读取图像字节
|
||
raw = await image.read()
|
||
if len(raw) > 20 * 1024 * 1024: # 20MB 上限
|
||
raise HTTPException(status_code=413, detail="图像文件过大（上限20MB）")
|
||
|
||
# 2. 解码为 numpy BGR
|
||
arr = np.frombuffer(raw, dtype=np.uint8)
|
||
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
|
||
if img is None:
|
||
raise HTTPException(status_code=422, detail="无法解码图像，请检查文件格式")
|
||
|
||
# 3. 推理
|
||
engine = get_engine()
|
||
result = engine.detect(img, conf=conf)
|
||
|
||
# 4. 可选：返回标注图
|
||
annotated_b64 = None
|
||
if return_annotated:
|
||
annotated = draw_boxes(img.copy(), result.defects)
|
||
_, buf = cv2.imencode(".jpg", annotated, [cv2.IMWRITE_JPEG_QUALITY, 85])
|
||
annotated_b64 = base64.b64encode(buf.tobytes()).decode()
|
||
|
||
return DetectResponse(
|
||
task_id=str(uuid.uuid4()),
|
||
line_id=line_id,
|
||
batch_id=batch_id,
|
||
duration_ms=result.duration_ms,
|
||
image_width=result.image_width,
|
||
image_height=result.image_height,
|
||
defect_count=result.defect_count,
|
||
max_severity=result.max_severity,
|
||
avg_confidence=result.avg_confidence,
|
||
defects=[d.__dict__ for d in result.defects],
|
||
annotated_image_b64=annotated_b64,
|
||
model_version=result.model_version,
|
||
)
|
||
```
|
||
|
||
### 中文路径问题（继承自现有代码）
|
||
|
||
现有 `inference.py` 已有 `_copy_to_ascii_temp_input()` 解决中文路径导致 OpenCV 无法读图的问题。`mingxi-vision` 改用 **内存读取**（`np.frombuffer` + `cv2.imdecode`），从源头消除路径问题，不再需要这个 workaround。
|
||
|
||
---
|
||
|
||
## 6. 后处理：标注图生成
|
||
|
||
```python
|
||
# engine/annotator.py
|
||
import cv2
|
||
import numpy as np
|
||
from typing import List
|
||
from .base import DefectBox
|
||
|
||
SEVERITY_COLORS = {
|
||
"fatal": (0, 0, 220), # 红（BGR）
|
||
"major": (0, 128, 255), # 橙
|
||
"minor": (0, 215, 255), # 黄
|
||
"rework": (255, 165, 0 ), # 蓝
|
||
"none": (180, 180, 180), # 灰
|
||
}
|
||
|
||
|
||
def draw_boxes(image: np.ndarray, defects: List[DefectBox]) -> np.ndarray:
|
||
for d in defects:
|
||
x1, y1, x2, y2 = [int(v) for v in d.box_xyxy]
|
||
color = SEVERITY_COLORS.get(d.severity, (180, 180, 180))
|
||
cv2.rectangle(image, (x1, y1), (x2, y2), color, 2)
|
||
|
||
label = f"{d.class_name_zh} {d.confidence:.2f}"
|
||
(tw, th), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.55, 1)
|
||
cv2.rectangle(image, (x1, y1 - th - 6), (x1 + tw + 4, y1), color, -1)
|
||
cv2.putText(
|
||
image, label, (x1 + 2, y1 - 4),
|
||
cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA
|
||
)
|
||
return image
|
||
```
|
||
|
||
---
|
||
|
||
## 7. 缺陷类别定义
|
||
|
||
```python
|
||
# engine/defect_classes.py
|
||
# 与 HRIPCB 数据集的 6 类对齐，扩展至 10 类（含真实产线样本迁移学习后）
|
||
|
||
DEFECT_CLASSES = {
|
||
# ── HRIPCB 基础 6 类（公开数据集，开箱即用）──
|
||
0: {"name": "missing_hole", "zh": "缺孔", "severity": "fatal"},
|
||
1: {"name": "mouse_bite", "zh": "鼠咬", "severity": "major"},
|
||
2: {"name": "open_circuit", "zh": "断路", "severity": "fatal"},
|
||
3: {"name": "short_circuit", "zh": "短路", "severity": "fatal"},
|
||
4: {"name": "spur", "zh": "毛刺", "severity": "minor"},
|
||
5: {"name": "spurious_copper", "zh": "余铜", "severity": "major"},
|
||
# ── 扩展类（真实产线样本微调后启用）──
|
||
6: {"name": "oxidation", "zh": "氧化", "severity": "minor"},
|
||
7: {"name": "solder_ball", "zh": "锡珠", "severity": "rework"},
|
||
8: {"name": "scratch", "zh": "划痕", "severity": "minor"},
|
||
9: {"name": "label_error", "zh": "标签错贴","severity": "rework"},
|
||
}
|
||
|
||
SEVERITY_ORDER = ["fatal", "major", "minor", "rework", "none"]
|
||
```
|
||
|
||
---
|
||
|
||
## 8. 配置管理
|
||
|
||
```python
|
||
# config.py
|
||
import os
|
||
from typing import Literal
|
||
from pydantic import BaseSettings # pydantic v1
|
||
|
||
|
||
class Settings(BaseSettings):
|
||
# 运行时选择
|
||
runtime: Literal["ultralytics", "onnx"] = "ultralytics"
|
||
model_path: str = "./models/pcb_defect_v1.pt"
|
||
device: Literal["cuda", "cpu"] = "cuda"
|
||
|
||
# 推理默认参数
|
||
default_conf: float = 0.45
|
||
max_image_size_mb: int = 20
|
||
|
||
# 服务配置
|
||
host: str = "0.0.0.0"
|
||
port: int = 8001
|
||
workers: int = 1 # 推理服务单 worker，GPU 不支持多进程共享
|
||
|
||
class Config:
|
||
env_file = ".env"
|
||
env_file_encoding = "utf-8"
|
||
|
||
|
||
settings = Settings()
|
||
```
|
||
|
||
`.env` 文件示例：
|
||
|
||
```bash
|
||
# 开发阶段（ultralytics .pt）
|
||
RUNTIME=ultralytics
|
||
MODEL_PATH=./models/pcb_defect_v1.pt
|
||
DEVICE=cuda
|
||
|
||
# 生产阶段（ONNX Runtime）
|
||
# RUNTIME=onnx
|
||
# MODEL_PATH=./models/pcb_defect_v1.onnx
|
||
# DEVICE=cuda
|
||
```
|
||
|
||
---
|
||
|
||
## 9. 健康检查与诊断接口
|
||
|
||
```python
|
||
# api/health.py
|
||
import platform
|
||
from fastapi import APIRouter
|
||
from engine.loader import get_engine
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/health")
|
||
def health():
|
||
try:
|
||
engine = get_engine()
|
||
return {
|
||
"status": "ok",
|
||
"runtime": engine.__class__.__name__,
|
||
"platform": platform.system(),
|
||
}
|
||
except RuntimeError as e:
|
||
return {"status": "error", "detail": str(e)}
|
||
|
||
|
||
@router.get("/health/gpu")
|
||
def gpu_info():
|
||
"""开发调试用，确认 GPU 是否被正确使用"""
|
||
info = {"cuda_available": False, "onnx_providers": []}
|
||
try:
|
||
import torch
|
||
info["cuda_available"] = torch.cuda.is_available()
|
||
if torch.cuda.is_available():
|
||
info["gpu_name"] = torch.cuda.get_device_name(0)
|
||
info["vram_total_gb"] = round(
|
||
torch.cuda.get_device_properties(0).total_memory / 1e9, 1
|
||
)
|
||
except ImportError:
|
||
pass
|
||
try:
|
||
import onnxruntime as ort
|
||
info["onnx_providers"] = ort.get_available_providers()
|
||
except ImportError:
|
||
pass
|
||
return info
|
||
```
|
||
|
||
---
|
||
|
||
## 10. 完整目录结构
|
||
|
||
```
|
||
mingxi-vision/
|
||
├── api/
|
||
│ ├── __init__.py
|
||
│ ├── detect.py # POST /api/detect（核心推理接口）
|
||
│ └── health.py # GET /api/health, /api/health/gpu
|
||
├── engine/
|
||
│ ├── __init__.py
|
||
│ ├── base.py # DefectBox, DetectResult, BaseInferenceEngine
|
||
│ ├── loader.py # 单例管理 + init_engine()
|
||
│ ├── ultralytics_adapter.py # .pt 推理（开发阶段）
|
||
│ ├── onnx_adapter.py # .onnx 推理（生产阶段）
|
||
│ ├── annotator.py # 在图像上绘制标注框
|
||
│ └── defect_classes.py # 缺陷类别 + 等级映射表
|
||
├── models/
|
||
│ ├── pcb_defect_v1.pt # ultralytics 训练产物（gitignore）
|
||
│ └── pcb_defect_v1.onnx # 导出的 ONNX 模型（gitignore）
|
||
├── scripts/
|
||
│ ├── export_onnx.py # pt → onnx 导出脚本
|
||
│ └── benchmark.py # 本地推理性能测试
|
||
├── tests/
|
||
│ ├── test_detect_api.py
|
||
│ └── fixtures/ # 测试用 PCB 图片
|
||
├── schema.py # Pydantic 响应模型
|
||
├── config.py # Settings（pydantic BaseSettings）
|
||
├── main.py # FastAPI app + lifespan
|
||
├── requirements.txt
|
||
├── .env.example
|
||
└── README.md
|
||
```
|
||
|
||
---
|
||
|
||
## 11. 响应 Schema
|
||
|
||
```python
|
||
# schema.py
|
||
from typing import List, Optional
|
||
from pydantic import BaseModel
|
||
|
||
|
||
class DefectItem(BaseModel):
|
||
class_id: int
|
||
class_name: str
|
||
class_name_zh: str
|
||
confidence: float
|
||
severity: str # fatal / major / minor / rework / none
|
||
box_xyxy: List[float] # [x1, y1, x2, y2]
|
||
|
||
|
||
class DetectResponse(BaseModel):
|
||
task_id: str
|
||
line_id: str
|
||
batch_id: str
|
||
duration_ms: float
|
||
image_width: int
|
||
image_height: int
|
||
defect_count: int
|
||
max_severity: str # 本次检测中最高等级缺陷
|
||
avg_confidence: Optional[float]
|
||
defects: List[DefectItem]
|
||
annotated_image_b64: Optional[str] # 仅 return_annotated=true 时有值
|
||
model_version: str
|
||
```
|
||
|
||
实际响应示例：
|
||
|
||
```json
|
||
{
|
||
"task_id": "a3f1c2d4-...",
|
||
"line_id": "LINE-01",
|
||
"batch_id": "BAT-20260524-001",
|
||
"duration_ms": 41.3,
|
||
"image_width": 3072,
|
||
"image_height": 2048,
|
||
"defect_count": 2,
|
||
"max_severity": "fatal",
|
||
"avg_confidence": 0.8762,
|
||
"defects": [
|
||
{
|
||
"class_id": 2,
|
||
"class_name": "open_circuit",
|
||
"class_name_zh": "断路",
|
||
"confidence": 0.9134,
|
||
"severity": "fatal",
|
||
"box_xyxy": [234.5, 891.2, 312.8, 943.7]
|
||
},
|
||
{
|
||
"class_id": 4,
|
||
"class_name": "spur",
|
||
"class_name_zh": "毛刺",
|
||
"confidence": 0.8390,
|
||
"severity": "minor",
|
||
"box_xyxy": [1204.1, 456.3, 1251.9, 489.0]
|
||
}
|
||
],
|
||
"annotated_image_b64": null,
|
||
"model_version": "./models/pcb_defect_v1.pt"
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## 12. 性能分析（Y7000P RTX 4060）
|
||
|
||
### 推理延迟预估
|
||
|
||
| 运行时 | 模型 | 分辨率 | 单次推理 | 含预处理+后处理 |
|
||
|--------|------|--------|---------|--------------|
|
||
| ultralytics | YOLOv8n | 640×640 | ~12ms | ~20ms |
|
||
| ultralytics | YOLOv8s | 640×640 | ~20ms | ~30ms |
|
||
| ONNX Runtime GPU | YOLOv8n | 640×640 | ~8ms | ~15ms |
|
||
| ONNX Runtime CPU | YOLOv8n | 640×640 | ~80ms | ~100ms |
|
||
|
||
> 数据来源：`pcb缺陷检测初步方案.md` 引用的 Y7000P 实测基准
|
||
|
||
**PCB 后道复判场景需求**：触发模式，每块板约 1-3 秒间隔，15ms 的推理延迟完全满足。
|
||
|
||
### 显存占用
|
||
|
||
| 模型 | 显存占用 |
|
||
|------|---------|
|
||
| YOLOv8n（ultralytics） | ~1.2GB |
|
||
| YOLOv8s（ultralytics） | ~2.4GB |
|
||
| YOLOv8n（ONNX CUDA EP） | ~0.8GB |
|
||
|
||
RTX 4060 8GB 剩余 6GB+ 可用，不构成瓶颈。
|
||
|
||
### 并发限制
|
||
|
||
```python
|
||
# main.py 启动参数
|
||
# workers=1：GPU 不支持多进程共享同一 CUDA Context
|
||
# 推理天然串行，配合 FastAPI 的 async 处理并发等待队列
|
||
uvicorn main:app --host 0.0.0.0 --port 8001 --workers 1
|
||
```
|
||
|
||
PCB 产线触发模式下，单工位请求天然串行，`workers=1` 完全够用。若需多产线并发，在 `mingxi-backend` 侧做请求队列即可。
|
||
|
||
---
|
||
|
||
## 13. 模型导出（pt → onnx）
|
||
|
||
```python
|
||
# scripts/export_onnx.py
|
||
from ultralytics import YOLO
|
||
|
||
model = YOLO("./models/pcb_defect_v1.pt")
|
||
model.export(
|
||
format="onnx",
|
||
imgsz=640,
|
||
opset=12, # onnxruntime 1.16 兼容
|
||
simplify=True, # onnx-simplifier 优化计算图
|
||
dynamic=False, # 固定 batch=1，推理服务不需要动态 batch
|
||
)
|
||
# 产物：pcb_defect_v1.onnx
|
||
```
|
||
|
||
切换到 ONNX 的时机：
|
||
|
||
```
|
||
1. YOLOv8n 在验证集 mAP@0.5 > 0.90 → 导出 ONNX
|
||
2. 运行 scripts/benchmark.py 对比两个运行时延迟
|
||
3. 确认输出一致后，修改 .env 的 RUNTIME=onnx
|
||
4. 重启服务，观察 /api/health 确认切换成功
|
||
```
|
||
|
||
---
|
||
|
||
## 14. 与现有代码的对应关系
|
||
|
||
| 现有 `inference.py` 函数 | mingxi-vision 对应位置 | 变化 |
|
||
|------------------------|----------------------|------|
|
||
| `_MODEL_CACHE` | `engine/loader.py` 单例 | 从懒加载改为启动时加载 |
|
||
| `_get_model()` | `loader.init_engine()` | 新增 ONNX 分支 |
|
||
| `_copy_to_ascii_temp_input()` | **已删除** | 改用内存读图，无需文件路径 |
|
||
| `_parse_result()` | `ultralytics_adapter.detect()` | 增加 severity 映射 |
|
||
| `run_image_inference()` | `engine/ultralytics_adapter.py` + `engine/onnx_adapter.py` | 拆成两个运行时 |
|
||
| `run_camera_inference()` | **不迁移** | 由 `mingxi-capture` 负责 |
|
||
| `run_video_inference()` | **暂不迁移** | 此版本不需要 |
|
||
| Django `MEDIA_ROOT` 文件落盘 | **已删除** | 推理服务无状态，不落盘 |
|
||
|
||
---
|
||
|
||
## 15. 关键设计决策汇总
|
||
|
||
| 决策 | 选择 | 理由 |
|
||
|------|------|------|
|
||
| Web 框架 | FastAPI | 轻量、无需 DB、原生 async、自动文档 |
|
||
| 模型运行时 | 双运行时（适配器模式） | 开发用 .pt 方便迭代，生产用 .onnx 稳定 |
|
||
| 模型加载时机 | 启动时加载 + warmup | 消除首次推理延迟，路演不卡顿 |
|
||
| 图像读取 | 内存读取（np.frombuffer） | 彻底消除中文路径问题 |
|
||
| 标注图返回 | 按需（return_annotated=true） | 默认不返回，减少响应体积和编码开销 |
|
||
| 并发模型 | workers=1，单进程 | GPU CUDA Context 不支持多进程共享 |
|
||
| 鉴权 | 无 | 内网服务，由 backend 代理调用 |
|
||
| 数据持久化 | 无 | 无状态推理服务，落库交给 backend |
|
||
| NMS 后处理 | ONNX 适配器自实现，ultralytics 自带 | 两路输出语义一致 |
|