SUMO交通仿真自动化：用Python构建高效车流生成与数据分析工作流

如果你曾经手动配置过SUMO的交通流文件，一遍遍地修改rou.xml里的车辆参数，然后等待漫长的仿真运行，最后再手动从一堆XML日志里扒拉数据，那你一定能理解那种重复劳动带来的疲惫感。对于交通规划、自动驾驶算法测试或者学术研究而言，这种低效的手工作坊模式，已经成为制约想法快速验证和实验规模化的主要瓶颈。

今天，我想和你分享一套我经过多个项目迭代后沉淀下来的全自动化工作流。这套方法的核心，是利用Python脚本将SUMO仿真的“配置-运行-分析”三个环节彻底打通。我们不再满足于单个路口的简单分析，而是着眼于如何批量生成不同交通强度的场景，并自动解析、聚合、可视化海量的仿真输出数据。无论是研究信号灯配时对拥堵的影响，还是评估新交通管控策略的效果，这套标准化流程都能让你从繁琐的重复操作中解放出来，将精力真正聚焦在问题本身和结果分析上。接下来，我将从环境搭建开始，一步步拆解这个自动化管道的每个关键组件。

1. 环境准备与基础概念澄清

在开始编写自动化脚本之前，确保你的工作环境是整洁且可复现的，这是所有高效工作的基石。很多人一开始就急于写代码，却忽略了环境配置的细节，导致后期出现各种“在我的机器上能运行”的诡异问题。

首先，你需要安装SUMO及其必要的工具集。我强烈建议使用包管理器（如Ubuntu的apt或macOS的brew）进行安装，以确保所有依赖项被正确管理。对于Windows用户，SUMO官网提供了安装程序，但请注意将其安装路径添加到系统的PATH环境变量中。一个验证安装是否成功的方法是，在终端中运行以下命令：

sumo --version
netconvert --version

如果能看到版本号输出，说明核心组件安装成功。接下来是Python环境。我推荐使用conda或venv创建一个独立的虚拟环境，专门用于SUMO相关的开发。这能避免不同项目间的库版本冲突。在这个环境中，我们需要安装几个关键的Python库：

• sumolib 和 traci: 这两个库通常随SUMO一同安装，但有时需要手动链接或通过pip install sumolib获取。sumolib用于读写SUMO的各类配置文件（如.net.xml, .rou.xml），而traci用于在仿真运行时进行实时交互控制。本文的自动化流程主要使用sumolib进行前期配置。
• pandas: 数据处理和分析的瑞士军刀，后续我们解析XML输出文件后，会将其转换为DataFrame进行高效操作。
• lxml 或 xml.etree.ElementTree: 用于解析SUMO输出的XML格式的结果文件。lxml功能更强大、解析速度更快，是生产环境的首选。
• matplotlib 或 seaborn: 用于数据可视化，将枯燥的数据转化为直观的图表。
• numpy: 提供基础的数值计算支持。

你可以通过一个requirements.txt文件来管理这些依赖：

pandas>=1.4.0
lxml>=4.9.0
matplotlib>=3.5.0
numpy>=1.22.0

使用pip install -r requirements.txt一键安装。完成这些后，你的项目目录结构应该初步规划如下：

your_project/
├── configs/          # 存放不同的仿真配置文件 (.sumocfg)
├── networks/         # 存放路网文件 (.net.xml)
├── scripts/          # 存放所有Python自动化脚本
├── output/           # 存放仿真生成的原始结果 (XML格式)
├── processed_data/   # 存放处理后的结构化数据 (CSV, Parquet)
└── visualizations/   # 存放生成的图表

这种结构化的目录管理，能让你的项目随着实验复杂度的增加而依然保持清晰。

注意：SUMO的tools目录下有许多实用脚本，如randomTrips.py。确保你知道它的路径（例如/usr/share/sumo/tools/或C:\Program Files\Eclipse\Sumo\tools\），因为在我们的自动化脚本中需要调用它。

2. 自动化车流生成：超越随机Trips

大多数入门教程会教你使用randomTrips.py脚本生成随机交通需求。这确实是一个快速的起点，但对于严肃的研究或工程应用来说，完全随机的车流往往缺乏现实意义。我们的自动化流程需要能生成可控、可重复、符合特定分布的车流。

2.1 构建参数化的车流生成脚本

