Python交通流仿真【含源碼】

雖然交通并不總是暢通無阻，但汽車無縫穿越交叉路口，在交通信號(hào)燈處轉(zhuǎn)彎和停車看起來相當(dāng)壯觀。這種沉思讓我思考交通流對人類文明的重要性。

在此之后，內(nèi)心的書呆子特質(zhì)讓我忍不住思考一種模擬交通流的方法。我在一個(gè)涉及交通流量的本科項(xiàng)目上投入了幾個(gè)星期的時(shí)間，深入到不同的模擬技術(shù)，最終選擇了一個(gè)。

在本文中，我將解釋為什么交通流模擬很重要，比較不同的方法來模擬交通流，并呈現(xiàn)仿真結(jié)果（以及源代碼）。

1、為什么要模擬交通流？

模擬交通流的主要原因是在沒有真實(shí)世界的情況下生成數(shù)據(jù)。你可以使用軟件運(yùn)行模型來預(yù)測交通流，而不需要在現(xiàn)實(shí)世界中安裝傳感器來測試新的管理思路如何影響現(xiàn)實(shí)世界。這有助于加速交通流系統(tǒng)的優(yōu)化和數(shù)據(jù)收集。仿真是實(shí)際測試更便宜、更快捷的替代方案。

訓(xùn)練機(jī)器學(xué)習(xí)模型需要巨大的數(shù)據(jù)集，這些數(shù)據(jù)集的收集和處理成本可能很高。通過模擬交通流在程序上生成數(shù)據(jù)可以很容易地適應(yīng)所需的數(shù)據(jù)類型。

2、建 模

要分析和優(yōu)化交通系統(tǒng)，我們首先必須對交通系統(tǒng)進(jìn)行數(shù)學(xué)建模。這種模型應(yīng)根據(jù)輸入?yún)?shù)（路網(wǎng)幾何、每分鐘車輛、車速等）真實(shí)地表示交通流量。

交通流系統(tǒng)模型通常分為三類，具體取決于它們在哪個(gè)級別上運(yùn)行：

• 微型模型：分別代表每輛車，并嘗試復(fù)現(xiàn)駕駛員行為。
• 宏觀模型：從交通密度（每公里車輛）和交通流量（車輛每分鐘）的角度描述車輛的整體移動(dòng)。它們通常類似于流體的流動(dòng)。
• 中觀模型：是結(jié)合微觀和宏觀模型特點(diǎn)的混合模型，將流量建模為車輛的"包"。

在本文中，我將使用微觀模型。

3、微觀模型

在微觀模型中使用駕駛員模型描述單個(gè)駕駛員/車輛的行為。因此，它必須是一個(gè)多代理系統(tǒng)，即每輛車都使用來自其環(huán)境的輸入自行運(yùn)行。

在微觀模型中，每輛車都編號(hào)為i。第i輛車跟隨第（i-1）輛車。對于第i輛車，我們將用xi表示其沿路的位置，用vi表示其速度，以及li表示其長度。每輛車都是如此。

我們將用si表示保險(xiǎn)杠到保險(xiǎn)杠的距離，用Δv? 表示第i輛車和前面的第i-1輛車之間的速度差異。

4、智能駕駛員模型 （IDM）

2000年，Treiber, Hennecke 和 Helbing開發(fā)了智能駕駛員模型，它描述第i輛車的加速度是其變量和前面車輛的變量的函數(shù)。動(dòng)態(tài)方程的定義為：

在我解釋這個(gè)模型背后的直覺之前，我應(yīng)該先解釋下這些符號(hào)代表什么。

我們已經(jīng)談到了s?, v?, andΔv?。其他參數(shù)包括：

• s0i：是第i輛車和第i-1輛車之間的最小期望距離。
• v0i ：是第i輛車的最大期望速度.
• δ：是加速度指數(shù)，它控制著加速度的"平滑度"。
• Ti：是第i輛車的駕駛員的反應(yīng)時(shí)間。
• ai ：是第i輛車的最大加速度。
• bi ：是第i輛車的舒適減速度.
• s* ： 是第i輛車和第i-1輛車之間的實(shí)際期望距離。

