自学内容网 自学内容网

Python应用指南:利用高德地图API实现路径规划

高德路径规划API是一套基于HTTP协议的接口服务,旨在为开发者提供便捷的路径规划解决方案。该API支持多种出行方式,包括步行、公交和驾车,能够满足不同场景下的路径查询需求。通过调用这些API,用户可以获得从起点到终点的最优路径建议,以及详细的行驶距离和预计时间等信息。API返回的数据格式灵活,支持JSON和XML两种格式,便于开发者根据自身应用的需求进行解析和处理。

本篇文章就来实现对高德路径API的调用,并展示如何处理返回的数据,最终实现路径规划结果的可视化呈现。

先讲一下方法思路,一共三个步骤;

方法思路

  1. 确认不同路径规划 API URL数据结构,并根据配置参数构建请求URL
  2. 坐标转换——高德坐标系(GCJ-02) to WGS84
  3. 输出shp图层和csv结果

我们先看步行的路径规划的请求API:路径规划-基础 API 文档-开发指南-Web服务 API|高德地图API

https://restapi.amap.com/v3/direction/walking?origin=116.434307,39.90909&destination=116.434446,39.90816&key=<用户的key>

参考服务示例,我们需要输入并配置的参数包括key,orign(起点坐标)、desination(终点坐标)、output 选json、origin_id (起点 POI ID)与destination_id (目的地 POI ID)填不填都可以,有需求的可以填一下,可以提高路径规划的精度。

        parameters = {
            'key': api_key,
            'origin': from_location,
            'destination': to_location,
            'output': 'json'
        }

首先通过高德API拾取坐标器拾取需要研究的路径的起末点坐标:坐标拾取器 | 高德地图API,这里挑了两个起末点坐标作为示例;

('121.430041,31.154579', '121.318910,31.193987'),  # 上海南站 -> 虹桥火车站
('121.492466,31.225879', '121.508419,31.084134'),  # 人民广场 -> 复旦大学

步行路径规划

步行路径规划 API 可以规划100km 以内的步行通勤方案,并且返回通勤方案的数据。最大支持 100km 的步行路线规划;

完整代码#运行环境Python 3.11

# -*- coding: utf-8 -*-
import requests
import json
import geopandas as gpd
import pandas as pd
from shapely.geometry import LineString, Point
import glob
import math

# 高德GCJ02(火星坐标系)转GPS84
x_pi = 3.14159265358979324 * 3000.0 / 180.0
pi = 3.1415926535897932384626  # π
a = 6378245.0  # 长半轴
ee = 0.00669342162296594323  # 扁率

def gcj02towgs84(lng, lat):
    """
    GCJ02(火星坐标系)转GPS84
    :param lng:火星坐标系的经度
    :param lat:火星坐标系纬度
    :return:
    """
    if out_of_china(lng, lat):
        return lng, lat
    dlat = transformlat(lng - 105.0, lat - 35.0)
    dlng = transformlng(lng - 105.0, lat - 35.0)
    radlat = lat / 180.0 * pi
    magic = math.sin(radlat)
    magic = 1 - ee * magic * magic
    sqrtmagic = math.sqrt(magic)
    dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * pi)
    dlng = (dlng * 180.0) / (a / sqrtmagic * math.cos(radlat) * pi)
    mglat = lat + dlat
    mglng = lng + dlng
    return [lng * 2 - mglng, lat * 2 - mglat]

def transformlat(lng, lat):
    ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat + 0.1 * lng * lat + 0.2 * math.sqrt(math.fabs(lng))
    ret += (20.0 * math.sin(6.0 * lng * pi) + 20.0 * math.sin(2.0 * lng * pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lat * pi) + 40.0 * math.sin(lat / 3.0 * pi)) * 2.0 / 3.0
    ret += (160.0 * math.sin(lat / 12.0 * pi) + 320 * math.sin(lat * pi / 30.0)) * 2.0 / 3.0
    return ret