我们创建一个名为generate_flows.py的脚本。它的核心思想是：将交通强度（如车辆到达率）、车辆组成（小车、卡车、公交的比例）、出发时间分布等参数化，然后动态生成对应的.rou.xml文件。

首先，我们定义一个函数，它不再仅仅调用randomTrips.py，而是更精细地构建车辆和路径。以下是一个生成具有时变到达率的车流示例：

import sumolib
import pandas as pd
from lxml import etree
import os

def generate_route_file(scenario_name, total_steps=3600, flow_params=None):
    """
    生成一个SUMO路由文件。
    :param scenario_name: 场景名称，用于命名输出文件
    :param total_steps: 仿真总步长（秒）
    :param flow_params: 一个字典，定义交通流参数，例如：
        {
            'edge_flows': [('incoming_edge', 'outgoing_edge', 600)], # (起始边， 目标边， 车辆数/小时)
            'vehicle_type': {'car': {'probability': 0.8, 'maxSpeed': '13.9'},
                             'truck': {'probability': 0.2, 'maxSpeed': '8.3'}},
            'depart_variation': 'uniform' # 出发时间分布：'uniform', 'gaussian'
        }
    """
    # 创建XML根节点
    routes = etree.Element('routes')

    # 1. 定义车辆类型
    vType_elem = etree.SubElement(routes, 'vType')
    vType_elem.set('id', 'car')
    vType_elem.set('accel', '2.6')
    vType_elem.set('decel', '4.5')
    vType_elem.set('sigma', '0.5')
    vType_elem.set('length', '5.0')
    vType_elem.set('maxSpeed', '13.89') # 约50 km/h

    # 可以添加更多车辆类型，如卡车
    vType_truck = etree.SubElement(routes, 'vType')
    vType_truck.set('id', 'truck')
    vType_truck.set('accel', '1.3')
    vType_truck.set('decel', '4.5')
    vType_truck.set('sigma', '0.5')
    vType_truck.set('length', '12.0')
    vType_truck.set('maxSpeed', '8.33') # 约30 km/h

    # 2. 根据参数生成车流
    vehicle_id = 0
    for start_edge, end_edge, veh_per_hour in flow_params.get('edge_flows', []):
        # 计算总车辆数
        total_vehicles = int(veh_per_hour * (total_steps / 3600.0))
        interval = total_steps / total_vehicles if total_vehicles > 0 else float('inf')

        for i in range(total_vehicles):
            vehicle = etree.SubElement(routes, 'vehicle')
            vehicle.set('id', f'veh_{vehicle_id}')
            vehicle_id += 1

            # 随机分配车辆类型
            import random
            if random.random() < flow_params.get('vehicle_type', {}).get('truck', {}).get('probability', 0.0):
                vehicle.set('type', 'truck')
            else:
                vehicle.set('type', 'car')

            # 设置出发时间，加入随机扰动
            depart_time = i * interval
            if flow_params.get('depart_variation') == 'gaussian':
                depart_time += random.gauss(0, interval * 0.2) # 20%的随机扰动
                depart_time = max(0, min(depart_time, total_steps - 1))
            vehicle.set('depart', f'{depart_time:.2f}')

            # 创建路径（这里简化处理，实际应用中可能需要路径查找算法）
            route = etree.SubElement(vehicle, 'route')
            # 此处仅为示例，实际应根据路网计算从start_edge到end_edge的路径
            # 可以使用sumolib.net读取路网并调用shortestPath函数
            route.set('edges', f'{start_edge} {end_edge}')

    # 3. 写入文件
    tree = etree.ElementTree(routes)
    file_path = f'./configs/{scenario_name}.rou.xml'
    tree.write(file_path, pretty_print=True, xml_declaration=True, encoding='UTF-8')
    print(f"路由文件已生成: {file_path}")
    return file_path

这个函数给了你极大的灵活性。你可以通过修改flow_params字典，轻松创建早高峰、晚高峰、平峰期等不同场景，或者混合不同类型的车辆。

2.2 批量生成多场景配置

单一场景的结论往往缺乏说服力。我们需要研究不同交通负荷下的系统表现。因此，编写一个批量生成的脚本至关重要。

