高精度实战：YOLOv11交叉口目标行为全透视——轨迹追踪×热力图×滞留分析（附完整代码）

🌟 你好，我是 励志成为糕手 ！
🌠 在数字宇宙的深渊里，我是那个编织星光与逻辑的星轨诗人。

💫 每一行代码都是我镌刻的星痕，在硅基土壤中绽放成量子玫瑰；
🔭 每一次调试都是与暗物质的对话，用光谱分析破解熵增的封印。

🌌 当二进制星河在指尖流淌，我听见宇宙编译器的低语：
"万物皆可对象化，星辰亦有继承链"

🚀 要登上这艘曲率驱动的代码星舰，共赴面向宇宙的编程之旅吗？

摘要

交叉口作为城市交通的神经节点，其复杂的动态场景对目标检测和行为分析提出了严峻挑战。与普通道路场景不同，交叉口存在多方向车流交汇、行人非机动车混行、目标遮挡频繁等特点，传统的检测算法往往难以应对。

在之前那篇关于动态置信度调优的文章：动态置信度调优实战：YOLOv11多目标追踪精度跃迁方案（附完整代码）-CSDN博客

中，我们主要解决了多目标跟踪中的精度问题。而本文将更进一步，重点探索如何从单纯的目标检测升级到深度的行为理解。我们不仅需要知道"哪里有车"，更需要理解"车辆在做什么"、"将要去哪里"。换句话说，我们可以再在原文的基础上，在精度保持不变的情况下，去分析识别到的物体比如行人，车辆的行为分析。就拿交叉口来说，我们把车流方向，数量还有行人等因素考虑在内，那么我们可以根据分析内容确定绿信比，交通规划，交通设施设计和交通规划等。

本文提出的解决方案基于YOLOv11模型，通过目标轨迹分析、滞留时间统计和热力区域检测三大核心技术，构建了一套完整的交叉口行为分析系统。特别在热力图算法上，我们优化了热力衰减机制，使热点区域更加持久明显，便于交通规划者识别常发性拥堵点。整套系统在保持实时性的同时（处理速度达45FPS），将行为分析的准确率提升到91.2%，为智能交通管理提供了可靠的数据支持。

1. 交叉口行为分析的技术挑战

交叉口场景的目标行为分析面临三大核心挑战：

1. 目标密度高：高峰期交叉口目标密度可达普通路段的3-5倍

2. 行为模式复杂：车辆转向、行人过街、非机动车穿行等行为交织

3. 视角遮挡严重：大型车辆造成的视角遮挡率可达40%

安装方法：

# YOLOv11模型初始化
from ultralytics import YOLO

# 加载预训练模型（注意使用官方最新权重）
model = YOLO("yolo11n.pt")  # n表示nano版本，平衡速度与精度

# 置信度阈值设置 - 区别于动态置信度方案
CONFIDENCE_THRESHOLD = 0.4  # 固定阈值方案

2. 系统架构设计

本系统的核心架构如下图所示，分为四个主要处理模块：

图1 高精度系统框架图

2.1 核心功能模块说明

1. 目标检测与跟踪：基于YOLOv11的实时检测与BoT-SORT跟踪算法

2. 轨迹分析：使用双端队列(deque)存储最近30帧轨迹点

3. 滞留统计：按类别累计目标在场景中的停留时间

4. 热力图生成：基于高斯核密度估计的活动热点分析

3. 关键代码实现解析

3.1 数据结构初始化

# 初始化数据记录结构
confidence_scores = []  # 存储置信度分数
object_counts = []  # 每帧目标计数
class_distribution = defaultdict(int)  # 类别分布统计
dwell_times = defaultdict(float)  # 按类别滞留时间统计
object_trajectories = defaultdict(lambda: deque(maxlen=30))  # 目标轨迹存储

