Python 中的交通流量模拟

2025年3月17日 | 阅读 24 分钟

众所周知，交通流量并非总是顺畅的；然而，车辆流畅地通过十字路口、转弯和在交通信号灯前停车，看起来会很壮观。这一观察让我们开始思考交通流量对人类文明的重要性。

在接下来的教程中，我们将了解交通模拟的重要性。我们还将比较模拟交通的各种可能方法，最后演示一个带有源代码的模拟。

了解交通流量模拟的重要性

交通模拟的关键目的是在真实世界之外生成数据。我们不需要在真实世界中测试管理交通系统的新想法或通过传感器收集数据，而是可以使用在软件上执行的模型来预测交通流量。该模型支持加速交通系统的优化和数据收集。模拟比真实世界测试更便宜、更快。

训练机器学习 (ML) 模型需要大量数据集，这些数据集可能复杂且昂贵。通过模拟交通流量程序化地生成数据，可以轻松地进行修改以满足所需数据的精确时间。

建模

我们将从对交通系统进行建模开始，以便以数学方式分析和优化交通系统。这样的模型应根据输入参数（道路网络几何形状、每分钟车辆数、车速等）真实地描绘交通流量。

根据操作的级别，交通系统模型通常分为三类：

微观模型：这些模型单独表示每辆车，并尝试复制驾驶员行为。
宏观模型：这些模型用交通密度（每公里车辆数）和交通流量（每分钟车辆数）来描述车辆的运动。它们通常类似于流体流动。
介观模型：这些模型是混合模型，结合了微观和宏观模型的特点。它们将流量开发为“车队”。

在接下来的教程中，我们将使用微观模型。

理解微观模型

微观驾驶员模型描述了单个驾驶员/车辆的行为。因此，它必须是一个多主体系统；也就是说，每辆车都独立运行，并从其环境中获取输入。

在微观模型中，每辆车都被编号为 *i*。第 *i* 辆车跟随第 (*i*-1) 辆车。我们将第 *i* 辆车沿道路的位置表示为 *x_i*，其速度表示为 *v_i*，其长度表示为 *l_i*。这对所有车辆都适用。

我们将车辆碰撞点之间的距离表示为 *s_i*，将第 *i* 辆车与其前车（编号为 *i*-1 的车辆）之间的速度差表示为 *∆v_i*。

理解智能驾驶员模型 (IDM)

2000 年，Treiber、Hennecke 和 Helbing 开发了一个名为智能驾驶员模型 (IDM) 的模型。该模型根据第 *i* 辆车的可变参数和其前车的可变参数来说明第 *i* 辆车的加速度。我们可以将动力学方程定义如下：

既然我们已经见过 *s_i*、*v_i* 和 ∆*v_i*，那么其他参数如下：

1. s₀ᵢ: 此参数是车辆 *i* 和 *i*-1 之间的最小期望距离。

2. v₀ᵢ: 此参数是车辆 *i* 的最大期望速度。

3. δ: 此参数是加速度指数，它控制加速度的“平滑度”。

4. Tᵢ: 此参数是车辆 *i* 驾驶员的反应时间。

5. aᵢ: 此参数是车辆 *i* 的最大加速度。

6. bᵢ: 此参数是车辆 *i* 的舒适减速度。

7. s^*: 此参数是车辆 *i* 和 *i*-1 之间的期望距离。

首先，我们将查看 *s*，它是一个由三项组成的距离。

8. s₀ᵢ: 如前所述，此参数是最小期望距离。

9. vᵢTᵢ: 此参数是反应时间安全距离。这是驾驶员在做出反应（刹车）之前车辆行驶的距离。

由于速度是距离除以时间，因此距离是速度乘以时间。

10. (vᵢ Δvᵢ)/√(2aᵢ bᵢ): 此参数是一个更复杂的项。它是一个基于速度差的安全距离。它表示车辆在不刹车过猛（减速度应小于 *bᵢ*）的情况下减速（不撞上前车）所需的时间。

理解智能驾驶员模型的工作原理

我们可以假设车辆沿直线路径移动并遵循以下方程：

为了更好地理解上述方程，我们可以将其分为两部分。我们有自由道路加速度和交互加速度。

自由道路加速度是在空旷道路上（没有前方车辆）的加速度。如果我们以车速 *v_i* 为函数绘制加速度，我们将得到以下结果：

图片：加速度与速度的关系图

在上图中，我们注意到当车辆静止（*v_i* = 0）时，加速度最大。当车速接近最大速度 *v₀₁* 时，加速度变为零。这个陈述意味着自由道路的加速度会将车辆加速到最大速度。

如果我们想为不同的 *δ* 值绘制 v-a 图，我们将注意到它控制着驾驶员在接近最大速度时减速的快慢，进而调节加速度/减速度的平滑度。