def batch_generate_scenarios(network_file, base_scenario_name, flow_intensities):
    """
    批量生成多个交通强度的仿真场景。
    :param network_file: 路网文件路径
    :param base_scenario_name: 基础场景名
    :param flow_intensities: 一个列表，包含不同的车辆到达率（辆/小时）
    """
    # 读取路网，获取所有可能的入口边和出口边（简化示例）
    net = sumolib.net.readNet(network_file)
    incoming_edges = [e.getID() for e in net.getEdges() if e.getLaneNumber() > 2] # 简单筛选
    outgoing_edges = incoming_edges # 简化处理

    scenarios_created = []
    for idx, intensity in enumerate(flow_intensities):
        scenario_id = f"{base_scenario_name}_flow_{intensity}"
        print(f"正在生成场景: {scenario_id}")

        # 为每个场景定义不同的流参数
        flow_params = {
            'edge_flows': [(incoming_edges[0], outgoing_edges[-1], intensity)], # 示例：一对OD对
            'vehicle_type': {'car': {'probability': 0.9}, 'truck': {'probability': 0.1}},
            'depart_variation': 'uniform'
        }

        # 生成路由文件
        rou_file = generate_route_file(scenario_id, flow_params=flow_params)

        # 生成对应的.sumocfg配置文件
        generate_config_file(scenario_id, network_file, rou_file)

        scenarios_created.append(scenario_id)
    return scenarios_created

def generate_config_file(scenario_id, net_file, rou_file):
    """生成SUMO配置文件"""
    config = etree.Element('configuration')
    input_elem = etree.SubElement(config, 'input')
    etree.SubElement(input_elem, 'net-file').set('value', os.path.basename(net_file))
    etree.SubElement(input_elem, 'route-files').set('value', os.path.basename(rou_file))

    time_elem = etree.SubElement(config, 'time')
    etree.SubElement(time_elem, 'begin').set('value', '0')
    etree.SubElement(time_elem, 'end').set('value', '3600') # 仿真1小时

    output_elem = etree.SubElement(config, 'output')
    # 输出车辆轨迹数据（FCD）
    etree.SubElement(output_elem, 'fcd-output').set('value', f'./output/{scenario_id}_fcd.xml')
    # 输出车辆旅行时间等统计信息
    etree.SubElement(output_elem, 'tripinfo-output').set('value', f'./output/{scenario_id}_tripinfo.xml')
    # 输出排队长度等车道级数据
    etree.SubElement(output_elem, 'lanearea-output').set('value', f'./output/{scenario_id}_lanearea.xml')

    tree = etree.ElementTree(config)
    config_path = f'./configs/{scenario_id}.sumocfg'
    tree.write(config_path, pretty_print=True, xml_declaration=True, encoding='UTF-8')
    print(f"配置文件已生成: {config_path}")

通过调用batch_generate_scenarios函数，并传入一个如[300, 600, 900, 1200]的流量列表，你就能一键生成从稀疏到拥堵的多个完整仿真场景。所有配置文件都会整齐地存放在configs/目录下，并以流量强度清晰命名。

3. 自动化仿真执行与结果提取

生成了几十个配置文件后，手动一个个运行SUMO是不现实的。我们需要一个“调度器”来批量执行仿真，并确保输出结果被妥善保存。

3.1 并行化仿真运行

我们可以利用Python的subprocess模块来调用SUMO命令行，并结合concurrent.futures库实现并行运行，充分利用多核CPU，将数小时的仿真时间压缩到几分钟。

import subprocess
from concurrent.futures import ProcessPoolExecutor, as_completed
import glob

def run_simulation(config_file):
    """运行单个仿真任务"""
    cmd = ['sumo', '-c', config_file, '--no-warnings', 'true']
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=300) # 设置5分钟超时
        print(f"成功完成: {config_file}")
        return config_file, True, None
    except subprocess.TimeoutExpired:
        print(f"超时: {config_file}")
        return config_file, False, "Timeout"
    except subprocess.CalledProcessError as e:
        print(f"运行失败: {config_file}, 错误: {e.stderr[:200]}")
        return config_file, False, e.stderr

