当前位置： 首页 > news >正文

告别手动抓取！用Python脚本5分钟批量下载Mapillary指定区域的街景图片

news 2026/6/6 17:16:14

5分钟极速获取Mapillary街景：Python自动化采集实战指南

当我们需要对特定区域进行空间分析或计算机视觉研究时，获取高质量的街景图像是第一步。传统手动下载方式不仅效率低下，还难以保证数据的一致性和完整性。本文将带你用Python打造一个全自动化的街景采集工具，只需定义目标区域坐标，5分钟内即可完成数百张Mapillary街景图像的批量下载。

1. 环境准备与核心工具链

在开始编写自动化脚本前，我们需要配置合适的开发环境。推荐使用Python 3.8+版本，这是目前最稳定的Python发行版之一。以下是必需的Python库及其作用：

pip install mercantile requests vt2geojson
  • mercantile：处理地图瓦片坐标转换的核心库，支持WGS84坐标与瓦片行列号的相互转换
  • requests：发送HTTP请求获取Mapillary API数据
  • vt2geojson：将Mapillary返回的矢量切片数据转换为易处理的GeoJSON格式

提示：建议使用虚拟环境隔离项目依赖，避免与其他Python项目产生冲突。可以使用python -m venv mapillary_env创建专属环境。

Mapillary的API访问需要身份验证令牌。登录Mapillary开发者平台后，可以在"我的应用"部分生成专属的access token。这个令牌将作为所有API请求的通行证，务必妥善保管。

2. 区域定义与瓦片计算原理

Mapillary的街景数据采用瓦片地图体系组织，理解这套体系是自动化采集的基础。我们以14级瓦片为例（这是Mapillary官方规定的街景数据层级），演示如何将地理坐标转换为瓦片索引。

2.1 定义目标区域边界

首先需要确定目标区域的边界坐标。推荐使用QGIS或Google Earth等工具获取精确的WGS84坐标。坐标格式为(西经,南纬,东经,北纬)：

# 示例：迈阿密某区域坐标 west, south, east, north = -80.134234, 25.773769, -80.126423, 25.788608

2.2 计算相交瓦片

使用mercantile库计算与目标区域相交的所有14级瓦片：

import mercantile tiles = list(mercantile.tiles(west, south, east, north, 14)) print(f"共找到{len(tiles)}个相交瓦片")

每个瓦片对象包含三个关键属性：

  • z：瓦片层级（固定为14）
  • x：瓦片列号
  • y：瓦片行号

这些参数将用于构建Mapillary的API请求URL。值得注意的是，由于地图投影和瓦片划分的特性，一个地理矩形区域可能对应多个瓦片。

3. 构建自动化请求管道

获取瓦片信息后，我们需要设计高效的数据请求和处理流程。这个阶段需要考虑API限流、错误处理和数据解析等多个方面。

3.1 构造请求URL模板

Mapillary的矢量切片API遵循固定格式：

https://tiles.mapillary.com/maps/vtp/mly1_public/2/{z}/{x}/{y}?access_token={token}

我们可以使用Python的f-string动态生成URL：

base_url = "https://tiles.mapillary.com/maps/vtp/mly1_public/2/{z}/{x}/{y}" access_token = "YOUR_ACCESS_TOKEN" # 替换为实际token for tile in tiles: tile_url = f"{base_url}?access_token={access_token}".format( z=tile.z, x=tile.x, y=tile.y )

3.2 处理API响应

Mapillary返回的是矢量切片数据，需要使用vt2geojson转换为GeoJSON格式：