图片：加速度与速度的关系图

交互加速度与与前车的交互有关。我们可以通过考虑以下情况来更好地理解其工作原理：

在自由道路上（*v_i* >> *s*^*)

当前车很远时，距离 *s_i* 支配着期望距离 *s*^*，交互几乎为 0。

这表明我们可以通过自由道路加速度来控制车辆。

在高速接近率下（∆*v_i*）

当速度差很高时，交互加速度试图通过（vi2 来实现这一点。

在小的距离差下（?_? << 1 且 ∆?_? = 0）

加速度变成一个简单的排斥力。

理解交通道路网络模型

图片：有向图示例

Set

我们必须对道路网络进行建模。我们可以通过有向图 *G = (V,E)* 来实现，其中：

*V* 被视为顶点（或节点）的集合。
*E* 被视为代表道路的边的集合。

每辆车都有一个由多条道路（边）组成的路径。我们将在同一条道路（同一条边）上对车辆应用智能驾驶员模型。当车辆到达道路末端时，我们可以将其从该道路移除并添加到其下一条道路。

我们不会在模拟中维护一个节点集（数组）。然而，每条道路都将通过其起始节点和结束节点的显式值来定义。

理解随机车辆生成器

我们有两种选择来将车辆添加到模拟中：

选项一：我们可以通过创建 `Vehicle` 类的实例并将其添加到车辆列表中来手动添加每辆车到模拟中。

选项二：我们也可以根据预定义的概率使用随机方法来添加车辆。

为了选择第二个选项，我们需要定义一个随机车辆生成器。

我们可以通过两个约束来定义随机车辆生成器：

车辆生成率 (τ)：（以每分钟车辆数为单位）此约束描述了平均每分钟应添加到模拟中的车辆数量。
车辆配置列表 (L)：此约束是元组列表，包含车辆的配置和概率。

随机车辆生成器以概率 *p_i* 生成车辆 *V_i*。

理解交通信号灯

交通信号灯放置在顶点，并由两个区域表征：

减速区：减速距离和因子表征此区域。减速区是车辆通过减速因子减速到最大速度的区域。
停车区：停车距离表征此区域。可以通过以下动力学方程中的阻尼力来实现停车区：

交通流量模拟的项目代码

对于这个项目，我们将采用面向对象的方法。因此，每条道路和每辆车都必须定义为一个类。

我们将在接下来的各种类中频繁使用以下初始化函数。此函数允许我们通过 `set_default_config` 函数设置当前类的默认配置。它还接收一个字典，并将字典中的每个属性设置为当前类实例的属性。这样，就不需要担心更新不同类的 `__init__` 函数或未来的更改了。

让我们看下面的代码片段来说明这一点：

文件：init.py

def __init__(self, config = {}):
    # Setting the default configuration
    self.set_default_config()

    # Updating the configuration
    for attr, val in config.items():
        setattr(self, attr, val)

说明

在上面的代码片段中，我们定义了接受 `config` 作为字典的 `__init__()` 函数。在此函数中，我们使用了 `set_default_config()` 函数来设置默认配置。然后，我们使用 `for` 循环遍历 `config` 字典中的属性和值，并使用 `setattr()` 函数更新不同类的配置。

Road

我们现在将创建一个 `Road` 类。让我们看下面的代码片段来演示这一点：

文件：road.py

# importing the distance function from the scipy package
from scipy.spatial import distance

# defining the Road class
class Road:
    def __init__(self, start, end):
        self.start = start
        self.end = end
        self.initProperties()

    def initProperties(self):
        self.length = distance.euclidean(self.start, self.end)
        self.angleSin = (self.end[1] - self.start[1]) / self.length
        self.angleCos = (self.end[0] - self.start[0]) / self.length

说明

在上面的代码片段中，我们从 SciPy 包导入了 `distance` 函数，并在此类中定义了一个名为 `Road` 的类。我们在 `__init__()` 函数中初始化了一些参数，如 `self`、`start` 和 `end`。然后，我们定义了另一个名为 `initProperties()` 的函数，该函数计算道路长度的左侧以及绘制它在屏幕上的角度的正弦和余弦。

Simulation

我们现在将创建一个 `Simulation` 类，并添加一些方法来将道路添加到模拟中。现在让我们看下面的代码片段来演示这一点：

文件：simulator.py

# importing the Road class
from .road import Road

# defining the Simulator class
class Simulator:
    def __init__(self, config = {}):
        # Setting default configuration
        self.set_default_config()

        # Updating configuration
        for attr, val in config.items():
            setattr(self, attr, val)

    def set_default_config(self):
        # Time keeping
        self.t = 0.0
        # Frame count keeping
        self.frame_count = 0
        # Simulation time step
        self.dt = 1/60
        # Array to store roads
        self.roads = []

    def createRoad(self, start, end):
        the_road = Road(start, end)
        self.roads.append(the_road)
        return the_road

    def createRoads(self, roadList):
        for the_road in roadList:
            self.createRoad(*the_road)

