基于Pygame框架的交通导流可视化模拟

目录标题

• 项目介绍
• 项目要求
• 关键技术
• 项目核心功能
• 代码
• 参考资料

源码下载地址

https://download.csdn.net/download/david2000999/85627883

项目介绍

本项目根据以下项目要求完成的一个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
        self.frame_count = 0
        self.dt = 1/60
        self.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.v1):
                n3+=1
        for veh in self.roads[6].vehicles:
            if(veh.v1):
                n6+=1

        if(self.canf):
            if (  n33 or n33 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

        idx = -1

        for road in self.roads:
            idx += 1
            if len(road.vehicles) == 0: continue
            if(idx>=0 and idx<4):

                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):

                    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

        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):

        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)

        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):

        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
            )

            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

Original: https://blog.csdn.net/david2000999/article/details/125257436
Author: 死磕的斯坦张
Title: 基于Pygame框架的交通导流可视化模拟