Python调用OpenRouteService API的速率限制问题及请求错误修复

我编写了一段Python代码，用于通过OpenRouteService（ORS）API处理包含多台车辆停靠点经纬度的CSV文件（格式为"EquipmentID","Latitude","Longitude"），生成路线并导出为ESRI Shapefile。但遇到三个核心问题：

1. 已尝试处理ORS每分钟40次的调用限制，但仍触发速率限制导致失败
2. 改用另一种简单间隔控制请求的方式，依旧触发速率限制
3. 用requests模块直接调用API时，收到500 Server Error

对应的三段代码如下：

初始速率控制尝试 #

import pandas as pd
import openrouteservice as ors
import fiona
from openrouteservice import exceptions
import requests
import time

csv_file = '24hourtrucks.csv'
api_key = 'your key here'

df = pd.read_csv(csv_file)
grouped_df = df.groupby('equipmentID')

def get_directions(group):
    coords = list(zip(group['longitude'], group['latitude']))
    client = ors.Client(key=api_key)
    rate_limit = 40
    counter = 0
    while counter < rate_limit:
        try:
        # Calculate directions using OpenRouteService
            directions = client.directions(
                coordinates=coords,
                profile='driving-car',
                format='geojson',
            )
            break
        except exceptions.ApiError as e:
            # Handle cases where a route is not available
            print(f"Error calculating route: {e}")
            time.sleep(60)
            return None
        counter += 1

directions_df = grouped_df.apply(get_directions)
polylines = [directions['features'][0]['geometry']['coordinates'] for directions in directions_df]
equipment_ids = [str(i) for i in range(len(polylines))]

schema = {
    'geometry': 'LineString',
    'properties': {
        'equipmentID': 'str',
    },
}

with fiona.open('output.shp', 'w', 'ESRI Shapefile', schema) as shp:
    features = [{
        'geometry': {
            'type': 'LineString',
            'coordinates': polyline,
        },
        'properties': {
            'equipmentID': equipment_id,
            'directions': directions_df.iloc[i],  # Add the associated set of directions
        },
    } for i, (polyline, equipment_id) in enumerate(zip(polylines, equipment_ids))]
    shp.writerecords(features)

简单间隔控制尝试 #

def get_directions(group):
    coords = list(zip(group['longitude'], group['latitude']))
    client = ors.Client(key=api_key)
    try:
        # Calculate directions using OpenRouteService
        directions = client.directions(
            coordinates=coords,
            profile='driving-car',
            format='geojson',
        )
        return directions
    except exceptions.ApiError as e:
        # Handle cases where a route is not available
        print(f"Error calculating route: {e}")
        return None, None
    #pause for api limit
    iteration = len(group)
    if iteration % 2 == 0:
        time.sleep(1)

requests调用尝试 #

def get_directions(group):
    coords = list(zip(group['longitude'], group['latitude']))[0]
    url = 'https://api.openrouteservice.org/v2/directions/driving-car/geojson'
    headers = {'Authorization': api_key}
    params = {
        'coordinates': coords,
    }
    try:
        # Send request using requests library and proxy
        response = requests.post(url, headers=headers, params=params, proxies=proxies)
        response.raise_for_status()
        directions = response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error calculating route: {e}")
        return None

1. 正确处理ORS速率限制 #

你的速率控制逻辑存在两个核心问题：

• 每个get_directions函数内的计数器只控制当前请求的重试次数，而非全局每分钟请求数
• 遇到错误时直接return None，没有完成重试逻辑

修正后的速率控制代码 #

# 全局速率控制变量
request_count = 0
last_reset_time = time.time()

client = ors.Client(key=api_key)  # 全局创建client，避免重复初始化

def get_directions(group):
    global request_count, last_reset_time
    coords = list(zip(group['longitude'], group['latitude']))
    
    # 速率控制：确保每分钟不超过40次请求
    current_time = time.time()
    if current_time - last_reset_time > 60:
        request_count = 0
        last_reset_time = current_time
    if request_count >= 40:
        sleep_time = 60 - (current_time - last_reset_time)
        time.sleep(sleep_time)
        request_count = 0
        last_reset_time = time.time()
    
    max_retries = 3
    for retry in range(max_retries):
        try:
            directions = client.directions(
                coordinates=coords,
                profile='driving-car',
                format='geojson',
            )
            request_count += 1
            time.sleep(1.5)  # 额外加1.5秒间隔，确保不会触发限制
            return directions
        except exceptions.ApiError as e:
            print(f"Retry {retry+1} failed for group {group.name}: {e}")
            if "rate limit" in str(e).lower():
                time.sleep(60)
                request_count = 0
                last_reset_time = time.time()
            else:
                time.sleep(2)
    print(f"All retries failed for group {group.name}")
    return None

2. 修复requests调用的500错误 #

500错误的原因是：

• 你只提取了coords的第一个元素（[0]），但ORS的directions API需要至少两个点才能生成路线
• 错误地将coordinates放在params里传递，POST请求的复杂参数应该放在请求体中
• coordinates格式需要是[[lon1, lat1], [lon2, lat2], ...]的嵌套列表

修正后的requests调用代码 #

def get_directions(group):
    coords = list(zip(group['longitude'], group['latitude']))
    if len(coords) < 2:
        print(f"Group {group.name} has fewer than 2 points, skipping")
        return None
    
    url = 'https://api.openrouteservice.org/v2/directions/driving-car/geojson'
    headers = {
        'Authorization': api_key,
        'Content-Type': 'application/json'
    }
    payload = {
        'coordinates': coords
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        directions = response.json()
        return directions
    except requests.exceptions.RequestException as e:
        print(f"Error calculating route for group {group.name}: {e}")
        return None

3. 修复代码中的其他潜在问题 #

• EquipmentID错误：原来的equipment_ids是生成的序号，而非CSV中的原始EquipmentID，应直接从group中获取
• Shapefile属性错误：Shapefile不支持存储directions这种复杂JSON对象，需移除该属性
• 空值处理：需过滤directions_df中的None值，避免后续索引报错

修正后的后续处理代码 #

directions_df = grouped_df.apply(get_directions)
# 过滤空值
valid_directions = directions_df.dropna()

# 获取原始EquipmentID
equipment_ids = valid_directions.index.astype(str)
polylines = [directions['features'][0]['geometry']['coordinates'] for directions in valid_directions]

schema = {
    'geometry': 'LineString',
    'properties': {
        'equipmentID': 'str',
    },
}

with fiona.open('output.shp', 'w', 'ESRI Shapefile', schema) as shp:
    for equipment_id, polyline in zip(equipment_ids, polylines):
        shp.write({
            'geometry': {
                'type': 'LineString',
                'coordinates': polyline,
            },
            'properties': {
                'equipmentID': equipment_id,
            },
        })

内容的提问来源于stack exchange，提问作者Jackson Dunn