说明

在上面的代码片段中，我们从 `road.py` 文件导入了 `Road` 类。然后，我们定义了一个名为 `Simulation` 的类。我们在该类中使用了之前定义的初始化函数。然后，我们定义了另一个名为 `set_default_config()` 的函数，并将一些属性值设置为默认设置。然后，我们定义了 `createRoad()` 和 `createRoads()` 函数来创建一条或多条道路。

Window

我们现在将实时在屏幕上显示模拟。为此，我们将使用 `pygame` 库并创建一个 `Window` 类，该类将 `Simulation` 类作为参数。

我们将定义支持绘制基本形状的各种绘图函数。

`loop` 方法创建一个 `pygame` 窗口，并在每一帧调用 `draw` 方法和 `loop` 参数。当模拟需要每帧更新时，这将非常有用。

让我们看下面的代码片段来说明这一点：

文件：window.py

import pygame
from pygame import gfxdraw
import numpy as np

class Window:
    def __init__(self, simulate, config = {}):
        # Simulation to draw
        self.simulate = simulate

        # Setting the default configurations
        self.set_default_config()

        # Updating the configurations
        for attr, val in config.items():
            setattr(self, attr, val)
        
    def set_default_config(self):
        """Setting the default configuration"""
        self.the_width = 1400
        self.the_height = 1000
        self.the_bgColor = (250, 250, 250)

        self.the_fps = 60
        self.the_zoom = 5
        self.the_offset = (0, 0)

        self.mouseLast = (0, 0)
        self.mouseDown = False

    def loop(self, loop = None):
        """Showing a window visualizing the simulation and runs the loop function."""
        # Creating a pygame window
        self.screen = pygame.display.set_mode((self.the_width, self.the_height))
        pygame.display.flip()

        # Fixed fps
        clock = pygame.time.Clock()

        # To draw text
        pygame.font.init()
        self.text_font = pygame.font.SysFont('Lucida Console', 16)

        # Drawing loop
        running = True
        while not self.simulate.stop_condition(self.simulate) and running:
            # Updating the simulation
            if loop: loop(self.simulate)

            # Drawing simulation
            self.draw()

            # Updating the window
            pygame.display.update()
            clock.tick(self.the_fps)

            # Handling all events
            for event in pygame.event.get():
                # Handling mouse drag and wheel events
                ...

    def convert(self, x, y = None):
        """Converting the simulation coordinates to screen coordinates"""
        ...

    def inverseConvert(self, x, y = None):
        """Converting the screen coordinates to simulation coordinates"""
        ...

    def the_background(self, r, g, b):
        """Filling the screen with one color."""
        ...
        
    def the_line(self, start_pos, end_pos, color):
        """Drawing a line."""
        ...

    def the_rect(self, pos, size, color):
        """Drawing a rectangle."""
        ...

    def the_box(self, pos, size, color):
        """Drawing a rectangle."""
        ...

    def the_circle(self, pos, radius, color, filled = True):
        """Drawing a circle"""
        ...

    def the_polygon(self, vertices, color, filled = True):
        """Drawing a polygon"""

    def the_rotated_box(self, pos, size, angle = None, cos = None, sin = None, centered = True, color = (0, 0, 255), filled = True):
        """Drawing a filled rectangle centered at *pos* with size *size* rotated anti-clockwise by *angle*."""

    def the_rotated_rect(self, pos, size, angle = None, cos = None, sin = None, centered = True, color = (0, 0, 255)):
        """Drawing a rectangle centered at *pos* with size *size* rotated anti-clockwise by *angle*."""

    def drawAxes(self, color = (100, 100, 100)):
        """Drawing x and y axis"""

    def drawGrid(self, unit = 50, color = (150, 150, 150)):
        """Drawing a grid"""

    def drawRoads(self):
        """Drawing every road"""

    def drawStatus(self):
        """Drawing status text"""

    def draw(self):
        # Filling the background
        self.the_background(*self.the_bgColor)

        # Major and minor grid and axes
        self.drawGrid(10, (220, 220, 220))
        self.drawGrid(100, (200, 200, 200))
        self.drawAxes()

        # Drawing roads
        self.drawRoads()

        # Drawing the status info
        self.drawStatus()

说明

在上面的代码片段中，我们导入了 `pygame` 库。然后，我们创建了一个名为 `Window` 的类。我们在该类中初始化了一些参数，并设置了默认配置。然后，我们定义了 `loop` 函数，该函数显示一个可视化模拟的窗口并执行 `loop` 函数。然后，我们定义了 `convert`、`inverseConvert`、`the_background`、`the_line`、`the_rect`、`the_box`、`the_circle`、`the_polygon`、`the_rotated_box`、`the_rotated_rect`、`drawAxes`、`drawGrid`、`drawRoads`、`drawStatus` 和 `draw` 等函数。

