💡前言

前一篇已经完成 QAIRT/QNN 的基础环境搭建与板端推理跑通验证。本篇将在该基础上，把“能跑通”推进到“可长期运行”：实现采集、推理、输出三段解耦的流水线结构，并落地事件截图与结构化结果落盘/上报。

本篇的工作重点从“提升峰值 FPS”转向“保证长期稳定运行”。工业现场更常遇到的问题主要包括：

• 延迟越来越大（堆积导致实时性崩溃）
• 输出写盘/上报阻塞主链路
• 运行几小时后卡死/内存涨/偶发大抖动

因此本文的目标是构建一个可长期运行、不卡死、不出现延迟累计的工程骨架：

• latest-only：采集侧永远只保留最新帧（宁可丢帧，不可堆积）
• 三线程流水线：采集、推理、输出解耦
• 输出可降级：事件截图限频；上报失败不阻塞

推理实现这块我先用 ONNXRuntime（CPU）兜底（好装、好排错）。你如果已经跑通了 QNN/HTP，就把 infer() 那一层换掉就行，工程结构不用动。

1. 目录结构

defect_demo/
 ├─ main.py
 ├─ capture.py
 ├─ infer_onnx.py
 ├─ postprocess.py
 ├─ configs.py
 ├─ requirements.txt
 ├─ models/
 │   └─ defect.onnx
 └─ runs/
     ├─ events/
     └─ logs/

requirements.txt：

numpy
opencv-python-headless
onnxruntime
requests

---

2. 设计要点 (latest-only)

工业实时系统最常见的“慢性死亡”就是堆积：队列越堆越多，延迟越积越大，最后表面上程序还活着，但输出其实已经“落后现实好几秒”。这比丢帧更要命。

因此本文采用的策略是：采集层仅保留最新一帧（新帧到来覆盖旧帧），并通过 drop 计数记录丢帧情况，用于运行状态观测。

下方流程图更直观地展示了本文的核心设计——三线程流水线 + latest-only 机制。

3. 关键代码：采集层（LatestFrame + 采集线程）

保存为 capture.py：

import time
import threading
from dataclasses import dataclass
from typing import Optional

import cv2
import numpy as np


@dataclass
class FramePacket:
    img_bgr: np.ndarray
    t_capture_ns: int


class LatestFrame:
    def __init__(self):
        self._lock = threading.Lock()
        self._pkt: Optional[FramePacket] = None
        self.drop = 0

    def put(self, pkt: FramePacket):
        with self._lock:
            if self._pkt is not None:
                self.drop += 1
            self._pkt = pkt

    def get(self) -> Optional[FramePacket]:
        with self._lock:
            pkt = self._pkt
            self._pkt = None
            return pkt


class OpenCVCapture(threading.Thread):
    def __init__(self, latest: LatestFrame, device="/dev/video0",
                 width=1280, height=720, fps=30):
        super().__init__(daemon=True)
        self.latest = latest
        self.device = device
        self.width = width
        self.height = height
        self.fps = fps
        self._stop = threading.Event()

    def stop(self):
        self._stop.set()

    def run(self):
        cap = cv2.VideoCapture(self.device)
        if not cap.isOpened():
            raise RuntimeError(f"Cannot open camera: {self.device}")

        cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(self.width))
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(self.height))
        cap.set(cv2.CAP_PROP_FPS, float(self.fps))

        while not self._stop.is_set():
            ok, frame = cap.read()
            if not ok:
                time.sleep(0.01)
                continue
            self.latest.put(FramePacket(frame, time.monotonic_ns()))

        cap.release()

4. 关键代码：推理层（ONNX版本，后续可替换QNN）

infer_onnx.py：核心接口infer(frame)->(dets, infer_ms)

import time
from dataclasses import dataclass
from typing import Dict, List, Tuple

import cv2
import numpy as np
import onnxruntime as ort

from postprocess import postprocess_yolov8


