用抽帧与单槽队列降低边缘视觉延迟

边缘设备跑 YOLOv8，未必需要把摄像头的每一帧都送进模型。对于植物健康、工业仪表这类变化较慢的对象，逐帧推理可能先耗尽 CPU/GPU，检测结果却没有增加多少价值。作者给出的方案是：限制推理频率，让采集和推理分开运行，只保留一个待处理帧。

这篇最有用的是一个可直接阅读和改造的 Python 包装类。但它没有提供性能对照，代码也没有完整实现文中提到的监控指标。理解它解决了什么，比照搬更重要。

为什么队列只留一帧

如果在摄像头采集循环里直接调用模型，推理耗时会拖住下一次采集。如果改成持续入队，模型处理速度又跟不上，队列里的画面就会越来越旧。

作者使用 queue.Queue(maxsize=1)，将采集与模型执行解耦：新帧到来时，先移除还没处理的旧帧，再放入新帧。后台线程独立取帧推理。

这套策略优先保证画面的新鲜度，代价是主动丢帧。

这里的「最新」有边界：它更新的是等待处理的帧，已经送进模型的画面不会被替换。对于缓慢变化的监测对象，这通常符合需求；如果任务要求保留每个瞬间，就不能直接采用这个前提。

默认每秒两次推理，代码如何控制节奏

下面这段代码同时解决待处理帧堆积和推理频率控制，保留原实现：

import threading
import queue
import time

class EdgeInferencePipeline:
    def __init__(self, model, target_fps: float = 2.0):
        self.model = model
        self.interval = 1.0 / target_fps
        self.frame_queue = queue.Queue(maxsize=1)
        self.running = False

    def push_frame(self, frame):
        if self.frame_queue.full():
            try:
                self.frame_queue.get_nowait()
            except queue.Empty:
                pass
        self.frame_queue.put(frame)

    def _worker(self):
        while self.running:
            start_time = time.perf_counter()
            try:
                frame = self.frame_queue.get(timeout=1.0)
                results = self.model.predict(frame, verbose=False)
                self.handle_detections(results)
            except queue.Empty:
                continue

            elapsed = time.perf_counter() - start_time
            sleep_time = max(0.0, self.interval - elapsed)
            time.sleep(sleep_time)

    def handle_detections(self, results):
        for r in results:
            boxes = r.boxes
            print(f"Detected {len(boxes)} objects at {time.strftime('%X')}")

    def start(self):
        self.running = True
        threading.Thread(target=self._worker, daemon=True).start()

    def stop(self):
        self.running = False

重点看 sleep_time = max(0.0, self.interval - elapsed)：默认 target_fps=2.0，对应每轮目标间隔 0.5 秒，处理后只补足剩余时间。如果处理已经超时，就不再休眠，它无法强行让慢模型达到每秒两次。

还有一个容易忽略的细节：计时从取帧之前开始，因此 elapsed 包含等待队列、模型推理和结果处理，并非纯推理耗时。get(timeout=1.0) 则让空队列最多等待 1.0 秒，随后捕获 queue.Empty 并重试。

作者还给出独立的 measure_inference，用来测量一次模型调用耗时：

def measure_inference(model, sample_image):
    t0 = time.perf_counter()
    _ = model.predict(sample_image, verbose=False)
    latency_ms = (time.perf_counter() - t0) * 1000
    print(f"Inference latency: {latency_ms:.2f} ms")
    return latency_ms

注意乘以 1000 的那一行，它将秒转换成毫秒。但这个函数只返回单次耗时，并没有计算平均值，也没有统计丢帧率。

我的判断

我认同「慢变化场景不必逐帧推理」这个取舍。相较于不断增加缓存，容量为 1 的队列能直接限制待处理帧数量；相较于在采集循环里同步推理，后台线程能让两项工作独立推进。

但这是一份实现思路，没有硬件配置、输入帧率、实际延迟或资源占用的对照数据，不能据此认定优化幅度。

作者还说解耦可以防止内存泄漏，这个表述过强。代码能抑制队列积压造成的内存增长，无法证明模型、摄像头或其他资源不存在泄漏。

「非阻塞」也需要限定：full()、移除、put() 是分开的操作，多生产者并发时，最终的 put() 仍可能等待。stop() 只修改标志，没有等待线程退出；start() 也没有防止重复启动。它适合作为原型，距离完整的生产服务还有生命周期和监控工作要补。

能拿来做什么

• 在植物或仪表监测原型中，把摄像头帧交给 push_frame()，先确认业务允许漏过哪些变化，再调整 target_fps。
• 用 measure_inference() 收集多次调用耗时，补上平均值和被替换帧计数，验证限频是否真正降低了设备负担。
• 如果由前端或 Node 服务控制这条 Python 管线，先补齐重复启动保护和停止等待逻辑，让页面上的运行状态能够对应实际线程状态。