我们将上述文件保存在一个名为 `trafficFlowSimulator` 的文件夹中。我们现在将创建另一个名为 `__init__.py` 的 python 文件，并从上述文件中导入类。

文件：__init__.py

from .road import *
from .simulator import *
from .window import *

说明

在上面的代码片段中，我们从之前创建的 python 文件导入了类。

现在让我们运行一个测试代码来查看输出。

文件：testCase1.py

from trafficFlowSimulator import *

# Creating simulation
firstSimulation = Simulator()

# Adding one road
firstSimulation.createRoad((300, 97), (0, 97))

# Adding multiple roads
firstSimulation.createRoads([
    ((300, 97), (0, 97)),
    ((0, 101), (300, 101)),
    ((180, 61), (0, 61)),
    ((220, 56), (180, 61)),
    ((300, 31), (220, 56)),
    ((180, 61), (160, 97)),
    ((158, 131), (300, 131)),
    ((0, 179), (300, 179)),
    ((300, 181), (0, 181)),
    ((160, 101), (156, 180))
    
])

# Starting the simulation
firstWindow = Window(firstSimulation)
firstWindow.loop()

输出

说明

在上面的代码片段中，我们从 `trafficFlowSimulator` 导入了类。然后，我们创建了 `Simulator()` 类的对象。然后，我们使用之前创建的 `createRoad()` 函数添加了一条道路。然后，我们使用 `createRoads()` 函数添加了多条道路。最后，我们通过创建 `Window()` 类的对象并使用 `loop()` 函数来启动模拟。

Vehicles

现在，我们将车辆添加到道路上。我们将使用泰勒级数来近似逼近我们在此教程建模部分中讨论过的动力学方程的解。

对于无限可微函数 f，其泰勒级数展开为：

我们将 a 替换为 x，并将 x 替换为 x + ∆x，得到：

现在我们将 f 替换为位置 x：

作为精度，我们将位置的阶数限制在 2，因为加速度是最高阶导数。我们得到方程 (2)：

方程 (2)

对于速度，我们将 x 替换为 v：

我们将阶数限制在 1，因为我们拥有的最高阶导数是加速度（速度的 1 阶）。方程 (2)：

方程 (1)

在每一次迭代（或每一帧）中，一旦我们使用 IDM 公式计算出加速度，我们将使用这两个方程来更新位置和速度：

方程 (1)

方程 (2)

让我们看下面的代码片段来说明这一点：

文件：numericalApprx.py

# defining the IDM formula
self.a = ...
self.v += self.a * dt
self.x += self.v * dt + self.a * dt * dt / 2

说明

由于上面的代码片段只是一个近似值，因此速度有时会变成负值（尽管模型不允许这样做）。当速度为负时会出现不稳定性，并且位置和速度会发散到负无穷。

我们可以通过预测负速度并将其设置为零来克服这个问题，然后从那里开始工作。

让我们看下面的代码片段来说明这一点：

文件：negativeSpeed.py

if self.v + self.a * dt < 0:
    self.x -= 1 / 2 * self.v * self.v / self.a
    self.v = 0
else:
    self.v += self.a * dt
    self.x += self.v * dt + self.a * dt * dt / 2

说明

在上面的代码片段中，我们使用了 `if-else` 条件语句来检查速度是否为负。

为了计算 IDM 加速度，我们将前车表示为 `leadVehicle`，并在 `leadVehicle` 不为 None 时计算交互项（表示为 `alpha`）。

让我们看下面的代码片段来说明这一点：

文件：leadVehicle.py

alpha = 0
if leadVehicle:
    del_x = leadVehicle.x - self.x - leadVehicle.l
    del_v = self.v - lead.v
    alpha = (self.s0 + max(0, self.T * self.v + del_v * self.v / self.sqrt_ab)) / del_x
self.a = self.a_max * (1 - (self.v / self.v_max)**4 - alpha**2)

说明

在上面的代码片段中，我们将 `alpha` 的值初始化为零。然后，我们使用了 `if` 条件语句来计算 `del_x`、`del_v` 和 `alpha` 值。

如果车辆已停止（例如，在交通信号灯前），我们将使用阻尼方程。之后，我们将所有内容合并到 `Vehicle` 类中的 `update` 方法中。

让我们看下面的代码片段来说明这一点：

文件：vehicle.py

import numpy as np

