嵌入式视觉模型进阶 30：综合项目：检测、跟踪与云台跟随验收

第 30 / 30 课。 前置：上一课。本篇的运行范围以实验说明和验证记录为准，构造样例不代表真实任务精度或板端性能。

项目最终交付什么

目标是把检测、跟踪、目标锁定和控制意图串成可追溯流程。包内包含固定检测回放 sequence.json、跟踪器、控制器、运行入口和 JSON 报告；另有 video.py，可用本地 YOLOv8 ONNX 与录像或显式指定的摄像头运行同一后半段。

本课不直接驱动电机。输出 speed 是第 17 课定义的归一化速度意图，实际驱动必须完成单位、方向和限位适配。学习路线到这里形成闭环，未在本次环境执行的板端与机械验收仍属于项目实施阶段。

先完成确定性回放

python3 main.py --output runs 读取本篇自带序列：正常移动、低分观测、短时丢失、过期结果和恢复。每帧保留采集时间、处理完成时间、检测、轨迹和控制命令。

验收检查包括 ID 延续、无观测停止、超时停止、速度上限和有限数值。输出报告中的 CPU 回放耗时仅表示处理这些 JSON 数据的耗时，不包括摄像头、图像推理或硬件动作。

再接真实视觉输入

准备第 11 课契约的模型后执行 python3 video.py --model model.onnx --video recording.mp4 --classes 80 --output runs-video。摄像头需显式使用 --camera 0。程序保存逐帧 JSON，不显示 GUI、不发送电机命令，也不下载权重。

录像回放使用帧序号/FPS 推进运动时间；摄像头使用单调时钟间隔。接收时刻不等于传感器曝光时刻，摄像头报告的本机处理耗时不能当作完整采集端到端延迟。接入真实媒体管线时必须携带设备采集时间戳。

四层验收逐步放行

层次	输入与检查	通过后才能做什么
数值	固定帧、同一预处理，比较 CPU 与 NPU 输出	接入连续帧
任务	独立录像，标注漏检、误跟、ID 切换与恢复	接控制意图
控制	模拟或电机禁能，检查方向、限幅和失效停止	低速机械测试
系统	真实设备持续运行，记录延迟、温度、内存与故障	按需求交付

每层保存模型哈希、配置、软件版本和失败样本。性能门限由产品需求预先确定，不能测完后再把门限调到刚好通过。

毕业练习

增加第二个目标交叉运动，观察仅靠几何时的身份问题；增加推理卡顿，验证过期输出停止；中断输入，确认驱动层独立看门狗也能停止。进程退出时本程序没有硬件通信，实际驱动集成必须另行实现退出与断连后的停机。

本课验收完成意味着能够读模型、跑任务、定位错误和交付可复现程序，不意味着所有模型和所有板卡已经学完。后续按 OCR、测距或工业质检项目增补专题即可。平台实现继续参考海思 NPU 系列和云台系列。

配套实验与运行记录

实验：综合项目：检测、跟踪与云台跟随验收

查看完整代码：main.py

"""交付可回放的跟踪控制程序、真实视频入口和分层验收报告，完成课程闭环。"""