首先，我們將看看s*，這是一個(gè)距離，它由三部分組成。

• s0i：如前所言，是所需的最小距離。
• viTi：是反應(yīng)時(shí)間的安全距離。這是車輛在駕駛員反應(yīng)（剎車）前行駛的距離。
  由于速度是隨著時(shí)間推移而保持的距離，距離是速度乘以時(shí)間。

• (v? Δv?)/√(2a? b?)：這個(gè)有點(diǎn)復(fù)雜，這是一個(gè)基于速度差的安全距離。它表示車輛在非緊急剎車時(shí)（減速度應(yīng)小于bi），不撞到前面的車輛的前提下所需的距離。

5、智能駕駛員模型的工作原理

車輛假定沿著一條直道行駛，并假定遵守以下方程：

為了更好地了解這個(gè)等式，我們可以將其一分為二。我們有一個(gè)自由道路加速度和交互加速度。

自由道路加速度是自由道路上的加速，即沒有車輛的空路。如果我們繪制加速度作為速度vi的函數(shù)，可以得到：

我們注意到，當(dāng)車輛靜止（vi=0）時(shí)，加速是最大的。當(dāng)車速接近最高速度時(shí)，加速變?yōu)?0。這表明，自由道路加速度將加速車輛到最高速度。

如果我們繪制不同值的δ的 v-a 圖，我們會(huì)注意到它控制駕駛員在接近最大速度時(shí)減速的速度。這反過來又控制了加速度/減速度的平滑度。

交互加速度與與前方車輛的交互有關(guān)。為了更好地了解它是如何工作的，讓我們考慮以下情況：

• 在自由路上（si >>s*）：
  當(dāng)前面的車輛很遠(yuǎn)時(shí)，即si遠(yuǎn)遠(yuǎn)大于車輛之間所需的安全距離s*，交互加速度幾乎為0。這意味著車輛將受到自由道路加速的制約。
  

• 在高接近速率（(Δv?）：
  當(dāng)速度差很大時(shí)，交互加速度試圖通過分子中的(v?Δv?)2項(xiàng)進(jìn)行制動(dòng)補(bǔ)償，但過于困難。這是通過分母中4b?s?2實(shí)現(xiàn)的（老實(shí)說，我不知道它如何精確限制減速到b?）。

• 在小距離差（si <<1和Δv?≈0）：
  加速度成為一個(gè)簡單的排斥力。

6、交通道路網(wǎng)絡(luò)模型

我們需要模擬道路網(wǎng)絡(luò)。為此，我們將使用有向圖 G =（V、E），其中：

• V是一組頂點(diǎn)（或節(jié)點(diǎn)）。
• E是代表道路的邊的幾何。

每輛車將沿著一條由多個(gè)路段（邊）組成的路徑形式。我們將在同一路段（邊）上的車輛應(yīng)用智能駕駛員模型。當(dāng)車輛到達(dá)路的盡頭時(shí)，我們會(huì)將其從該路段移開并附加到下一路段上。

在仿真中，我們不會(huì)保留一組節(jié)點(diǎn)（陣列），相反，每條道路都將由其開始和結(jié)束節(jié)點(diǎn)的值明確定義。

7、隨機(jī)車輛發(fā)生器

為了將車輛添加到我們的模擬中，我們有兩個(gè)選項(xiàng)：

• 通過創(chuàng)建新的Vehicle類實(shí)例并將其添加到車輛列表中，手動(dòng)將每輛車添加到模擬中。
• 根據(jù)預(yù)先定義的概率，隨機(jī)添加車輛。

對于第二個(gè)選項(xiàng)，我們必須定義一個(gè)隨機(jī)車輛發(fā)生器。

隨機(jī)車輛發(fā)生器由兩個(gè)約束定義：

• 車輛生成速率（τ）： 平均每分鐘應(yīng)添加到仿真中的車輛數(shù)。
• 車輛配置列表 （L）：車輛配置和生成概率的元組列表。
  
  L = [（p1，V1），（p2，V2），（p3，V3]），...

隨機(jī)車輛發(fā)生器以概率pi生成車輛Vi。

8、交通信號(hào)燈

紅綠燈位于路段端點(diǎn)上，其特點(diǎn)是包含以下兩個(gè)區(qū)域：

• 減速區(qū)：以減速距離和減速系數(shù)為特征，是車輛使用減速系數(shù)減速的區(qū)域。

• 停車區(qū)：以停車距離為特征，是車輛停車的區(qū)域。這是通過此動(dòng)態(tài)方程使用阻尼力實(shí)現(xiàn)的：

9、仿真

我們將采用面向?qū)ο蟮姆椒āＣ枯v車和道路將被定義為一個(gè)類。

