【BetterBench博士】2024年华为杯E题：高速公路应急车道紧急启用模型 Python代码实现

在这里插入图片描述

题目

【BetterBench博士】2024 “华为杯”第二十一届中国研究生数学建模竞赛选题分析

【BetterBench博士】2024年中国研究生数学建模竞赛 E题：高速公路应急车道紧急启用模型问题分析

【BetterBench博士】2024年中国研究生数学建模竞赛 C题：数据驱动下磁性元件的磁芯损耗建模问题分析

1 问题一

（1）第一小问

首先，需要将视频文件转换为帧，以便进行后续处理。
然后，使用算法对视频帧进行检测车辆。代码实现了YOLO5算法,尝试了FastRCNN算法非常耗时，不建议运行。
使用跟踪算法跟踪车辆。以下采用的SORT，或者DeepSORT等算法。
计算特征

时间戳：通过视频帧的位置获取。
车流密度：车流密度通常是检测到的车辆数量与区域面积的比值。
车流量：在一定时间段内通过的车辆数量。
平均速度：首先计算车辆的位移：通过跟踪的车辆ID在每一帧之间的位置信息，计算每辆车在相邻帧之间的位移。然后计算时间间隔：根据时间戳计算相邻帧之间的时间间隔。最后计算速度：使用位移除以时间间隔得到车辆的速度。

（2）第二、三小问

可以使用时序预测模型（例如LSTM、ARIMA）或传统的回归模型（如线性回归、决策树）来预测交通流拥堵情况。下面是一个基于随机森林回归的实现，提取的车流参数（车流密度、车流量、速度）来预测未来是否会发生拥堵。使用从四个观测点视频中提取的车流密度、车流量和速度等数据作为模型输入，进行时序建模，并预测未来的交通拥堵状态。基于历史数据预测下一时间点的交通流参数（车流密度、车流量和速度），并结合定义的拥堵条件，发出实时预警。

定义拥堵状态：当车流密度大于0.7且速度小于40 km/h时

使用测试集验证模型预测的车流密度、车流量和速度，并结合可视化方法来评估模型的表现。计算混淆矩阵和ROC曲线，分析模型的拥堵预测能力。

import cv2
import torch
from sort.sort import Sort
import numpy as np
import pandas as pd
from datetime import datetime
from tqdm import tqdm  # 用于显示进度条
device = torch.device("cuda" if True else "cpu")
# 使用 YOLOv5 作为目标检测模型
def load_model():
    # 使用 Ultralytics 的 YOLOv5 模型，确保已安装 'yolov5' 包
    # pip install yolov5
    model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True)
    model.to(device)
    model.eval()
    return model

# 将检测结果格式化为 SORT 算法接受的格式
def format_detections(predictions, score_threshold=0.5):
    detections = []
    for *box, score, cls in predictions:
        if score > score_threshold and int(cls) == 2:  # COCO中 'car' 的类别ID通常为2
            x1, y1, x2, y2 = box
            detections.append([x1, y1, x2, y2, score])
    return detections

# 计算车辆速度 (像素/秒)
def calculate_speed(track, prev_track, fps):
    if prev_track is None:
        return 0
    # 计算位移 (欧几里得距离)
    x1, y1, _, _ = track[:4]
    prev_x1, prev_y1, _, _ = prev_track[:4]
    displacement = np.sqrt((x1 - prev_x1)**2 + (y1 - prev_y1)**2)
    time_interval = 1 / fps  # 时间间隔（秒）
    return displacement / time_interval  # 像素/秒

# 处理视频，提取车辆流量信息
def detect_and_track(video_path, model, tracker, device, score_threshold=0.5, resize_width=640):
    # 打开视频
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise IOError(f"无法打开视频文件 {video_path}")
    
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    
    # 初始化数据存储
    timestamps = []
    densities = []
    flows = []
    speeds = []
    
    tracked_objects = {}  # 存储每个车辆的历史轨迹
    total_flow = 0  # 总车流量
    
    # 使用 tqdm 显示处理进度
    for _ in tqdm(range(frame_count), desc=f"Processing {video_path}"):
        ret, frame = cap.read()
        if not ret:
            break
        
        # 记录时间戳（可以选择只记录关键帧的时间戳以减少开销）
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
        
        # 调整图像大小以加快处理速度
        if resize_width:
            height, width = frame.shape[:2]
            scale = resize_width / width
            frame_resized = cv2.resize(frame, (resize_width, int(height * scale)))
        else:
            frame_resized = frame
        
        # 转换颜色格式（BGR -> RGB）
        img_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)
        
        # 使用模型进行检测
        results = model(img_rgb, size=640)  # size可以根据需求调整
        
        # 获取预测结果
        detections = results.xyxy[0].cpu().numpy()  # [x1, y1, x2, y2, confidence, class]
        detections = format_detections(detections, score_threshold)
        
        if len(detections) > 0:
            ...略
        else:
            # 如果当前帧没有检测到车辆，存储默认值
            timestamps.append(timestamp)
            densities.append(0)
            flows.append(0)
            speeds.append(0)
    
    cap.release()
    
    # 返回结果 DataFrame
    data = pd.DataFrame({
        'timestamp': timestamps,
        'density': densities,
        'flow': flows,
        'speed': speeds
    })
    
    return data