class Vehicle:
    def __init__(self, config = {}):
        # Setting default configuration
        self.set_default_config()

        # Updating the configuration
        for attr, val in config.items():
            setattr(self, attr, val)

        # Calculate properties
        self.initProperties()

    def set_default_config(self):
        self.l = 4
        self.s_0 = 4
        self.T = 1
        self.vMax = 16.6
        self.aMax = 1.44
        self.bMax = 4.61

        self.path = []
        self.currentRoadIndex = 0

        self.x = 0
        self.v = self.vMax
        self.a = 0
        self.stopped = False

    def initProperties(self):
        self.sqrt_ab = 2 * np.sqrt(self.aMax * self.bMax)
        self._vMax = self.vMax

    def update(self, leadVehicle, dt):
        # Updating the position and velocity
        if self.v + self.a * dt < 0:
            self.x -= 1/2 * self.v * self.v / self.a
            self.v = 0
        else:
            self.v += self.a * dt
            self.x += self.v * dt + self.a * dt * dt/2
        
        # Updating the acceleration
        alpha = 0
        if leadVehicle:
            del_x = leadVehicle.x - self.x - leadVehicle.l
            del_v = self.v - leadVehicle.v

            alpha = (self.s_0 + max(0, self.T * self.v + del_v * self.v / self.sqrt_ab)) / del_x

        self.a = self.aMax * (1 - (self.v / self.vMax) ** 4 - alpha ** 2)

        if self.stopped: 
            self.a = - self.bMax * self.v / self.vMax
        
    def stopVehicle(self):
        self.stopped = True

    def unstopVehicle(self):
        self.stopped = False

    def slowVehicle(self, v):
        self.vMax = v

    def fastVehicle(self):
        self.vMax = self._vMax

说明

在上面的代码片段中，我们导入了所需的模块，并定义了一个名为 `Vehicle` 的类。我们使用了 `__init__()` 函数并在该类中设置了默认配置。然后，我们定义了 `initProperty()` 函数以及更新位置、速度和加速度的 `update()` 函数。我们还定义了停止、解除停止、减速和增加车辆速度的函数。

在 `Road` 类中，我们将包含一个双端队列（也称为双端队列）以跟踪车辆。像队列这样的数据结构更适合存储车辆，因为队列中的第一辆车是道路上最远的车，也是我们可以从队列中移除的第一辆车。我们可以使用 `self.vehicles.popleft()` 从双端队列中删除第一个数据元素。

我们将在 `Road` 类中包含一个 `update` 方法。让我们看下面的代码片段来理解这一点：

文件：road.py

def update(self, dt):
        num = len(self.vehicles)
        
        if num > 0:
            # Updating the first vehicle
            self.vehicles[0].update(None, dt)
            # Updating the other vehicles
            for i in range(1, num):
                lead = self.vehicles[i - 1]
                self.vehicles[i].update(lead, dt)

说明

在上面的代码片段中，我们为 `Road` 类定义了一个 `update` 函数。在此函数中，我们将双端队列中车辆的数量分配给一个名为 `num` 的变量，并使用 `if` 条件检查它是否大于零并更新第一辆车。然后，我们使用从 `1` 到 `num` 的 `for` 循环，并更新双端队列中其余的车辆。

现在，让我们也向 `Simulation` 类添加一个 `update` 方法。这是演示它的代码片段：

文件：simulation.py

def update(self):
        # Updating every road
        for road in self.roads:
            road.update(self.dt)

        # Checking the roads for out of bounds vehicle
        for road in self.roads:
            # If road does not have vehicles, then continue
            if len(road.vehicles) == 0: continue
            # If not
            vehicle = road.vehicles[0]
            # If the first vehicle is out of road bounds
            if vehicle.x >= road.length:
                # If vehicle has a next road
                if vehicle.currentRoadIndex + 1 < len(vehicle.path):
                    # Updating the current road to next road
                    vehicle.currentRoadIndex += 1
                    # Creating a copy and reseting some vehicle properties
                    newVehicle = deepcopy(vehicle)
                    newVehicle.x = 0
                    # Adding it to the next road
                    nextRoadIndex = vehicle.path[vehicle.currentRoadIndex]
                    self.roads[nextRoadIndex].vehicles.append(newVehicle)
                # In all cases, removing it from its road
                road.vehicles.popleft()

说明

在上面的代码片段中，我们在 `Simulation` 类中定义了一个 `update()` 函数。在此函数中，我们更新了每条道路。然后，我们检查道路是否存在越界车辆并执行相应操作。

现在，让我们回到 `Window` 类并添加一个 `run` 方法来实时更新模拟：

文件：window.py

    def run(self, stepsPerUpdate = 1):
        """Running the simulation by updating in every loop."""
        def loop(sim):
            sim.run(stepsPerUpdate)
        self.loop(loop)

说明

在上面的代码片段中，我们定义了一个 `run` 方法，该方法在每次循环时更新模拟。

现在，我们将手动包含车辆：

文件：testCase2.py

