基于SimPy的多方向信号交叉口交通仿真代码修改咨询

我正在用SimPy设计一个包含三个进口道的信号交叉口交通仿真，现有代码实现了单车道单方向车辆驶入、信号灯红绿切换的逻辑。现在需要修改代码，实现同一交叉口处多组不同到达率的信号灯按序工作，目标是把多条道路连接到交叉口，用信号灯管控车流。

原仿真代码 #

# Section 1: 导入模块并定义工具类
from collections import deque
from numpy import random
import simpy
from simpy.util import start_delayed


class Struct(object):
    """
    简单工具类，通过关键字参数初始化属性，后续可修改
    """
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)


# Section 2: 初始化参数
random.seed([1, 2, 3])

# 仿真总时长（秒）
end_time = 200.0

# 车辆到达率（每秒0.2辆）
arrival_rate = 0.2
t_interarrival_mean = 1.0 / arrival_rate

# 信号灯红绿时长
t_green = 30.0; t_red = 40.0

# 车辆通过交叉口的时间（三角分布）
t_depart_left = 1.6; t_depart_mode = 2.0; t_depart_right = 2.4

# 初始队列
queue = deque()

# 车辆计数
arrival_count = departure_count = 0

# 统计数据
Q_stats = Struct(count=0, cars_waiting=0)
W_stats = Struct(count=0, waiting_time=0.0)


# Section 3: 车辆到达事件
def arrival():
    """
    模拟车辆到达，服从泊松分布
    """
    global arrival_count, env, light, queue

    while True:
        arrival_count += 1

        if light == 'red' or len(queue):
            # 红灯或有队列，车辆加入排队
            queue.append((arrival_count, env.now))
            print(f"车辆#{arrival_count} 在时间{env.now:.3f}到达，加入队列第{len(queue)}位。")
        else:
            # 绿灯且无队列，直接通过
            print(f"车辆#{arrival_count} 在时间{env.now:.3f}到达，绿灯无等待直接通过。")
            W_stats.count += 1

        # 调度下一辆车到达
        yield env.timeout(random.exponential(t_interarrival_mean))


# Section 4: 事件函数定义
# Section 4.1: 车辆离开事件
def departure():
    """
    模拟车辆离开交叉口，即排队车辆通过路口
    """
    global env, queue

    while True:
        car_number, t_arrival = queue.popleft()
        print(f"车辆#{car_number} 在时间{env.now:.3f}离开，队列剩余{len(queue)}辆车。")

        # 更新等待时间统计
        W_stats.count += 1
        W_stats.waiting_time += env.now - t_arrival

        # 红灯或队列为空时，停止调度离开事件
        if light == 'red' or len(queue) == 0:
            return

        # 生成下一辆车的离开延迟
        delay = random.triangular(left=t_depart_left, mode=t_depart_mode, right=t_depart_right)
        yield env.timeout(delay)


# Section 4.2: 信号灯状态切换事件
def light():
    """
    模拟信号灯红绿切换，无黄灯状态
    """
    global env, light

    while True:
        # 切换为绿灯
        light = 'green'
        print(f"\n信号灯在时间{env.now:.3f}变为绿灯。")

        # 队列有车则调度离开事件
        if len(queue):
            delay = random.triangular(left=t_depart_left, mode=t_depart_mode, right=t_depart_right)
            start_delayed(env, departure(), delay=delay)

        # 绿灯持续时长
        yield env.timeout(t_green)

        # 切换为红灯
        light = 'red'
        print(f"\n信号灯在时间{env.now:.3f}变为红灯。")

        # 红灯持续时长
        yield env.timeout(t_red)


# Section 4.3: 队列状态监控事件
def monitor():
    """
    每隔1秒收集一次队列状态统计数据
    """
    global env, Q_stats

    while True:
        Q_stats.count += 1
        Q_stats.cars_waiting += len(queue)
        yield env.timeout(1.0)


# Section 5: 初始化并运行仿真
print("\n单方向信号灯交叉口车辆仿真\n\n")

# 创建仿真环境
env = simpy.Environment()

# 调度信号灯初始切换事件
env.process(light())

# 调度第一辆车到达
t_first_arrival = random.exponential(t_interarrival_mean)
start_delayed(env, arrival(), delay=t_first_arrival)

# 调度监控事件
env.process(monitor())

# 运行仿真
env.run(until=end_time)


# Section 6: 输出统计结果
print("\n\n      *** 统计结果 ***\n\n")
print(f"平均等待车辆数: {Q_stats.cars_waiting / float(Q_stats.count):.3f}")
print(f"平均等待时间（秒）: {W_stats.waiting_time / float(W_stats.count):.3f}")

改造后的代码实现 #

核心改动说明 #

1. 用Road类封装每条进口道的属性（到达率、队列、信号灯、统计数据），避免全局变量冲突
2. 新增SignalController类实现信号灯相位调度，按预设顺序切换各道路的绿灯
3. 每个道路独立处理车辆到达和离开事件，互不干扰
4. 全局监控器汇总所有道路的统计数据

from collections import deque
from numpy import random
import simpy
from simpy.util import start_delayed


class Struct(object):
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)


