數(shù)學(xué)建模是人工智能技術(shù)的重要組成部分。

題目

【BetterBench博士】2024 “華為杯”第二十一屆中國研究生數(shù)學(xué)建模競賽 選題分析

【BetterBench博士】2024年中國研究生數(shù)學(xué)建模競賽 E題：高速公路應(yīng)急車道緊急啟用模型 問題分析

【BetterBench博士】2024年中國研究生數(shù)學(xué)建模競賽 C題：數(shù)據(jù)驅(qū)動下磁性元件的磁芯損耗建模 問題分析

1 問題一

（1） 第一小問

1. 首先，需要將視頻文件轉(zhuǎn)換為幀，以便進行后續(xù)處理。
2. 然后，使用算法對視頻幀進行檢測車輛。代碼實現(xiàn)了YOLO5算法,嘗試了FastRCNN算法非常耗時，不建議運行。
3. 使用跟蹤算法跟蹤車輛。以下采用的SORT，或者DeepSORT等算法。
4. 計算特征

• 時間戳：通過視頻幀的位置獲取。
• 車流密度：車流密度通常是檢測到的車輛數(shù)量與區(qū)域面積的比值。
• 車流量：在一定時間段內(nèi)通過的車輛數(shù)量。
• 平均速度：首先計算車輛的位移：通過跟蹤的車輛ID在每一幀之間的位置信息，計算每輛車在相鄰幀之間的位移。然后計算時間間隔：根據(jù)時間戳計算相鄰幀之間的時間間隔。最后計算速度：使用位移除以時間間隔得到車輛的速度。

（2） 第二、三小問

可以使用時序預(yù)測模型（例如LSTM、ARIMA）或傳統(tǒng)的回歸模型（如線性回歸、決策樹）來預(yù)測交通流擁堵情況。下面是一個基于隨機森林回歸的實現(xiàn)，提取的車流參數(shù)（車流密度、車流量、速度）來預(yù)測未來是否會發(fā)生擁堵。使用從四個觀測點視頻中提取的車流密度、車流量和速度等數(shù)據(jù)作為模型輸入，進行時序建模，并預(yù)測未來的交通擁堵狀態(tài)。基于歷史數(shù)據(jù)預(yù)測下一時間點的交通流參數(shù)（車流密度、車流量和速度），并結(jié)合定義的擁堵條件，發(fā)出實時預(yù)警。

定義擁堵狀態(tài)：當(dāng)車流密度大于0.7且速度小于40 km/h時

使用測試集驗證模型預(yù)測的車流密度、車流量和速度，并結(jié)合可視化方法來評估模型的表現(xiàn)。計算混淆矩陣和ROC曲線，分析模型的擁堵預(yù)測能力。

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 作為目標(biāo)檢測模型
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

# 將檢測結(jié)果格式化為 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))

    # 初始化數(shù)據(jù)存儲
    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

        # 記錄時間戳（可以選擇只記錄關(guān)鍵幀的時間戳以減少開銷）
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")

        # 調(diào)整圖像大小以加快處理速度
        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

        # 轉(zhuǎn)換顏色格式（BGR -> RGB）
        img_rgb = cv2.cvtColor(frame_resized, cv2.COLOR_BGR2RGB)

        # 使用模型進行檢測
        results = model(img_rgb, size=640)  # size可以根據(jù)需求調(diào)整

        # 獲取預(yù)測結(jié)果
        detections = results.xyxy[0].cpu().numpy()  # [x1, y1, x2, y2, confidence, class]
        detections = format_detections(detections, score_threshold)

        if len(detections) > 0:
            ...略
        else:
            # 如果當(dāng)前幀沒有檢測到車輛，存儲默認(rèn)值
            timestamps.append(timestamp)
            densities.append(0)
            flows.append(0)
            speeds.append(0)

    cap.release()

    # 返回結(jié)果 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)

    # 合并所有視頻的數(shù)據(jù)
    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' # 換成自己環(huán)境下的中文字體，比如'SimHei'

# 第一個觀測點數(shù)據(jù)可視化

# 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個觀測點數(shù)據(jù)可視化.png',dpi=200)
plt.show()

問題二 、三

傳統(tǒng)應(yīng)急車道的啟用多是靜態(tài)的，且缺少區(qū)域路網(wǎng)聯(lián)動分析，難以動態(tài)應(yīng)對不同場景下的交通需求。因此，模型需要集成多情景的動態(tài)分析，包括節(jié)假日高峰、突發(fā)事故、排隊消散等情景。 對不同情景的處理邏輯，如在節(jié)假日高峰期通過流量監(jiān)測系統(tǒng)提前識別大流量，啟用應(yīng)急車道；對于已經(jīng)擁堵的路段，采取臨時開放應(yīng)急車道的策略來消散車流，恢復(fù)主線運行。

虛擬應(yīng)急車道能夠通過主動交通管理技術(shù)提升救援效率，縮短事故響應(yīng)時間。因此，模型可考慮集成虛擬應(yīng)急車道的控制策略，利用車輛通信系統(tǒng)動態(tài)引導(dǎo)車輛臨時進入虛擬應(yīng)急車道，提升通行效率。 在緊急情況下，模型通過仿真預(yù)測未來交通流量，并決定是否虛擬啟用應(yīng)急車道，結(jié)合事故發(fā)生時的流量、速度等關(guān)鍵參數(shù)決定應(yīng)急車道的開放時長與區(qū)域。