@dataclass
class Det:
    name: str
    score: float
    bbox: Tuple[int, int, int, int]


def letterbox(im, new_shape=640, color=(114, 114, 114)):
    h, w = im.shape[:2]
    if isinstance(new_shape, int):
        new_shape = (new_shape, new_shape)
    r = min(new_shape[0] / h, new_shape[1] / w)
    nh, nw = int(round(h * r)), int(round(w * r))
    im_resized = cv2.resize(im, (nw, nh), interpolation=cv2.INTER_LINEAR)

    top = (new_shape[0] - nh) // 2
    bottom = new_shape[0] - nh - top
    left = (new_shape[1] - nw) // 2
    right = new_shape[1] - nw - left

    im_padded = cv2.copyMakeBorder(
        im_resized, top, bottom, left, right,
        cv2.BORDER_CONSTANT, value=color
    )
    return im_padded, r, (left, top)


class OnnxDetector:
    def __init__(self, model_path: str, img_size=640, conf=0.35, iou=0.45,
                 class_map: Dict[int, str] = None):
        self.img_size = img_size
        self.conf = conf
        self.iou = iou
        self.class_map = class_map or {}
        self.sess = ort.InferenceSession(model_path,
                                         providers=["CPUExecutionProvider"])
        self.in_name = self.sess.get_inputs()[0].name

    def infer(self, frame_bgr: np.ndarray) -> Tuple[List[Det], float]:
        img, r, (padw, padh) = letterbox(frame_bgr, self.img_size)
        rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
        inp = rgb.astype(np.float32) / 255.0
        inp = np.transpose(inp, (2, 0, 1))[None, ...]

        t1 = time.time()
        outputs = self.sess.run(None, {self.in_name: inp})
        boxes, scores, cls_ids = postprocess_yolov8(outputs, self.conf, self.iou)
        t2 = time.time()

        dets: List[Det] = []
        if boxes.shape[0] > 0:
            boxes[:, [0, 2]] -= padw
            boxes[:, [1, 3]] -= padh
            boxes /= r
            boxes[:, [0, 2]] = np.clip(boxes[:, [0, 2]], 0, frame_bgr.shape[1] - 1)
            boxes[:, [1, 3]] = np.clip(boxes[:, [1, 3]], 0, frame_bgr.shape[0] - 1)
            for (x1, y1, x2, y2), sc, cid in zip(boxes, scores, cls_ids):
                name = self.class_map.get(int(cid), f"class_{int(cid)}")
                dets.append(Det(
                    name=name,
                    score=float(sc),
                    bbox=(int(x1), int(y1), int(x2), int(y2))
                ))

        return dets, (t2 - t1) * 1000.0

postprocess.py：

import numpy as np


def xywh2xyxy(x):
    y = x.copy()
    y[..., 0] = x[..., 0] - x[..., 2] / 2
    y[..., 1] = x[..., 1] - x[..., 3] / 2
    y[..., 2] = x[..., 0] + x[..., 2] / 2
    y[..., 3] = x[..., 1] + x[..., 3] / 2
    return y


def nms(boxes, scores, iou_thres):
    x1, y1, x2, y2 = boxes.T
    areas = (x2 - x1) * (y2 - y1)
    order = scores.argsort()[::-1]
    keep = []
    while order.size > 0:
        i = order[0]
        keep.append(i)
        if order.size == 1:
            break
        xx1 = np.maximum(x1[i], x1[order[1:]])
        yy1 = np.maximum(y1[i], y1[order[1:]])
        xx2 = np.minimum(x2[i], x2[order[1:]])
        yy2 = np.minimum(y2[i], y2[order[1:]])
        w = np.maximum(0.0, xx2 - xx1)
        h = np.maximum(0.0, yy2 - yy1)
        inter = w * h
        iou = inter / (areas[i] + areas[order[1:]] - inter + 1e-9)
        inds = np.where(iou <= iou_thres)[0]
        order = order[inds + 1]
    return keep