import requests from vt2geojson.tools import vt_bytes_to_geojson response = requests.get(tile_url) if response.status_code == 200: geojson_data = vt_bytes_to_geojson( response.content, tile.x, tile.y, tile.z, layer="image" # 指定获取图像层数据 )

GeoJSON数据中包含每个街景点的完整元信息，包括：

  • 几何坐标（经度、纬度）
  • 图像ID
  • 拍摄时间
  • 朝向角度等属性

4. 高级过滤与批量下载

获取原始数据后，我们需要进行精确筛选和高效下载，这是保证数据质量的关键步骤。

4.1 精确空间过滤

虽然我们已经通过瓦片进行了初步区域筛选，但为了确保每张图像都确实位于目标区域内，还需要进行二次验证：

def is_in_bbox(lng, lat, bbox): west, south, east, north = bbox return west <= lng <= east and south <= lat <= north valid_images = [] for feature in geojson_data['features']: lng, lat = feature['geometry']['coordinates'] if is_in_bbox(lng, lat, (west, south, east, north)): valid_images.append(feature)

4.2 并行下载优化

当需要下载数百张图像时，串行请求效率极低。我们可以使用concurrent.futures实现多线程下载：

from concurrent.futures import ThreadPoolExecutor import os def download_image(image_id, access_token, save_dir="images"): os.makedirs(save_dir, exist_ok=True) headers = {'Authorization': f'OAuth {access_token}'} url = f'https://graph.mapillary.com/{image_id}?fields=thumb_original_url' try: response = requests.get(url, headers=headers) image_url = response.json()['thumb_original_url'] image_data = requests.get(image_url).content with open(f"{save_dir}/{image_id}.jpg", 'wb') as f: f.write(image_data) return True except Exception as e: print(f"下载{image_id}失败: {str(e)}") return False with ThreadPoolExecutor(max_workers=8) as executor: futures = [ executor.submit( download_image, feature['properties']['id'], access_token ) for feature in valid_images ] results = [f.result() for f in futures]

注意：Mapillary API有请求频率限制，建议合理设置max_workers数量（通常4-8个线程为宜），避免触发限流。

5. 错误处理与日志记录

在实际运行中，网络波动、API限制等问题不可避免。健壮的脚本需要完善的错误处理机制。

5.1 重试机制

对于失败的请求，可以实现自动重试逻辑：

from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) def safe_request(url, headers=None): response = requests.get(url, headers=headers) response.raise_for_status() return response

5.2 操作日志

记录脚本运行情况有助于后期排查问题：

