代码收藏家技术教程 2024-06-11

使用Python计算交叉口车道的饱和流量

饱和流量指的是在一次连续的绿灯时间内，交叉口某车道上的连续车队通过停止线的最大流量。

而获取饱和流量是计算交叉口通行能力与服务水平的一环。一般来说，饱和流量应使用实测数据，其调查方法大致如下：

实测饱和流量的方法

确定观测时段（如高峰期15min），两人观察一根车道，并以信号周期为单元进行观测。

一人打开手机备忘录，一人打开手机秒表

站在停止线处，绿灯启亮时，每有一辆车驶出停止线时做一次记录：

一次观测结束后，截图/滚动截图，事后再将车型和驶过时刻进行匹配。如果两者数量不匹配，你可以拷打拷打你的同伴。

要求（1）选取相同车型连续通过停止线的数据。这里前10辆车均为小车0

（2）前4辆车由于需要反应和起动加速时间，其车头时距是不饱和的。在计算时不考虑

可知第4辆车和第10辆车驶出停止线时间为12s和25.2s

故该车队平均饱和车头时距为:

$\bar{h}=\frac{25.2-12}{10-4}=2.2s/pcu$

车队饱和流量为：

$S=\frac{3600}{\bar{h}}=1636 pcu/(h\cdot ln)$

估算饱和流量

当无法获取实测数据时（如新建交叉口设计、懒得实测），才考虑使用估算方法。

估算过程较为繁琐，涉及有大量修正系数的计算，并需要考虑交叉口几何因素、渠化方式、信号配时、各流向交通流冲突。

其基本流程为：

（1）获取实测基本饱和流量 $S$ （无干扰交叉口，规范中有给定值）

（2）根据各种参数（如车道宽度、坡度，干扰交通流流量等），计算出各类修正系数 $f_i$

（3）乘算后得到估算的饱和流量 $S'=f_1f_2 \cdots f_nS$

为此，这里基于 DGJ08-96-2013：城市道路平面交叉口规划与设计规程 设计了一组python函数：

详见：所以我要打广告了o(*￣▽￣*)ブpython库——traffictoolshttps://www.yuque.com/u33978426/traffictools?#%20%E3%80%8Atraffictools%E6%96%87%E6%A1%A3%E3%80%8B

借助该函数，可以聚焦于所需参数的输入，无须关心繁琐的计算过程。其基本流程如下：

你要做的：

（1）命名交叉口

（2）输入相关参数

（3）少量代码

from traffictools.indicator import manaul

# 生成多种车道计算器，包括某交叉口所有的车道类型
s_zhi = manual.Sq('直行车道')  # 直行车道类
s_zuo = manual.Sq('左转车道', 'L')  # 左转车道类
s_you = manual.Sq('右转车道', 'R')  # 右转车道类
s_zhi_zuo = manual.Sq('直左', 'SL')  # 直左车道类

# 获取参数表格模板，表格将生成在当前目录下
s_zhi.form_generate()
s_zuo.form_generate()
s_you.form_generate()
s_zhi_zuo.form_generate()

# 计算结果
res_zhi = s_zhi.conclude()
res_zuo = s_zuo.conclude()
res_you = s_you.conclude()
res_zhi_zuo = s_zhi_zuo.conclude()

print(res_you)  # 直接打印
# res_you.to_excel('./res_zhi')  # 保存

你得到的

饱和流量估算结果以及相关的修正系数

copy源码

如果你不想下载第三方库，那你就之间在下边copy源码吧

copy后，按照使用文档食用

traffictools.indicator.manual.Sqhttps://www.yuque.com/u33978426/traffictools/lgftwucgpce1ogx7?singleDoc#%20%E3%80%8Aindicator.manual.Sq%E3%80%8B

"""
@File : manual.py
@Author : 哈回差米拉
@Time : 2024/04/18 23:20
"""
import pandas as pd
import os
import numpy as np