def transformlng(lng, lat):
    ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng + 0.1 * lng * lat + 0.1 * math.sqrt(math.fabs(lng))
    ret += (20.0 * math.sin(6.0 * lng * pi) + 20.0 * math.sin(2.0 * lng * pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lng * pi) + 40.0 * math.sin(lng / 3.0 * pi)) * 2.0 / 3.0
    ret += (150.0 * math.sin(lng / 12.0 * pi) + 300.0 * math.sin(lng / 30.0 * pi)) * 2.0 / 3.0
    return ret

def out_of_china(lng, lat):
    """
    判断是否在国内,不在国内不做偏移
    :param lng:
    :param lat:
    :return:
    """
    if lng < 72.004 or lng > 137.8347:
        return True
    if lat < 0.8293 or lat > 55.8271:
        return True
    return False

def coordinates(c):
    lng, lat = c.split(',')
    lng, lat = float(lng), float(lat)
    wlng, wlat = gcj02towgs84(lng, lat)
    return wlng, wlat

def route_planning(api_key, routes):
    all_gdfs = []
    all_dfs = []

    for i, (from_location, to_location) in enumerate(routes):
        url = 'https://restapi.amap.com/v3/direction/walking'
        parameters = {
            'key': api_key,
            'origin': from_location,
            'destination': to_location,
            'output': 'json'
        }

        response = requests.get(url, params=parameters)
        data = json.loads(response.text)

        if data['status'] == '1':
            route = data['route']
            paths = route['paths']
            if paths:
                path = paths[0]
                distance = path['distance']
                duration = path['duration']
                steps = path['steps']
                print(f"步行路线规划 {i+1}:")
                all_coordinates = []
                step_data = []

                for step in steps:
                    instruction = step['instruction']
                    distance = step['distance']
                    duration = step['duration']
                    polyline = step['polyline']
                    print(f"指令: {instruction}, 距离: {distance}米, 时间: {duration}秒, 路径: {polyline}")

                    # 转换坐标
                    coordinates_list = [coordinates(c) for c in polyline.split(';')]
                    all_coordinates.extend(coordinates_list)

                    # 将数据添加到列表
                    step_data.append({
                        '指令': instruction,
                        '距离': distance,
                        '时间': duration,
                        '路径': polyline
                    })

                # 创建LineString
                geometry = LineString(all_coordinates)

                # 创建GeoDataFrame
                gdf = gpd.GeoDataFrame(index=[i], geometry=[geometry])
                gdf['distance'] = distance
                gdf['duration'] = duration
                gdf['route_id'] = i  # 添加一个标识符来区分不同的路线

                # 保存为SHP文件
                gdf.to_file(f'route_{i}.shp', driver='ESRI Shapefile')

                # 保存为CSV文件
                df = pd.DataFrame(step_data)
                df['route_id'] = i  # 添加一个标识符来区分不同的路线
                df['label'] = f'步行路线规划{i+1}'  # 添加标签
                df.to_csv(f'route_{i}.csv', index=False, encoding='utf-8-sig')

                all_gdfs.append(gdf)
                all_dfs.append(df)

                print(f"路径 {i+1} 已保存为 route_{i}.shp 和 route_{i}.csv")
            else:
                print(f"未找到路径 {i+1}")
        else:
            print(f"路线规划失败 {i+1}: {data['info']}")

    # 合并所有GeoDataFrame
    combined_gdf = gpd.GeoDataFrame(pd.concat(all_gdfs, ignore_index=True))
    combined_gdf.to_file('combined_routes.shp', driver='ESRI Shapefile')

    # 合并所有DataFrame
    combined_df = pd.concat(all_dfs, ignore_index=True)
    combined_df.to_csv('combined_routes.csv', index=False, encoding='utf-8-sig')

    print("所有路径已合并并保存为 combined_routes.shp 和 combined_routes.csv")