我們將反復(fù)使用以下的__init__函數(shù)。它通過一個(gè)函數(shù)set_default_config設(shè)置當(dāng)前類的默認(rèn)配置，并將一個(gè)字典中的屬性設(shè)置為當(dāng)前類實(shí)例的屬性。這樣，我們不必?fù)?dān)心更新不同類別的__init__函數(shù)或?qū)淼淖兓?/p>

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

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

我們將為路段創(chuàng)建一個(gè)Road類：

from scipy.spatial import distance

class Road:
    def __init__(self, start, end):
        self.start = start
        self.end = end

        self.init_porperties()

    def init_properties(self):
        self.length = distance.euclidean(self.start, self.end)
        self.angle_sin = (self.end[1]-self.start[1]) / self.length
        self.angle_cos = (self.end[0]-self.start[0]) / self.length

當(dāng)在屏幕上繪制道路及其角度的正弦和余弦時(shí)，我們需要道路的長度length。

還有一個(gè)Simulation類，讓我們添加了一些方法來將道路添加到仿真里。

from .road import Road

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

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

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

    def create_road(self, start, end):
        road = Road(start, end)
        self.roads.append(road)
        return road

    def create_roads(self, road_list):
        for road in road_list:
            self.create_road(*road)

我們必須在屏幕上實(shí)時(shí)顯示我們的仿真。為此，我們將使用pygame。我將創(chuàng)建一個(gè)Window類，以類Simulation作為參數(shù)。

我定義了多個(gè)繪圖函數(shù)，有助于繪制基本形狀。

該loop方法創(chuàng)建一個(gè)pygame窗口，并在每一幀調(diào)用draw方法和loop參數(shù)。當(dāng)我們的仿真需要逐幀更新時(shí)，這將變得有用。

