边缘端 YOLO 推理——SSIM 变化检测 + bbox 签名去重，别把相似的帧都上传了

巡检车上的 YOLO 推理节点跑在树莓派上，每秒处理 2 帧。检测到病虫害就往后端上传图片。问题在于：车停着不动的时候，相机拍到的画面几乎一样，YOLO 检测到的框也几乎一样——但每帧都上传，后端收到的全是重复图片。

一帧 JPEG 85 质量大约 200KB，每秒 2 帧 = 400KB/s。果园 WiFi 带宽有限，YOLO 上传把其他数据（心跳、传感器、地图）全挤掉了。

解法：两重变化检测——SSIM 图像相似度 + bbox 签名去重。相似的帧不传。

SSIM 变化检测：20 行实现结构相似度

SSIM（Structural Similarity Index）衡量两张图的相似程度，范围 0~1，1 表示完全相同。阈值设为 0.985——低于这个值才认为”画面变了”。

关键是：不拿原图算 SSIM，先把图缩到 160×90 灰度再算。640×400 的 RGB 图算 SSIM 大概 20ms，160×90 灰度图只要 0.5ms——在树莓派上差别很大。

def _to_change_gray(self, frame):
    small = cv2.resize(frame, (160, 90), interpolation=cv2.INTER_AREA)
    return cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)

SSIM 的计算：

@staticmethod
def _compute_ssim(prev_gray, curr_gray):
    a = prev_gray.astype(np.float32)
    b = curr_gray.astype(np.float32)
    c1 = (0.01 * 255) ** 2
    c2 = (0.03 * 255) ** 2

    mu_a = cv2.GaussianBlur(a, (11, 11), 1.5)
    mu_b = cv2.GaussianBlur(b, (11, 11), 1.5)
    mu_a2 = mu_a * mu_a
    mu_b2 = mu_b * mu_b
    mu_ab = mu_a * mu_b

    sigma_a2 = cv2.GaussianBlur(a * a, (11, 11), 1.5) - mu_a2
    sigma_b2 = cv2.GaussianBlur(b * b, (11, 11), 1.5) - mu_b2
    sigma_ab = cv2.GaussianBlur(a * b, (11, 11), 1.5) - mu_ab

    num = (2 * mu_ab + c1) * (2 * sigma_ab + c2)
    den = (mu_a2 + mu_b2 + c1) * (sigma_a2 + sigma_b2 + c2)
    den = np.where(den == 0.0, 1e-6, den)
    ssim_map = num / den
    return float(np.mean(ssim_map))

这是 SSIM 的标准公式，用 GaussianBlur 替代显式窗口卷积——OpenCV 的 GaussianBlur 底层做了优化，比手写卷积快得多。11×11 高斯核、σ=1.5 是论文推荐值。

整个计算没有调用 skimage 或其他重型库，纯 NumPy + OpenCV，部署简单。

bbox 签名去重：框没变就不传

SSIM 检测画面变化，但有时候画面变了框没变——车稍微挪了一下，SSIM 降到了 0.98，但 YOLO 检测到的病虫害位置完全一样。这种帧传上去也没用。

bbox 签名：把检测框的位置归一化后编码成 tuple，跟上次上传的签名比对。

@staticmethod
def _build_detection_signature(detections, frame_w, frame_h):
    inv_w = 1.0 / float(max(1, frame_w))
    inv_h = 1.0 / float(max(1, frame_h))
    signature = []
    for det in detections:
        x1, y1, x2, y2 = det.box_xyxy
        signature.append((
            int(det.class_id),
            round(float(det.confidence), 2),
            round(float(x1) * inv_w, 3),
            round(float(y1) * inv_h, 3),
            round(float(x2) * inv_w, 3),
            round(float(y2) * inv_h, 3),
        ))
    signature.sort()
    return tuple(signature)

归一化到 [0, 1] 范围是因为：同一场景在不同分辨率下检测，框的像素坐标不同但归一化位置相同。round(..., 3) 精度是千分之几像素，足够区分不同位置但容许微小抖动。

双重判定逻辑

def _should_upload_changed(self, frame, detections):
    curr_gray = self._to_change_gray(frame)
    curr_sig = self._build_detection_signature(detections, frame_w, frame_h)

    if self._last_uploaded_gray_small is None:
        return True, curr_gray, curr_sig  # 第一帧，必须传

    boxes_changed = curr_sig != self._last_uploaded_signature
    if self.upload_change_method == 'boxes':
        return boxes_changed, curr_gray, curr_sig  # 纯框签名模式

    ssim = self._compute_ssim(self._last_uploaded_gray_small, curr_gray)
    changed = boxes_changed or (ssim < 0.985)
    return changed, curr_gray, curr_sig

两种模式可选：

• boxes 模式：只在检测框变化时上传。适合检测目标很少、位置变化明显的场景
• ssim 模式（默认）：框变了或画面变了都上传。覆盖面更广，适合画面变化频繁的场景

两个条件是 OR 关系：只要有一个变化就传。宁可多传几张，也不能漏掉真正的新检测结果。

上传队列：满了丢旧的

变化检测之后，符合条件的帧进入上传队列。队列 maxsize=32，单独线程消费：

def _enqueue_upload(self, payload):
    try:
        self._upload_queue.put_nowait(payload)
        return
    except queue.Full:
        pass

    if self._upload_drop_oldest:
        try:
            self._upload_queue.get_nowait()  # 丢弃最旧的
            self._upload_queue.task_done()
        except queue.Empty:
            pass
        try:
            self._upload_queue.put_nowait(payload)
        except queue.Full:
            pass

upload_queue_drop_oldest=True：队列满了丢弃最旧的帧。为什么不丢最新的？因为最新帧包含最新的位置信息——位置对了才能在地图上正确标注病虫害。旧帧的位置信息已经过时了。

模型热替换：换模型不重启

后端可以通过运行时配置下发新的 ONNX 模型 URL，推理节点下载并替换模型，不用重启 Docker：

def _download_file(self, uri, target_path, label):
    temp = target.with_suffix(target.suffix + '.tmp')
    resp = requests.get(uri, timeout=20)
    temp.write_bytes(resp.content)
    temp.replace(target)  # 原子替换
    return True

tempfile + os.replace 保证原子性——下载中途断电，.tmp 文件是不完整的，但原始模型文件不受影响。下次启动用旧模型继续跑。

下载完后重建 ONNX session：

def _rebuild_engine(self, model_path, conf_threshold, iou_threshold):
    engine = YoloOnnxEngine(model_path=model_path, ...)
    self.engine = engine

_rebuild_engine 在 _on_image 之外被调用，不会中断正在执行的推理。下一次 _on_image 回调时自动用新模型。

实际效果

指标	无变化检测	SSIM + 签名去重
上传帧率（静止时）	2 fps	~0.1 fps
上传帧率（移动时）	2 fps	~1.5 fps
带宽占用	~400 KB/s	~30 KB/s（静止时）
漏检率	0	≈0（OR 逻辑兜底）

静止时带宽从 400KB/s 降到 30KB/s——降了 92%。心跳和传感器数据再也不被挤掉了。

0.985 的 SSIM 阈值是调出来的：0.99 太敏感（光照微变就传），0.97 太迟钝（病虫害移出画面了还不传）。0.985 在树莓派上的实际测试效果最好。