import logging logging.basicConfig( filename='mapillary_downloader.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def download_image(image_id, access_token): try: # ...下载逻辑... logging.info(f"成功下载图像 {image_id}") except Exception as e: logging.error(f"下载 {image_id} 失败: {str(e)}")

6. 完整脚本封装与使用

将上述各模块整合为一个完整的Python类，提高代码复用性：

import json import logging import os from concurrent.futures import ThreadPoolExecutor import mercantile import requests from vt2geojson.tools import vt_bytes_to_geojson class MapillaryDownloader: def __init__(self, access_token, max_workers=4): self.access_token = access_token self.max_workers = max_workers self.setup_logging() def setup_logging(self): logging.basicConfig( filename='mapillary_downloader.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def get_tiles(self, bbox, zoom=14): west, south, east, north = bbox return list(mercantile.tiles(west, south, east, north, zoom)) def fetch_geojson_data(self, tile): url = f"https://tiles.mapillary.com/maps/vtp/mly1_public/2/{tile.z}/{tile.x}/{tile.y}" response = requests.get(f"{url}?access_token={self.access_token}") return vt_bytes_to_geojson(response.content, tile.x, tile.y, tile.z, layer="image") def filter_images_in_bbox(self, features, bbox): west, south, east, north = bbox return [ f for f in features if (west <= f['geometry']['coordinates'][0] <= east and south <= f['geometry']['coordinates'][1] <= north) ] def download_image(self, image_id, save_dir="images"): os.makedirs(save_dir, exist_ok=True) try: headers = {'Authorization': f'OAuth {self.access_token}'} metadata_url = f'https://graph.mapillary.com/{image_id}?fields=thumb_original_url' metadata = requests.get(metadata_url, headers=headers).json() image_url = metadata['thumb_original_url'] image_data = requests.get(image_url).content with open(f"{save_dir}/{image_id}.jpg", 'wb') as f: f.write(image_data) logging.info(f"下载成功: {image_id}") return True except Exception as e: logging.error(f"下载失败 {image_id}: {str(e)}") return False def download_area(self, bbox, save_dir="images", zoom=14): tiles = self.get_tiles(bbox, zoom) all_features = [] for tile in tiles: try: geojson = self.fetch_geojson_data(tile) all_features.extend(geojson['features']) except Exception as e: logging.warning(f"获取瓦片 {tile} 数据失败: {str(e)}") valid_images = self.filter_images_in_bbox(all_features, bbox) logging.info(f"找到 {len(valid_images)} 张有效图像") with ThreadPoolExecutor(max_workers=self.max_workers) as executor: results = list(executor.map( lambda f: self.download_image(f['properties']['id'], save_dir), valid_images )) success_count = sum(results) logging.info(f"下载完成: 成功 {success_count}/{len(valid_images)}") return success_count

使用示例：

if __name__ == "__main__": # 配置参数 ACCESS_TOKEN = "YOUR_ACCESS_TOKEN" BBOX = [-80.134234, 25.773769, -80.126423, 25.788608] # 目标区域边界 SAVE_DIR = "miami_streetview" # 保存目录 # 创建下载器实例 downloader = MapillaryDownloader(ACCESS_TOKEN, max_workers=6) # 开始下载 downloader.download_area(BBOX, SAVE_DIR)

在实际项目中，这个脚本帮助我高效采集了多个城市的街景数据集。一个常见的优化是添加进度条显示，可以使用tqdm库增强用户体验。对于特别大的区域，可以考虑分块处理，避免一次性请求过多数据导致内存问题。

http://www.jsqmd.com/news/563835/

相关文章：

  • 别让临时存储拖垮集群！K8s中emptyDir的正确使用姿势与替代方案
  • 07 从 MLP 到 LeNet：感知机到底解决了什么问题？
  • IEEE会议论文避雷指南：如何用GSview+Photoshop搞定EPS图片压缩与特殊字符命名
  • 超级千问语音设计世界实战：一句话轻松变出英雄、魔王四种声音
  • 避坑指南：ESP32+MicroPython混合编程时C库编译的3个常见错误
  • 大恒相机硬触发实战：从IO配置到回调函数处理的完整流程（附避坑指南）
  • Python自动化操作Synology群晖文件：从下载到上传的完整实践
  • 别再让串口打印卡死你的STM32了！用FreeRTOS队列实现异步日志（附完整代码）
  • 快速排序图解：5分钟搞懂分治法的核心思想（含动态演示）
  • ZYNQ UART中断的四种工作模式详解：除了回环，还能怎么玩？
  • 2026年超低压钢带管优质品牌推荐榜：防腐钢带管、高压钢带管、SFB钢带管、SF钢带管、WF屋顶钢带管、低噪声钢带管选择指南 - 优质品牌商家
  • Linux 内核中的网络协议栈：从数据包到应用程序
  • 2026除甲醛果壳活性炭优质生产厂家推荐指南：除甲醛活性炭、除甲醛粉末活性炭、除甲醛粉状活性炭、净水木质活性炭选择指南 - 优质品牌商家
  • 第六章、Isaacsim中的USD资产：从零开始构建自定义机器人模型
  • DASD-4B-Thinking在Ubuntu系统管理中的智能助手应用
  • 收藏！一张图带你入门AIAgent全流程：从提问到结果返回的17步详解（小白程序员必备）
  • 简单几步，让通义千问3-4B-Instruct-2507支持外部设备访问
  • Qwen3-VL-8B效果惊艳展示：识别电路图并解释工作原理与元器件作用
  • 组态王与施耐德M580 PLC的Modbus TCP通信实战指南
  • 2026年比较好的舒适独立弹簧床垫/弹簧床垫源头工厂推荐 - 品牌宣传支持者
  • 2026年热门的全国MABR污水处理设备选型服务商/全国MABR污水处理运维解决方案提供商靠谱公司推荐 - 品牌宣传支持者
  • 2026医药食品GMP超细粉碎设备评测报告：实验室气流磨/实验室气流粉碎机/小型气流磨/小型气流粉碎机/新型气流磨/选择指南 - 优质品牌商家
  • 从Shiro到Spring Security：在若依(RuoYi)不同版本中，免登录访问配置的‘踩坑’与‘填坑’指南
  • LLM+运筹优化：工业级多机器人协同控制软件生成新范式
  • Linux文件系统介绍
  • 告别UnsatisfiedLinkError！OpenCV Java版环境配置的终极避坑指南（含Maven/Gradle依赖）
  • Sambert语音合成镜像快速入门：环境配置、模型加载、语音生成三步走
  • Verilog实战：从零搭建D锁存器与D触发器的5个关键步骤（附代码）
  • 【NoC片上网络 On-Chip Network】从总线到NoC：多核芯片通信架构的演进与设计权衡
  • SVN 启动模式详解