用python实现获取道路实时信息和速度

实时信息和速度"/>

用python实现获取道路实时信息和速度

在爬取之前，要先申请自己的key,申请的方法可以参见

获取百度地图开发平台的key_lei吼吼的博客-CSDN博客=1001.2014.3001.5501

初始化每条道路四个方向（we,ew,sn,ns）的速度信息于内存的 allRoadInfo 字典对象中。因为城市道路交通条例的规定，市区普通道路车速限速 60km/h，因此设置每个方向上的车速默认值为 60。将道路名称和各方向速度作为一个字典对象插入到 allRoadInfo 对象中

# 功能：读取定义道路编号的文本，对每条道路进行四个方向的初始化，插入到allRoadInfo的字典中
# 因为城市中的车速为0到60，所以在初始化的时候就设置为60
def initRoadInfo():# 读取文件，获取查询的道路的名称f = open("./road_name.txt", 'r', encoding="utf-8")lines = f.readlines()# 遍历文件中的道路的信息，创建一个roadInfo字典并初始化i = 1for line in lines:roadInfo = {'name': line, 'we': 60, 'ew': 60, 'sn': 60, 'ns': 60, 'time': '0'}# 创建一个道路信息的字典allRoadInfo[i] = roadInfoi += 1print("初始化道路信息成功！")print(allRoadInfo)f.close()

初始化道路信息后便可以根据收集到的数据更 新对应的道路信息。首先，通过拼接字符串的方式获得想要查询道路的 URL 地址（这个拼接URL的方法可以参见百度地图服务文档），利用 get()方法返回 response 类型的数据，使用 load()方法把数据转换为 Json 格式，便于后续处理根据返回结果，获取到该道路各个方向上车流平均速度，然后更新allRoadInfo 字典中相应的车速数据。

# 功能：调用百度地图API获取数据，解析数据后数将据流写入文件中
def updateRoadInfo():f = open('./实时速度.txt', "a+")road_num = 44  # 这个需要根据你们自己的想要获取的道路信息的个数定（就是道路数量）for num in range(road_num):  # 遍历所有道路，调用API获取相应的数据pre_baidu = "?"# 拼接字符串形成URLaddress_daidu = "{}road_name={}&city={}&ak={}".format(pre_baidu,allRoadInfo[num + 1]["name"].replace("\n", "").replace("\t", ""), "重庆市", key_baidu)# 在爬取数据的同时写入时间信息curr_time = datetime.datetime.now()time_str = datetime.datetime.strftime(curr_time, '%Y-%m-%d %H:%M:%S')allRoadInfo[num + 1]['time'] = time_strresults = requests.get(address_daidu)  # 发出get请求，返回数据data = json.loads(results.text)# print(data)if data["status"] != 0:  # 判断信息是否获取成功，0表示获取数据成功print("\n{}的信息获取失败！".format((allRoadInfo[num + 1]["name"])))  # 若信息获取失败就不做任何的操作else:print("\n{}的信息获取成功！".format(allRoadInfo[num + 1]["name"]))temp1 = data["evaluation"]['status_desc']  # 获取道路描述temp2 = data["description"]  # 获取道路描述if temp1.find(u"拥堵") != -1 or temp2.find(u"拥堵") != -1:  # 从道路描述中判断道侣是否发生拥堵print("该道路正在发生拥堵！")print("拥堵详情:%s ,%s" % (temp1, temp2))# 因为百度地图返回的信息中，道路拥堵和道路没有拥堵时候，返回的信息不一样，所以这里要分一下if "congestion_sections" in data["road_traffic"][0]:  # 如果出现道路拥堵信息则遍历详情信息length = len(data["road_traffic"][0]["congestion_sections"])  # 获取道路拥堵信息长度for i in range(length):speed = data["road_traffic"][0]["congestion_sections"][i]["speed"]direct = getDirection(data["road_traffic"][0]["congestion_sections"][i]["section_desc"])# 获取当前道路信息的方向代码allRoadInfo[num + 1][direct] = speed  # 储存对应道路，对应方向，速度的信息else:print("该道路双向通畅！")# 将道路信息写入文件中f.write(("%d  %s  %.2f  %.2f  %.2f  %.2f  %s\n" % (num + 1,allRoadInfo[num + 1]["name"].replace('\n', ''),allRoadInfo[num + 1]["we"],allRoadInfo[num + 1]["ew"],allRoadInfo[num + 1]["sn"], allRoadInfo[num + 1]["ns"],allRoadInfo[num + 1]['time'])))print("该道路个方向速度写入成功！")f.close()print(allRoadInfo)print("\n************结束调用API获取道路信息***************")

在这段代码里面有一个getDirection()的函数，函数的描述如下：

因为API 所获得数据中描述道路信息的字符串为中文，因此需要定义一个函数来返回这条描述所对应的方向代码。

最后主函数

while True:initRoadInfo()  # 初始化道路信息updateRoadInfo()  # 更新道路信息（将实时速度写入）print("成功获取第%d帧信息！" % (time_num))time_num += 1time.sleep(60)  # 等待gap_time*60秒再次执行

整体代码

