1500字范文 > 使用百度地图API计算两点直线距离道路距离和行车时间

使用百度地图API计算两点直线距离道路距离和行车时间

时间：2022-11-22 05:07:02

使用百度地图API计算两点直线距离、道路距离和行车时间

摘要关键词介绍数据展示工具介绍根据经纬度计算两点之间的距离根据经纬度计算两点之间的道路距离和行车时间多进程运行中结果展示问题收集参考文献

摘要

在本篇博客中，将通过调用百度地图的API计算两点间的距离和行车时间。处理的csv文件包含30万条数据的经纬度，通过爬虫即可获得；为了更加高效的处理数据，使用到了多进程的方式。

关键词

Python，多进程，距离计算

介绍

数据展示

工具介绍

json包：用于读取json文件内容multiprocessing包：python专用多进程包requests包：用于网页信息爬取，是一个很强大的包tqdm包：进度展示time包：time.sleep()，设置进程休眠时间socket包：socket.setdefaulttimeout()，防止因为爬取网页时间过长带来的卡顿

import jsonfrom multiprocessing.pool import Poolfrom urllib.request import urlopenimport pandas as pdfrom random import choiceimport numpy as npfrom tqdm import tqdmimport requestsfrom requests import adaptersimport timeimport socket

根据经纬度计算两点之间的距离

def getWorkDistance(lon_a, lat_a, lon_b, lat_b):if abs(lon_a - lon_b) < 0.000001 and abs(lat_a - lat_b) < 0.000001:return 0re = 6378140 # 赤道半径 (m)rp = 6356755 # 极半径 (m)oblateness = (re - rp) / re # 地球扁率rad_lat_a = radians(lat_a)rad_lon_a = radians(lon_a)rad_lat_b = radians(lat_b)rad_lon_b = radians(lon_b)atan_a = atan(rp / re * tan(rad_lat_a))atan_b = atan(rp / re * tan(rad_lat_b))tmp = acos(sin(atan_a) * sin(atan_b) + cos(atan_a) * cos(atan_b) * cos(rad_lon_a - rad_lon_b))if tmp == 0:return 0c1 = (sin(tmp) - tmp) * (sin(atan_a) + sin(atan_b)) ** 2 / cos(tmp / 2) ** 2c2 = (sin(tmp) + tmp) * (sin(atan_a) - sin(atan_b)) ** 2 / sin(tmp / 2) ** 2dr = oblateness / 8 * (c1 - c2)distance = re * (tmp + dr)return distance

根据经纬度计算两点之间的道路距离和行车时间

在调用百度地图API的时候需要用到API KEY作为密钥，每个key每天有使用次数的限制，所以我们需要通过收集许多keys来保证爬取过程的流畅和成功率。

ak_pool = ['k936lbWYFPwG1LEoKb9faZ8MEizFwh60','jBSeo7Mu3M8c1YGRPwegojz6G77K1XQo','0Acp9mZxe08BaGRt51cN3XAl4qCjEsrG',...]

def getDistAndTime(var):"""The input var is a list contains both latitude and longitude of new and old places. Try different AK to catch the information. If AKs run out, return 'No AK', 'No AK' to represent that all today's opportunities are exhausted. If there's some wrong happened when requesting, then return 'wrong request', 'wrong request'. To deal with these invalid returns, we can have a try in another time."""s = requests.session()s.keep_alive = False # 关闭多余连接requests.adapters.DEFAULT_RETRIES = 5 # 增加重连次数i = 0for i in range(len(ak_pool)):ak = ak_pool[i]try:url_drive = r"http://api./direction/v2/driving?output=json&origin={0},{1}&destination={2},{3}&ak={4}".format(var[0], var[1], var[2], var[3], ak)result_drive = json.loads(urlopen(url_drive).read()) # json转dictstatus_drive = result_drive['status']if status_drive == 0: # 状态码为0：无异常distance_drive = result_drive['result']['routes'][0]['distance'] # 里程(米)timesec_drive = result_drive['result']['routes'][0]['duration'] # 耗时(秒)return distance_drive, timesec_driveexcept:time.sleep(5)return 'wrong request', 'wrong request'return 'No AK', 'No AK'

多进程

variables = zip(new_lat, new_long, old_lat, old_long)pool1 = Pool(10) # 设置进程数newToOldDis, newToOldTime = [], []for result1, result2 in tqdm(pool1.imap(getWorkRoad, variables), total=1000): # 加入进度条查看进度newToOldDis.append(result1)newToOldTime.append(result2)pool1.close()pool1.join()

运行中

结果展示

id,new_lat,new_long,work_lat,work_long,current_lat,current_long,newToWork_Distance,curToWork_Distance,newToWork_RoadDis,curToWork_RoadDis,newToWork_TimeByCar,curwToWork_TimeByCar