【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 yolov5model = 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通常为2x1, y1, x2, y2 = boxdetections.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 / widthframe_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()# 返回结果 DataFramedata = pd.DataFrame({'timestamp': timestamps,'density': densities,'flow': flows,'speed': speeds})return datadef 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 warningswarnings.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).hourreturn (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 Truereturn 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("现有采集点布置合理，无需新增采集点")