應(yīng)急車道管控不僅需要考慮實時流量與交通狀況，還需結(jié)合歷史數(shù)據(jù)，分析擁堵易發(fā)節(jié)點，構(gòu)建擁堵中斷模型。對于應(yīng)急車道的開放時機和時長，需基于具體的交通狀況預(yù)測。 集成實時交通數(shù)據(jù)與歷史擁堵數(shù)據(jù)，構(gòu)建交通流概率中斷模型，動態(tài)調(diào)整應(yīng)急車道的開放時長和開放區(qū)域。

可以通過VISSIM仿真分析不同車道配置下虛擬應(yīng)急車道的效益優(yōu)化，表明在不同事故情景下，適時開啟虛擬應(yīng)急車道可有效減少平均延誤和排隊長度。模型可以根據(jù)車道封閉數(shù)量和流量密度優(yōu)化啟用條件。引入VISSIM仿真結(jié)果，基于流量密度、封閉車道數(shù)和v/C值優(yōu)化應(yīng)急車道的啟用條件，進一步提升模型的精度。

import pandas as pd
import numpy as np
import warnings

warnings.filterwarnings('ignore')
# 設(shè)置閾值
density_threshold = 0.7  # 車流密度閾值
speed_threshold = 40  # 速度閾值，單位：km/h

# 擁堵風(fēng)險計算函數(shù)
def compute_congestion_risk(density, speed, density_threshold, speed_threshold):
   ...略
    return congestion_risk

# 添加高峰期和事故判斷函數(shù)
def is_peak_hour(timestamp):
    """假設(shè)高峰時段為每天的7-9點和17-19點"""
    hour = pd.to_datetime(timestamp).hour
    return (7 <= hour <= 9) or (17 <= hour <= 19)

def is_accident_scene(density, speed):
    """假設(shè)當(dāng)速度為0且密度很高時為事故"""
    return (speed == 0) and (density > density_threshold)

# 動態(tài)管控模型：應(yīng)急車道啟用判斷
def dynamic_control(data, density_threshold, speed_threshold, time_window=10):
    # 計算每個時刻的擁堵風(fēng)險
    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)

    # 判斷是否為事故現(xiàn)場
    data['accident_scene'] = data.apply(lambda row: is_accident_scene(row['density'], row['speed']), axis=1)

    # 滑動窗口計算未來time_window分鐘內(nèi)的平均擁堵風(fēng)險
    data['avg_risk'] = data['congestion_risk'].rolling(window=time_window).mean()

    # 如果未來的平均風(fēng)險超過1，且為高峰期或事故現(xiàn)場，啟用虛擬應(yīng)急車道
    data['virtual_lane'] = np.where((data['avg_risk'] > 1) & (data['peak_hours'] | data['accident_scene']), True, False)
    return data



# 讀取數(shù)據(jù)并應(yīng)用模型
data_point = pd.read_csv('data/data_point1.csv')
data = data_point[(data_point['speed'] > 0) & (data_point['density'] > 0)]


# 應(yīng)用動態(tài)控制模型
result = dynamic_control(data, density_threshold, speed_threshold)
# 輸出結(jié)果
print(result[['timestamp', 'virtual_lane']])

問題四

1. 問題分析

原有四個采集點的布置方案中，前三個采集點之間相隔 1000m，最后一個點與第三個點之間相隔 3000m。但是這樣的布置在前1000m 的路段可以較好地捕捉交通狀態(tài)變化，但在距離較長的第三點和第四點之間，可能會丟失大量交通流變化的信息。

優(yōu)化數(shù)據(jù)采集點的布置，確保在第三點和第四點之間的 3000m 路段上能夠精確采集數(shù)據(jù)，以便做出更科學(xué)的應(yīng)急車道啟用決策。優(yōu)化模型的核心目標(biāo)是：

• 精度：確保關(guān)鍵路段（第三點到第四點）的交通狀態(tài)能夠被準(zhǔn)確地監(jiān)控。
• 成本控制：在精度要求下，盡量減少新采集點的數(shù)量，降低安裝與維護的成本。

1. 建模算法步驟

（1）初始化參數(shù)：

• 設(shè)定采集點的最小間隔 $D_{\text{min}}=1000m$。
• 設(shè)定應(yīng)急響應(yīng)時間 分鐘$T_{\text{response}}=10\text{分鐘}$。
• 獲取現(xiàn)有四個采集點的交通流量、速度和密度數(shù)據(jù)。

（2）計算每個路段的交通狀態(tài)變異性：
根據(jù)變異性公式，計算各個路段的變異性 $V_i$。

（3）判斷是否需要增加采集點：
如果某個路段的變異性 $Vi超過預(yù)定閾值$超過預(yù)定閾值$V{\text{threshold}}$，則需要在該路段新增數(shù)據(jù)采集點。

（4）優(yōu)化布點方案：

• 在第三點到第四點之間，若變異性較大（$V2 > V{\text{threshold}}$），則在1500m處增加新的采集點。
• 保證數(shù)據(jù)采集點的間隔 $D{\text{new}} \geq D{\text{min}}$。

（5）輸出新的采集點布置方案。

import pandas as pd
import numpy as np

# 初始化參數(shù)
D_min = 1000  # 最小采集點間隔，單位：米
T_response = 10  # 應(yīng)急響應(yīng)時間，單位：分鐘
V_threshold = 0.2  # 變異性閾值

# 讀取數(shù)據(jù)
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')

# 計算交通狀態(tài)變異性
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)

# 輸出優(yōu)化的布置方案
if new_point_needed:
    print("建議在第三點到第四點之間新增采集點")
else:
    print("現(xiàn)有采集點布置合理，無需新增采集點")

完整資料

轉(zhuǎn)到B站視頻介紹
bilibili.com/video/BV16AskeMEPN/?share_source=copy_web&vd_source=d2dd5fcbeeeec396792650b25c110a13