def postprocess_yolov8(outputs, conf_thres=0.35, iou_thres=0.45):
    out = outputs[0]
    out = np.squeeze(out, axis=0)
    if out.shape[0] < out.shape[1]:
        out = out.T
    boxes = out[:, :4]
    cls_scores = out[:, 4:]
    cls_conf = cls_scores.max(axis=1)
    cls_id = cls_scores.argmax(axis=1)

    mask = cls_conf >= conf_thres
    boxes, cls_conf, cls_id = boxes[mask], cls_conf[mask], cls_id[mask]

    if boxes.shape[0] == 0:
        return np.empty((0, 4)), np.empty((0,)), np.empty((0,), dtype=np.int32)

    boxes = xywh2xyxy(boxes)
    keep = nms(boxes, cls_conf, iou_thres)
    return boxes[keep], cls_conf[keep], cls_id[keep].astype(np.int32)

5. 关键代码：输出层（事件截图 + JSONL + HTTP 上报开关）

configs.py（现场要改参数，就改这里）：

from pathlib import Path

MODEL_PATH = "models/defect.onnx"

CAM_DEVICE = "/dev/video0"
CAM_W = 1280
CAM_H = 720
CAM_FPS = 30

IMG_SIZE = 640
CONF_THRES = 0.35
IOU_THRES = 0.45

EVENT_THRES = 0.60
EVENT_MIN_INTERVAL_SEC = 0.5

RUNS_DIR = Path("runs")
EVENT_DIR = RUNS_DIR / "events"
LOG_DIR = RUNS_DIR / "logs"
JSONL_PATH = LOG_DIR / "result.jsonl"

HTTP_REPORT_ENABLE = False   # 先关掉，跑通后再开（避免网络问题干扰主链路）
HTTP_ENDPOINT = "http://127.0.0.1:8080/report"
DEVICE_ID = "qcs9100-001"
CAMERA_ID = "cam0"

main.py（三线程骨架）：

import json
import time
import threading
from queue import Queue, Empty

import cv2
import requests

from capture import LatestFrame, OpenCVCapture
from infer_onnx import OnnxDetector
import configs as C


def ensure_dirs():
    C.RUNS_DIR.mkdir(parents=True, exist_ok=True)
    C.EVENT_DIR.mkdir(parents=True, exist_ok=True)
    C.LOG_DIR.mkdir(parents=True, exist_ok=True)