if __name__ == '__main__':
    api_key = '你的key'  # 替换为你的高德地图API密钥

    # 定义多个起终点对
    routes = [
        ('121.430041,31.154579', '121.318910,31.193987'),  # 上海南站 -> 虹桥火车站
        ('121.492466,31.225879', '121.508419,31.084134'),  # 人民广场 -> 复旦大学
        # 可以继续添加更多的起终点对
    ]

    route_planning(api_key, routes)

结果会输出每个路径的单独shp和csv,和合并后的结果为 combined_routes.shp 和 combined_outes.csv;

在调用高德路径规划API并获取到步行路径的详细数据后,我们可以将这些数据在地图上进行可视化,看到步行的具体路径;

同样的,驾车路径规划参考服务示例,我们需要输入并配置的参数包括key、orign(起点坐标)、desination(终点坐标)、output 选json,其他的例如waypoints(途径点)、province(车牌)、strategy(驾车选择策略,驾车策略有20种,详见文档,本脚本选择是''11'')等一些参数,可以参考高德的API文档进行配置,这里不再赘述;

驾车路径规划

驾车路径规划 API 可以规划以小客车、轿车通勤出行的方案,并且返回通勤方案的数据。

完整代码#运行环境Python 3.11

# -*- coding: utf-8 -*-
import requests
import json
import geopandas as gpd
import pandas as pd
from shapely.geometry import LineString, Point
import math

# 高德GCJ02(火星坐标系)转GPS84
x_pi = 3.14159265358979324 * 3000.0 / 180.0
pi = 3.1415926535897932384626  # π
a = 6378245.0  # 长半轴
ee = 0.00669342162296594323  # 扁率

def gcj02towgs84(lng, lat):
    """
    GCJ02(火星坐标系)转GPS84
    :param lng:火星坐标系的经度
    :param lat:火星坐标系纬度
    :return:
    """
    if out_of_china(lng, lat):
        return lng, lat
    dlat = transformlat(lng - 105.0, lat - 35.0)
    dlng = transformlng(lng - 105.0, lat - 35.0)
    radlat = lat / 180.0 * pi
    magic = math.sin(radlat)
    magic = 1 - ee * magic * magic
    sqrtmagic = math.sqrt(magic)
    dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * pi)
    dlng = (dlng * 180.0) / (a / sqrtmagic * math.cos(radlat) * pi)
    mglat = lat + dlat
    mglng = lng + dlng
    return [lng * 2 - mglng, lat * 2 - mglat]

def transformlat(lng, lat):
    ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat + 0.1 * lng * lat + 0.2 * math.sqrt(math.fabs(lng))
    ret += (20.0 * math.sin(6.0 * lng * pi) + 20.0 * math.sin(2.0 * lng * pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lat * pi) + 40.0 * math.sin(lat / 3.0 * pi)) * 2.0 / 3.0
    ret += (160.0 * math.sin(lat / 12.0 * pi) + 320 * math.sin(lat * pi / 30.0)) * 2.0 / 3.0
    return ret

def transformlng(lng, lat):
    ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng + 0.1 * lng * lat + 0.1 * math.sqrt(math.fabs(lng))
    ret += (20.0 * math.sin(6.0 * lng * pi) + 20.0 * math.sin(2.0 * lng * pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lng * pi) + 40.0 * math.sin(lng / 3.0 * pi)) * 2.0 / 3.0
    ret += (150.0 * math.sin(lng / 12.0 * pi) + 300.0 * math.sin(lng / 30.0 * pi)) * 2.0 / 3.0
    return ret

def out_of_china(lng, lat):
    """
    判断是否在国内,不在国内不做偏移
    :param lng:
    :param lat:
    :return:
    """
    if lng < 72.004 or lng > 137.8347:
        return True
    if lat < 0.8293 or lat > 55.8271:
        return True
    return False

def coordinates(c):
    lng, lat = c.split(',')
    lng, lat = float(lng), float(lat)
    wlng, wlat = gcj02towgs84(lng, lat)
    return wlng, wlat