def batch_run_simulations(config_dir='./configs/', max_workers=4):
    """批量并行运行仿真"""
    config_files = glob.glob(os.path.join(config_dir, '*.sumocfg'))
    print(f"找到 {len(config_files)} 个配置文件待运行。")

    results = []
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        future_to_config = {executor.submit(run_simulation, cfg): cfg for cfg in config_files}
        for future in as_completed(future_to_config):
            config_file = future_to_config[future]
            try:
                result = future.result()
                results.append(result)
            except Exception as exc:
                print(f'{config_file} 产生了异常: {exc}')
                results.append((config_file, False, str(exc)))

    # 汇总结果
    success_count = sum(1 for _, success, _ in results if success)
    print(f"仿真完成。成功: {success_count}, 失败: {len(results)-success_count}")
    return results

提示：并行运行仿真时，请确保你的输出文件路径（在.sumocfg中定义）是唯一的，或者将输出重定向到不同的子目录，避免文件写入冲突。

3.2 从XML到结构化数据的智能解析

仿真结束后，我们得到的是原始的XML文件。手动查看这些文件无异于大海捞针。我们需要编写解析器，将XML中我们关心的指标（如车辆速度、旅行时间、排队长度）提取出来，并转换为pandas DataFrame，为后续分析做准备。

以下是一个解析tripinfo输出文件的函数示例，它提取每次出行的关键性能指标：

import pandas as pd
from lxml import etree

def parse_tripinfo_xml(xml_file_path):
    """
    解析tripinfo输出文件，提取每次出行的统计信息。
    """
    trips_data = []
    tree = etree.parse(xml_file_path)
    root = tree.getroot()

    for tripinfo in root.findall('tripinfo'):
        trip_dict = {
            'id': tripinfo.get('id'),
            'depart': float(tripinfo.get('depart')),
            'arrival': float(tripinfo.get('arrival')) if tripinfo.get('arrival') else None,
            'duration': float(tripinfo.get('duration')),
            'routeLength': float(tripinfo.get('routeLength')),
            'waitingTime': float(tripinfo.get('waitingTime')) if tripinfo.get('waitingTime') else 0.0,
            'timeLoss': float(tripinfo.get('timeLoss')) if tripinfo.get('timeLoss') else 0.0,
            'departLane': tripinfo.get('departLane'),
            'arrivalLane': tripinfo.get('arrivalLane'),
            'vehicle_type': tripinfo.get('vType')
        }
        # 计算平均速度
        if trip_dict['duration'] > 0:
            trip_dict['avgSpeed'] = trip_dict['routeLength'] / trip_dict['duration']
        else:
            trip_dict['avgSpeed'] = 0.0
        trips_data.append(trip_dict)

    df = pd.DataFrame(trips_data)
    # 从文件名推断场景信息（例如流量强度）
    scenario_name = os.path.basename(xml_file_path).replace('_tripinfo.xml', '')
    df['scenario'] = scenario_name
    return df

def aggregate_scenario_results(output_dir='./output/'):
    """聚合所有场景的tripinfo结果"""
    all_dfs = []
    tripinfo_files = glob.glob(os.path.join(output_dir, '*_tripinfo.xml'))

    for file in tripinfo_files:
        print(f"正在处理: {file}")
        try:
            df = parse_tripinfo_xml(file)
            all_dfs.append(df)
        except Exception as e:
            print(f"解析文件 {file} 时出错: {e}")

    if all_dfs:
        combined_df = pd.concat(all_dfs, ignore_index=True)
        # 保存为更高效的Parquet格式，便于后续快速读取
        combined_df.to_parquet('./processed_data/all_tripinfo.parquet')
        print("所有场景数据已聚合并保存。")
        return combined_df
    else:
        print("未找到可解析的tripinfo文件。")
        return pd.DataFrame()

这个DataFrame包含了每个车辆每次出行的详细信息。你可以轻松地按场景（scenario列）进行分组，计算整个路网的平均旅行时间、平均速度、总延误等宏观指标。

4. 数据可视化与深度分析洞察

数据聚合完成后，真正的乐趣——分析开始了。可视化的目的不仅仅是画图，而是为了发现规律、验证假设和讲述故事。

4.1 关键性能指标（KPI）可视化

我们首先计算每个场景的宏观KPI，并绘制其随交通流量（到达率）变化的趋势图。这是评估路网性能的基础。

import matplotlib.pyplot as plt
import seaborn as sns

