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

題目

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

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

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

1 問(wèn)題一

（1） 第一小問(wèn)

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

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

（2） 第二、三小問(wèn)

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

定義擁堵?tīng)顟B(tài)：當(dāng)車流密度大于0.7且速度小于40 km/h時(shí)

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

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  # 用于顯示進(jìn)度條
device = torch.device("cuda" if True else "cpu")
# 使用 YOLOv5 作為目標(biāo)檢測(cè)模型
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

# 將檢測(cè)結(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

# 計(jì)算車輛速度 (像素/秒)
def calculate_speed(track, prev_track, fps):
    if prev_track is None:
        return 0
    # 計(jì)算位移 (歐幾里得距離)
    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  # 時(shí)間間隔（秒）
    return displacement / time_interval  # 像素/秒

# 處理視頻，提取車輛流量信息
def detect_and_track(video_path, model, tracker, device, score_threshold=0.5, resize_width=640):
    # 打開(kāi)視頻
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise IOError(f"無(wú)法打開(kāi)視頻文件 {video_path}")

    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    # 初始化數(shù)據(jù)存儲(chǔ)
    timestamps = []
    densities = []
    flows = []
    speeds = []

    tracked_objects = {
   }  # 存儲(chǔ)每個(gè)車輛的歷史軌跡
    total_flow = 0  # 總車流量

    # 使用 tqdm 顯示處理進(jìn)度
    for _ in tqdm(range(frame_count), desc=f"Processing {video_path}"):
        ret, frame = cap.read()
        if not ret:
            break

        # 記錄時(shí)間戳（可以選擇只記錄關(guān)鍵幀的時(shí)間戳以減少開(kāi)銷）
        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)

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

        # 獲取預(yù)測(cè)結(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)前幀沒(méi)有檢測(cè)到車輛，存儲(chǔ)默認(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

可視化提取出來(lái)的特征

import matplotlib.pyplot as plt
import pandas as pd
plt.rcParams['font.sans-serif'] = 'SimSun' # 換成自己環(huán)境下的中文字體，比如'SimHei'

# 第一個(gè)觀測(cè)點(diǎn)數(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('車流密度隨時(shí)間變化')
plt.xlabel('時(shí)間戳 (ms)')
plt.ylabel('車流密度')
plt.legend()

plt.subplot(3, 1, 2)
plt.plot(range(len(result_df)), result_df['flow'], label='車流量', color='orange')
plt.title('車流量隨時(shí)間變化')
plt.xlabel('時(shí)間戳 (ms)')
plt.ylabel('車流量')
plt.legend()

plt.subplot(3, 1, 3)
plt.plot(range(len(result_df)), result_df['speed'], label='平均速度', color='green')
plt.title('平均速度隨時(shí)間變化')
plt.xlabel('時(shí)間戳 (ms)')
plt.ylabel('平均速度 (像素/秒)')
plt.legend()

plt.tight_layout()
plt.savefig('img/第4個(gè)觀測(cè)點(diǎn)數(shù)據(jù)可視化.png',dpi=200)
plt.show()

問(wèn)題二 、三

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

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

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

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

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)險(xiǎn)計(jì)算函數(shù)
def compute_congestion_risk(density, speed, density_threshold, speed_threshold):
   ...略
    return congestion_risk

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

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

# 動(dòng)態(tài)管控模型：應(yīng)急車道啟用判斷
def dynamic_control(data, density_threshold, speed_threshold, time_window=10):
    # 計(jì)算每個(gè)時(shí)刻的擁堵風(fēng)險(xiǎn)
    data['congestion_risk'] = data.apply(
        lambda row: compute_congestion_risk(row['density'], row['speed'], density_threshold, speed_threshold), axis=1)

    # 判斷是否為高峰時(shí)段
    data['peak_hours'] = data['timestamp'].apply(is_peak_hour)

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

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

    # 如果未來(lái)的平均風(fēng)險(xiǎn)超過(guò)1，且為高峰期或事故現(xiàn)場(chǎng)，啟用虛擬應(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)用動(dòng)態(tài)控制模型
result = dynamic_control(data, density_threshold, speed_threshold)
# 輸出結(jié)果
print(result[['timestamp', 'virtual_lane']])

問(wèn)題四

1. 問(wèn)題分析

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

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

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

1. 建模算法步驟

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

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

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

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

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

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

（5）輸出新的采集點(diǎn)布置方案。

import pandas as pd
import numpy as np

# 初始化參數(shù)
D_min = 1000  # 最小采集點(diǎn)間隔，單位：米
T_response = 10  # 應(yīng)急響應(yīng)時(shí)間，單位：分鐘
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')

# 計(jì)算交通狀態(tài)變異性
def compute_variability(data):
   ...略
# 計(jì)算每個(gè)路段的變異性
V1 = compute_variability(data_point1)
V2 = compute_variability(data_point2)
V3 = compute_variability(data_point3)
V4 = compute_variability(data_point4)

# 判斷是否需要增加采集點(diǎn)
def check_new_point(V, V_threshold, D_current, D_min):
    if V > V_threshold and D_current > D_min:
        return True
    return False

# 路段間隔（第三點(diǎn)到第四點(diǎn)為3000m，其他為1000m）
D1, D2, D3 = 1000, 1000, 3000

# 判斷第三點(diǎn)和第四點(diǎn)之間是否需要新增采集點(diǎn)
new_point_needed = check_new_point(V3, V_threshold, D3, D_min)

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

完整資料

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