chore: 初始化四个子项目骨架目录结构
- 新增 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>
This commit is contained in:
Claude Sonnet 4.6
@@ -0,0 +1,826 @@
|
||||
# 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 自带 | 两路输出语义一致 |
|
||||
Reference in New Issue
Block a user