import pygame
from pygame import gfxdraw
import numpy as np

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

        # Set default configurations
        self.set_default_config()

        # Update configurations
        for attr, val in config.items():
            setattr(self, attr, val)
        
    def set_default_config(self):
        """Set default configuration"""
        self.width = 1400
        self.height = 1000
        self.bg_color = (250, 250, 250)

        self.fps = 60
        self.zoom = 5
        self.offset = (0, 0)

        self.mouse_last = (0, 0)
        self.mouse_down = False


    def loop(self, loop=None):
        """Shows a window visualizing the simulation and runs the loop function."""
        # Create a pygame window
        self.screen = pygame.display.set_mode((self.width, self.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)

        # Draw loop
        running = True
        while not self.sim.stop_condition(self.sim) and running:
            # Update simulation
            if loop: loop(self.sim)

            # Draw simulation
            self.draw()

            # Update window
            pygame.display.update()
            clock.tick(self.fps)

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


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

    def inverse_convert(self, x, y=None):
        """Converts screen coordinates to simulation coordinates"""
        ...


    def background(self, r, g, b):
        """Fills screen with one color."""
        ...
        
    def line(self, start_pos, end_pos, color):
        """Draws a line."""
        ...

    def rect(self, pos, size, color):
        """Draws a rectangle."""
        ...

    def box(self, pos, size, color):
        """Draws a rectangle."""
        ...

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

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

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

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


    def draw_axes(self, color=(100, 100, 100)):
        """Draw x and y axis"""

    def draw_grid(self, unit=50, color=(150,150,150)):
        """Draws a grid"""

    def draw_roads(self):
        """Draws every road"""

    def draw_status(self):
        """Draws status text"""


    def draw(self):
        # Fill background
        self.background(*self.bg_color)

        # Major and minor grid and axes
        self.draw_grid(10, (220,220,220))
        self.draw_grid(100, (200,200,200))
        self.draw_axes()

        # Draw roads
        self.draw_roads()

        # Draw status info
        self.draw_status()

我將trafficSimulator文件夾中的每一個(gè)文件使用__init__.py組合在一起。

from .road import *
from .simulation import *
from .window import *

每當(dāng)定義新類時(shí)，應(yīng)在此文件中導(dǎo)入該類。

將trafficSimulator文件夾放置在我們的項(xiàng)目文件夾中將以便使用仿真模塊。

from trafficSimulator import *

# Create simulation
sim = Simulation()

# Add one road
sim.create_road((300, 98), (0, 98))

# Add multiple roads
sim.create_roads([
    ((300, 98), (0, 98)),
    ((0, 102), (300, 102)),
    ((180, 60), (0, 60)),
    ((220, 55), (180, 60)),
    ((300, 30), (220, 55)),
    ((180, 60), (160, 98)),
    ((158, 130), (300, 130)),
    ((0, 178), (300, 178)),
    ((300, 182), (0, 182)),
    ((160, 102), (155, 180))
    
])

# Start simulation
win = Window(sim)
win.loop()

10、車輛

現(xiàn)在，我們必須增加車輛。

我們將使用Taylor級數(shù)來近似求解本文建模部分所討論的動(dòng)態(tài)方程。

一個(gè)微分方程f的泰勒級數(shù)展開是：

使用▲x替換a，使用x+▲x替換x， 我們得到：

使用位置x替換f：

作為一個(gè)近似，對位置我們將在2階截止，因?yàn)榧铀俣仁亲罡唠A導(dǎo)數(shù)。我們得到方程（2）：

對于速度，我們將用v代替x：

我們將在1階 停止，因?yàn)樗俣仁?階導(dǎo)數(shù)。方程（2）：

在每個(gè)迭代（或幀）中，使用 IDM 公式計(jì)算加速度后，我們將使用以下兩個(gè)方程更新位置和速度：

在代碼中看起來像這樣：

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

由于這只是一個(gè)近似值，速度有時(shí)可能會(huì)變?yōu)樨?fù)值（但模型不允許這樣）。當(dāng)速度為負(fù)時(shí)，會(huì)產(chǎn)生不穩(wěn)定性，位置和速度會(huì)分化為負(fù)無窮大。

為了克服這個(gè)問題，每當(dāng)我們預(yù)測負(fù)速度時(shí)，我們就會(huì)將其設(shè)定為零，并從那里找出方法：

在代碼中，此實(shí)現(xiàn)如下：

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

要計(jì)算 IDM 加速度，我們將引導(dǎo)車輛表示為lead，并當(dāng)lead不是None時(shí)，計(jì)算一個(gè)交互項(xiàng)（用alpha表示）。

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

如果車輛停止（例如在紅綠燈處），我們將使用阻尼方程：

if self.stopped: 
  self.a = -self.b_max*self.v/self.v_max

然后，我們將所有內(nèi)容整合在Vehicle類的update方法中：

import numpy as np

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

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

        # Calculate properties
        self.init_properties()

    def set_default_config(self):    
        self.l = 4
        self.s0 = 4
        self.T = 1
        self.v_max = 16.6
        self.a_max = 1.44
        self.b_max = 4.61

        self.path = []
        self.current_road_index = 0

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

    def init_properties(self):
        self.sqrt_ab = 2*np.sqrt(self.a_max*self.b_max)
        self._v_max = self.v_max

    def update(self, lead, dt):
        # Update 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
        
        # Update acceleration
        alpha = 0
        if lead:
            delta_x = lead.x - self.x - lead.l
            delta_v = self.v - lead.v

            alpha = (self.s0 + max(0, self.T*self.v + delta_v*self.v/self.sqrt_ab)) / delta_x

        self.a = self.a_max * (1-(self.v/self.v_max)**4 - alpha**2)

        if self.stopped: 
            self.a = -self.b_max*self.v/self.v_max
        
    def stop(self):
        self.stopped = True

    def unstop(self):
        self.stopped = False

    def slow(self, v):
        self.v_max = v

    def unslow(self):
        self.v_max = self._v_max