def plot_kpi_trends(combined_df):
    """绘制关键性能指标随场景变化的趋势图"""
    if combined_df.empty:
        print("数据框为空，无法绘图。")
        return

    # 从场景名中提取流量强度（假设场景名格式为 `base_flow_600`）
    combined_df['flow_rate'] = combined_df['scenario'].str.extract('flow_(\d+)').astype(float)

    # 按流量分组计算聚合指标
    kpi_df = combined_df.groupby('flow_rate').agg({
        'duration': 'mean',
        'avgSpeed': 'mean',
        'timeLoss': 'mean',
        'waitingTime': 'mean',
        'id': 'count'  # 车辆总数
    }).rename(columns={'id': 'vehicle_count'}).reset_index()

    # 创建多子图
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    fig.suptitle('路网性能指标 vs 交通流量', fontsize=16)

    # 平均旅行时间
    axes[0, 0].plot(kpi_df['flow_rate'], kpi_df['duration'], marker='o', linewidth=2)
    axes[0, 0].set_xlabel('到达率 (辆/小时)')
    axes[0, 0].set_ylabel('平均旅行时间 (秒)')
    axes[0, 0].grid(True, linestyle='--', alpha=0.7)
    axes[0, 0].set_title('平均旅行时间')

    # 平均速度
    axes[0, 1].plot(kpi_df['flow_rate'], kpi_df['avgSpeed'], marker='s', color='green', linewidth=2)
    axes[0, 1].set_xlabel('到达率 (辆/小时)')
    axes[0, 1].set_ylabel('平均速度 (米/秒)')
    axes[0, 1].grid(True, linestyle='--', alpha=0.7)
    axes[0, 1].set_title('平均速度')

    # 总时间损失
    axes[1, 0].bar(kpi_df['flow_rate'], kpi_df['timeLoss'] * kpi_df['vehicle_count'], color='orange')
    axes[1, 0].set_xlabel('到达率 (辆/小时)')
    axes[1, 0].set_ylabel('总时间损失 (秒)')
    axes[1, 0].grid(True, linestyle='--', alpha=0.7, axis='y')
    axes[1, 0].set_title('系统总时间损失')

    # 平均等待时间
    axes[1, 1].plot(kpi_df['flow_rate'], kpi_df['waitingTime'], marker='^', color='red', linewidth=2)
    axes[1, 1].set_xlabel('到达率 (辆/小时)')
    axes[1, 1].set_ylabel('平均等待时间 (秒)')
    axes[1, 1].grid(True, linestyle='--', alpha=0.7)
    axes[1, 1].set_title('平均等待时间（如遇红灯）')

    plt.tight_layout()
    plt.savefig('./visualizations/kpi_trends.png', dpi=300)
    plt.show()

    return kpi_df

通过这张综合图表，你可以一眼看出路网的“崩溃点”——当流量增加到某个值时，平均旅行时间和时间损失会急剧上升，而平均速度会骤降。这个点就是你当前路网设计或信号配时的容量极限。

4.2 时空轨迹分析与热点识别

除了宏观KPI，微观的车辆轨迹数据（FCD输出）能揭示更细致的问题。例如，我们可以找出常发性拥堵路段或交叉口。

def analyze_congestion_hotspots(fcd_file_pattern='./output/*_fcd.xml'):
    """分析FCD数据，识别拥堵时空热点"""
    import numpy as np
    from collections import defaultdict

    hotspot_data = defaultdict(list) # 键为 (timestep, lane)，值为速度列表
    fcd_files = glob.glob(fcd_file_pattern)

    for file in fcd_files:
        tree = etree.parse(file)
        root = tree.getroot()
        for timestep_elem in root.findall('timestep'):
            time = float(timestep_elem.get('time'))
            for vehicle_elem in timestep_elem.findall('vehicle'):
                lane = vehicle_elem.get('lane')
                speed = float(vehicle_elem.get('speed'))
                hotspot_data[(time, lane)].append(speed)

    # 计算每个时空单元的平均速度
    congestion_map = []
    for (time, lane), speeds in hotspot_data.items():
        avg_speed = np.mean(speeds)
        congestion_map.append({
            'time': time,
            'lane': lane,
            'avg_speed': avg_speed,
            'vehicle_count': len(speeds)
        })

    congestion_df = pd.DataFrame(congestion_map)
    # 定义拥堵阈值，例如速度低于5米/秒（18公里/小时）
    congestion_threshold = 5.0
    congestion_df['is_congested'] = congestion_df['avg_speed'] < congestion_threshold

    # 找出最拥堵的车道（拥堵次数最多）
    lane_congestion = congestion_df[congestion_df['is_congested']].groupby('lane').size().sort_values(ascending=False)
    print("拥堵最频繁的车道TOP 5:")
    print(lane_congestion.head())

    # 找出最拥堵的时间段
    time_congestion = congestion_df[congestion_df['is_congested']].groupby('time').size()
    peak_congestion_time = time_congestion.idxmax()
    print(f"拥堵最严重的时间点: {peak_congestion_time} 秒")

    return congestion_df