def driving_planning(api_key, routes):
    all_gdfs = []
    all_dfs = []

    for i, (from_location, to_location) in enumerate(routes):
        url = 'https://restapi.amap.com/v3/direction/driving'
        parameters = {
            'key': api_key,
            'origin': from_location,
            'destination': to_location,
            'strategy': '11',
            'output': 'json'
        }

        response = requests.get(url, params=parameters)
        data = json.loads(response.text)

        if data['status'] == '1':
            route = data['route']
            paths = route['paths']
            if paths:
                path = paths[0]
                distance = path['distance']
                duration = path['duration']
                steps = path['steps']
                print(f"驾车路线规划 {i+1}:")
                all_coordinates = []
                step_data = []

                for step in steps:
                    instruction = step['instruction']
                    distance = step['distance']
                    duration = step['duration']
                    polyline = step['polyline']
                    print(f"指令: {instruction}, 距离: {distance}米, 时间: {duration}秒, 路径: {polyline}")

                    # 转换坐标
                    coordinates_list = [coordinates(c) for c in polyline.split(';')]
                    all_coordinates.extend(coordinates_list)

                    # 将数据添加到列表
                    step_data.append({
                        '指令': instruction,
                        '距离': distance,
                        '时间': duration,
                        '路径': polyline
                    })

                # 创建LineString
                geometry = LineString(all_coordinates)

                # 创建GeoDataFrame
                gdf = gpd.GeoDataFrame(index=[i], geometry=[geometry])
                gdf['distance'] = distance
                gdf['duration'] = duration
                gdf['route_id'] = i  # 添加一个标识符来区分不同的路线

                # 保存为SHP文件
                gdf.to_file(f'driving_route_{i}.shp', driver='ESRI Shapefile')

                # 保存为CSV文件
                df = pd.DataFrame(step_data)
                df['route_id'] = i  # 添加一个标识符来区分不同的路线
                df['label'] = f'驾车路线规划{i+1}'  # 添加标签
                df.to_csv(f'driving_route_{i}.csv', index=False, encoding='utf-8-sig')

                all_gdfs.append(gdf)
                all_dfs.append(df)

                print(f"路径 {i+1} 已保存为 driving_route_{i}.shp 和 driving_route_{i}.csv")
            else:
                print(f"未找到路径 {i+1}")
        else:
            print(f"路线规划失败 {i+1}: {data['info']}")

    # 合并所有GeoDataFrame
    combined_gdf = gpd.GeoDataFrame(pd.concat(all_gdfs, ignore_index=True))
    combined_gdf.to_file('combined_driving_routes.shp', driver='ESRI Shapefile')

    # 合并所有DataFrame
    combined_df = pd.concat(all_dfs, ignore_index=True)
    combined_df.to_csv('combined_driving_routes.csv', index=False, encoding='utf-8-sig')

    print("所有路径已合并并保存为 combined_driving_routes.shp 和 combined_driving_routes.csv")

if __name__ == '__main__':
    api_key = '你的key'  # 替换为你的高德地图API密钥

    # 定义多个起终点对
    routes = [
        ('121.596168,31.194734', '121.42837,31.230395'),  # 示例起终点
        ('121.387047,31.108427', '121.506295,31.281965'),  # 示例起终点
        # 可以继续添加更多的起终点对
    ]

    driving_planning(api_key, routes)
    

同样的 ,结果会输出每个路径的单独shp和csv,和合并后的结果为 combined_driving_routes.shp 和 combined_driving_routes.csv,在调用高德路径规划API并获取到驾车路径的详细数据后,我们可以将这些数据在地图上进行可视化,看到驾车的具体路径;

骑行路径规划参考服务示例,我们需要参数就更少了,只需要输入并配置的参数包括key、orign(起点坐标)、desination(终点坐标)、output 选json即可;

骑行路径规划

骑行路径规划用于规划骑行通勤方案,规划时会考虑天桥、单行线、封路等情况。最大支持 500km 的骑行路线规划。

