Python 脚本批量下载 Sentinel-2 数据：基于 USGS API 实现 100 景自动化

Python 脚本批量下载 Sentinel-2 数据基于 Copernicus Open Access Hub 实现 100 景自动化遥感数据处理的第一步往往是获取高质量的卫星影像。对于需要处理大量 Sentinel-2 数据的研究人员和工程师来说手动下载不仅效率低下还容易出错。本文将介绍如何利用 Python 脚本实现 Sentinel-2 数据的自动化搜索、筛选和批量下载显著提升数据获取效率。1. Sentinel-2 数据获取现状与挑战Sentinel-2 卫星由欧洲航天局ESA发射提供高分辨率多光谱影像广泛应用于植被监测、土地利用分类和环境变化研究等领域。其数据特点包括13 个光谱波段覆盖可见光、近红外和短波红外多种空间分辨率10 米、20 米和 60 米高重访频率全球覆盖周期为 5 天免费开放政策数据对公众完全开放传统手动下载方式面临的主要问题效率低下每次只能下载少量数据筛选困难需要人工检查每景影像的云量、质量等网络不稳定大文件下载容易中断管理复杂大量数据难以有效组织# 典型手动下载流程示例 1. 登录 Copernicus Open Access Hub 2. 绘制研究区域 3. 设置时间范围和云量阈值 4. 逐页浏览搜索结果 5. 逐个点击下载链接 6. 等待下载完成 7. 检查文件完整性2. 自动化下载技术方案设计2.1 核心工具选择我们推荐使用sentinelsatPython 包这是目前最成熟的 Sentinel 数据下载工具之一。其主要优势包括完整 API 覆盖支持所有搜索和下载功能断点续传自动处理网络中断进度显示实时查看下载状态批量操作支持同时处理多个请求安装方法pip install sentinelsat2.2 系统架构设计完整的自动化下载系统应包含以下模块查询模块构建搜索条件获取可用数据列表筛选模块根据质量指标过滤数据下载模块管理下载队列和断点续传日志模块记录操作过程和错误信息通知模块任务完成时发送提醒graph TD A[定义搜索参数] -- B[执行API查询] B -- C[筛选结果] C -- D[加入下载队列] D -- E[监控下载进度] E -- F{完成?} F --|是| G[发送通知] F --|否| E3. 实战构建自动化下载脚本3.1 基础配置首先设置认证信息和基本参数from sentinelsat import SentinelAPI # 连接Copernicus Open Access Hub api SentinelAPI( user您的用户名, password您的密码, api_urlhttps://scihub.copernicus.eu/dhus ) # 定义搜索区域WKT格式 area_of_interest POLYGON(( 116.2 39.8, 116.5 39.8, 116.5 40.0, 116.2 40.0, 116.2 39.8 )) 3.2 高级搜索参数Sentinel-2 数据搜索支持多种精细控制参数类别可选值说明平台Sentinel-2A, Sentinel-2B选择特定卫星产品级别L1C, L2A原始数据或大气校正后数据云量0-100%设置最大允许云覆盖率时间范围YYYYMMDD精确到天的过滤处理基线04.00 等指定数据处理版本# 构建复杂查询条件 products api.query( area_of_interest, date(20230101, 20231231), platformnameSentinel-2, producttypeS2MSI2A, # Level-2A cloudcoverpercentage(0, 30) # 云量0-30% )3.3 结果筛选与排序获取的搜索结果需要进一步处理import pandas as pd # 转换为DataFrame方便处理 products_df api.to_dataframe(products) # 按云量和日期排序 products_df_sorted products_df.sort_values([ cloudcoverpercentage, beginposition ], ascending[True, False]) # 显示关键信息 print(products_df_sorted[[ title, cloudcoverpercentage, size, beginposition ]].head(10))3.4 批量下载实现实现可靠的大批量下载需要考虑以下因素并发控制避免同时发起过多请求断点续传自动恢复中断的下载速度限制遵守服务器使用政策本地校验确保文件完整性from tqdm import tqdm import os # 创建下载目录 download_dir sentinel2_downloads os.makedirs(download_dir, exist_okTrue) # 分批下载函数 def batch_download(product_ids, max_workers4): failed_downloads [] with tqdm(totallen(product_ids)) as pbar: for product_id in product_ids: try: # 检查是否已下载 output_path os.path.join(download_dir, f{product_id}.zip) if os.path.exists(output_path): pbar.update(1) continue # 执行下载 api.download( product_id, directory_pathdownload_dir, checksumTrue # 自动校验文件 ) except Exception as e: print(f下载失败 {product_id}: {str(e)}) failed_downloads.append(product_id) pbar.update(1) return failed_downloads # 执行下载示例取前100景 selected_ids products_df_sorted.index[:100] failed batch_download(selected_ids) if failed: print(f{len(failed)}个文件下载失败将重试...) batch_download(failed)4. 高级功能实现4.1 断点续传机制sentinelsat内置了断点续传功能但我们可以进一步优化def robust_download(product_id, max_retries3): for attempt in range(max_retries): try: api.download( product_id, directory_pathdownload_dir, checksumTrue, n_concurrent_dl2 # 并发连接数 ) return True except Exception as e: print(f尝试 {attempt1} 失败: {str(e)}) time.sleep(60 * (attempt 1)) # 指数退避 return False4.2 自动化通知集成任务完成后发送邮件通知import smtplib from email.mime.text import MIMEText def send_notification(subject, message): msg MIMEText(message) msg[Subject] subject msg[From] senderexample.com msg[To] receiverexample.com with smtplib.SMTP(smtp.example.com, 587) as server: server.starttls() server.login(username, password) server.send_message(msg) # 在下载完成后调用 send_notification( Sentinel-2下载完成, f成功下载{len(selected_ids)-len(failed)}个文件失败{len(failed)}个 )4.3 与云存储集成直接将数据下载到云存储如AWS S3import boto3 def upload_to_s3(local_path, bucket_name): s3 boto3.client(s3) filename os.path.basename(local_path) try: s3.upload_file( local_path, bucket_name, fsentinel2/{filename}, ExtraArgs{Metadata: {source: copernicus}} ) return True except Exception as e: print(f上传失败: {str(e)}) return False # 下载后自动上传 for product_id in selected_ids: local_file os.path.join(download_dir, f{product_id}.zip) if os.path.exists(local_file): upload_to_s3(local_file, my-s3-bucket)5. 性能优化与最佳实践5.1 查询优化技巧缩小时间窗口分批查询比大范围查询更高效精确空间范围避免查询不必要的大区域利用产品类型过滤明确指定 L1C 或 L2A缓存结果对重复查询保存中间结果# 分月查询示例 from datetime import datetime, timedelta def query_by_month(start_date, end_date, aoi): current datetime.strptime(start_date, %Y%m%d) end datetime.strptime(end_date, %Y%m%d) all_products {} while current end: next_month current timedelta(days31) month_end min(next_month.replace(day1) - timedelta(days1), end) print(f查询 {current.date()} 至 {month_end.date()}) products api.query( aoi, date(current.strftime(%Y%m%d), month_end.strftime(%Y%m%d)), platformnameSentinel-2, producttypeS2MSI2A ) all_products.update(products) current next_month return all_products5.2 错误处理策略完善的错误处理应包含API 限流处理捕获 429 错误并等待网络异常自动重试机制磁盘空间检查避免因空间不足失败日志记录详细记录错误上下文import time import logging logging.basicConfig( filenamedownload.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def safe_download(product_id): try: # 检查磁盘空间 stat os.statvfs(download_dir) if stat.f_bavail * stat.f_frsize 2 * 1024**3: # 小于2GB raise Exception(磁盘空间不足) api.download(product_id, directory_pathdownload_dir) logging.info(f成功下载 {product_id}) return True except Exception as e: logging.error(f下载失败 {product_id}: {str(e)}) if 429 in str(e): # Too Many Requests time.sleep(300) # 等待5分钟 return False5.3 资源监控仪表板使用 Prometheus 和 Grafana 监控下载任务from prometheus_client import start_http_server, Gauge # 创建监控指标 download_progress Gauge( sentinel_download_progress, 当前下载进度, [product_id] ) download_speed Gauge( sentinel_download_speed, 下载速度(MB/s) ) # 在下载循环中更新指标 for product_id in product_ids: with tqdm(...) as pbar: # ...下载逻辑... download_progress.labels(product_id).set(pbar.n) download_speed.set(current_speed)6. 扩展应用场景6.1 与数据处理流程集成典型的端到端处理流程自动下载本文介绍的脚本预处理大气校正、辐射校正特征提取NDVI、NDWI 等指数计算分类分析土地利用/覆盖分类变化检测时序变化分析# 示例下载后自动计算NDVI import rasterio import numpy as np def calculate_ndvi(product_path): with rasterio.open(product_path) as src: red src.read(4).astype(float) nir src.read(8).astype(float) ndvi (nir - red) / (nir red) return ndvi # 下载完成后处理 for product_id in completed_downloads: product_path os.path.join(download_dir, f{product_id}.zip) ndvi calculate_ndvi(product_path) np.save(f{product_id}_ndvi.npy, ndvi)6.2 定期自动更新方案设置定时任务保持数据最新Linux cron 作业0 3 * * * /usr/bin/python3 /path/to/sentinel_downloader.py --updatePython 定时器import schedule import time def daily_update(): # 获取昨天到今天的新数据 yesterday (datetime.now() - timedelta(days1)).strftime(%Y%m%d) today datetime.now().strftime(%Y%m%d) products api.query( area_of_interest, date(yesterday, today), platformnameSentinel-2 ) batch_download(products.keys()) # 每天凌晨3点执行 schedule.every().day.at(03:00).do(daily_update) while True: schedule.run_pending() time.sleep(60)7. 替代方案与工具比较当 Copernicus Open Access Hub 不可用时可以考虑以下替代方案数据源特点Python工具AWS Sentinel-2直接S3访问速度快boto3Google Earth Engine云端处理无需下载earthengine-apiMundi Web Services商业服务稳定性高requestsSCIHUB Mirror备用镜像站点sentinelsat# 使用AWS开放数据集的示例 import boto3 from botocore import UNSIGNED from botocore.config import Config s3 boto3.client( s3, configConfig(signature_versionUNSIGNED) ) # 列出可用数据 response s3.list_objects_v2( Bucketsentinel-s2-l2a, Prefixtiles/10/S/DG/2023/1/ ) # 下载特定文件 s3.download_file( sentinel-s2-l2a, tiles/10/S/DG/2023/1/28/0/B01.jp2, B01.jp2 )实际项目中我们通常会结合多种数据源和工具构建健壮的自动化流程。关键是根据具体需求选择最适合的技术方案并充分考虑可维护性和扩展性。