firstSimulation.roads[4].vehicles.append(
    Vehicle({
        "path" : [4, 3, 2]
    })
)

firstSimulation.roads[0].vehicles.append(Vehicle())
firstSimulation.roads[1].vehicles.append(Vehicle())
firstSimulation.roads[6].vehicles.append(Vehicle())
firstSimulation.roads[7].vehicles.append(Vehicle())

输出

说明

在上面的代码片段中，我们手动将车辆添加到之前创建的道路上。我们使用了 `append()` 函数将车辆插入不同的道路。

Vehicle Generators

文件：vehicleGenerator.py

from .vehicle import Vehicle
from numpy.random import randint

class VehicleGenerators:
    def __init__(self, sim, config = {}):
        ...

    def set_default_config(self):
        self.vehicleRate = 20
        self.vehicles = [
            (1, {})
        ]

    def initProperties(self):
        self.upcomingVehicle = self.generateVehicle()

    def generateVehicle(self):
        """Returning a random vehicle from self.vehicles with random proportions"""
        ...

    def update(self):
        """Adding vehicles"""
        ...

说明

在上面的代码片段中，我们从 `vehicle.py` 文件导入了 `Vehicle` 类，从 `numpy` 库导入了 `randint` 函数。然后，我们创建了一个名为 `VehicleGenerators` 的类，并定义了 `__init__` 函数来设置默认配置和 `initProperties`。然后，我们包含了 `generateVehicle` 和 `update` 等函数，以返回来自 `self.vehicles` 的随机车辆，并以随机比例添加车辆。

`VehicleGenerators` 类有一个元组数组 `(odds, vehicle)`。元组的第一个数据元素是车辆在同一元组中生成的权重（不是概率）。我们使用权重，因为它们便于处理，因为我们可以使用整数。

例如，如果我们有三种车辆，权重分别为 3、1、2。这对应于 3/6、1/6、2/6，其中 6 (= 3 + 1 + 2)。

我们可以使用以下算法来实现这一点：

生成一个介于 1 和所有权重之和之间的数字 n。
当 n 非负时
遍历所有可能的车辆，并在每次迭代中减去它们的权重。
返回最后使用的车辆。

假设我们有权重 W1、W2、W3。以下算法将允许我们将 1 到 W1 之间的数字分配给第一辆车，将 W1 到 W1 + W2 之间的数字分配给第二辆车，将 W1 + W2 + W3 之间的数字分配给第三辆车。

文件：vehicleGenerator.py

def generateVehicle(self):
        """Returning a random vehicle from self.vehicles with random proportions"""
        total = sum(pair[0] for pair in self.vehicles)
        n = randint(1, total + 1)
        for (weight, config) in self.vehicles:
            n -= weight
            if n <= 0:
                return Vehicle(config)

说明

在上面的代码片段中，我们在 `generateVehicle()` 函数中定义了一个函数。我们在此函数中计算了 `self.vehicles` 中元组的之和，并返回了具有随机比例的随机车辆。

我们包含了一个名为 `lastAddedTime` 的属性，这样每当我们添加一辆车时，生成器执行函数时当前时间就会更新。当当前时间与 `lastAddedTime` 之间的时间差大于车辆生成周期时，就会添加一辆车。

添加车辆的周期是 `60/vehicleRate`，因为 `vehicleRate` 的单位是每分钟车辆数，而 `60` 是 1 分钟或 60 秒。

我们还将检查道路上是否有空间来添加即将到来的车辆。我们通过检查道路上最后一辆车与即将到来的车辆的长度和安全距离之和之间的距离来执行此操作。

让我们看下面的代码片段来说明这一点：

文件：vehicleGenerator.py

def update(self):
        """Adding vehicles"""
        if self.simulation.t - self.lastAddedTime >= 60 / self.vehicleRate:
            # If time elapsed after the last added vehicle is
            # greater than vehicle period; then generate a vehicle
            road = self.simulation.roads[self.upcomingVehicle.path[0]]      
            if len(road.vehicles) == 0\
            or road.vehicles[-1].x > self.upcomingVehicle.s_0 + self.upcomingVehicle.l:
                # If there is space for the generated vehicle; then add it
                self.upcomingVehicle.timeAdded = self.simulation.t
                road.vehicles.append(self.upcomingVehicle)
                # Resetting lastAddedTime and upcomingVehicle
                self.lastAddedTime = self.simulation.t
            self.upcomingVehicle = self.generateVehicle()

说明

在上面的代码片段中，我们定义了一个 `update()` 函数来将车辆添加到模拟中。我们使用了 `if` 条件语句来检查自上次添加车辆以来经过的时间是否大于车辆周期，并为此生成了一辆车。我们还检查了生成的车辆是否有空间，并添加了它。最后，我们重置了上次添加的时间和即将到来的车辆。

