企业数字化笔记头像
关注
用YOLO做道路视频车流统计：从目标检测到越线计数的实现方案封面图

用YOLO做道路视频车流统计：从目标检测到越线计数的实现方案

在道路视频上画出一堆检测框并不难，难的是回答一个更接近交付的问题：这段视频里到底有多少辆车从左向右通过，又有多少辆车从右向左通过？如果同一辆车连续 300 帧都被检测到，不能计 300 次；如果车辆在计数线附近抖动、短暂遮挡或折返，也不能随意加一。

本文用一段脱敏固定机位道路视频说明车流统计的最小工程闭环。输入是一段 10 分钟 MP4 和一条人工标定的计数线；输出是带标注的视频、按方向和类别统计的结果，以及每次计数可回看的事件记录。示例只讨论通用的 YOLO 检测、目标跟踪、越线判断和验收，不公开真实道路位置、完整产品编排、模型权重或部署参数。

示例环境为 Python 3、Ultralytics YOLO、OpenCV、FFmpeg 和 MySQL 8.x。模型版本、类别映射、置信度阈值和追踪器参数必须按镜头条件实测，文中的值不应直接复制到生产环境。

目录

  1. 先把检测框和统计结果区分开
  2. 固定案例和验收目标
  3. 从视频帧到越线事件的处理链路
  4. 数据模型：轨迹与计数事件分开保存
  5. Python实现：稳定编号后再判断越线
  6. 服务层实现：保存可回看的统计结果
  7. 预期输出和自动测试
  8. SQL验证：怎样发现重复计数
  9. 误差边界和上线验收
  10. 小结和延伸阅读

一、先把检测框和统计结果区分开

目标检测回答“这一帧可能有什么”，目标跟踪回答“这几帧是不是同一个对象”，越线计数才回答“某个对象是否完成了一次方向明确的通过”。把三件事混在一起，最常见的结果是检测框越多，统计数越离谱。

层次输入输出不能直接当成
检测单帧图像边界框、类别、置信度一辆真实车辆
跟踪连续帧检测结果稳定的 track_id 与轨迹一次通行事件
越线轨迹中心点与计数线左向右或右向左事件所有经过线附近的框
汇总已确认的越线事件分类、方向、时间段统计模型精度指标

固定计数线把画面分成 A、B 两侧。某辆车的稳定中心点从 A 侧首次跨到 B 侧，且该 track_id 尚未产生过同方向事件时，系统才登记一次 LEFT_TO_RIGHT。车辆只是在边界附近晃动、检测短暂丢失或离线后又出现，不能因此重复计数。

在这里插入图片描述

图1：检测解决“看见什么”，跟踪解决“是不是同一辆”，越线事件才形成业务统计。

二、固定案例和验收目标

任务编号：TV-20261009-001
输入视频：road-demo.mp4，10 分钟，1920x1080，25 fps
检测类别：CAR、MOTO、BUS、BIKE
计数线：画面坐标 (960, 320) -> (960, 920)
输出一：road-annotated.mp4，显示框、编号、线和实时统计
输出二：按类别和方向汇总的统计表
输出三：越线事件，含 track_id、帧号、时间点和中心点

验收不要求“绝对零误差”，而是要让误差可见、可抽检。本文把可交付标准定为：输出视频可播放；每个计数事件都能定位到视频时间点；同一个 track_id 同方向最多计一次；统计总数等于事件表按类别、方向聚合的结果；人工抽检片段中能够解释遮挡、误检和漏检。

三、从视频帧到越线事件的处理链路

车流统计不是把模型输出直接累加，而是逐层收紧事实：

步骤处理动作关键约束
读取视频OpenCV 逐帧读取并保留帧号、fps输入文件可播放，帧号连续
目标检测与跟踪返回类别、边界框、置信度和 track_id只接受目标类别与足够置信度的结果
更新轨迹计算中心点，保存前后两次位置不用单帧框直接计数
判断越线计算中心点在线两侧的符号变化加入最小移动距离，过滤贴线抖动
写入事件同一轨迹同方向只登记一次使用唯一约束防重复
导出结果生成标注视频和聚合统计总数与事件表可相互核对

在这里插入图片描述

图2：计数发生在稳定轨迹跨线之后，不发生在单帧检测时。

四、数据模型：轨迹与计数事件分开保存

轨迹是过程数据，事件是业务事实。保存时不必把每一帧的全部检测框永久写入数据库，但至少要保留最终事件的定位信息和必要轨迹摘要：

