高德地图API v3与Python数据处理5步实现公交站点坐标批量解析与存储公交站点数据是城市交通规划和智能出行应用的基础。本文将介绍如何通过Python调用高德地图API v3版本实现公交站点信息的自动化采集、解析与存储。整个过程分为5个关键步骤最终输出结构化的CSV文件为后续的空间分析和可视化提供数据支持。1. 环境准备与API配置在开始之前我们需要准备好开发环境并获取必要的高德地图API访问权限。以下是具体操作开发环境要求Python 3.7或更高版本requests库用于HTTP请求pandas库用于数据处理json库用于解析API响应安装依赖库的命令pip install requests pandas获取高德地图API Key访问 高德开放平台 并注册开发者账号进入控制台创建新应用为应用添加Web服务类型的Key记录下生成的Key字符串后续请求中需要用到提示高德地图API对免费用户有每日调用限额商业项目建议购买更高配额的服务套餐。配置Python环境变量import os # 替换为你的实际API Key AMAP_API_KEY 你的高德API Key CITY_NAME 北京 # 目标城市 BUS_LINE 1路 # 目标公交线路2. 公交线路数据获取高德地图提供了专门的公交线路查询接口我们可以通过构造特定URL获取线路的详细信息。API接口分析 高德地图v3版本的公交线路查询接口为https://restapi.amap.com/v3/bus/linename关键请求参数key: 开发者Keycity: 城市名称keywords: 公交线路名称extensions: 设置为all获取完整信息output: 返回格式通常使用jsonPython实现代码import requests import json def fetch_bus_line_info(city, line, api_key): url fhttps://restapi.amap.com/v3/bus/linename?key{api_key}city{city}keywords{line}extensionsalloutputjson try: response requests.get(url) response.raise_for_status() # 检查请求是否成功 return response.json() except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None # 示例调用 bus_data fetch_bus_line_info(CITY_NAME, BUS_LINE, AMAP_API_KEY) print(json.dumps(bus_data, indent2, ensure_asciiFalse))返回数据结构解析 API返回的JSON数据包含以下关键信息buslines: 公交线路数组包含线路基本信息name: 线路名称start_stop: 始发站end_stop: 终点站busstops: 途经站点数组name: 站点名称location: 经纬度坐标格式经度,纬度polyline: 行驶轨迹坐标串3. 数据结构解析与清洗获取原始数据后我们需要从中提取有用的信息并进行结构化处理。站点数据提取def extract_bus_stops(bus_data): if not bus_data or buslines not in bus_data or not bus_data[buslines]: return None busline bus_data[buslines][0] stops [] for stop in busline.get(busstops, []): stop_info { name: stop.get(name, ), location: stop.get(location, ) } stops.append(stop_info) return { line_name: busline.get(name, ), start_stop: busline.get(start_stop, ), end_stop: busline.get(end_stop, ), stops: stops, polyline: busline.get(polyline, ) } # 示例调用 parsed_data extract_bus_stops(bus_data)坐标拆分处理 原始数据中的坐标是经度,纬度格式的字符串我们需要将其拆分为单独的经度和纬度字段。import pandas as pd def process_coordinates(data): if not data or stops not in data: return None # 创建DataFrame df pd.DataFrame(data[stops]) # 拆分坐标 df[[longitude, latitude]] df[location].str.split(,, expandTrue) df[longitude] pd.to_numeric(df[longitude]) df[latitude] pd.to_numeric(df[latitude]) # 添加线路信息 df[line_name] data[line_name] df[start_stop] data[start_stop] df[end_stop] data[end_stop] return df # 示例调用 processed_df process_coordinates(parsed_data) print(processed_df.head())4. 数据存储与导出处理后的数据可以保存为多种格式这里我们选择CSV格式因为它兼容性好且易于后续处理。CSV文件导出def save_to_csv(df, city, line): if df is None or df.empty: return False filename f{city}_{line}_公交站点信息.csv try: df.to_csv(filename, indexFalse, encodingutf-8-sig) print(f数据已保存到 {filename}) return True except Exception as e: print(f保存文件时出错: {e}) return False # 示例调用 save_to_csv(processed_df, CITY_NAME, BUS_LINE)数据质量检查 在导出前我们应该检查数据的完整性和准确性检查是否有缺失值验证坐标范围是否合理确保站点顺序正确def validate_data(df): if df is None: return False # 检查缺失值 if df.isnull().values.any(): print(警告数据中存在缺失值) # 检查坐标范围以北京为例 if not ((df[longitude] 115.7).all() and (df[longitude] 117.4).all() and (df[latitude] 39.4).all() and (df[latitude] 41.6).all()): print(警告部分坐标超出北京大致范围) # 检查站点数量 if len(df) 2: print(警告站点数量过少) return True # 示例调用 validate_data(processed_df)5. 批量处理与自动化实际应用中我们通常需要处理多条公交线路。下面介绍如何批量处理多条线路的数据。批量处理实现def batch_process_bus_lines(city, lines, api_key): all_data [] for line in lines: print(f正在处理线路: {line}) # 获取数据 raw_data fetch_bus_line_info(city, line, api_key) if not raw_data: continue # 解析数据 parsed_data extract_bus_stops(raw_data) if not parsed_data: continue # 处理数据 processed_df process_coordinates(parsed_data) if processed_df is not None: all_data.append(processed_df) # 避免频繁调用API time.sleep(0.5) # 合并所有线路数据 if all_data: final_df pd.concat(all_data, ignore_indexTrue) return final_df return None # 示例调用多条线路 bus_lines [1路, 2路, 52路] # 示例线路列表 batch_df batch_process_bus_lines(CITY_NAME, bus_lines, AMAP_API_KEY) if batch_df is not None: save_to_csv(batch_df, CITY_NAME, 多条线路)自动化脚本优化 为了提高效率和稳定性我们可以添加以下功能错误重试机制进度显示结果日志记录import time from datetime import datetime def enhanced_batch_process(city, lines, api_key, max_retries3): log_entries [] all_data [] for i, line in enumerate(lines, 1): retries 0 success False while retries max_retries and not success: try: print(f\n处理进度: {i}/{len(lines)} - {line}) start_time time.time() # 获取数据 raw_data fetch_bus_line_info(city, line, api_key) if not raw_data: raise ValueError(API返回数据为空) # 解析和处理数据 parsed_data extract_bus_stops(raw_data) processed_df process_coordinates(parsed_data) if processed_df is not None: all_data.append(processed_df) elapsed time.time() - start_time log_msg f{datetime.now()} - 成功: {line} (耗时: {elapsed:.2f}s) print(log_msg) log_entries.append(log_msg) success True else: raise ValueError(数据处理失败) except Exception as e: retries 1 err_msg f{datetime.now()} - 尝试 {retries}/{max_retries}: {line} 失败 - {str(e)} print(err_msg) log_entries.append(err_msg) if retries max_retries: time.sleep(2 ** retries) # 指数退避 # 保存日志 with open(processing_log.txt, w, encodingutf-8) as f: f.write(\n.join(log_entries)) # 合并数据 if all_data: return pd.concat(all_data, ignore_indexTrue) return None # 示例调用 lines_to_process [1路, 2路, 52路, 300路, 特8路] # 扩展的线路列表 final_data enhanced_batch_process(CITY_NAME, lines_to_process, AMAP_API_KEY) if final_data is not None: save_to_csv(final_data, CITY_NAME, 批量公交线路) print(批量处理完成数据已保存) else: print(批量处理未获得有效数据)高级应用与扩展掌握了基础的数据采集和处理方法后我们可以进一步扩展应用场景。数据可视化 使用Python的folium库可以轻松实现公交线路在地图上的可视化。import folium def visualize_bus_line(df, line_name): if df.empty: return # 创建地图以第一个站点为中心 first_stop df.iloc[0] m folium.Map(location[first_stop[latitude], first_stop[longitude]], zoom_start13) # 添加站点标记 for _, stop in df.iterrows(): folium.Marker( [stop[latitude], stop[longitude]], popupf{stop[name]}br经度: {stop[longitude]}br纬度: {stop[latitude]} ).add_to(m) # 添加线路连接线 line_coords df[[latitude, longitude]].values.tolist() folium.PolyLine(line_coords, colorblue, weight2.5, opacity1).add_to(m) # 保存地图 map_file f{line_name.replace( , _)}_地图.html m.save(map_file) print(f地图已保存为 {map_file}) # 示例调用需先筛选特定线路 if final_data is not None: line_1_data final_data[final_data[line_name] 1路] visualize_bus_line(line_1_data, 1路)性能优化技巧使用多线程/协程并发请求注意API限流缓存已获取的数据避免重复请求使用更高效的数据结构处理大规模数据import concurrent.futures from functools import partial def concurrent_fetch(city, lines, api_key, max_workers5): with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 使用偏函数固定部分参数 fetch_func partial(fetch_bus_line_info, city, api_keyapi_key) results list(executor.map(fetch_func, lines)) valid_results [r for r in results if r is not None] print(f成功获取 {len(valid_results)}/{len(lines)} 条线路数据) return valid_results # 示例调用 lines [1路, 2路, 52路, 300路, 特8路, 4路, 5路, 10路] all_raw_data concurrent_fetch(CITY_NAME, lines, AMAP_API_KEY)数据质量监控 建立自动化监控机制确保数据的准确性和及时更新。def monitor_data_quality(df): metrics { total_stops: len(df), missing_values: df.isnull().sum().to_dict(), coordinate_ranges: { min_longitude: df[longitude].min(), max_longitude: df[longitude].max(), min_latitude: df[latitude].min(), max_latitude: df[latitude].max() }, unique_lines: df[line_name].nunique() } # 保存质量报告 report_file data_quality_report.json with open(report_file, w, encodingutf-8) as f: json.dump(metrics, f, indent2, ensure_asciiFalse) print(f数据质量报告已保存到 {report_file}) return metrics # 示例调用 if final_data is not None: quality_metrics monitor_data_quality(final_data)通过以上5个步骤我们建立了一个完整的公交站点数据采集、处理和分析流程。这套方法不仅适用于高德地图API经过适当调整也可以应用于其他地图服务平台的数据处理工作。
