目录
- 一、卡口数据简介
- 二、路段基本图
一、卡口数据简介
| 数据名称 | 含义 |
|---|---|
| DEVICEID | 设备ID |
| TRAVELID | 车辆ID，可分辨车辆 |
| hpzl | 车辆类型，1表示大型车，2表示小汽车 |
| SJ | 时间 |
| LANEID | 车道编号，较长编号表示该位置处于上下游车道过渡区域，无划线；-1表示由于技术原因导致的无效计数，可忽略该检测值 |
| LANNUM | 车道数 |
| SPEED | 瞬时车速 |
| TURN | 车道方向信息，LSR分别表示左直右及其组合，LU和RU分别表示左掉头和右掉头，LC表示上下游车道连接过渡段，表明此断面包含车道变化区域 |
| FTNODE | 起始节点 |
卡口布设简图
二、路段基本图
绘出东进口道（下游检测器编号504，上游检测器编号508）的路段基本图，与断面基本图对比。
交通流特性（流速密）在断面、路段上的区别
利用卡口数据绘制断面基本图——Python交通数据分析
import numpy as np
import pandas as pd
from pandas import DataFrame
from numpy import *
import scipy
from scipy import stats
import matplotlib.pyplot as plt
from matplotlib.font_manager import FontProperties
import matplotlib.dates as mdate
import time
kk_data = pd.read_csv('data/kk_data/kk.csv')
kk_data['SJ'] = pd.to_datetime(kk_data['SJ'])
kk_data['TRAVELID'] = kk_data['TRAVELID'].astype(str) # 转化为字符串
kk_data_504 = kk_data[kk_data['DEVICEID'] == 504].copy()
kk_data_504 = kk_data_504.sort_values(by = 'SJ')
kk_data_504.rename(columns={'SJ':'OUT_SJ'},inplace=True) # 重命名
kk_data_508 = kk_data[kk_data['DEVICEID']==508].copy()
kk_data_508 = kk_data_508.sort_values(by = 'SJ')
kk_data_508.rename(columns={'SJ':'IN_SJ'},inplace=True)
data_merge = pd.merge(kk_data_508,kk_data_504,on='TRAVELID',how='inner')
data_merge['time'] = data_merge['OUT_SJ'] - data_merge['IN_SJ']
#清洗数据，有的时间过短，有的时间过长，还有的进口道时间晚于出口道时间
wrong_data = data_merge[(data_merge['time'] < pd.Timedelta(seconds=5)) | (data_merge['time'] >= pd.Timedelta(minutes=3))]
wrong_data_TRAVELID = wrong_data['TRAVELID'].tolist()
for i in wrong_data_TRAVELID:
kk_data_508 = kk_data_508[kk_data_508['TRAVELID']!=i]
kk_data_504 = kk_data_504[kk_data_504['TRAVELID']!=i]
data_merge = data_merge[(data_merge['time'] > pd.Timedelta(seconds=5)) & (data_merge['time'] <= pd.Timedelta(minutes=3))]
kk_data_508 = kk_data_508.reset_index() # 清洗了时间异常的数据之后，索引会出现问题，需要重新设定索引
kk_data_504 = kk_data_504.reset_index()
data_merge = data_merge.reset_index()
# 进口道下游数据远多于进口道上游数据，不合理，需要对上下游的数据进行再次清洗
data_merge_TRAVELID = data_merge['TRAVELID'].tolist()
labels_list = []
# 避免清洗在路段上的原始车辆,从第六辆车开始遍历
for i in range(5,len(kk_data_504)):
ID = kk_data_504['TRAVELID'].iloc[i]
if ID not in data_merge_TRAVELID:
labels_list.append(i)
kk_data_504.drop(labels=labels_list,axis=0,inplace=True)
kk_data_504 = kk_data_504.reset_index()
# 避免删除在最后时刻进入路段的车辆，最后十辆车不遍历
labels_list = []
for i in range(0, (len(kk_data_508) - 10)):
ID = kk_data_508['TRAVELID'].iloc[i]
if ID not in data_merge_TRAVELID:
labels_list.append(i)
kk_data_508.drop(labels=labels_list,axis=0,inplace=True)
kk_data_508 = kk_data_508.reset_index()
# 计算每辆车的速度
time = data_merge['time']
sec = []
V = []
for i in range(0,len(time),1):
sec.append(time[i].seconds)
V.append((300/time[i].seconds)*3.6)
data_merge['sec'] = sec
data_merge['V'] = V
# 路段基本图
# 出入量法，统计每一分钟路段上的车辆数
# 结合处理过后的数据与原始进出口数据，原始车辆数E=2
E = 2
K = []
Q = []
time = pd.datetime(2019,7,9,6,0)
while(time < pd.datetime(2019,7,9,10,0)):
data_IN = kk_data_508[(kk_data_508['IN_SJ']>=time) & (kk_data_508['IN_SJ']<(time+pd.Timedelta(minutes=1)))].copy()
data_OUT = kk_data_504[(kk_data_504['OUT_SJ']>=time) & (kk_data_504['OUT_SJ']<(time+pd.Timedelta(minutes=1)))].copy()
data_V = data_merge[(data_merge['OUT_SJ']>(time+pd.Timedelta(minutes=1))) & (data_merge['IN_SJ']<=(time+pd.Timedelta(minutes=1)))].copy()
count = len(data_IN) - len(data_OUT)
E = E + count
V_mean = data_V['V'].mean()
K.append(E / (2 * 0.3)) # 辆/(千米*车道)
Q.append(V_mean * (E / (2 * 0.3))) # Q=K*V
time = time + pd.Timedelta(minutes=1)
# 断面基本图
def kk(kk_data):
kk_data['1/SPEED'] = 1 / kk_data['SPEED']
kk_data_group = kk_data.groupby(['DEVICEID',pd.Grouper(key='IN_SJ',freq='5min')])
kk_data_flow = kk_data_group['DEVICEID'].count()
kk_data_flow = kk_data_flow * 12 # 在交通测量平台的显示中，东进口上游的车道数为1，但在数据中，车道数为2，此处设定为1
kk_data_speed = 1 / (kk_data_group['1/SPEED'].mean())
return kk_data_flow,kk_data_speed
kk_flow,kk_speed = kk(kk_data_508)
kk_density = kk_flow / kk_speed
plt.xlabel('K: veh/km/lane')
plt.ylabel('Q: veh/h/lane', rotation=360, horizontalalignment='right', verticalalignment='center')
plt.scatter(K,Q)
plt.scatter(kk_density,kk_flow)
plt.legend(['Road','Section']) # Road路段，Section断面
plt.show()
注意，东进口道可能并不是一条封闭的路段，才需要清洗这么多数据
水平有限，仅供参考，有错误请指出