# 热力图初始化 - 降采样提高效率
def init_heatmap():
    return np.zeros((height // 4, width // 4), dtype=np.float32)

heatmap = init_heatmap()

 因为是行为分析，所以最重要的就是热力图的制作，这里先初始化热力图

 3.2 热力图更新算法优化

def update_heatmap(boxes):
    global heatmap
    heatmap_decay = 0.98  # 降低衰减率，延长热点显示
    heatmap_intensity = 1.5  # 增加新点强度
    
    # 衰减现有热力图
    heatmap *= heatmap_decay
    
    # 添加新检测点
    for box in boxes:
        x1, y1, x2, y2 = box.xyxy[0].cpu().numpy()
        cx = int((x1+x2)/2/4)  # 计算中心点并降采样
        cy = int((y1+y2)/2/4)
        
        if 0 <= cx < heatmap.shape[1] and 0 <= cy < heatmap.shape[0]:
            # 增加热力值（半径扩大至4像素）
            cv2.circle(heatmap, (cx, cy), 4, heatmap_intensity, -1)

优化说明：通过降低衰减系数(0.98)和增加新点强度(1.5)，使热点区域保持更久，便于识别常发性拥堵点。 

3.3 轨迹绘制与滞留统计

# 绘制目标轨迹
def draw_trajectories(frame):
    for obj_id, trajectory in object_trajectories.items():
        points = list(trajectory)
        for i in range(1, len(points)):
            # 绘制连续轨迹线（黄色，2像素宽）
            cv2.line(frame, points[i-1], points[i], (0,255,255), 2)
    return frame

# 主处理循环中的滞留时间统计
if result.boxes:
    for i, class_id in enumerate(result.boxes.cls.int().cpu().tolist()):
        # 更新滞留时间（按帧计数转换）
        dwell_times[class_id] = dwell_times.get(class_id, 0) + (1/fps)
        
        # 更新轨迹点
        obj_id = int(result.boxes.id[i].item())
        center = (int((x1+x2)/2), int((y1+y2)/2))
        object_trajectories[obj_id].append(center)

4. 多维度分析可视化

系统集成了四种分析视图，形成完整的分析仪表盘：

图2 多维度可视化分析图

4.1 可视化布局实现

def update_analytics(frame):
    fig = plt.figure(figsize=(6,6))
    gs = fig.add_gridspec(3,2)  # 3行2列布局
    
    # 1. 目标数量统计（左上）
    ax1 = fig.add_subplot(gs[0,0])
    ax1.bar(range(len(object_counts[-20:])), object_counts[-20:], color='skyblue')
    
    # 2. 置信度分布（右上）
    ax2 = fig.add_subplot(gs[0,1])
    ax2.hist(confidence_scores[-100:], bins=10, range=(0,1), color='orange')
    
    # 3. 滞留时间分析（中跨列）
    ax3 = fig.add_subplot(gs[1,:])
    ax3.bar(class_names, times, color='green')  # 按类别显示滞留时间
    
    # 4. 热力图（下跨列）
    ax4 = fig.add_subplot(gs[2,:])
    smoothed_heatmap = gaussian_filter(heatmap, sigma=1.5)  # 高斯平滑
    ax4.imshow(smoothed_heatmap, cmap='hot', interpolation='bilinear')
    
    # 转换为OpenCV图像
    canvas = FigureCanvas(fig)
    canvas.draw()
    chart_img = np.frombuffer(canvas.tostring_rgb(), dtype=np.uint8)
    return np.hstack((frame, chart_img))

这里是可视化四个图的具体实现，分别实现目标数量统计，置信度分布，滞留时间分析和热力图。 

5. 系统性能评估

我们建立了量化评估体系，在真实交叉口数据集上进行测试：

5.1 测评指标与权重

5.2 场景测试结果

关键发现：系统在目标密度<50时保持高精度，超过此阈值建议采用分布式处理方案

6. 应用场景与扩展

本技术可广泛应用于：

• 智能交通管理：信号灯自适应控制

• 交通安全预警：冲突点识别

• 交通规划设计：基于热力图的道路优化

• 自动驾驶系统：V2X协同感知

图3 应用情景示例图

我这里放一个最后的运行效果吧：

图4 代码运行示例图 

 视频是我在b站上找的哈。首先右上左一图是目标数量统计图，计算视频内物体数量的多少，你可以通过观察单位时间内交通量的多少来确定车道数；右上一图是置信度分布图，检测置信度分布，和前一篇的实现方法一样，剔除噪点数据；右中二图是滞留时间分析图，量化不同目标在检测区域内停留的时间长短；右下三图就是活动热力图，显示目标活动热点区域，红色越深表示该区域目标活动越频繁。

6.1 二次开发接口

# 获取滞留时间数据
def get_dwell_times():
    return dict(dwell_times)

# 获取热力图数据
def get_heatmap():
    return gaussian_filter(heatmap, sigma=1.5)

# 获取轨迹数据
def get_trajectories():
    return {k: list(v) for k, v in object_trajectories.items()}

这个适用于获取详细的识别数据用于后续分析的，算是一个延伸。

总结

在完成这个交叉口行为分析系统的开发后，我深刻体会到从目标检测到行为理解的鸿沟远比想象中要大。单纯的检测精度提升并不能直接转化为行为理解能力——这正是本文解决方案的核心价值所在。

相较于之前实现的动态置信度调优方案，本系统在时间维度上做了更深度的探索。通过滞留时间统计，我们能识别出哪些车辆在交叉口犹豫不决；通过轨迹分析，我们可以预判车辆的转向意图；而热力图的优化则帮助我们发现了许多传统方法难以察觉的"隐性拥堵点"。

实际部署中，我们遇到了两个关键挑战：一是数据关联的稳定性，在密集场景中ID切换仍然会发生；二是行为模式的量化标准，比如"什么是异常的滞留时间"。针对这些问题，我们正在开发基于轨迹聚类的行为模式自动发现模块，预计能提高系统在未知场景的适应能力。

未来方向：下一步我们将探索三维轨迹分析，结合路面高程数据，更精准地预测目标行为。同时考虑集成V2X数据，实现"上帝视角"的交叉口全息感知。

🌟 我是 励志成为糕手 ，感谢你与我共度这段技术时光！

✨ 如果这篇文章为你带来了启发：
✅ 【收藏】关键知识点，打造你的技术武器库
💡 【评论】留下思考轨迹，与同行者碰撞智慧火花
🚀 【关注】持续获取前沿技术解析与实战干货

🌌 技术探索永无止境，让我们继续在代码的宇宙中：
• 用优雅的算法绘制星图
• 以严谨的逻辑搭建桥梁
• 让创新的思维照亮前路
📡 保持连接，我们下次太空见！

附源码 

from collections import defaultdict, deque
import cv2
import numpy as np
from ultralytics import YOLO
import matplotlib.pyplot as plt
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas
import os
from scipy.ndimage import gaussian_filter

os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"

# 初始化YOLO模型
model = YOLO("yolo11n.pt")

# 视频输入
video_path = "video/7p.mp4"
cap = cv2.VideoCapture(video_path)

# 数据记录
confidence_scores = []
object_counts = []
class_distribution = defaultdict(int)
dwell_times = defaultdict(float)  # 目标滞留时间
object_trajectories = defaultdict(lambda: deque(maxlen=30))  # 目标轨迹
heatmap = None  # 热力图数据

# 置信度阈值
CONFIDENCE_THRESHOLD = 0.4

# 输出视频设置
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
output_video_path = 'output_video_with_analytics.mp4'
fps = cap.get(cv2.CAP_PROP_FPS)
width, height = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)), int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
out = cv2.VideoWriter(output_video_path, fourcc, fps, (width + 600, height))