在Road類中，我們將添加一個(gè)deque（雙端隊(duì)列）來跟蹤車輛。隊(duì)列是存儲(chǔ)車輛的較好的數(shù)據(jù)結(jié)構(gòu)，因?yàn)殛?duì)列中的第一輛車是路上最遠(yuǎn)的車輛，它是第一個(gè)可以從隊(duì)列中刪除的車輛。要從deque 中刪除第一個(gè)車輛，我們可以使用self.vehicles.popleft() 。

我們將在Road類中添加一個(gè)update方法：

def update(self, dt):
  n = len(self.vehicles)

  if n > 0:
    # Update first vehicle
    self.vehicles[0].update(None, dt)
    # Update other vehicles
    for i in range(1, n):
      lead = self.vehicles[i-1]
      self.vehicles[i].update(lead, dt

在Simulation類中添加一個(gè)update方法：

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

        # Check roads for out of bounds vehicle
        for road in self.roads:
            # If road has no vehicles, continue
            if len(road.vehicles) == 0: continue
            # If not
            vehicle = road.vehicles[0]
            # If first vehicle is out of road bounds
            if vehicle.x >= road.length:
                # If vehicle has a next road
                if vehicle.current_road_index + 1 < len(vehicle.path):
                    # Update current road to next road
                    vehicle.current_road_index += 1
                    # Create a copy and reset some vehicle properties
                    new_vehicle = deepcopy(vehicle)
                    new_vehicle.x = 0
                    # Add it to the next road
                    next_road_index = vehicle.path[vehicle.current_road_index]
                    self.roads[next_road_index].vehicles.append(new_vehicle)
                # In all cases, remove it from its road
                road.vehicles.popleft() 

回到Window類，添加了一個(gè)run方法來實(shí)時(shí)更新仿真：

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

現(xiàn)在我們將手動(dòng)添加車輛：

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

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

車輛生成器代碼如下：

from .vehicle import Vehicle
from numpy.random import randint

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

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

    def init_properties(self):
        self.upcoming_vehicle = self.generate_vehicle()

    def generate_vehicle(self):
        """Returns a random vehicle from self.vehicles with random proportions"""
        ...

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

VehicleGenerator包含(odds, vehicle) 元組的列表。

元組第一個(gè)元素是在同一元組中生成車輛的權(quán)重（不是概率）。我使用權(quán)重， 因?yàn)樗鼈兏菀坠ぷ鳎?因?yàn)槲覀兛梢灾皇褂谜麛?shù)。

例如，如果我們有3輛車權(quán)重分別為132，這相當(dāng)于生成概率分別為1/6 ， 3/6 ，2/66 。

為了實(shí)現(xiàn)這一點(diǎn)，我們使用以下算法

• 生成1到權(quán)重和之間的數(shù)字r。
• 當(dāng)r是非負(fù)數(shù)：循環(huán)所有可能的車輛，并在每次迭代時(shí)減去其權(quán)重。
• 返回最后使用的車輛。

如果我們有權(quán)重：W1，W2，W3。 此算法將在1和W 3之間的數(shù)字分配到第一輛車，將W? 到 W?+W? 之間的數(shù)字分配給第二輛車，將W?+W?+W?到結(jié)束的數(shù)字分配給第三輛車。

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

每次生成器添加車輛時(shí)，last_added_time屬性都會(huì)更新到當(dāng)前時(shí)間。當(dāng)當(dāng)前時(shí)間和last_added_time之差值大于車輛生成周期時(shí)，添加車輛。

添加車輛的周期是60/vehicle_rate，因?yàn)?code>vehicle_rate 的單位是車輛/每分鐘，60表示1分鐘或60秒。

我們還必須檢查道路是否還有空間來添加即將行駛的車輛。我們通過檢查道路上最后一輛車之間的距離和即將行駛的車輛的長度和安全距離的總和來做到這一點(diǎn)。

