本项目根据以下项目要求完成的一个python课程的期末大作业，本项目主要就是采用pygame框架进行的一个交通导流的可视化模拟。
自主确定相应的假设或前提，如通过合流处的车辆数量（这个是随机的，你要考虑如何随机地产生车辆），车辆的速度等；你要模拟的合流处，主车道有n个车道（n>=3），匝道是1个车道，合流后，主车道还是至少n个车道；
你模拟的道路长度必须满足一定的条件，从而有可能让一些在最右侧车道上的车，在你程序的引导下，变道（不在最右侧车道）通过合流处，从而不需要交替通行，降低合流处的压力；
每一辆车，在你的可视化模拟中，都应该尽可能和其他车进行区分，比方说可以通过颜色、数字等方式进行标识，从而能够让大家观察到每一辆车的通行状况；
通过你的引导，你应该能够大致确定，按你制定的假设或前提，道路的最大通行能力是怎么样的，它应该是一个数值，这个数值怎么定义，由你来确定，要求是科学、合理；
要求使用OOP编程，对车辆进行建模，是不是要多线程，你自己考虑；
可视化的实现，可以使用OpenCV或其他的图形包，你根据你的需要来选取，报告里要求给出为什么选型这些包的理由
（1）Pygame
（2）智能驾驶员算法
（3）Scipy
（4）Numpy
1.pygame构建整体可视化界面
2.车辆换道策略
3.交替合流算法
4.智能驾驶员策略
(1) pygame构建整体可视化界面
在可视化展示方面，我们选择了简单容易上手的pygame进行展示和实现。在实现方面，主要就是先画出一个大致的窗口，然后在窗口定义一些图形和标志，然后不断进行界面数据的更新使得画面上的元素动起来。其中画出四个车道作为我们的道路模拟，一个匝道三个主干道，然后不断刷新出车辆元素进行交通流的模拟。其中在匝道和主干道合流处进行检查点设置，预防合流处出现两车抢道出现追尾事件。
同时为了方便查看以更加宏观方式检测和观看交通导流的模拟，我们还在界面上定义了一些鼠标的拖动事件和放大缩小功能，以便我们可以更细节或者更宏观进行观看模拟情况。
最后在界面的左上角定义了一些数据检测的实时动态展示，其中包括运行时间，运行帧数，合流运行策略，单位时间交通通行率。
(2) 车辆换道策略
在设计车辆变道策略中，首先我们要解决车辆什么时候进行边变道，主要需要满足两个条件，就是下一车道相比自身车道适合我们变道，还有变道后车辆行驶的安全性。
我们采用的逻辑是计算两车道运行车辆数量差距，在一定数量差距后就判定适合变道，然后变道之前，对将要进行变道的车道的对等位置的前后进行位置探测，检测到在变道安全距离内没有车辆后才进行变道。一旦算法判定进行变道以后，本车道运行的本车将会将自身运行的车道数据改为将要变道的车道，同时更新页面，这以后本车道上车辆画面消失，对等位置的变道车道出现变道成功后的车辆完成变道。
(3) 交替合流算法
在交替合流算法的设计中主要考虑两方面的问题，一个是什么时候进行交替合流，一个是如何进行交替合流。
在判定何时进行交替合流的策略中，我们设计了根据匝道和主干道车辆的各自的数量来衡量两车道的拥挤程度，一旦达到拥挤的阈值，我们判定两车道进入拥堵状态，可能会发生互相挣抢车道的问题，所以我们进入交替合流的策略。反之，一旦两车道拥挤程度下降，就会转换为正常通行模式。
在控制交替合流中，我们对于两车道设置了两个信号量flag1和flag2，在进入交替策略的同时，触发主干道的flag1数值置为1运行通过，flag2为0禁止通过，然后主干道通过一辆后，flag1为0，此时flag2不会马上改为1，而是等主干道车辆开过匝道和主干道交汇点后，flag2才会变为1允许匝道运行。之后反复上述过程实现交替通行策略。
(4) 智能驾驶员策略
智能驾驶员模型以统一的形式描述了车辆在道路上的不同情况加速度的变化，考虑了相邻车辆的速度和间距，其加速度方程为：
公式中各个符号的意义如下：
n：车辆编号
amax(n):车辆n的最大加速度
vmax(n):车辆n在前方没有车的状态下的期望速度
：加速度指数，通常取1到5之间
vn:车辆n的速度
vn=vn-vn-1:车辆n与前面的车辆n-1的速度差
Sn=xn-1-xn-ln-1:车辆n与前车的净间距
Xn:车辆n的位置
Ln-1:车辆(n-1)的长度
其中，加速度方程可有两部分加速度组成：
S*是当前状态下驾驶员的期望间距，定义如下
运行截图
road.py
from scipy.spatial import distance
from collections import deque
class Road:
def __init__(self, start, end):
self.start = start
self.end = end
self.aver_v=0
self.vehicles = deque()
self.init_properties()
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
self.has_traffic_signal = False
def update(self, dt):
n = len(self.vehicles)
if n > 0:
#更新首车
self.vehicles[0].update(None, dt)
#更新其余车，依据前车
sum = 0;
for i in range(1, n):
lead = self.vehicles[i-1]
self.vehicles[i].update(lead, dt)
sum +=self.vehicles[i].v
self.aver_v=sum/n;
self.vehicles[0].unstop()
for vehicle in self.vehicles:
vehicle.unslow()
simulation.py
import random
from .road import Road
from copy import deepcopy
from .vehicle_generator import VehicleGenerator
class Simulation:
def __init__(self, config={}):
# 配置基本项
self.set_default_config()
# 更新配置
for attr, val in config.items():
setattr(self, attr, val)
self.flagx = 1 #合流道可通过标志
self.flagy = 0 #匝道可通过标志
self.canf=1 #是否进行交替
self.lastc=0 #上一次变道出现时间
self.startf=0 #交替变化开始时间
self.num = 0 #出线车辆数量
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
self.generators = []
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)
def create_gen(self, config={}): #创建车辆生成策略
gen = VehicleGenerator(self, config)
self.generators.append(gen)
return gen
def update(self): #更新仿真器状态
# 每条路更新
for road in self.roads:
road.update(self.dt)
# 根据策略增加车辆
for gen in self.generators:
gen.update()
#根据车流量更改合流策略
n3=len(self.roads[3].vehicles)
n6=len(self.roads[6].vehicles)
for veh in self.roads[3].vehicles:
if(veh.v<=1):
n3+=1
for veh in self.roads[6].vehicles:
if(veh.v<=1):
n6+=1
if(self.canf):
if ( n3<=3 or n3<=3 and ((self.roads[3].vehicles[0].v==0)!= (self.roads[6].vehicles[0].v==0)) and self.t - self.startf >= 50):
self.canf = 0;
self.startf=self.t;
else:
if ( n3>=5 and n6>=5 and(self.roads[3].vehicles[0].v==0 and self.roads[6].vehicles[0].v==0) and self.t - self.startf >= 50):
self.canf = 1;
self.startf = self.t;
self.flagx = 0
if(len(self.roads[7].vehicles)==0 and len(self.roads[5].vehicles)==0):
self.flagx = 1
self.flagy = 0
#print(self.t)
idx = -1
#遍历每条路里的车更新状态
for road in self.roads:
idx += 1
if len(road.vehicles) == 0: continue
if(idx>=0 and idx<4):
#print(road.aver_v)
if(idx>=1 and len(road.vehicles)>len(self.roads[idx-1].vehicles) and len(self.roads[idx-1].vehicles)!=0 and self.t-self.lastc>=5):
# print(self.lastc)
# print("1:")
# print(road.aver_v)
# print("2:")
# print(self.roads[idx - 1].aver_v)
siz = len(road.vehicles)
veh = road.vehicles[random.randint(0, siz - 1)]
now = veh.x
pre = 0
f = 0
for r in self.roads[idx-1].vehicles:
if(r.x<now) :
if(now-r.x>5):
road.vehicles.remove(veh)
veh.path=self.roads[idx-1].vehicles[0].path
self.roads[idx-1].vehicles.insert(pre, veh)
self.lastc=self.t
f=1
break
pre += 1
if(f==0):
road.vehicles.remove(veh)
veh.path = self.roads[idx - 1].vehicles[0].path
self.roads[idx - 1].vehicles.append(veh)
self.lastc = self.t
if len(road.vehicles) == 0: continue
vehicle = road.vehicles[0]
#路的首辆车超出范围
if vehicle.x >= road.length:
# 如果当前车还未到路线末路
if vehicle.current_road_index + 1 < len(vehicle.path):
# 根据合流策略决定是否更新
if(self.canf==1): #交替合流状态
id=vehicle.current_road_index
if(vehicle.path[id]==3 and vehicle.path[id+1]==7):
if(self.flagx):
vehicle.unstop()
self.flagx=0
vehicle.current_road_index += 1
#创建车对象并赋值
new_vehicle = deepcopy(vehicle)
new_vehicle.x = 0
#加入当前车路线的邻路
next_road_index = vehicle.path[vehicle.current_road_index]
self.roads[next_road_index].vehicles.append(new_vehicle)
road.vehicles.popleft()
else:
vehicle.v = 0
vehicle.stop()
elif(vehicle.path[id]==6 and vehicle.path[id+1]==5):
if (self.flagy==0):
vehicle.v = 0
vehicle.stop()
else:
self.flagy=0
vehicle.unstop()
vehicle.current_road_index += 1
new_vehicle = deepcopy(vehicle)
new_vehicle.x = 0
next_road_index = vehicle.path[vehicle.current_road_index]
self.roads[next_road_index].vehicles.append(new_vehicle)
road.vehicles.popleft()
else:
if (vehicle.path[id] == 7 and vehicle.path[id + 1] == 4): self.flagy = 1
elif (vehicle.path[id] == 5 and vehicle.path[id + 1] == 4): self.flagx = 1
vehicle.current_road_index += 1
new_vehicle = deepcopy(vehicle)
new_vehicle.x = 0
next_road_index = vehicle.path[vehicle.current_road_index]
self.roads[next_road_index].vehicles.append(new_vehicle)
road.vehicles.popleft()
else:#抢道合流策略
if((idx==3 or idx==6) and (len(self.roads[7].vehicles)>0 or len(self.roads[5].vehicles)>0)):
vehicle.v = 0
vehicle.stop()
else:
vehicle.unstop()
vehicle.current_road_index += 1
new_vehicle = deepcopy(vehicle)
new_vehicle.x = 0
next_road_index = vehicle.path[vehicle.current_road_index]
self.roads[next_road_index].vehicles.append(new_vehicle)
road.vehicles.popleft()
else:#已经到路线末路
road.vehicles.popleft()
self.num += 1
# 更新时间
self.t += self.dt
self.frame_count += 1
def run(self, steps):
for _ in range(steps):
self.update()
vehicle.py
import random
import numpy as np
class Vehicle:
def __init__(self, config={}):
# 设置
self.set_default_config()
# 更新
for attr, val in config.items():
setattr(self, attr, val)
# 计算基本值
self.init_properties()
def set_default_config(self):
self.l = 4
self.s0 = 6
self.T = 1
self.v_max = 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
#颜色随机配置
ca=0
cb=0
cc=0
if(random.randint(0,1)): ca=255
if (random.randint(0, 1)): cb=255;
if (random.randint(0, 1)): cc = 255;
self.color=(ca,cb,cc)
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):
# 更新 位置、速度
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
vehicle_generator.py
from .vehicle import Vehicle
from numpy.random import randint
class VehicleGenerator:
def __init__(self, sim, config={}):
self.sim = sim
# 设置默认配置
self.set_default_config()
# 更新配置
for attr, val in config.items():
setattr(self, attr, val)
# 计算属性
self.init_properties()
def set_default_config(self):
self.vehicle_rate = 20
self.vehicles = [
(1, {})
]
self.last_added_time = 0
def init_properties(self):
self.upcoming_vehicle = self.generate_vehicle()
def generate_vehicle(self):
#返回随机车辆
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)
def update(self):#增加车
if self.sim.t - self.last_added_time >= 60 / self.vehicle_rate:
# 如果进入下一个加入车的周期
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:
# 如果和道路末端车辆有足够安全的距离
self.upcoming_vehicle.time_added = self.sim.t
road.vehicles.append(self.upcoming_vehicle)
#重置最佳更新时间和将要加入的车辆
self.last_added_time = self.sim.t
self.upcoming_vehicle = self.generate_vehicle()
window.py
import pygame
from pygame import gfxdraw
import numpy as np
class Window:
def __init__(self, sim, config={}):
# 注入，模拟控制类
self.sim = sim
# 设置默认属性
self.set_default_config()
# 更新属性
for attr, val in config.items():
setattr(self, attr, val)
def set_default_config(self):
# 设置基础属性
self.width = 1400
self.height = 700
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):
# 展示属性结果和运行loop函数
# 创建一个窗口
self.screen = pygame.display.set_mode((self.width, self.height))
pygame.display.flip()
# 固定帧率
clock = pygame.time.Clock()
# 定义字体
pygame.font.init()
self.text_font = pygame.font.SysFont('Lucida Console', 16)
# 执行loop
running = True
while running:
# 更新模拟器
if loop: loop(self.sim)
# 绘画模拟结果
self.draw()
# 更新窗口
pygame.display.update()
clock.tick(self.fps)
# 定义和处理窗口事件
for event in pygame.event.get():
# 关闭窗口
if event.type == pygame.QUIT:
running = False
# 处理鼠标事件
elif event.type == pygame.MOUSEBUTTONDOWN:
# 如果鼠标被按下
if event.button == 1:
# 左键
x, y = pygame.mouse.get_pos()
x0, y0 = self.offset
self.mouse_last = (x-x0*self.zoom, y-y0*self.zoom)
self.mouse_down = True
if event.button == 4:
# 鼠标滚轮
self.zoom *= (self.zoom**2+self.zoom/4+1) / (self.zoom**2+1)
if event.button == 5:
# 鼠标滚轮被按下
self.zoom *= (self.zoom**2+1) / (self.zoom**2+self.zoom/4+1)
elif event.type == pygame.MOUSEMOTION:
# 拖拉界面
if self.mouse_down:
x1, y1 = self.mouse_last
x2, y2 = pygame.mouse.get_pos()
self.offset = ((x2-x1)/self.zoom, (y2-y1)/self.zoom)
elif event.type == pygame.MOUSEBUTTONUP:
self.mouse_down = False
def run(self, steps_per_update=1):
# 通过更新loop来进行模拟
def loop(sim):
sim.run(steps_per_update)
self.loop(loop)
def convert(self, x, y=None):
# 将模拟坐标转换为屏幕坐标
if isinstance(x, list):
return [self.convert(e[0], e[1]) for e in x]
if isinstance(x, tuple):
return self.convert(*x)
return (
int(self.width/2 + (x + self.offset[0])*self.zoom),
int(self.height/2 + (y + self.offset[1])*self.zoom)
)
def inverse_convert(self, x, y=None):
# 将屏幕坐标转换为模拟坐标
if isinstance(x, list):
return [self.convert(e[0], e[1]) for e in x]
if isinstance(x, tuple):
return self.convert(*x)
return (
int(-self.offset[0] + (x - self.width/2)/self.zoom),
int(-self.offset[1] + (y - self.height/2)/self.zoom)
)
def background(self, r, g, b):
# 填充背景颜色
self.screen.fill((r, g, b))
def line(self, start_pos, end_pos, color):
# 画线
gfxdraw.line(
self.screen,
*start_pos,
*end_pos,
color
)
def rect(self, pos, size, color):
# 画矩形
gfxdraw.rectangle(self.screen, (*pos, *size), color)
def box(self, pos, size, color):
# 画矩形
gfxdraw.box(self.screen, (*pos, *size), color)
# 画多边形
def polygon(self, vertices, color, filled=True):
gfxdraw.aapolygon(self.screen, vertices, color)
if filled:
gfxdraw.filled_polygon(self.screen, vertices, color)
# 绘制容器
def rotated_box(self, pos, size, angle=None, cos=None, sin=None, centered=True, color=(0, 0, 255), filled=True):
"""Draws a rectangle center at *pos* with size *size* rotated anti-clockwise by *angle*."""
x, y = pos
l, h = size
if angle:
cos, sin = np.cos(angle), np.sin(angle)
vertex = lambda e1, e2: (
x + (e1*l*cos + e2*h*sin)/2,
y + (e1*l*sin - e2*h*cos)/2
)
if centered:
vertices = self.convert(
[vertex(*e) for e in [(-1,-1), (-1, 1), (1,1), (1,-1)]]
)
else:
vertices = self.convert(
[vertex(*e) for e in [(0,-1), (0, 1), (2,1), (2,-1)]]
)
self.polygon(vertices, color, filled=filled)
def rotated_rect(self, pos, size, angle=None, cos=None, sin=None, centered=True, color=(0, 0, 255)):
self.rotated_box(pos, size, angle=angle, cos=cos, sin=sin, centered=centered, color=color, filled=False)
# 绘制箭头
def arrow(self, pos, size, angle=None, cos=None, sin=None, color=(150, 150, 190)):
if angle:
cos, sin = np.cos(angle), np.sin(angle)
self.rotated_box(
pos,
size,
cos=(cos - sin) / np.sqrt(2),
sin=(cos + sin) / np.sqrt(2),
color=color,
centered=False
)
self.rotated_box(
pos,
size,
cos=(cos + sin) / np.sqrt(2),
sin=(sin - cos) / np.sqrt(2),
color=color,
centered=False
)
# 画道路
def draw_roads(self):
for road in self.sim.roads:
self.rotated_box(
road.start,
(road.length, 3.7),
cos=road.angle_cos,
sin=road.angle_sin,
color=(180, 180, 220),
centered=False
)
# Draw road lines
# self.rotated_box(
# road.start,
# (road.length, 0.25),
# cos=road.angle_cos,
# sin=road.angle_sin,
# color=(0, 0, 0),
# centered=False
# )
# 画道路上的箭头
if road.length > 5:
for i in np.arange(-0.5*road.length, 0.5*road.length, 10):
pos = (
road.start[0] + (road.length/2 + i + 3) * road.angle_cos,
road.start[1] + (road.length/2 + i + 3) * road.angle_sin
)
self.arrow(
pos,
(-1.25, 0.2),
cos=road.angle_cos,
sin=road.angle_sin
)
# 画车
def draw_vehicle(self, vehicle, road):
l, h = vehicle.l, 2
sin, cos = road.angle_sin, road.angle_cos
x = road.start[0] + cos * vehicle.x
y = road.start[1] + sin * vehicle.x
color = vehicle.color
self.rotated_box((x, y), (l, h), cos=cos, sin=sin, centered=True,color=color)
def draw_vehicles(self):
for road in self.sim.roads:
for vehicle in road.vehicles:
self.draw_vehicle(vehicle, road)
# 属性提醒
def draw_status(self):
text_fps = self.text_font.render(f't={self.sim.t:.5}', False, (0, 0, 0))
text_frc = self.text_font.render(f'n={self.sim.frame_count}', False, (0, 0, 0))
text_ftc = self.text_font.render(f'canf={self.sim.canf}', False, (0, 0, 0))
text_fcc = self.text_font.render(f'rate={self.sim.num*100/self.sim.frame_count}', False, (0, 0, 0))
self.screen.blit(text_fps, (0, 0))
self.screen.blit(text_frc, (0, 20))
self.screen.blit(text_ftc, (0, 40))
self.screen.blit(text_fcc, (0, 60))
def draw(self):
self.background(*self.bg_color)
self.draw_roads()
self.draw_vehicles()
self.draw_status()
system.py（相当于main函数，入口函数，直接跑这个调用全局）
import random
from trafficSimulator import *
# 创建策略
sim = Simulation()
# 增加路
sim.create_roads([
( (290, 106),(-10, 106)),
( (290, 102),(-10, 102)),
((290, 98), (-10, 98)),
((290, 94), (100, 94)),
((80, 94), (-10, 94)),
((101, 90), (80, 94)),
((290, 90), (100, 90)),
((100, 94), (80, 94)),
])
sim.create_gen({
'vehicle_rate': 40,
'vehicles': [
[1, {"path": [0]}],
[1, {"path": [1]}],
[6, {"path": [3, 7,4]}],
[1, {"path": [2]}],
[6, {"path": [6, 5, 4]}]
]
})
# 开始仿真
win = Window(sim)
win.offset = (-145, -95)
win.zoom = 8
win.run(steps_per_update=5)
pygame官方教程
https://www.pygame.org/
个人博客 交通流的模拟
https://towardsdatascience.com/simulating-traffic-flow-in-python-ee1eab4dd20f
朱婷,杨鸿泰,钟心志,邹亚杰.基于智能驾驶员模型的危险跟驰行为特征分析[J].交通与运输,2021,37(03):87-91.
金世文,何海成,王海晨.高速公路混合车流对交通的影响因素分析[J].电子设计工程,2021,29(18):121-125+130.
杭州推出三种新交通组织措施
https://xw.qq.com/cmsid/20210731A02GIG00?f=newdc