最后，我们应该通过从 `Simulation` 类调用 `update` 方法来更新车辆生成器。让我们看下面的代码片段来理解这一点：

文件：testCase3.py

firstSimulation.createGen({
    'vehicleRate' : 60,
    'vehicles' : [
        [1, {"path" : [3, 2, 1]}],
        [1, {"path" : [4]}],
        [1, {"path" : [5]}],
        [1, {"path" : [6]}],
        [1, {"path" : [7]}]
    ]
})

输出

说明

在上面的代码片段中，我们初始化了 `Simulation` 类的 `createGen` 函数，并为 `vehicleRate` 和 `vehicles` 等参数提供了所需的值。我们还指定了车辆的路径并执行了程序。

Traffic Lights

现在，让我们将交通信号灯及其属性添加到模拟中。交通信号灯的默认属性如下：

文件：trafficLight.py

class TrafficSignal:
    def __init__(self, roads, config = {}):
        # Initializing roads
        self.roads = roads
        # Setting the default configuration
        self.set_default_config()
        # Updating the configuration
        for attr, val in config.items():
            setattr(self, attr, val)
        # Calculating the properties
        self.initProperties()

    def set_default_config(self):
        self.cycle = [(False, True), (True, False)]
        self.slowDistance = 40
        self.slowFactor = 10
        self.stopDistance = 15

        self.currentCycleIndex = 0
        self.last_t = 0

说明

在上面的代码片段中，我们定义了一个名为 `TrafficSignal` 的类。我们使用了 `__init__()` 函数来初始化此函数中的一些变量和函数。然后，我们定义了另一个函数来设置默认配置。

`self.cycle` 变量是一个元组数组，包含设置在 `self.roads` 中的每条道路的状态（例如，`True` 表示绿色，`False` 表示红色）。

在默认配置中，数据元素 `(False, True)` 表示第一组道路为红色，第二组为绿色；`(True, False)` 则相反。

我们将使用这种方法，因为它易于扩展。我们可以创建包含多条道路的交通信号灯，具有不同转向信号的交通信号灯，甚至不同交叉路口的同步交通信号灯。

交通信号灯的 update 函数将是可定制的。此函数的默认行为是对称的固定时间循环。