class Sq:
    def __init__(self, name, direction: str = 'S'):
        self.ST = 1650
        self.SL = 1550
        self.SR = 1550
        self.STS = 3600
        self.STD = 1600
        self.I = 4
        self.fold = 'Sq'
        self.KSI = {1: 1, 2: 0.625, 3: 0.51, 4: 0.44}
        self.FPL = {60: 0.88, 90: 0.87, 120: 0.86}
        self.FPH = {60: 0.42, 90: 0.38, 120: 0.36}
        self.__direction = direction.upper()
        self.__filename = name

    def form_generate(self):
        general_col = ['id', 'W', 'G', 'HV', 'C', 'j']
        s_col = ['ge(可缺省)', 'bL(二选一)', 'qL(二选一)']
        l_col = ['n', 'qt0', 'lambda(可缺省)']
        r_col = ['r', 'p', 'bTD', 'bTS', 'Wb']
        t_col = ['qT', 'qturn']

        # 生成参数表
        if self.__direction == 'S':
            form = pd.DataFrame(columns=general_col + s_col)
        elif self.__direction == 'L':
            form = pd.DataFrame(columns=general_col + l_col)
        elif self.__direction == 'R':
            form = pd.DataFrame(columns=general_col + r_col)
        elif self.__direction == 'SL':
            form = pd.DataFrame(columns=general_col + s_col + l_col + t_col)
        elif self.__direction == 'SR':
            form = pd.DataFrame(columns=general_col + s_col + l_col + t_col)
        else:
            raise ValueError(f'direction参数设置不正确：{self.__direction} 不在[S、L、R、SL、SR]内')

        # 检索是否存在文件，若不存在则生成一个form
        try:
            pd.read_excel(f'./{self.fold}/{self.__filename}_{self.__direction}.xlsx')
        except OSError or FileNotFoundError:
            if not os.path.exists(f'./{self.fold}'):
                os.mkdir(f'{self.fold}')
            form.to_excel(f'./{self.fold}/{self.__filename}_{self.__direction}.xlsx', index=False)

    def conclude(self):
        # 列名
        output = ['饱和流量|pcu/h']
        f_general_col = ['id', 'fw', 'fg']
        f_s_col = ['fb']
        f_l_col = ['fL']
        f_r_col = ['fr', 'fp', 'fb_r', 'fpb']
        f_t_col = ['fTT']

        # 通用系数
        form = pd.read_excel(f'./{self.fold}/{self.__filename}_{self.__direction}.xlsx')
        form['fw'] = form['W'].apply(Sq.__fw)
        form['fg'] = form.apply(lambda x: Sq.__fg(x['G'], x['HV']), axis=1)

        # 直行车道求解过程
        def s_process(op=output[0]):
            form['fb'] = form.apply(lambda x: Sq.__fb(x['bL(二选一)'], x['ge(可缺省)'],
                                                      x['C'], x['j'], x['qL(二选一)'], self.I), axis=1)
            form[op] = (form['fw'] * form['fg'] * form['fb'] * self.ST).astype(int)

        # 左转车道求解过程
        def l_process(op=output[0]):
            form['fL'] = form.apply(lambda x: Sq.__fl(x['n'], x['qt0'], x['lambda(可缺省)'], self.KSI,
                                                      x['C'], x['j'], self.I), axis=1)
            form[op] = (form['fw'] * form['fg'] * form['fL'] * self.SL).astype(int)

        # 右转车道求解过程
        def r_process(op=output[0]):
            form['fr'] = form.apply(lambda x: Sq.__fr(x['r']), axis=1)
            form['fp'] = form.apply(lambda x: Sq.__fp(x['p'], x['C'], self.FPL, self.FPH), axis=1)
            form['fb_r'] = form.apply(lambda x: Sq.__fb_r(x['C'], x['bTD'], x['bTS'], self.STD, self.STS, x['Wb'])
                                      , axis=1)
            form['fpb'] = form[['fp', 'fb_r']].min(axis=1)
            form[op] = (form['fw'] * form['fg'] * form['fr'] * form['fpb'] * self.SR).astype(int)

        # 专用系数
        if self.__direction == 'S':
            s_process()
            res = form[f_general_col + f_s_col + output].copy()
        elif self.__direction == 'L':
            l_process()
            res = form[f_general_col + f_l_col + output].copy()
        elif self.__direction == 'R':
            r_process()
            res = form[f_general_col + f_r_col + output]
        elif self.__direction == 'SL':
            s_process('ST')
            l_process('SL')
            form['fTT'] = form.apply(lambda x: Sq.__ft_turn(x['qT'], x['qturn'], x['ST'], x['SL']), axis=1)
            form[output[0]] = (form['ST'] * form['fTT']).astype(int)
            res = form[f_general_col + f_s_col + f_l_col + f_t_col + output]
        elif self.__direction == 'SR':
            s_process('ST')
            r_process('SR')
            form['fTT'] = form.apply(lambda x: Sq.__ft_turn(x['qT'], x['qturn'], x['ST'], x['SR']), axis=1)
            form[output[0]] = (form['ST'] * form['fTT']).astype(int)
            res = form[f_general_col + f_s_col + f_r_col + f_t_col + output]
        else:
            res = -1
        return res

    # 车道宽度修正
    @staticmethod
    def __fw(w):
        if 3 <= w <= 3.5:
            return 1
        elif 2.7 <= w < 3:
            return 0.4 * (w - 0.5)
        elif w > 3.5:
            return 0.05 * (w + 16.5)
        else:
            raise ValueError(f'输入的车道宽度w不符合规范：(w = {w} < 2.7m)')

    # 坡度与大车修正
    @staticmethod
    def __fg(g, hv):
        if g < 0:  # 下坡时，取0
            g = 0
        if hv <= 0.5:
            return 1 - (g + hv)
        else:
            raise ValueError(f'大车率hv超出规范要求：hv = {hv} > 0.5')

    # 直行-非机动车修正
    @staticmethod
    def __fb(bl, ge, c, j, ql, i):
        if np.isnan(ge):
            ge = c / j - i
        if np.isnan(bl):
            bl = int(ql * (c - ge) / 3600)
        if bl > 0:
            return round(1 - (1 + np.sqrt(bl)) / ge, 3)
        else:
            return 1

    # 左转校正系数
    @staticmethod
    def __fl(n, qt0, _lambda, ksi, c, j, i):
        if np.isnan(_lambda):
            ge = c / j - i
            _lambda = ge / c
        if qt0 == 0:
            return 1
        else:
            n = ksi[n]
            return np.exp(-0.001 * n * qt0 / _lambda)

    # 右转-转弯半径校正系数
    @staticmethod
    def __fr(r):
        if r > 15:
            return 1
        elif 0 <= r <= 15:
            return 0.5 + r / 30
        else:
            ValueError(f'右转转弯半径r为负数：r = {r} < 0 ')

    # 右转-行人校正系数(近似)
    @staticmethod
    def __fp(p, c, fpl, fph):
        basic_c = np.array([60, 90, 120])
        c = basic_c[np.argmin(np.abs(c - basic_c))]  # 获取近似信号周期
        if p <= 20:
            return fpl[c]
        else:
            return fph[c]

    # 右转-非机动车校正系数
    @staticmethod
    def __fb_r(c, btd, bts, std, sts, wb):
        tt = 3600 * (bts / sts + btd / std) / wb
        return 1 - tt / c

    # 直转合流校正系数
    @staticmethod
    def __ft_turn(qt, q_turn, st, sturn):
        k = st / sturn
        qt1 = k * q_turn + qt
        return (qt + q_turn) / qt1

作者：哈回差米拉