CREATE TABLE traffic_analysis_job (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  job_no VARCHAR(64) NOT NULL,
  source_file_key VARCHAR(500) NOT NULL,
  model_profile VARCHAR(64) NOT NULL,
  line_config_json JSON NOT NULL,
  job_status VARCHAR(24) NOT NULL,
  annotated_video_key VARCHAR(500) NULL,
  total_events INT NOT NULL DEFAULT 0,
  create_time DATETIME NOT NULL,
  update_time DATETIME NULL,
  UNIQUE KEY uk_traffic_job_no (job_no),
  CHECK (job_status IN ('PENDING', 'RUNNING', 'READY', 'FAILED'))
);

CREATE TABLE traffic_crossing_event (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  analysis_job_id BIGINT NOT NULL,
  track_id BIGINT NOT NULL,
  vehicle_type VARCHAR(16) NOT NULL,
  direction VARCHAR(24) NOT NULL,
  crossing_frame INT NOT NULL,
  crossing_ms BIGINT NOT NULL,
  center_x INT NOT NULL,
  center_y INT NOT NULL,
  confidence DECIMAL(5,4) NOT NULL,
  create_time DATETIME NOT NULL,
  UNIQUE KEY uk_crossing_once (analysis_job_id, track_id, direction),
  KEY idx_crossing_summary (analysis_job_id, vehicle_type, direction)
);

唯一键不是模型准确率的替代品，但它能保护系统免受重复写入、重跑逻辑错误和贴线反复抖动的影响。输出视频则是事件的视觉证据：人工能跳到 crossing_ms 附近，检查这次计数是否合理。

在这里插入图片描述

图3：分析任务保存输入与配置，越线事件保存可计数事实，两者共同指向标注视频。

五、Python实现：稳定编号后再判断越线

示例使用检测器提供的持续编号。实现重点不是某一个追踪器名称，而是：中心点必须有前后位置；跨线必须有方向；同一轨迹和方向只能产生一次事件。

from dataclasses import dataclass

@dataclass(frozen=True)
class Point:
    x: float
    y: float

def side_of_vertical_line(point: Point, line_x: float) -> int:
    return -1 if point.x < line_x else 1

def crosses_line(previous: Point, current: Point, line_x: float, min_move: float = 12) -> str | None:
    if abs(current.x - previous.x) < min_move:
        return None
    before, after = side_of_vertical_line(previous, line_x), side_of_vertical_line(current, line_x)
    if before == after:
        return None
    return "LEFT_TO_RIGHT" if before < after else "RIGHT_TO_LEFT"

class CrossingCounter:
    def __init__(self, line_x: float):
        self.line_x = line_x
        self.last_center: dict[int, Point] = {}
        self.counted: set[tuple[int, str]] = set()

    def observe(self, track_id: int, center: Point) -> str | None:
        previous = self.last_center.get(track_id)
        self.last_center[track_id] = center
        if previous is None:
            return None
        direction = crosses_line(previous, center, self.line_x)
        key = (track_id, direction)
        if direction and key not in self.counted:
            self.counted.add(key)
            return direction
        return None

在真实帧循环中，Worker 只把经过类别筛选和置信度筛选的检测结果喂给 observe。当它返回方向时，再写入事件、绘制方向标记并更新实时统计。计数线、最小移动距离和类别映射都属于版本化配置，不能散落在代码常量里。

在这里插入图片描述

图4：中心点跨线只是候选条件；方向、最小移动距离和唯一约束一起抑制重复计数。

六、服务层实现：保存可回看的统计结果

服务层在启动前锁定任务和配置，Worker 返回事件及标注视频的临时路径。只有视频可播放、事件能落库、汇总能对上时，任务才进入 READY：

@Transactional(rollbackFor = Exception.class)
public TrafficResult analyze(TrafficCommand command) {
    TrafficAnalysisJob job = jobRepository.lock(command.jobNo())
            .orElseGet(() -> jobRepository.createPending(command));
    if ("READY".equals(job.status())) return TrafficResult.reused(job.annotatedVideoKey());

    jobRepository.markRunning(job.id());
    WorkerTrafficResult result = trafficWorker.analyze(command.sourceKey(), command.lineConfig());
    if (!result.success() || !result.videoInfo().playable()) {
        jobRepository.markFailed(job.id(), result.errorCode());
        return TrafficResult.failed(result.errorCode());
    }

    eventRepository.insertIgnoreDuplicates(job.id(), result.events());
    int eventTotal = eventRepository.countByJobId(job.id());
    if (eventTotal != result.summary().total()) {
        jobRepository.markFailed(job.id(), "EVENT_SUMMARY_MISMATCH");
        return TrafficResult.failed("EVENT_SUMMARY_MISMATCH");
    }
    String videoKey = fileRepository.commitTemporary(result.annotatedTemporaryPath(), "ANNOTATED_VIDEO");
    jobRepository.markReady(job.id(), videoKey, eventTotal);
    return TrafficResult.ready(videoKey, result.summary());
}