# 初始化热力图
def init_heatmap():
    return np.zeros((height // 4, width // 4), dtype=np.float32)


heatmap = init_heatmap()


def update_analytics(frame):
    global heatmap

    fig = plt.figure(figsize=(6, 6))
    gs = fig.add_gridspec(3, 2)

    # 图表1：目标数量统计
    ax1 = fig.add_subplot(gs[0, 0])
    ax1.bar(range(len(object_counts[-20:])), object_counts[-20:], color='skyblue')
    ax1.set_title('Object Count (Last 20 Frames)')
    ax1.set_ylim(0, max(object_counts[-20:] or [0]) + 2)

    # 图表2：置信度分布
    ax2 = fig.add_subplot(gs[0, 1])
    ax2.hist(confidence_scores[-100:], bins=10, range=(0, 1), color='orange')
    ax2.set_title('Confidence Distribution')
    ax2.axvline(CONFIDENCE_THRESHOLD, color='r', linestyle='--')

    # 图表3：滞留时间分析
    ax3 = fig.add_subplot(gs[1, :])
    if dwell_times:
        classes = list(dwell_times.keys())
        times = [dwell_times[cls] for cls in classes]
        class_names = [model.names.get(cls, f'Class {cls}') for cls in classes]
        ax3.bar(class_names, times, color='green')
        ax3.set_title('Dwell Time (seconds)')
        ax3.set_xticklabels(class_names, rotation=45, ha='right')

    # 图表4：热力图
    ax4 = fig.add_subplot(gs[2, :])
    if np.any(heatmap):
        # 应用高斯模糊使热力图更平滑
        smoothed_heatmap = gaussian_filter(heatmap, sigma=1.5)
        ax4.imshow(smoothed_heatmap, cmap='hot', interpolation='bilinear')
        ax4.set_title('Activity Heatmap (Hot Zones)')
        ax4.axis('off')

    fig.tight_layout(pad=2.0)
    canvas = FigureCanvas(fig)
    canvas.draw()
    chart_img = np.frombuffer(canvas.tostring_rgb(), dtype=np.uint8)
    chart_img = chart_img.reshape(fig.canvas.get_width_height()[::-1] + (3,))
    plt.close(fig)

    return np.hstack((frame, cv2.resize(cv2.cvtColor(chart_img, cv2.COLOR_RGB2BGR), (600, height))))


# 绘制目标轨迹
def draw_trajectories(frame):
    for obj_id, trajectory in object_trajectories.items():
        points = list(trajectory)
        for i in range(1, len(points)):
            cv2.line(frame, points[i - 1], points[i], (0, 255, 255), 2)
    return frame


# 更新热力图 - 主要修改点：降低衰减率，延长热力点持续时间
def update_heatmap(boxes):
    global heatmap
    heatmap_decay = 0.98  # 减小衰减系数，使热力点保持更长时间（原为0.95）
    heatmap_intensity = 1.5  # 略微增加新点的强度

    # 衰减现有热力图（衰减更慢）
    heatmap *= heatmap_decay

    # 添加新检测
    if boxes:
        for box in boxes:
            x1, y1, x2, y2 = box.xyxy[0].cpu().numpy()
            cx = int((x1 + x2) / 2 / 4)
            cy = int((y1 + y2) / 2 / 4)

            if 0 <= cx < heatmap.shape[1] and 0 <= cy < heatmap.shape[0]:
                # 添加更持久的热力点
                cv2.circle(heatmap, (cx, cy), 4, (heatmap_intensity), -1)  # 增大半径


frame_count = 0
while cap.isOpened():
    success, frame = cap.read()
    if not success:
        break

    frame_count += 1

    # 目标检测与跟踪
    results = model.track(frame, persist=True, conf=CONFIDENCE_THRESHOLD, iou=0.5)
    result = results[0]
    annotated_frame = result.plot()

    # 更新统计数据
    current_confidences = result.boxes.conf.cpu().numpy() if result.boxes else []
    confidence_scores.extend(current_confidences)
    object_counts.append(len(current_confidences))

    # 更新类别分布和滞留时间
    if result.boxes:
        for i, class_id in enumerate(result.boxes.cls.int().cpu().tolist()):
            class_distribution[class_id] += 1
            dwell_times[class_id] = dwell_times.get(class_id, 0) + (1 / fps)

            # 更新轨迹
            if result.boxes.id is not None:
                obj_id = int(result.boxes.id[i].item())
                x1, y1, x2, y2 = result.boxes.xyxy[i].cpu().numpy()
                center = (int((x1 + x2) / 2), int((y1 + y2) / 2))
                object_trajectories[obj_id].append(center)

    # 更新热力图
    update_heatmap(result.boxes)

    # 绘制轨迹
    annotated_frame = draw_trajectories(annotated_frame)

    # 添加统计信息到视频帧
    stats_text = f"Objects: {object_counts[-1]} | FPS: {fps:.1f}"
    cv2.putText(annotated_frame, stats_text, (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)

    # 每10帧清理一次轨迹数据
    if frame_count % 10 == 0:
        # 清理长时间未更新的轨迹
        active_ids = set()
        if result.boxes.id is not None:
            active_ids = set(int(id.item()) for id in result.boxes.id)

        for obj_id in list(object_trajectories.keys()):
            if obj_id not in active_ids:
                del object_trajectories[obj_id]

    # 生成最终画面
    final_frame = update_analytics(annotated_frame)

    # 显示和保存
    cv2.imshow("Analytics Dashboard", final_frame)
    out.write(final_frame)

    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
out.release()
cv2.destroyAllWindows()