def extract_vehicle_data(video_paths, score_threshold=0.5, resize_width=640):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = load_model()
    tracker = Sort()
    
    all_vehicle_data = []
    for video in video_paths:
        vehicle_data = detect_and_track(
            video_path=video,
            model=model,
            tracker=tracker,
            device=device,
            score_threshold=score_threshold,
            resize_width=resize_width
        )
        all_vehicle_data.append(vehicle_data)
    
    # 合并所有视频的数据
    all_vehicle_data = pd.concat(all_vehicle_data, ignore_index=True)
    
    return all_vehicle_data

可视化提取出来的特征

import matplotlib.pyplot as plt
import pandas as pd
plt.rcParams['font.sans-serif'] = 'SimSun' # 换成自己环境下的中文字体，比如'SimHei'

# 第一个观测点数据可视化

# result_df = pd.read_csv('data/data_point1.csv')
# result_df = pd.read_csv('data/data_point1.csv')
# result_df = pd.read_csv('data/data_point3.csv')
result_df = pd.read_csv('data/data_point4.csv')
# 可视化车流密度、流量和平均速度
plt.figure(figsize=(10, 8))
plt.subplot(3, 1, 1)
plt.plot(range(len(result_df)), result_df['density'], label='车流密度', color='blue')
plt.title('车流密度随时间变化')
plt.xlabel('时间戳 (ms)')
plt.ylabel('车流密度')
plt.legend()

plt.subplot(3, 1, 2)
plt.plot(range(len(result_df)), result_df['flow'], label='车流量', color='orange')
plt.title('车流量随时间变化')
plt.xlabel('时间戳 (ms)')
plt.ylabel('车流量')
plt.legend()

plt.subplot(3, 1, 3)
plt.plot(range(len(result_df)), result_df['speed'], label='平均速度', color='green')
plt.title('平均速度随时间变化')
plt.xlabel('时间戳 (ms)')
plt.ylabel('平均速度 (像素/秒)')
plt.legend()

plt.tight_layout()
plt.savefig('img/第4个观测点数据可视化.png',dpi=200)
plt.show()

在这里插入图片描述

问题二、三

传统应急车道的启用多是静态的，且缺少区域路网联动分析，难以动态应对不同场景下的交通需求。因此，模型需要集成多情景的动态分析，包括节假日高峰、突发事故、排队消散等情景。对不同情景的处理逻辑，如在节假日高峰期通过流量监测系统提前识别大流量，启用应急车道；对于已经拥堵的路段，采取临时开放应急车道的策略来消散车流，恢复主线运行。

虚拟应急车道能够通过主动交通管理技术提升救援效率，缩短事故响应时间。因此，模型可考虑集成虚拟应急车道的控制策略，利用车辆通信系统动态引导车辆临时进入虚拟应急车道，提升通行效率。在紧急情况下，模型通过仿真预测未来交通流量，并决定是否虚拟启用应急车道，结合事故发生时的流量、速度等关键参数决定应急车道的开放时长与区域。

应急车道管控不仅需要考虑实时流量与交通状况，还需结合历史数据，分析拥堵易发节点，构建拥堵中断模型。对于应急车道的开放时机和时长，需基于具体的交通状况预测。集成实时交通数据与历史拥堵数据，构建交通流概率中断模型，动态调整应急车道的开放时长和开放区域。

可以通过VISSIM仿真分析不同车道配置下虚拟应急车道的效益优化，表明在不同事故情景下，适时开启虚拟应急车道可有效减少平均延误和排队长度。模型可以根据车道封闭数量和流量密度优化启用条件。引入VISSIM仿真结果，基于流量密度、封闭车道数和v/C值优化应急车道的启用条件，进一步提升模型的精度。

import pandas as pd
import numpy as np
import warnings

warnings.filterwarnings('ignore')
# 设置阈值
density_threshold = 0.7  # 车流密度阈值
speed_threshold = 40  # 速度阈值，单位：km/h

# 拥堵风险计算函数
def compute_congestion_risk(density, speed, density_threshold, speed_threshold):
   ...略
    return congestion_risk