完整代码#运行环境Python 3.11

# -*- coding: utf-8 -*-
import requests
import json
import geopandas as gpd
import pandas as pd
from shapely.geometry import LineString, Point
import math

# 高德GCJ-02坐标转换为WGS-84坐标
x_pi = 3.14159265358979324 * 3000.0 / 180.0
pi = 3.1415926535897932384626  # π
a = 6378245.0  # 长半轴
ee = 0.00669342162296594323  # 扁率

def gcj02towgs84(lng, lat):
    """
    将GCJ-02坐标转换为WGS-84坐标。
    :param lng: 火星坐标系的经度
    :param lat: 火星坐标系的纬度
    :return: WGS-84坐标系的经度和纬度
    """
    if out_of_china(lng, lat):
        return lng, lat
    dlat = transformlat(lng - 105.0, lat - 35.0)
    dlng = transformlng(lng - 105.0, lat - 35.0)
    radlat = lat / 180.0 * pi
    magic = math.sin(radlat)
    magic = 1 - ee * magic * magic
    sqrtmagic = math.sqrt(magic)
    dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * pi)
    dlng = (dlng * 180.0) / (a / sqrtmagic * math.cos(radlat) * pi)
    mglat = lat + dlat
    mglng = lng + dlng
    return [lng * 2 - mglng, lat * 2 - mglat]

def transformlat(lng, lat):
    """
    计算坐标偏移量(纬度方向)。
    :param lng: 经度
    :param lat: 纬度
    :return: 纬度偏移量
    """
    ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat + 0.1 * lng * lat + 0.2 * math.sqrt(math.fabs(lng))
    ret += (20.0 * math.sin(6.0 * lng * pi) + 20.0 * math.sin(2.0 * lng * pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lat * pi) + 40.0 * math.sin(lat / 3.0 * pi)) * 2.0 / 3.0
    ret += (160.0 * math.sin(lat / 12.0 * pi) + 320 * math.sin(lat * pi / 30.0)) * 2.0 / 3.0
    return ret

def transformlng(lng, lat):
    """
    计算坐标偏移量(经度方向)。
    :param lng: 经度
    :param lat: 纬度
    :return: 经度偏移量
    """
    ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng + 0.1 * lng * lat + 0.1 * math.sqrt(math.fabs(lng))
    ret += (20.0 * math.sin(6.0 * lng * pi) + 20.0 * math.sin(2.0 * lng * pi)) * 2.0 / 3.0
    ret += (20.0 * math.sin(lng * pi) + 40.0 * math.sin(lng / 3.0 * pi)) * 2.0 / 3.0
    ret += (150.0 * math.sin(lng / 12.0 * pi) + 300.0 * math.sin(lng / 30.0 * pi)) * 2.0 / 3.0
    return ret

def out_of_china(lng, lat):
    """
    判断坐标是否在中国范围内,不在国内不做偏移。
    :param lng: 经度
    :param lat: 纬度
    :return: 布尔值,是否在中国范围内
    """
    if lng < 72.004 or lng > 137.8347:
        return True
    if lat < 0.8293 or lat > 55.8271:
        return True
    return False

def coordinates(c):
    """
    将字符串形式的坐标转换为浮点数,并进行坐标转换。
    :param c: 坐标字符串,格式为 "经度,纬度"
    :return: 转换后的WGS-84坐标
    """
    lng, lat = c.split(',')
    lng, lat = float(lng), float(lat)
    wlng, wlat = gcj02towgs84(lng, lat)
    return wlng, wlat

