video_yolov4_tiny

基于 PNNA NPU 的 YOLOv4-tiny 实时视频目标检测系统。通过 TCP 与 NPU 设备通信，将 YOLOv4-tiny 模型推理卸载到专用硬件上执行。上位机负责视频读取、图像预处理、后处理（DIoU NMS）和结果可视化，采用 3 线程流水线架构将预处理、NPU 通信（发送数据并等待 NPU 完成推理）、后处理并行化，提升吞吐量。支持 NPU 断线自动重连、指数退避重试、实时 FPS 显示和键盘交互控制。

流程图

flowchart TD
    A["视频采集"] --> B["pre_q"]
    B --> C["线程1: 预处理
resize 416×416×3"]
    C --> D["infer_q"]
    D --> E["线程2: 发送数据至NPU
等待返回结果"]
    E --> F["post_q"]
    F --> G["线程3: 后处理 + 画框
13×13×255 + 26×26×255"]
    G --> H["display_q"]
    H --> I["主线程: 显示"]
    E -->|断线| J["重连循环"]
    J --> E

架构

graph LR
    A[主线程
I/O + 显示] -->|pre_q| B[线程1
预处理]
    B -->|infer_q| C[线程2
NPU通信]
    C -->|post_q| D[线程3
后处理 + 画框]
    D -->|display_q| A

队列设计

所有队列使用 queue.Queue(maxsize=N) 限制内存：

队列	容量	用途
pre_q	2	主线程 → 预处理线程
infer_q	2	预处理线程 → NPU通信线程
post_q	2	NPU通信线程 → 后处理线程
display_q	4	后处理线程 → 主线程显示

线程职责

线程1 — 预处理 (_preprocess_worker)

• 从 pre_q 取原始帧
• resize 到 416×416 → BGR→RGB → HWC→CHW → 扁平化为 uint8 字节
• 保留原始分辨率帧，送入 infer_q

线程2 — NPU通信 (_infer_worker)

• 从 infer_q 取预处理数据
• 调用 PNNAClient.infer() 发送到 NPU 并阻塞等待结果
• 唯一访问 socket 的线程，天然串行化 NPU 访问（NPU 不支持流水线）
• 检测到断连时，触发协调关闭所有工作线程，通知主线程重连

线程3 — 后处理 + 画框 (_postprocess_worker)

• 从 post_q 取 NPU 原始输出
• 解码 uint8 张量 → YOLO 后处理（含 DIoU NMS）→ 检测结果
• 在原始分辨率图像上绘制边界框和标签
• 送入 display_q

主线程

• 打开视频 → 连接 NPU → 启动 3 个 daemon 线程
• 循环：读帧 → 检查连接状态 → push_frame() 超时轮询 → 排空 display_q 取最新结果 → 显示
• 检测到断连后，停止管线 → 重连 → 创建新管线继续
• 键盘控制：q/ESC 退出，+/- 缩放，0 重置缩放

关键设计决策

1. 阻塞推送：push_frame 使用阻塞 put()，管线满时主线程等待，自动将读取速率与处理速率对齐，不丢帧
2. NPU 串行化：仅线程2 持有 PNNAClient socket 引用，阻塞式 NPU 通信天然保证同一时刻只有一个请求在飞
3. 显示队列排空：主线程每次循环排空 display_q，只显示最新完成帧，避免积压
4. OpenCV GUI：cv2.imshow 和 cv2.waitKey 必须在主线程调用（Windows 要求），已遵守
5. 有界队列：防止预处理/NPU通信/后处理速率不匹配导致内存暴涨
6. 模型预加载：DSP 端启动时一次性加载模型并常驻内存，上位机不再发送模型数据（model_preloaded=True）

断线重连机制

对应 DSP 端固件的断线重连改造（模型常驻 + 循环 accept + TCP Keep-Alive + 链路监测），上位机实现了对应的自动重连功能。

DSP 端关键参数

DSP 参数	值	对上位机的意义
TCP Keep-Alive (idle/intvl/cnt)	3s / 1s / 3次	DSP 最快 ~6s 检测到死连接并关闭 socket
SO_RCVTIMEO	5s	DSP recv() 等待数据超时
链路监测轮询	每 3s	硬件断线检测延迟
accept() 循环	即时	断连后 DSP 立刻回到 accept 等待新连接

上位机重连架构

1. communication.py — 核心重连层

• ConnectionLostError 异常类：区分”连接断开”和”超时”，前者需重连，后者可重试
• Socket Keep-Alive：启用 SO_KEEPALIVE（跨平台），在支持的平台设置 TCP_KEEPIDLE=3s / TCP_KEEPINTVL=1s / TCP_KEEPCNT=3
• **非阻塞发送 + select()**：_send_request() 将 socket 设为非阻塞，使用 send() + select() 实现 5 秒硬截止时间的可靠发送。解决了 Windows 上 sendall() 在 TCP 缓冲区满时不响应 settimeout() 的问题
• **connect_with_retry()**：指数退避重连（1s → 2s → 4s → 8s → 15s 封顶），max_attempts=0 表示无限重试
• **reconnect()**：关闭旧 socket → 无限重试连接 → 成功后重建 msgpack.Unpacker
• **connection_lost**：threading.Event，线程2 (NPU通信) 在检测到断连时置位，主线程轮询此标志
• is_connected 属性：快速查询连接状态
• **model_preloaded**：为 True 时 add_model() / release_model() 变为空操作

2. thread_pipeline.py — 线程2 (NPU通信) 感知断连

• _infer_worker 捕获 ConnectionLostError：
  • 设置 connection_lost.set() 通知主线程
  • 设置 shutdown_event.set() 协调关闭所有工作线程
  • 发送 None 哨兵值解除下游线程阻塞
  • 退出线程
• 超时处理：infer() 返回 None（推理超时）仅跳过当前帧，不触发关闭
• **push_frame(timeout=0.5)**：队列满时 0.5 秒超时返回，主线程可轮询连接状态和键盘事件

3. main_video.py — 外层重连循环

run_once() 外层 while True:   ← 重连循环
  ├── 创建 VideoPipeline
  ├── pipeline.start()
  ├── connection_lost.clear()
  │
  └── 内层 while True:        ← 逐帧处理
        ├── 检查 connection_lost（前置检测）
        ├── cap.read()
        ├── push_frame(timeout=0.5)  ← 超时轮询 connection_lost + 键盘
        ├── 检查 connection_lost（push 后检测）
        ├── 排空 display_q → 显示
        ├── cv2.waitKey(1) → 键盘处理
        └── 若 connection_lost.is_set():
              → pipeline.stop()
              → client.reconnect()
              → break 回到外层

• 断连检测前置：推帧前检查 connection_lost，避免向已断开管线推帧
• 超时推帧：0.5s 超时，期间轮询连接状态和键盘，界面始终响应
• 视频位置保持：重连期间 cap 和 frame_idx 保持不变，恢复后从断点继续

4. main.py — 单图检测重试

• 推理包裹重试循环（最多 3 次）：
  • ConnectionLostError → client.reconnect() → 重试
  • 推理超时 → time.sleep(1) → 重试

完整重连时序

T0:     正常工作
T1:     网线断开
T2:     DSP Keep-Alive 检测（~6s）→ DSP 关闭 socket → 回到 accept()
T3:     上位机 infer() → recv() 超时 10s → 返回 None → 跳过帧
T4:     下一帧 infer() → _send_request() → 非阻塞 send → 缓冲区满
        → select() 等待 5s → deadline 到期 → ConnectionLostError
T5:     _infer_worker 捕获 → connection_lost.set() + shutdown_event.set()
T6:     主线程 push_frame() 超时 → 检测到 connection_lost.is_set()
        → pipeline.stop() → client.reconnect()
        → connect 指数退避重试: 1s, 2s, 4s, 8s, 15s...
T7:     DSP 网络恢复，accept() 接受新连接
T8:     主机重连成功 → 创建新 pipeline → 继续检测

总恢复时间：约 15-17 秒（最坏情况）

线程安全

共享资源	访问者	安全机制
client.sock	主线程(reconnect) + 线程2(infer)	reconnect 前先 pipeline.stop() 确保线程2已退出
client._connected	多线程	CPython GIL 下 bool 赋值是原子的
client.connection_lost	主线程 + 线程2	threading.Event 内置线程安全
client.unpacker	仅线程2	单线程访问，reconnect 后新建
Pipeline 队列	主线程 + 工作线程	queue.Queue 内置线程安全
pipeline._shutdown	主线程 + 线程2	threading.Event 内置线程安全

项目结构

├── main.py                  # 单图检测入口（支持重试）
├── main_video.py            # 视频检测入口（3线程并行 + 断线重连）
├── thread_pipeline.py       # 3线程流水线核心实现
├── communication.py         # PNNA NPU TCP 通信协议（含重连机制）
├── preprocess.py            # 图像预处理（resize/色彩转换/布局变换）
├── yolov4_tiny.py           # YOLOv4-tiny 后处理（DIoU NMS）
├── visualization.py         # 检测结果可视化 + 缩放显示
├── resource/
│   └── yolov4-tiny/
│       ├── yolov4-tiny_u8.nb   # 量化模型文件
│       ├── 3.mp4               # 测试视频
│       ├── dog.jpg             # 测试图片
│       └── input_0.dat         # 测试输入
└── README.md

配置

在 main_video.py 中修改：

TARGET_IP = "192.168.23.102"   # NPU 设备 IP
MODEL_PATH = "./resource/yolov4-tiny/yolov4-tiny_u8.nb"
VIDEO_PATH = "./resource/yolov4-tiny/3.mp4"

DSP 端固件需为新版（模型常驻 + 循环 accept），上位机使用 model_preloaded=True 模式。

运行

# 视频检测（3线程并行 + 断线重连）
python main_video.py

# 单图检测
python main.py

键盘快捷键

按键	功能
q / ESC	退出
+ / =	放大显示
-	缩小显示
0	重置缩放

数据流

cap.read() 原始帧 (H×W×3)
  │
  ├─ push_frame(timeout=0.5) ──> [pre_q:2]
  │                                    │
  │                                    └─ 线程1: resize(416×416) → BGR→RGB → HWC→CHW → bytes
  │                                        输入数据量: 416×416×3 = 519,168 bytes (uint8)
  │                                                              │
  │                                     保留原始帧 + bytes ──> [infer_q:2]
  │                                                              │
  │                                                              └─ 线程2: client.infer(bytes) ──> TCP发送至NPU
  │                                                                   等待NPU返回推理结果
  │                                                                                   │
  │                                     原始帧 + raw_outputs ──> [post_q:2]
  │                                     NPU输出: 13×13×255 (43,095 bytes)
  │                                            + 26×26×255 (172,380 bytes)
  │                                                                                   │
  │                                                                                   └─ 线程3: decode → YOLO后处理(DIoU NMS) → 在原图上画框
  │                                                                                                        │
  │                                                                     annotated_frame ──> [display_q:4]
  │                                                                                             │
  └─ get_result() 排空取最新 ── resize(scale) ── cv2.imshow() <──────────────────────────────────┘

测试结果

日志如下所示：

[OK] Connected to 192.168.23.102
[Log] Received 2 output tensors.
Tensor 0: shape=(13, 13, 255), dtype=uint8, size=43095 bytes
Tensor 1: shape=(26, 26, 255), dtype=uint8, size=172380 bytes

=== Detection Results ===
boxes number: 4
16 0.895512 0.289805 0.649326 0.259186 0.569164
7 0.740758 0.759681 0.218824 0.305246 0.157431
2 0.414782 0.759821 0.222086 0.286552 0.162486
1 0.635977 0.423077 0.508929 0.660444 0.612446

Result saved: H:\work_space\PycharmProjects\video_yolov4_tiny\dog_result.jpg
Press any key to close window...
[Log] Model preloaded on DSP, skipping release_model.