# 添加高峰期和事故判断函数
def is_peak_hour(timestamp):
    """假设高峰时段为每天的7-9点和17-19点"""
    hour = pd.to_datetime(timestamp).hour
    return (7 <= hour <= 9) or (17 <= hour <= 19)

def is_accident_scene(density, speed):
    """假设当速度为0且密度很高时为事故"""
    return (speed == 0) and (density > density_threshold)

# 动态管控模型：应急车道启用判断
def dynamic_control(data, density_threshold, speed_threshold, time_window=10):
    # 计算每个时刻的拥堵风险
    data['congestion_risk'] = data.apply(
        lambda row: compute_congestion_risk(row['density'], row['speed'], density_threshold, speed_threshold), axis=1)

    # 判断是否为高峰时段
    data['peak_hours'] = data['timestamp'].apply(is_peak_hour)

    # 判断是否为事故现场
    data['accident_scene'] = data.apply(lambda row: is_accident_scene(row['density'], row['speed']), axis=1)

    # 滑动窗口计算未来time_window分钟内的平均拥堵风险
    data['avg_risk'] = data['congestion_risk'].rolling(window=time_window).mean()

    # 如果未来的平均风险超过1，且为高峰期或事故现场，启用虚拟应急车道
    data['virtual_lane'] = np.where((data['avg_risk'] > 1) & (data['peak_hours'] | data['accident_scene']), True, False)
    return data



# 读取数据并应用模型
data_point = pd.read_csv('data/data_point1.csv')
data = data_point[(data_point['speed'] > 0) & (data_point['density'] > 0)]


# 应用动态控制模型
result = dynamic_control(data, density_threshold, speed_threshold)
# 输出结果
print(result[['timestamp', 'virtual_lane']])

在这里插入图片描述

问题四

问题分析

原有四个采集点的布置方案中，前三个采集点之间相隔 1000m，最后一个点与第三个点之间相隔 3000m。但是这样的布置在前1000m 的路段可以较好地捕捉交通状态变化，但在距离较长的第三点和第四点之间，可能会丢失大量交通流变化的信息。

优化数据采集点的布置，确保在第三点和第四点之间的 3000m 路段上能够精确采集数据，以便做出更科学的应急车道启用决策。优化模型的核心目标是：

精度：确保关键路段（第三点到第四点）的交通状态能够被准确地监控。
成本控制：在精度要求下，尽量减少新采集点的数量，降低安装与维护的成本。

建模算法步骤

（1）初始化参数：

设定采集点的最小间隔 $D_{\text{min}} = 1000 \, m$ 。
设定应急响应时间 $T_{\text{response}} = 10 \, \text{分钟}$ 。
获取现有四个采集点的交通流量、速度和密度数据。

（2）计算每个路段的交通状态变异性：
根据变异性公式，计算各个路段的变异性 $V_i$ 。

（3）判断是否需要增加采集点：
如果某个路段的变异性 $V_i$ 超过预定阈值 $V_{\text{threshold}}$ ，则需要在该路段新增数据采集点。

（4）优化布点方案：

在第三点到第四点之间，若变异性较大（ $V_2 > V_{\text{threshold}}$ ），则在1500m处增加新的采集点。
保证数据采集点的间隔 $D_{\text{new}} \geq D_{\text{min}}$ 。

（5）输出新的采集点布置方案。

import pandas as pd
import numpy as np

# 初始化参数
D_min = 1000  # 最小采集点间隔，单位：米
T_response = 10  # 应急响应时间，单位：分钟
V_threshold = 0.2  # 变异性阈值

# 读取数据
data_point1 = pd.read_csv('data/data_point1.csv')
data_point2 = pd.read_csv('data/data_point2.csv')
data_point3 = pd.read_csv('data/data_point3.csv')
data_point4 = pd.read_csv('data/data_point4.csv')

# 计算交通状态变异性
def compute_variability(data):
   ...略
# 计算每个路段的变异性
V1 = compute_variability(data_point1)
V2 = compute_variability(data_point2)
V3 = compute_variability(data_point3)
V4 = compute_variability(data_point4)

# 判断是否需要增加采集点
def check_new_point(V, V_threshold, D_current, D_min):
    if V > V_threshold and D_current > D_min:
        return True
    return False

# 路段间隔（第三点到第四点为3000m，其他为1000m）
D1, D2, D3 = 1000, 1000, 3000

# 判断第三点和第四点之间是否需要新增采集点
new_point_needed = check_new_point(V3, V_threshold, D3, D_min)

# 输出优化的布置方案
if new_point_needed:
    print("建议在第三点到第四点之间新增采集点")
else:
    print("现有采集点布置合理，无需新增采集点")