# Author:lei吼吼
# -*- coding=utf-8 -*-
# @Time :2022/3/20 15:00
# @File: 实时速度.py
# @Software:PyCharmimport datetime
import time
import requests
import jsoncurr_time = datetime.datetime.now()
time_str = datetime.datetime.strftime(curr_time, '%Y-%m-%d %H:%M:%S')
key_baidu = "自己的AK"  # 这个就是自己的AK，你们要自己改了
allRoadInfo = {}
time_num = 1# 功能：读取定义道路编号的文本，对每条道路进行四个方向的初始化，插入到allRoadInfo的字典中
# 因为城市中的车速为0到60，所以在初始化的时候就设置为60
def initRoadInfo():# 读取文件，获取查询的道路的名称f = open("./road_name.txt", 'r', encoding="utf-8")lines = f.readlines()# 遍历文件中的道路的信息，创建一个roadInfo字典并初始化i = 1for line in lines:roadInfo = {'name': line, 'we': 60, 'ew': 60, 'sn': 60, 'ns': 60, 'time': '0'}# 创建一个道路信息的字典allRoadInfo[i] = roadInfoi += 1print("初始化道路信息成功！")print(allRoadInfo)f.close()# 功能：将中文的道路信息的描述转化成英文代码
def getDirection(desc):direction = "default"# 根据中文的描述进行初始化if desc.find(u"西向东") != -1:direction = "we"if desc.find(u"东向西") != -1:direction = "ew"if desc.find(u"南向北") != -1:direction = "sn"if desc.find(u"北向南") != -1:direction = 'ns'return direction# 功能：调用百度地图API获取数据，解析数据后数将据流量写入文件中
def updateRoadInfo():f = open('./实时速度.txt', "a+")road_num = 44  # 这个需要根据你们自己的想要获取的道路信息的个数定（就是道路数量）for num in range(road_num):  # 遍历所有道路，调用API获取相应的数据pre_baidu = "?"# 拼接字符串形成URL，这个你们要根据你们自己的具体情况，自己改一下address_daidu = "{}road_name={}&city={}&ak={}".format(pre_baidu,allRoadInfo[num + 1]["name"].replace("\n", "").replace("\t", ""), "重庆市", key_baidu)# 在爬取数据的同时写入时间信息curr_time = datetime.datetime.now()time_str = datetime.datetime.strftime(curr_time, '%Y-%m-%d %H:%M:%S')allRoadInfo[num + 1]['time'] = time_strresults = requests.get(address_daidu)  # 发出get请求，返回数据data = json.loads(results.text)# print(data)if data["status"] != 0:  # 判断信息是否获取成功，0表示获取数据成功print("\n{}的信息获取失败！".format((allRoadInfo[num + 1]["name"])))  # 若信息获取失败就不做任何的操作else:print("\n{}的信息获取成功！".format(allRoadInfo[num + 1]["name"]))temp1 = data["evaluation"]['status_desc']  # 获取道路描述temp2 = data["description"]  # 获取道路描述if temp1.find(u"拥堵") != -1 or temp2.find(u"拥堵") != -1:  # 从道路描述中判断道侣是否发生拥堵print("该道路正在发生拥堵！")print("拥堵详情:%s ,%s" % (temp1, temp2))# 因为百度地图返回的信息中，道路拥堵和道路没有拥堵时候，返回的信息不一样，所以这里要分一下if "congestion_sections" in data["road_traffic"][0]:  # 如果出现道路拥堵信息则遍历详情信息length = len(data["road_traffic"][0]["congestion_sections"])  # 获取道路拥堵信息长度for i in range(length):speed = data["road_traffic"][0]["congestion_sections"][i]["speed"]direct = getDirection(data["road_traffic"][0]["congestion_sections"][i]["section_desc"])# 获取当前道路信息的方向代码allRoadInfo[num + 1][direct] = speed  # 储存对应道路，对应方向，速度的信息else:print("该道路双向通畅！")# 将道路信息写入文件中f.write(("%d  %s  %.2f  %.2f  %.2f  %.2f  %s\n" % (num + 1,allRoadInfo[num + 1]["name"].replace('\n', ''),allRoadInfo[num + 1]["we"],allRoadInfo[num + 1]["ew"],allRoadInfo[num + 1]["sn"], allRoadInfo[num + 1]["ns"],allRoadInfo[num + 1]['time'])))print("该道路个方向速度写入成功！")f.close()print(allRoadInfo)print("\n************结束调用API获取道路信息***************")while True:initRoadInfo()  # 初始化道路信息updateRoadInfo()  # 更新道路信息（将实时速度写入）print("成功获取第%d帧信息！" % (time_num))time_num += 1time.sleep(60)  # 等待gap_time*60秒再次执行

最后说明一下我在代码中用到的文档

road_name.txt是道路名字的文本文件（就是你要获取实时速度道路的名字）如下：

 最后得到的实时速度.txt 的样子如下：第一排是道路的名字，第二排是西向东的道路速度，第三排是东向西道路速度，第四排是南向北道路速度，第五排是北向南道路速度，最后一排是爬取数据时的时间