def update(self):
    """Add vehicles"""
    if self.sim.t - self.last_added_time >= 60 / self.vehicle_rate:
        # If time elasped after last added vehicle is
        # greater than vehicle_period; generate a vehicle
        road = self.sim.roads[self.upcoming_vehicle.path[0]]      
        if len(road.vehicles) == 0           or road.vehicles[-1].x > self.upcoming_vehicle.s0 + self.upcoming_vehicle.l:
            # If there is space for the generated vehicle; add it
            self.upcoming_vehicle.time_added = self.sim.t
            road.vehicles.append(self.upcoming_vehicle)
            # Reset last_added_time and upcoming_vehicle
            self.last_added_time = self.sim.t
        self.upcoming_vehicle = self.generate_vehicle()

最后，我們應(yīng)該通過調(diào)用Simulation 的update方法來更新車輛生成。

sim.create_gen({
    'vehicle_rate': 60,
    'vehicles': [
        [1, {"path": [4, 3, 2]}],
        [1, {"path": [0]}],
        [1, {"path": [1]}],
        [1, {"path": [6]}],
        [1, {"path": [7]}]
    ]
})

11、紅綠燈

交通信號(hào)燈的默認(rèn)屬性是：

class TrafficSignal:
    def __init__(self, roads, config={}):
        # Initialize roads
        self.roads = roads
        # Set default configuration
        self.set_default_config()
        # Update configuration
        for attr, val in config.items():
            setattr(self, attr, val)
        # Calculate properties
        self.init_properties()

    def set_default_config(self):
        self.cycle = [(False, True), (True, False)]
        self.slow_distance = 40
        self.slow_factor = 10
        self.stop_distance = 15

        self.current_cycle_index = 0
        self.last_t = 0

self.cycle是self.roads 的一個(gè)數(shù)組，每個(gè)元組包含道路的狀態(tài)（True綠色，False紅色）。

在默認(rèn)配置中，(False, True)表示第一組道路為紅色，第二組道路為綠色。 (True, False) 則恰恰相反。

所以使用此方法之，是因?yàn)樗子跀U(kuò)展。我們創(chuàng)建紅綠燈，包括超過 2 條道路、帶有左右轉(zhuǎn)彎單獨(dú)信號(hào)的紅綠燈，甚至用于多個(gè)交叉路口的同步交通信號(hào)燈。

交通信號(hào)燈的update函數(shù)應(yīng)該是可定制的。其默認(rèn)行為是對稱的固定時(shí)間循環(huán)。

def init_properties(self):
    for i in range(len(self.roads)):
        for road in self.roads[i]:
            road.set_traffic_signal(self, i)

@property
def current_cycle(self):
    return self.cycle[self.current_cycle_index]