import argparse, json, math, time
from pathlib import Path
from tracker import Tracker
from controller import Controller


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--sequence", type=Path, default=Path(__file__).with_name("sequence.json")
    )
    parser.add_argument("--output", type=Path, default=Path("runs"))
    args = parser.parse_args()
    source = json.loads(args.sequence.read_text())
    tracker = Tracker(max_age=2)
    controller = Controller(timeout=0.2)
    rows = []
    previous = None
    start = time.perf_counter()
    for frame in source["frames"]:
        timestamp = frame["capture_s"]
        dt = timestamp - previous if previous is not None else 0.1
        if dt <= 0:
            raise ValueError("capture timestamps must increase")
        previous = timestamp
        detections = frame["detections"]
        high = [d for d in detections if d["score"] >= 0.6]
        low = [d for d in detections if 0.1 <= d["score"] < 0.6]
        tracks = tracker.step(high, dt=dt, low=low)
        age = frame["finish_s"] - timestamp
        command = controller.step(tracks, source["width"], dt, age)
        assert (
            math.isfinite(command["speed"])
            and abs(command["speed"]) <= controller.max_speed
        )
        if not detections or age > controller.timeout:
            assert command["speed"] == 0
        if "expected_id" in frame:
            assert command["target_id"] == frame["expected_id"]
        rows.append(
            {
                "frame_id": frame["id"],
                "age_s": age,
                "tracks": tracks,
                "command": command,
            }
        )
    report = {
        "mode": "constructed detection replay",
        "frames": len(rows),
        "checks": "passed",
        "wall_seconds": time.perf_counter() - start,
        "hardware_status": "not measured; no motor commands sent",
        "results": rows,
    }
    args.output.mkdir(parents=True, exist_ok=True)
    (args.output / "report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    )
    print(
        "frames",
        len(rows),
        "checks passed",
        "lost/stale stops",
        sum(r["command"]["reason"] == "stale_or_lost" for r in rows),
    )
    print("report:", args.output / "report.json")


if __name__ == "__main__":
    main()

查看完整代码：tracker.py

"""教学版恒速 Kalman + IoU/Hungarian 跟踪；非官方 SORT/ByteTrack 实现。"""

import numpy as np
from scipy.optimize import linear_sum_assignment


def iou(a, b):
    a = np.asarray(a, dtype=float)
    b = np.asarray(b, dtype=float)
    lo = np.maximum(a[:2], b[:2])
    hi = np.minimum(a[2:], b[2:])
    inter = np.maximum(hi - lo, 0).prod()
    union = (
        np.maximum(a[2:] - a[:2], 0).prod()
        + np.maximum(b[2:] - b[:2], 0).prod()
        - inter
    )
    return float(inter / union) if union > 0 else 0.0


def measure(box):
    box = np.asarray(box, dtype=float)
    return np.r_[(box[:2] + box[2:]) / 2, box[2:] - box[:2]]


class Track:
    def __init__(self, ident, detection):
        self.id = ident
        self.label = detection.get("class_id", 0)
        self.x = np.r_[measure(detection["box"]), np.zeros(4)]
        self.p = np.diag([10.0] * 4 + [100.0] * 4)
        self.missed = 0
        self.hits = 1
        self.matched = True

    @property
    def box(self):
        cx, cy, w, h = self.x[:4]
        w, h = max(w, 1), max(h, 1)
        return np.array([cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2])

    def predict(self, dt):
        f = np.eye(8)
        f[:4, 4:] = np.eye(4) * dt
        self.x = f @ self.x
        self.p = f @ self.p @ f.T + np.eye(8) * 0.1
        self.missed += 1
        self.matched = False

    def update(self, detection):
        h = np.eye(4, 8)
        r = np.eye(4)
        innovation = measure(detection["box"]) - h @ self.x
        s = h @ self.p @ h.T + r
        k = np.linalg.solve(s, (self.p @ h.T).T).T
        self.x += k @ innovation
        a = np.eye(8) - k @ h
        self.p = a @ self.p @ a.T + k @ r @ k.T
        self.missed = 0
        self.hits += 1
        self.matched = True


class Tracker:
    def __init__(self, max_age=2, threshold=0.2):
        self.max_age = max_age
        self.threshold = threshold
        self.tracks = []
        self.next_id = 1

    def associate(self, indices, detections):
        if not indices or not detections:
            return [], list(indices), list(range(len(detections)))
        cost = np.full((len(indices), len(detections)), 1e6)
        for r, index in enumerate(indices):
            track = self.tracks[index]
            for c, det in enumerate(detections):
                overlap = iou(track.box, det["box"])
                if track.label == det.get("class_id", 0) and overlap >= self.threshold:
                    cost[r, c] = 1 - overlap
        rows, cols = linear_sum_assignment(cost)
        pairs = [(indices[r], int(c)) for r, c in zip(rows, cols) if cost[r, c] < 1e6]
        return (
            pairs,
            [i for i in indices if i not in {a for a, b in pairs}],
            [j for j in range(len(detections)) if j not in {b for a, b in pairs}],
        )

    def step(self, detections, dt=1.0, low=None):
        if not np.isfinite(dt) or dt <= 0:
            raise ValueError("positive dt required")
        for det in list(detections) + list(low or []):
            b = np.asarray(det["box"])
            if b.shape != (4,) or not np.isfinite(b).all() or np.any(b[2:] <= b[:2]):
                raise ValueError("invalid xyxy")
        for track in self.tracks:
            track.predict(dt)
        pairs, unmatched, new = self.associate(
            list(range(len(self.tracks))), detections
        )
        for i, j in pairs:
            self.tracks[i].update(detections[j])
        if low:
            pairs, _, _ = self.associate(unmatched, low)
            for i, j in pairs:
                self.tracks[i].update(low[j])
        for j in new:
            self.tracks.append(Track(self.next_id, detections[j]))
            self.next_id += 1
        self.tracks = [t for t in self.tracks if t.missed <= self.max_age]
        return [
            {
                "id": t.id,
                "box": t.box.tolist(),
                "matched": t.matched,
                "missed": t.missed,
                "class_id": t.label,
            }
            for t in self.tracks
        ]

查看完整代码：controller.py

"""输出归一化角速度意图，不直接驱动电机。"""

import math


class Controller:
    def __init__(
        self, kp=1.2, deadzone=0.05, max_speed=0.6, max_accel=2.0, timeout=0.2
    ):
        self.kp = kp
        self.deadzone = deadzone
        self.max_speed = max_speed
        self.max_accel = max_accel
        self.timeout = timeout
        self.previous = 0.0
        self.target = None

    def step(self, rows, width, dt, age):
        if (
            width <= 0
            or dt <= 0
            or not all(math.isfinite(v) for v in [width, dt, age])
            or age < 0
        ):
            raise ValueError("invalid timing or dimensions")
        visible = [r for r in rows if r["matched"]]
        chosen = next((r for r in visible if r["id"] == self.target), None)
        if chosen is None and self.target is None and visible:
            chosen = min(
                visible, key=lambda r: abs((r["box"][0] + r["box"][2]) / 2 - width / 2)
            )
            self.target = chosen["id"]
        if age > self.timeout or chosen is None:
            self.previous = 0.0
            return {"target_id": self.target, "speed": 0.0, "reason": "stale_or_lost"}
        center = (chosen["box"][0] + chosen["box"][2]) / 2
        error = (center - width / 2) / (width / 2)
        desired = (
            0.0
            if abs(error) <= self.deadzone
            else max(-self.max_speed, min(self.max_speed, self.kp * error))
        )
        delta = self.max_accel * dt
        speed = max(self.previous - delta, min(self.previous + delta, desired))
        self.previous = speed
        return {"target_id": self.target, "speed": speed, "reason": "tracking"}

    def release(self):
        self.target = None
        self.previous = 0.0

查看完整代码：sequence.json

{
  "width": 320,
  "frames": [
    {
      "id": 0,
      "capture_s": 0,
      "finish_s": 0.03,
      "detections": [
        {
          "box": [
            220,
            80,
            260,
            120
          ],
          "score": 0.9,
          "class_id": 0
        }
      ],
      "expected_id": 1
    },
    {
      "id": 1,
      "capture_s": 0.1,
      "finish_s": 0.13,
      "detections": [
        {
          "box": [
            216,
            80,
            256,
            120
          ],
          "score": 0.9,
          "class_id": 0
        }
      ],
      "expected_id": 1
    },
    {
      "id": 2,
      "capture_s": 0.2,
      "finish_s": 0.23,
      "detections": [
        {
          "box": [
            212,
            80,
            252,
            120
          ],
          "score": 0.3,
          "class_id": 0
        }
      ],
      "expected_id": 1
    },
    {
      "id": 3,
      "capture_s": 0.3,
      "finish_s": 0.33,
      "detections": [],
      "expected_id": 1
    },
    {
      "id": 4,
      "capture_s": 0.4,
      "finish_s": 0.43,
      "detections": [
        {
          "box": [
            204,
            80,
            244,
            120
          ],
          "score": 0.9,
          "class_id": 0
        }
      ],
      "expected_id": 1
    },
    {
      "id": 5,
      "capture_s": 0.5,
      "finish_s": 0.81,
      "detections": [
        {
          "box": [
            200,
            80,
            240,
            120
          ],
          "score": 0.9,
          "class_id": 0
        }
      ],
      "expected_id": 1
    },
    {
      "id": 6,
      "capture_s": 0.6,
      "finish_s": 0.84,
      "detections": [
        {
          "box": [
            196,
            80,
            236,
            120
          ],
          "score": 0.9,
          "class_id": 0
        }
      ],
      "expected_id": 1
    },
    {
      "id": 7,
      "capture_s": 0.7,
      "finish_s": 0.87,
      "detections": [
        {
          "box": [
            192,
            80,
            232,
            120
          ],
          "score": 0.9,
          "class_id": 0
        }
      ],
      "expected_id": 1
    }
  ]
}

下载代码

在项目根目录执行；下载包解压后，在解压目录执行 python3 main.py。

cd 'content/posts/AI-深度学习/嵌入式视觉模型进阶/30-综合项目/code' && python3 main.py

在本篇代码目录执行 python3 main.py。依赖安装、可选入口见包内 README.md；实际运行结果见 verification.md。下载包包含本课使用的辅助模块与样例。

参考与进一步阅读

原始论文或官方文档用于核对结构、接口和适用范围。本篇代码是围绕教学目标编写的最小实验；除明确注明的完整入口外，不视为对应论文模型的完整复现。