本文旨在指导读者如何使用Python高效地通过外部API计算地理位置间的驾驶距离。内容涵盖了从多源列表数据中提取信息、构建API请求、集成OSRM路由服务进行距离计算的核心方法，并重点介绍了如何利用Python的contextlib.contextmanager实现健壮的API请求速率限制机制，以避免超出API调用频率限制。最终，文章将展示如何将原始数据与计算结果整合至Pandas DataFrame，形成结构化的输出。
1. 引言：地理距离计算与API交互挑战
在地理信息处理和位置服务领域，计算两点间的驾驶距离是常见的需求。通常，我们会拥有多组起点和终点的经纬度数据，需要通过地图服务api进行批量查询。然而，直接进行大量api调用可能会遇到服务提供商的速率限制，导致请求失败甚至ip被封禁。因此，在进行批量api请求时，实现有效的速率限制和错误处理机制至关重要。
2. 核心功能：基于OSRM API的距离计算
本教程将以开源路由服务OSRM (Open Source Routing Machine) 为例，展示如何通过其API获取驾驶距离。OSRM提供了一个简洁的HTTP接口，允许用户查询两点间的路线信息，包括距离和持续时间。
首先，定义一个基础函数来执行API请求并解析结果：
import requestsimport jsonimport pandas as pdfrom time import sleepfrom contextlib import contextmanagerdef get_driving_distance(lat1, lon1, lat2, lon2): """ 通过OSRM API计算两点间的驾驶距离。 Args: lat1 (float): 起点纬度 lon1 (float): 起点经度 lat2 (float): 终点纬度 lon2 (float): 终点经度 Returns: float: 驾驶距离（英里），如果API请求失败或无有效路线则返回None。 """ # OSRM API的URL格式为 /route/v1/car/{lon1},{lat1};{lon2},{lat2} # 注意：OSRM API的坐标顺序是经度在前，纬度在后。 url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" try: r = requests.get(url) # 检查HTTP状态码，如果不是2xx，则抛出HTTPError异常 r.raise_for_status() routes = json.loads(r.content) # 检查API响应中是否存在'routes'键以及其是否为空 if routes and "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info.get('distance') if driving_distance_meters is not None: # 将米转换为英里 (1 英里 ≈ 1609.34 米) return driving_distance_meters / 1609.34 else: print(f"API响应未包含有效路线信息: {url}, 响应: {routes}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误: {e.response.status_code} - {e.response.text} for URL: {url}") return None except requests.exceptions.ConnectionError as e: print(f"连接错误: {e} for URL: {url}") return None except requests.exceptions.Timeout as e: print(f"请求超时: {e} for URL: {url}") return None except json.JSONDecodeError as e: print(f"JSON解析错误: {e} for URL: {url}, 响应内容: {r.content}") return None except Exception as e: print(f"发生未知错误: {e} for URL: {url}") return None
在上述代码中，我们增加了try-except块来捕获可能发生的网络请求异常（如HTTPError、ConnectionError、Timeout）以及JSON解析错误，这使得函数更加健壮。r.raise_for_status()是一个非常实用的方法，它会在HTTP请求返回错误状态码（如4xx或5xx）时自动抛出异常。
3. 高效处理多源地理数据
通常，我们的地理位置数据会以列表的形式存储，例如：
立即学习“Python免费学习笔记（深入）”；
location_latitudes = [34.0522, 34.0522]location_longitudes = [-118.2437, -118.2437]station_latitudes = [34.0689, 34.0753]station_longitudes = [-118.2942, -118.2575]
为了将这些列表中的对应元素传递给get_driving_distance函数，可以使用Python内置的zip函数来同时迭代多个列表：
# 示例：迭代并调用函数（暂不考虑速率限制）# for lat1, lon1, lat2, lon2 in zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes):# distance = get_driving_distance(lat1, lon1, lat2, lon2)# print(f"距离: {distance} 英里")
4. API速率限制与鲁棒性实践
当需要进行大量API调用时，必须考虑API提供商的速率限制。不加限制的频繁请求可能导致服务拒绝或IP被暂时封禁。Python的contextlib.contextmanager提供了一种优雅的方式来封装资源管理逻辑，非常适合实现API速率限制。
我们将创建一个上下文管理器，它会在每次API调用前检查计数器，如果达到预设限制，则暂停一段时间。
api_calls_counter = 0 # 全局计数器，用于跟踪API调用次数@contextmanagerdef rate_limited(limit=500, delay=5): """ 一个上下文管理器，用于实现API调用速率限制。 当API调用次数达到指定限制时，暂停指定时间。 Args: limit (int): 达到此次数后暂停。 delay (int): 暂停的秒数。 """ global api_calls_counter # 在进入上下文之前检查并处理速率限制 if api_calls_counter + 1 >= limit: print(f"达到 {limit} 次API调用限制，暂停 {delay} 秒...") sleep(delay) # 暂停后，将计数器重置，以允许新的批次调用 api_calls_counter = 0 # 或者 api_calls_counter -= limit，取决于具体策略 print("暂停结束，继续调用。") # 增加API调用计数 api_calls_counter += 1 try: yield # 执行被包装的代码块 finally: # 可以在这里执行一些清理工作，但对于速率限制不是必需的 pass
rate_limited 上下文管理器详解：
api_calls_counter: 一个全局变量，用于记录从上次重置以来已经进行了多少次API调用。limit 和 delay: 上下文管理器的参数，分别定义了触发暂停的调用次数阈值和暂停的秒数。if api_calls_counter + 1 >= limit:: 在每次进入上下文（即每次API调用前）检查，如果当前调用将使总数达到或超过限制，则触发暂停。sleep(delay): 执行实际的暂停操作。api_calls_counter = 0: 暂停结束后，将计数器重置为0。这意味着每500次调用后，会暂停5秒，然后重新开始计数。这是一种简单的固定窗口计数器实现。api_calls_counter += 1: 无论是否暂停，每次进入上下文，调用计数器都会增加。yield: 这是上下文管理器的核心。它将控制权交给with语句块内的代码。当with块执行完毕或抛出异常时，控制权会回到finally块。
接下来，我们将这个速率限制逻辑集成到get_driving_distance函数中：
# 修改后的 get_driving_distance 函数，集成速率限制def get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2): """ 通过OSRM API计算两点间的驾驶距离，并集成速率限制。 """ with rate_limited(limit=5, delay=1): # 示例：为演示目的将限制设低，实际使用请根据API规则调整 return get_driving_distance(lat1, lon1, lat2, lon2)
注意事项： 在实际应用中，limit 和 delay 参数应根据所使用的API服务提供商的具体速率限制策略进行调整。OSRM的公共实例通常有非常宽松的限制，但商业API服务通常会更严格。
5. 数据整合与结果输出：构建Pandas DataFrame
最终目标是将原始的经纬度数据与计算出的驾驶距离整合到一个Pandas DataFrame中，便于后续分析。
# 示例数据location_latitudes = [34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522, 34.0522] # 60个点location_longitudes = [-118.2437] * 60station_latitudes = [34.0689 + i*0.001 for i in range(60)]station_longitudes = [-118.2942 + i*0.001 for i in range(60)]# 用于存储结果的列表distances = []# 迭代所有点对并计算距离for i, (lat1, lon1, lat2, lon2) in enumerate(zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes)): print(f"正在处理第 {i+1} 对数据: ({lat1},{lon1}) -> ({lat2},{lon2})") distance = get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2) distances.append(distance)# 创建DataFramedata = { 'Location_Latitude': location_latitudes, 'Location_Longitude': location_longitudes, 'Station_Latitude': station_latitudes, 'Station_Longitude': station_longitudes, 'Driving_Distance_Miles': distances}df = pd.DataFrame(data)print("n生成的DataFrame:")print(df.head())print(f"nDataFrame包含 {len(df)} 条记录。")
6. 完整示例与注意事项
以下是包含所有组件的完整代码示例，便于读者直接运行和理解：
import requestsimport jsonimport pandas as pdfrom time import sleepfrom contextlib import contextmanager# 1. 全局API调用计数器api_calls_counter = 0# 2. 速率限制上下文管理器@contextmanagerdef rate_limited(limit=500, delay=5): global api_calls_counter if api_calls_counter + 1 >= limit: print(f"达到 {limit} 次API调用限制，暂停 {delay} 秒...") sleep(delay) api_calls_counter = 0 # 重置计数器 print("暂停结束，继续调用。") api_calls_counter += 1 yield# 3. 核心API请求函数def get_driving_distance(lat1, lon1, lat2, lon2): url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" try: r = requests.get(url) r.raise_for_status() # 检查HTTP状态码 routes = json.loads(r.content) if routes and "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info.get('distance') if driving_distance_meters is not None: return driving_distance_meters / 1609.34 else: print(f"API响应未包含有效路线信息: {url}, 响应: {routes}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误: {e.response.status_code} - {e.response.text} for URL: {url}") return None except requests.exceptions.ConnectionError as e: print(f"连接错误: {e} for URL: {url}") return None except requests.exceptions.Timeout as e: print(f"请求超时: {e} for URL: {url}") return None except json.JSONDecodeError as e: print(f"JSON解析错误: {e} for URL: {url}, 响应内容: {r.content}") return None except Exception as e: print(f"发生未知错误: {e} for URL: {url}") return None# 4. 集成速率限制的API请求函数def get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2): # 注意：这里的limit和delay可以根据实际API限制和测试需求调整 # 为演示目的，我们设置一个较低的限制（例如，每5次调用暂停1秒） with rate_limited(limit=5, delay=1): return get_driving_distance(lat1, lon1, lat2, lon2)# 5. 示例数据（增加数据量以触发速率限制）location_latitudes = [34.0522] * 20location_longitudes = [-118.2437] * 20station_latitudes = [34.0689 + i * 0.001 for i in range(20)]station_longitudes = [-118.2942 + i * 0.001 for i in range(20)]# 6. 计算并整合结果distances = []for i, (lat1, lon1, lat2, lon2) in enumerate(zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes)): print(f"处理第 {i+1} 对数据: ({lat1},{lon1}) -> ({lat2},{lon2})") distance = get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2) distances.append(distance)# 7. 构建DataFramedata = { 'Location_Latitude': location_latitudes, 'Location_Longitude': location_longitudes, 'Station_Latitude': station_latitudes, 'Station_Longitude': station_longitudes, 'Driving_Distance_Miles': distances}df = pd.DataFrame(data)print("n--- 结果DataFrame ---")print(df.head(10)) # 打印前10行print(f"nDataFrame包含 {len(df)} 条记录。")
重要注意事项：
API密钥： 某些商业API服务（如Google Maps API）需要API密钥进行认证。如果使用此类服务，请确保将密钥安全地包含在请求中（通常通过URL参数或HTTP头）。OSRM公共实例通常不需要密钥。错误处理： 尽管我们已经添加了基本的错误处理，但生产级的应用程序可能需要更复杂的重试逻辑（例如，指数退避）来处理瞬时网络问题或API服务暂时性错误。数据验证： 在将经纬度传递给API之前，最好进行数据验证，确保它们是有效的浮点数且在合理范围内。并发： 对于非常大的数据集，可以考虑使用多线程或异步IO（如asyncio配合aiohttp）来并行化API请求，但这会使速率限制逻辑变得更复杂，需要更精细的控制。OSRM服务： 本教程使用的是公共OSRM路由服务。对于生产环境或高并发需求，建议搭建自己的OSRM服务器或使用商业地理服务提供商。
7. 总结
本教程详细介绍了如何使用Python处理多源地理位置数据，通过OSRM API计算驾驶距离，并重点演示了如何利用contextlib.contextmanager实现优雅且健壮的API请求速率限制。通过将原始数据、计算逻辑和结果整合到Pandas DataFrame中，我们能够高效地管理和分析地理空间数据。掌握这些技术将有助于您在处理大规模地理数据时，构建更稳定、
以上就是使用Python通过API计算地理距离：数据整合与API速率限制实践的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1366974.html
微信扫一扫 
支付宝扫一扫 