def update(self, sim):
    # Goes through all cycles every cycle_length and repeats
    cycle_length = 30
	k = (sim.t // cycle_length) % 2
	self.current_cycle_index = int(k)

我們需要在Road類添加以下方法：

def set_traffic_signal(self, signal, group):
    self.traffic_signal = signal
    self.traffic_signal_group = group
    self.has_traffic_signal = True

@property
def traffic_signal_state(self):
    if self.has_traffic_signal:
        i = self.traffic_signal_group
        return self.traffic_signal.current_cycle[i]
    return True

而這個(gè)，在Road的update方法。

# Check for traffic signal
if self.traffic_signal_state:
    # If traffic signal is green or doesn't exist
    # Then let vehicles pass
    self.vehicles[0].unstop()
    for vehicle in self.vehicles:
        vehicle.unslow()
else:
    # If traffic signal is red
    if self.vehicles[0].x >= self.length - self.traffic_signal.slow_distance:
        # Slow vehicles in slowing zone
        self.vehicles[0].slow(self.traffic_signal.slow_speed)
    if self.vehicles[0].x >= self.length - self.traffic_signal.stop_distance and       self.vehicles[0].x <= self.length - self.traffic_signal.stop_distance / 2:
        # Stop vehicles in the stop zone
        self.vehicles[0].stop()

在Simulation 的update 方法中檢查交通燈狀態(tài)：

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

12、曲線

在現(xiàn)實(shí)世界中，道路有曲線。雖然從技術(shù)上講，我們可以通過手寫很多道路的坐標(biāo)來接近曲線來創(chuàng)建此模擬中的曲線，但我們可以在程序上實(shí)現(xiàn)同樣的事情。

我們將使用貝賽爾曲線來達(dá)到這個(gè)效果。

我創(chuàng)建了一個(gè)curve.py文件，其中包含有助于創(chuàng)建曲線并按其道路序號(hào)引用曲線的功能。

def curve_points(start, end, control, resolution=5):
	# If curve is a straight line
	if (start[0] - end[0])*(start[1] - end[1]) == 0:
		return [start, end]

	# If not return a curve
	path = []

	for i in range(resolution+1):
		t = i/resolution
		x = (1-t)**2 * start[0] + 2*(1-t)*t * control[0] + t**2 *end[0]
		y = (1-t)**2 * start[1] + 2*(1-t)*t * control[1] + t**2 *end[1]
		path.append((x, y))

	return path

def curve_road(start, end, turn_direction, resolution=15):
	points = curve_points(start, end, turn_direction, resolution)
	return [(points[i-1], points[i]) for i in range(1, len(points))]

測試：

from trafficSimulator import *

# Create simulation
sim = Simulation()

# Add multiple roads
sim.create_roads([
    ((0, 100), (140, 100)),
    ((150, 110), (150, 200)),

    *curve_road((140, 100), (150, 110), (150, 100))
])

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


# Start simulation
win = Window(sim)
win.run(steps_per_update=5)

13、示例

這些示例的代碼可在 Github 中找到。

公路匝道入口

雙向交叉路口

環(huán)形交叉路口

鉆石型分流交叉路口

14、局限性

雖然我們可以修改Simulation類來存儲(chǔ)有關(guān)我們以后可以使用的模擬數(shù)據(jù)，但如果數(shù)據(jù)收集過程更加簡化，則更好。

這種模擬仍然缺乏很多。曲線的實(shí)現(xiàn)是糟糕和低效的，并導(dǎo)致車輛和交通信號(hào)之間的相互作用的問題。

雖然有些人可能認(rèn)為智能驅(qū)動(dòng)模型有點(diǎn)過頭了，但重要的是要有一個(gè)模型，可以復(fù)制現(xiàn)實(shí)世界的現(xiàn)象，如交通波（又名幽靈交通蛇）和司機(jī)的反應(yīng)時(shí)間的影響。因此，我選擇使用智能驅(qū)動(dòng)模型。但對于精度和極端現(xiàn)實(shí)主義不重要的仿真，就像在視頻游戲中一樣，IDM可以被一個(gè)更簡單的基于邏輯的模型所取代。

完全依賴基于仿真的數(shù)據(jù)會(huì)增加過度擬合的風(fēng)險(xiǎn)。你的 ML 模型可以優(yōu)化用于僅存在于仿真中的處理，并且在現(xiàn)實(shí)世界中不存在。

15、結(jié)論

仿真是數(shù)據(jù)科學(xué)和機(jī)器學(xué)習(xí)的重要組成部分。有時(shí)，從現(xiàn)實(shí)世界中收集數(shù)據(jù)是不可能的，或者成本很高。生成數(shù)據(jù)有助于以更好的價(jià)格構(gòu)建巨大的數(shù)據(jù)集。仿真還有助于填補(bǔ)真實(shí)數(shù)據(jù)中的空白。在某些情況下，現(xiàn)實(shí)世界數(shù)據(jù)集缺少對開發(fā)模型至關(guān)重要的邊緣案例。

這個(gè)仿真是我參與的本科學(xué)校項(xiàng)目的一部分。目的是優(yōu)化城市交叉路口的交通信號(hào)。我制作了此仿真來測試和驗(yàn)證我的優(yōu)化方法。

我從來沒有想過發(fā)表這篇文章，直到我看到特斯拉的AI展示日，其中他們談到他們?nèi)绾问褂梅抡鎭砩蓴?shù)據(jù)的邊緣樣本。

原文鏈接：Simulating Traffic Flow in Python

BimAnt翻譯整理，轉(zhuǎn)載請表明出處