def draw(frame, dets, infer_ms, fps_est):
    vis = frame.copy()
    for d in dets:
        x1, y1, x2, y2 = d.bbox
        cv2.rectangle(vis, (x1, y1), (x2, y2), (0, 255, 0), 2)
        cv2.putText(vis, f"{d.name} {d.score:.2f}", (x1, max(0, y1 - 6)),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
    cv2.putText(vis, f"infer={infer_ms:.1f}ms fps~{fps_est:.1f}", (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 0, 0), 2)
    return vis


def main():
    ensure_dirs()

    latest = LatestFrame()
    cap_th = OpenCVCapture(latest, C.CAM_DEVICE, C.CAM_W, C.CAM_H, C.CAM_FPS)

    detector = OnnxDetector(
        model_path=C.MODEL_PATH,
        img_size=C.IMG_SIZE,
        conf=C.CONF_THRES,
        iou=C.IOU_THRES,
        class_map={}
    )

    out_q: Queue = Queue(maxsize=2)
    stop = threading.Event()

    infer_count = 0
    infer_fps_win = []
    last_event_t = 0.0

    def infer_thread():
        nonlocal infer_count, infer_fps_win
        while not stop.is_set():
            pkt = latest.get()
            if pkt is None:
                time.sleep(0.001)
                continue

            t1 = time.time()
            dets, infer_ms = detector.infer(pkt.img_bgr)
            t2 = time.time()

            fps = 1.0 / max(1e-6, (t2 - t1))
            infer_fps_win.append(fps)
            if len(infer_fps_win) > 30:
                infer_fps_win.pop(0)

            payload = {
                "ts_ms": int(time.time() * 1000),
                "device": C.DEVICE_ID,
                "camera": C.CAMERA_ID,
                "perf": {
                    "infer_ms": round(infer_ms, 3),
                    "infer_fps": round(sum(infer_fps_win) / len(infer_fps_win), 2),
                    "drop": latest.drop,
                },
                "defects": [
                    {"name": d.name, "score": round(d.score, 4), "bbox": list(d.bbox)}
                    for d in dets
                ],
                "capture": {"t_capture_ns": pkt.t_capture_ns},
            }
            try:
                if out_q.full():
                    out_q.get_nowait()
                out_q.put_nowait((pkt.img_bgr, dets, infer_ms, payload))
            except Exception:
                pass
            infer_count += 1

    def output_thread():
        nonlocal last_event_t
        while not stop.is_set():
            try:
                frame, dets, infer_ms, payload = out_q.get(timeout=0.2)
            except Empty:
                continue

            with open(C.JSONL_PATH, "a", encoding="utf-8") as f:
                f.write(json.dumps(payload, ensure_ascii=False) + "\n")

            now = time.time()
            is_event = any(d["score"] >= C.EVENT_THRES for d in payload["defects"])
            if is_event and (now - last_event_t) >= C.EVENT_MIN_INTERVAL_SEC:
                last_event_t = now
                fps_est = payload["perf"]["infer_fps"]
                vis = draw(frame, dets, infer_ms, fps_est)

                out_path = C.EVENT_DIR / f"event_{payload['ts_ms']}.jpg"
                cv2.imwrite(str(out_path), vis)

                payload["event_image"] = str(out_path)
                if C.HTTP_REPORT_ENABLE:
                    try:
                        requests.post(C.HTTP_ENDPOINT, json=payload, timeout=1.0)
                    except Exception:
                        pass

    cap_th.start()
    th1 = threading.Thread(target=infer_thread, daemon=True)
    th2 = threading.Thread(target=output_thread, daemon=True)
    th1.start()
    th2.start()
    print("defect demo started. Ctrl+C to stop.")
    try:
        while True:
            time.sleep(5)
            print(f"infer_count={infer_count} drop={latest.drop}")
    except KeyboardInterrupt:
        pass
    finally:
        stop.set()
        cap_th.stop()
        print("stopping...")


if __name__ == "__main__":
    main()

运行：

python3 -m pip install -r requirements.txt
python3 main.py

你会得到：

• runs/logs/result.jsonl：持续追加结构化结果
• runs/events/*.jpg：事件触发截图（限频）

6. systemd 服务化部署

保存为 /etc/systemd/system/defectd.service：

[Unit]
Description=QCS9100 defect detection daemon
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=YOUR_USER
WorkingDirectory=/home/YOUR_USER/defect_demo
ExecStart=/usr/bin/python3 /home/YOUR_USER/defect_demo/main.py
Restart=always
RestartSec=2
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target

启用：

sudo systemctl daemon-reload
sudo systemctl enable defectd
sudo systemctl start defectd
journalctl -u defectd -f

7. 常见问题

• 延迟越跑越大：通常是未启用 latest-only 或队列无界。处理方式：采集覆盖旧帧、队列限长、输出可降级。
• 写盘拖慢：别每帧落图，只留事件截图，而且要限频；普通帧留在内存里就好。
• 上报把主流程卡住：HTTP 必须短超时，失败就当没发生（日志里记一下就行）。
• CPU 被颜色转换吃满：将取流切换为 GStreamer app sink + NV12 路线，减少不必要的转换与拷贝。

📢下一篇介绍

**下一篇**将在该骨架上完成两项工作：通过 profiling 拆解端到端耗时并定位瓶颈；基于 QNN/HTP 推理链路完成 INT8 量化与精度回归。