文件：trafficLight.py

 def initProperties(self):
        for i in range(len(self.roads)):
            for the_road in self.roads[i]:
                the_road.setTrafficSignal(self, i)

    @property
    def currentCycle(self):
        return self.cycle[self.currentCycleIndex]

    def update(self, simulation):
        # Going through all cycles every cycleLength and repeats
        cycleLength = 30
        m = (simulation.t // cycleLength) % 2
        self.currentCycleIndex = int(m)

说明

在上面的代码片段中，我们定义了 `initProperties()` 函数。然后，我们使用 `for` 循环遍历数组中的每条道路，并在该函数中设置所有信号。然后，我们定义了一个函数来返回当前周期索引。然后，我们定义了一个 update 函数，我们在其中遍历所有周期并重复。

现在，我们将这些方法包含在 `Road` 类中。

文件：road.py

def setTrafficSignal(self, signal, group):
        self.trafficSignal = signal
        self.trafficSignalGroup = group
        self.hasTrafficSignal = True

    @property
    def trafficSignalState(self):
        if self.hasTrafficSignal:
            i = self.trafficSignalGroup
            return self.trafficSignal.currentCycle[i]
        return True

说明

在上面的代码片段中，我们定义了为每个周期设置交通信号灯及其状态的函数。

我们现在将在 `Road` 类的 update 函数中添加以下代码片段。

文件：road.py

# Checking for traffic signal
        if self.trafficSignalState:
            # In case the traffic signal is green or does not exist
            # Then let the vehicles pass
            self.vehicles[0].unstopVehicle()
            for the_vehicle in self.vehicles:
                the_vehicle.fastVehicle()
        else:
            # In case the traffic signal is red
            if self.vehicles[0].x >= self.length - self.trafficSignal.slowDistance:
                # Slowing vehicles down in slowing zone
                self.vehicles[0].slowVehicle(self.trafficSignal.slowSpeed)
            if self.vehicles[0].x >= self.length - self.trafficSignal.stopDistance and\
            self.vehicles[0].x <= self.length - self.trafficSignal.stopDistance / 2:
                # Stopping vehicles in the stop zone
                self.vehicles[0].stopVehicle()

说明

在上面的代码片段中，我们检查了交通信号灯，并为特定信号执行了特定的任务集。

现在，我们将在 `Simulation` 类的 update 方法中检查交通信号灯的状态。

文件：simulator.py

for signal in self.trafficSignals:
            signal.update(self)

说明

在上面的代码片段中，我们使用 `for` 循环遍历 `trafficSignals` 数组中的每个信号并更新它们。

这是相同的一个输出：

输出

Curves

现实世界中的道路有曲线。由于我们可以在技术上通过手动编写大量道路的坐标来估计曲线来创建曲线，因此我们可以以程序化的方式执行相同的操作。

我们将为此使用贝塞尔曲线。我们将创建一个 `curve.py` 文件，其中包含用于创建曲线并按道路索引引用的函数。

让我们看下面的代码片段来说明这一点：

文件：curve.py

def curvePoints(begin, end, ctrl, res = 5):
    # If curve is a straight line
    if (begin[0] - end[0]) * (begin[1] - end[1]) == 0:
        return [begin, end]

    # If not return a curve
    path = []

    for i in range(res + 1):
        p = i / res
        a = (1 - p)**2 * begin[0] + 2*(1 - p)*p * ctrl[0] + p**2 *end[0]
        b = (1 - p)**2 * begin[1] + 2*(1 - p)*p * ctrl[1] + p**2 *end[1]
        path.append((a, b))

    return path

def curveRoad(begin, end, turnDirection, res = 15):
    pts = curvePoints(begin, end, turnDirection, res)
    return [(pts[i - 1], pts[i]) for i in range(1, len(pts))]

说明

在上面的代码片段中，我们定义了一个名为 `curvePoints()` 的函数。然后，我们检查曲线是否是直线，并相应地执行操作。然后，我们定义了另一个名为 `curveRoad()` 的函数，该函数返回弯曲的路径。

现在，让我们测试上面的代码以获得更好的说明。

文件：testCase3.py

from trafficFlowSimulation import *

# Creating the simulation
firstSimulation = Simulator()

# Adding multiple roads
firstSimulation.createRoads([
    ((0, 105), (145, 105)),
    ((155, 115), (155, 205)),

    *curveRoad((145, 105), (155, 115), (155, 105))
])

firstSimulation.create_gen({
    'vehicleRate' : 20,
    'vehicles' : [
        [1, {"path" : [0, *range(2, 17), 1]}]
    ]
})

# Starting simulation
win = Window(firstSimulation)
win.run(steps_per_update = 5)

说明

在上面的代码片段中，我们导入了包含我们之前创建的类的文件夹。然后，我们创建了 `Simulator()` 类的实例。然后，我们添加了多条道路。我们还包含了 `curveRoad()` 函数来创建弯曲的道路。然后，我们将车辆添加到模拟中并执行了程序。

输出

局限性

虽然我们可以修改 `Simulation` 类来存储与模拟相关的数据以供以后使用，但如果数据收集更顺畅会更好。

此模拟仍有许多不足之处。曲线实现很糟糕且效率低下，并导致车辆与交通信号灯之间的交互问题。

虽然有些人可能担心智能驾驶员模型有点过度，但拥有一个能够复制真实世界现象（如交通波（也称为幽灵交通蛇）和驾驶员反应时间的影响）的模型非常重要。出于同样的原因，我们选择使用智能驾驶员模型。但是，为了创建不需要精确度和极端现实主义的模拟，例如在电子游戏中，我们可以用更简单的基于逻辑的模型来替代 IDM。

完全依赖基于模拟的数据会增加过度拟合的风险。ML 模型可能会针对模拟中存在但真实世界中不存在的特征进行优化。

结论

模拟是数据科学和机器学习的重要组成部分。有时，从现实世界收集数据既不可能也不昂贵。数据生成支持以相对更好的价格构建大型数据集。模拟还可以帮助填补真实世界数据的空白。在某些情况下，真实世界的数据集可能缺少对已开发模型至关重要的边缘情况。

下一主题如何在 Python 中合并和排序两个列表

Python 中的交通流量模拟

了解交通流量模拟的重要性

建模

理解微观模型

理解智能驾驶员模型 (IDM)

理解智能驾驶员模型的工作原理

理解交通道路网络模型

理解随机车辆生成器

理解交通信号灯

交通流量模拟的项目代码

Road

Simulation

Window

Vehicles

Vehicle Generators

Traffic Lights

Curves

局限性

结论

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Python 问题

Python 中的交通流量模拟

了解交通流量模拟的重要性

建模

理解微观模型

理解智能驾驶员模型 (IDM)

理解智能驾驶员模型的工作原理

理解交通道路网络模型

理解随机车辆生成器

理解交通信号灯

交通流量模拟的项目代码

Road

Simulation

Window

Vehicles

Vehicle Generators

Traffic Lights

Curves

局限性

结论

相关帖子

检查给定链表是否为循环链表

Python 中的计数器

Python 中的数据结构和算法 | 第一部分

NumPy ufunc - Python 的通用函数

使用 Python 字典的一些高级方法

Python 中的 SAX 算法

Python 中的最长公共子序列

Python 矩阵

编写 Python 程序查找完美和

Python 社交媒体数据实时分析

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器