def bicycling_planning(api_key, routes):
    """
    实现骑行路径规划,并将结果保存为SHP和CSV文件。
    :param api_key: 高德地图API密钥
    :param routes: 起终点对列表,每个元素是一个元组 (from_location, to_location)
    """
    all_gdfs = []
    all_dfs = []

    for i, (from_location, to_location) in enumerate(routes):
        # 构建请求URL和参数
        url = 'https://restapi.amap.com/v4/direction/bicycling'
        parameters = {
            'key': api_key,
            'origin': from_location,
            'destination': to_location,
            'output': 'json'
        }

        # 发送请求
        response = requests.get(url, params=parameters)
        data = json.loads(response.text)

        if data['errcode'] == 0:
            # 获取路径规划数据
            paths = data['data']['paths']
            if paths:
                path = paths[0]
                distance = path['distance']
                duration = path['duration']
                steps = path['steps']
                print(f"骑行路线规划 {i+1}:")
                all_coordinates = []
                step_data = []

                for step in steps:
                    instruction = step['instruction']
                    road = step['road']
                    distance = step['distance']
                    duration = step['duration']
                    polyline = step['polyline']
                    print(f"指令: {instruction}, 道路: {road}, 距离: {distance}米, 时间: {duration}秒, 路径: {polyline}")

                    # 转换坐标
                    coordinates_list = [coordinates(c) for c in polyline.split(';')]
                    all_coordinates.extend(coordinates_list)

                    # 将数据添加到列表
                    step_data.append({
                        '指令': instruction,
                        '道路': road,
                        '距离': distance,
                        '时间': duration,
                        '路径': polyline
                    })

                # 创建LineString
                geometry = LineString(all_coordinates)

                # 创建GeoDataFrame
                gdf = gpd.GeoDataFrame(index=[i], geometry=[geometry])
                gdf['distance'] = distance
                gdf['duration'] = duration
                gdf['route_id'] = i  # 添加一个标识符来区分不同的路线

                # 保存为SHP文件
                gdf.to_file(f'bicycling_route_{i}.shp', driver='ESRI Shapefile')

                # 保存为CSV文件
                df = pd.DataFrame(step_data)
                df['route_id'] = i  # 添加一个标识符来区分不同的路线
                df['label'] = f'骑行路线规划{i+1}'  # 添加标签
                df.to_csv(f'bicycling_route_{i}.csv', index=False, encoding='utf-8-sig')

                all_gdfs.append(gdf)
                all_dfs.append(df)

                print(f"路径 {i+1} 已保存为 bicycling_route_{i}.shp 和 bicycling_route_{i}.csv")
            else:
                print(f"未找到路径 {i+1}")
        else:
            print(f"路线规划失败 {i+1}: {data['errmsg']}")

    # 合并所有GeoDataFrame
    combined_gdf = gpd.GeoDataFrame(pd.concat(all_gdfs, ignore_index=True))
    combined_gdf.to_file('combined_bicycling_routes.shp', driver='ESRI Shapefile')

    # 合并所有DataFrame
    combined_df = pd.concat(all_dfs, ignore_index=True)
    combined_df.to_csv('combined_bicycling_routes.csv', index=False, encoding='utf-8-sig')

    print("所有路径已合并并保存为 combined_bicycling_routes.shp 和 combined_bicycling_routes.csv")

if __name__ == '__main__':
    api_key = '你的key'  # 替换为你的高德地图API密钥

    # 定义多个起终点对
    routes = [
        ('121.481453,31.154145', '121.534596,31.233022'),  # 示例起终点
        ('121.49956,31.240278', '121.569872,31.328836'),  # 示例起终点
        # 可以继续添加更多的起终点对
    ]

    # 调用骑行路径规划函数
    bicycling_planning(api_key, routes)

同样的 ,结果会输出每个路径的单独shp和csv,和合并后的结果为 combined_bicycling_routes.shp 和 combined_bicycling_routes.csv,在调用高德路径规划API并获取到骑行路径的详细数据后,我们可以将这些数据在地图上进行可视化,看到骑行的具体路径;

文章仅用于分享个人学习成果与个人存档之用,分享知识,如有侵权,请联系作者进行删除。所有信息均基于作者的个人理解和经验,不代表任何官方立场或权威解读。


原文地址:https://blog.csdn.net/weixin_45812624/article/details/143180592

免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!