这里不把“模型返回了多少框”当成功指标。对业务而言，可交付的是能播放的标注视频、可回看的越线事件和能相互核对的统计结果。

七、预期输出和自动测试

CAR 左向右：18
CAR 右向左：11
MOTO 左向右：4
BUS 右向左：1
总越线事件：34
每条事件均含 track_id、方向、帧号和视频时间点

def test_crossing_is_counted_once_per_direction():
    counter = CrossingCounter(line_x=100)
    assert counter.observe(7, Point(80, 300)) is None
    assert counter.observe(7, Point(115, 302)) == "LEFT_TO_RIGHT"
    assert counter.observe(7, Point(130, 304)) is None

def test_small_jitter_near_line_is_not_crossing():
    counter = CrossingCounter(line_x=100)
    counter.observe(9, Point(98, 300))
    assert counter.observe(9, Point(103, 301)) is None

@Test
void shouldFailWhenEventSummaryDoesNotMatchWorkerResult() {
    worker.stubSuccessWithEvents(12, 13);

    TrafficResult result = service.analyze(new TrafficCommand("TV-001", "road-demo.mp4", lineConfig));

    assertEquals("EVENT_SUMMARY_MISMATCH", result.errorCode());
    assertFalse(jobRepository.isReady("TV-001"));
}

八、SQL验证：怎样发现重复计数

-- 同一任务、同一轨迹、同一方向不能出现两次，预期结果为空
SELECT analysis_job_id, track_id, direction, COUNT(*) AS cnt
FROM traffic_crossing_event
GROUP BY analysis_job_id, track_id, direction
HAVING COUNT(*) > 1;

-- 任务总数必须等于事件表聚合结果，预期结果为空
SELECT j.job_no, j.total_events, COUNT(e.id) AS actual_events
FROM traffic_analysis_job j
LEFT JOIN traffic_crossing_event e ON e.analysis_job_id = j.id
WHERE j.job_status = 'READY'
GROUP BY j.id, j.job_no, j.total_events
HAVING j.total_events <> COUNT(e.id);

-- 查没有标注视频的完成任务，预期结果为空
SELECT job_no FROM traffic_analysis_job
WHERE job_status = 'READY'
  AND (annotated_video_key IS NULL OR annotated_video_key = '');

在这里插入图片描述

图5：统计值必须能从事件表重算，并能回到标注视频复核；否则只是不可解释的数字。

九、误差边界和上线验收

现象影响应对方式
车辆遮挡或并行跟踪编号可能切换抽检密集片段，记录跟踪丢失率
夜景、逆光、雨雾检测置信度下降按场景分组评测，必要时降低自动使用范围
计数线太靠近停车区同一车辆反复穿线调整线位置，增加最小移动距离与方向规则
镜头移动或抖动固定坐标线失效仅用于固定机位；镜头变化需重新标定
类别误判分类统计失真用抽样人工复核，不把类别计数当作绝对真值

上线前应完成：固定视频复跑结果一致；抽检事件能回看；同一轨迹同方向不重复；任务总数能由事件表重算；视频、事件和统计任一部分失败时任务不进入 READY。这比单独报一个“模型准确率”更接近真实交付。

十、小结和延伸阅读

道路视频车流统计的核心不是检测框数量，而是把连续检测转换成稳定轨迹，再把轨迹转换成方向明确、可回看、可重算的越线事件。检测、跟踪、计数和交付各有边界，才能在遮挡、误检和镜头差异存在时仍然解释统计结果来自哪里。

  1. Ultralytics YOLO Track Mode
  2. OpenCV: Getting Started with Videos
  3. FFmpeg Documentation
  4. Spring Framework: Transaction Management
  5. MySQL 8.0 Reference Manual: CREATE TABLE

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/weixin_53653412/article/details/166845303

文章来源转载

评论

赞0

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--