这个分析能帮你精准定位问题所在。也许你会发现，某个左转车道在特定时间段总是拥堵，这可能是车道划分或信号相位设置不合理导致的。

4.3 自动化报告生成

最后，我们可以将上述所有分析步骤整合，并生成一个简单的文本或HTML报告，汇总本次批量实验的核心发现。

def generate_analysis_report(kpi_df, congestion_df, report_path='./visualizations/analysis_report.txt'):
    """生成简单的文本分析报告"""
    with open(report_path, 'w') as f:
        f.write("="*60 + "\n")
        f.write("SUMO批量仿真自动化分析报告\n")
        f.write("="*60 + "\n\n")

        f.write("1. 宏观性能概览\n")
        f.write("-"*40 + "\n")
        if not kpi_df.empty:
            # 找到性能拐点（例如，旅行时间开始显著增长的点）
            duration_increase = kpi_df['duration'].pct_change().dropna()
            if not duration_increase.empty:
                critical_flow_idx = duration_increase[duration_increase > 0.5].index.min() # 定义增长超过50%为拐点
                if pd.notna(critical_flow_idx):
                    critical_flow = kpi_df.loc[critical_flow_idx, 'flow_rate']
                    f.write(f"   * 预估路网服务容量拐点约在 {critical_flow:.0f} 辆/小时 附近。\n")
                else:
                    f.write("   * 在测试流量范围内，未发现明显的性能拐点。\n")
            f.write(f"   * 测试流量范围: {kpi_df['flow_rate'].min():.0f} - {kpi_df['flow_rate'].max():.0f} 辆/小时。\n")
            f.write(f"   * 平均旅行时间范围: {kpi_df['duration'].min():.1f} - {kpi_df['duration'].max():.1f} 秒。\n")
            f.write(f"   * 平均速度范围: {kpi_df['avgSpeed'].min():.1f} - {kpi_df['avgSpeed'].max():.1f} 米/秒。\n\n")

        f.write("2. 拥堵热点分析\n")
        f.write("-"*40 + "\n")
        if not congestion_df.empty:
            top_congested_lanes = congestion_df[congestion_df['is_congested']]['lane'].value_counts().head(3)
            if not top_congested_lanes.empty:
                f.write("   * 最常发生拥堵的车道（前3）：\n")
                for lane, count in top_congested_lanes.items():
                    f.write(f"      - 车道 {lane}: 拥堵记录 {count} 次\n")
            else:
                f.write("   * 未检测到严重拥堵（低于阈值）。\n")

        f.write("\n3. 建议与后续步骤\n")
        f.write("-"*40 + "\n")
        f.write("   * 针对识别出的拥堵热点车道，检查其上游信号配时或车道功能设置。\n")
        f.write("   * 若在较低流量下即出现性能拐点，需考虑优化路网拓扑或交叉口设计。\n")
        f.write("   * 可进一步引入不同车型比例、驾驶行为参数（如跟驰模型），进行敏感性分析。\n")
        f.write("   * 本报告基于自动化流程生成，详细数据与图表请查阅 `processed_data/` 和 `visualizations/` 目录。\n")

    print(f"分析报告已生成: {report_path}")

从环境搭建、参数化车流生成，到批量仿真、数据解析，再到可视化和报告生成，我们构建了一个完整的、可扩展的SUMO自动化分析闭环。这套流程的价值在于其可重复性和可扩展性。当你需要测试一个新的信号控制算法时，只需修改车流生成逻辑或替换路网文件，然后重新运行整个管道即可。所有中间数据和最终图表都会自动更新。这彻底改变了我们与SUMO交互的方式，从一次性的、手动的“实验”转变为高效的、数据驱动的“分析流水线”。