class Road:
    """
    进口道类，封装单条道路的车辆到达、排队、信号灯逻辑
    """
    def __init__(self, env, road_id, arrival_rate, depart_params):
        self.env = env
        self.road_id = road_id  # 道路编号
        self.arrival_rate = arrival_rate  # 车辆到达率
        self.t_interarrival_mean = 1.0 / arrival_rate
        self.depart_params = depart_params  # 车辆通过时间参数
        self.light = 'red'  # 初始红灯
        self.queue = deque()  # 车辆队列
        self.arrival_count = 0  # 本道路到达车辆数
        # 统计数据
        self.Q_stats = Struct(count=0, cars_waiting=0)
        self.W_stats = Struct(count=0, waiting_time=0.0)

        # 启动车辆到达事件
        t_first_arrival = random.exponential(self.t_interarrival_mean)
        start_delayed(env, self.arrival(), delay=t_first_arrival)
        # 启动本道路监控事件
        env.process(self.monitor())

    def arrival(self):
        """本道路车辆到达事件"""
        while True:
            self.arrival_count += 1
            car_id = f"{self.road_id}-{self.arrival_count}"

            if self.light == 'red' or len(self.queue):
                # 红灯或有队列，加入排队
                self.queue.append((car_id, self.env.now))
                print(f"[{self.env.now:.3f}] 道路{self.road_id} 车辆{car_id}到达，加入队列第{len(self.queue)}位")
            else:
                # 绿灯无等待，直接通过
                print(f"[{self.env.now:.3f}] 道路{self.road_id} 车辆{car_id}到达，绿灯无等待直接通过")
                self.W_stats.count += 1

            # 调度下一辆车到达
            yield self.env.timeout(random.exponential(self.t_interarrival_mean))

    def departure(self):
        """本道路车辆离开事件"""
        while True:
            if len(self.queue) == 0 or self.light == 'red':
                return

            car_id, t_arrival = self.queue.popleft()
            print(f"[{self.env.now:.3f}] 道路{self.road_id} 车辆{car_id}离开，队列剩余{len(self.queue)}辆车")

            # 更新等待时间统计
            self.W_stats.count += 1
            self.W_stats.waiting_time += self.env.now - t_arrival

            # 生成下一辆车的通过延迟
            delay = random.triangular(**self.depart_params)
            yield self.env.timeout(delay)

    def turn_green(self):
        """切换为绿灯，触发车辆放行"""
        self.light = 'green'
        print(f"\n[{self.env.now:.3f}] 道路{self.road_id} 信号灯变为绿灯")
        if len(self.queue) > 0:
            delay = random.triangular(**self.depart_params)
            start_delayed(self.env, self.departure(), delay=delay)

    def turn_red(self):
        """切换为红灯"""
        self.light = 'red'
        print(f"\n[{self.env.now:.3f}] 道路{self.road_id} 信号灯变为红灯")

    def monitor(self):
        """监控本道路队列状态"""
        while True:
            self.Q_stats.count += 1
            self.Q_stats.cars_waiting += len(self.queue)
            yield self.env.timeout(1.0)


class SignalController:
    """
    信号灯控制器，按相位顺序调度各道路的绿灯
    """
    def __init__(self, env, roads, phase_sequence):
        self.env = env
        self.roads = roads  # 所有道路对象
        self.phase_sequence = phase_sequence  # 相位序列：[(道路ID, 绿灯时长), ...]
        env.process(self.run_phases())

    def run_phases(self):
        """循环执行相位序列"""
        while True:
            for phase in self.phase_sequence:
                road_id, green_duration = phase
                target_road = next(r for r in self.roads if r.road_id == road_id)

                # 切换目标道路为绿灯
                target_road.turn_green()
                # 绿灯持续时长
                yield self.env.timeout(green_duration)
                # 切换目标道路为红灯
                target_road.turn_red()
                # 全红间隔（可选，避免冲突）
                yield self.env.timeout(3.0)


def global_monitor(env, roads):
    """全局监控，汇总所有道路的统计数据"""
    while True:
        total_waiting = sum(len(r.queue) for r in roads)
        print(f"\n[{env.now:.3f}] 全局队列状态：各道路等待车辆数分别为{[len(r.queue) for r in roads]}，总计{total_waiting}辆")
        yield env.timeout(10.0)


# 主仿真逻辑
if __name__ == "__main__":
    random.seed([1, 2, 3])
    end_time = 300.0

    # 定义车辆通过时间参数
    depart_params = {
        "left": 1.6,
        "mode": 2.0,
        "right": 2.4
    }

    # 创建仿真环境
    env = simpy.Environment()

    # 创建三条进口道，设置不同到达率
    roads = [
        Road(env, "北进口", 0.2, depart_params),
        Road(env, "东进口", 0.3, depart_params),
        Road(env, "南进口", 0.15, depart_params)
    ]

    # 定义信号灯相位序列：按北→东→南的顺序切换绿灯
    phase_sequence = [
        ("北进口", 30),
        ("东进口", 25),
        ("南进口", 20)
    ]

    # 启动信号灯控制器
    SignalController(env, roads, phase_sequence)

    # 启动全局监控
    env.process(global_monitor(env, roads))

    # 运行仿真
    print("\n三进口道信号灯交叉口仿真启动\n")
    env.run(until=end_time)

    # 输出各道路统计结果
    print("\n\n      *** 各道路统计结果 ***\n")
    for road in roads:
        avg_wait_cars = road.Q_stats.cars_waiting / float(road.Q_stats.count) if road.Q_stats.count > 0 else 0
        avg_wait_time = road.W_stats.waiting_time / float(road.W_stats.count) if road.W_stats.count > 0 else 0
        print(f"道路{road.road_id}:")
        print(f"  平均等待车辆数: {avg_wait_cars:.3f}")
        print(f"  平均等待时间: {avg_wait_time:.3f}秒")
        print("-"*30)

内容的提问来源于stack exchange，提问作者AAT