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. 效率低下：每次只能下载少量数据
2. 筛选困难：需要人工检查每景影像的云量、质量等
3. 网络不稳定：大文件下载容易中断
4. 管理复杂：大量数据难以有效组织

# 典型手动下载流程示例 1. 登录 Copernicus Open Access Hub 2. 绘制研究区域 3. 设置时间范围和云量阈值 4. 逐页浏览搜索结果 5. 逐个点击下载链接 6. 等待下载完成 7. 检查文件完整性

2. 自动化下载技术方案设计

2.1 核心工具选择

我们推荐使用sentinelsatPython 包，这是目前最成熟的 Sentinel 数据下载工具之一。其主要优势包括：

• 完整 API 覆盖：支持所有搜索和下载功能
• 断点续传：自动处理网络中断
• 进度显示：实时查看下载状态
• 批量操作：支持同时处理多个请求

安装方法：

pip install sentinelsat

2.2 系统架构设计

完整的自动化下载系统应包含以下模块：

1. 查询模块：构建搜索条件，获取可用数据列表
2. 筛选模块：根据质量指标过滤数据
3. 下载模块：管理下载队列和断点续传
4. 日志模块：记录操作过程和错误信息
5. 通知模块：任务完成时发送提醒

graph TD A[定义搜索参数] --> B[执行API查询] B --> C[筛选结果] C --> D[加入下载队列] D --> E[监控下载进度] E --> F{完成?} F -->|是| G[发送通知] F -->|否| E

3. 实战：构建自动化下载脚本

3.1 基础配置

首先设置认证信息和基本参数：

from sentinelsat import SentinelAPI # 连接Copernicus Open Access Hub api = SentinelAPI( user='您的用户名', password='您的密码', api_url='https://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 数据搜索支持多种精细控制：

# 构建复杂查询条件 products = api.query( area_of_interest, date=('20230101', '20231231'), platformname='Sentinel-2', producttype='S2MSI2A', # 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 批量下载实现

实现可靠的大批量下载需要考虑以下因素：

1. 并发控制：避免同时发起过多请求
2. 断点续传：自动恢复中断的下载
3. 速度限制：遵守服务器使用政策
4. 本地校验：确保文件完整性

from tqdm import tqdm import os # 创建下载目录 download_dir = 'sentinel2_downloads' os.makedirs(download_dir, exist_ok=True) # 分批下载函数 def batch_download(product_ids, max_workers=4): failed_downloads = [] with tqdm(total=len(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_path=download_dir, checksum=True # 自动校验文件 ) 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_retries=3): for attempt in range(max_retries): try: api.download( product_id, directory_path=download_dir, checksum=True, n_concurrent_dl=2 # 并发连接数 ) return True except Exception as e: print(f"尝试 {attempt+1} 失败: {str(e)}") time.sleep(60 * (attempt + 1)) # 指数退避 return False

4.2 自动化通知集成

任务完成后发送邮件通知：

import smtplib from email.mime.text import MIMEText def send_notification(subject, message): msg = MIMEText(message) msg['Subject'] = subject msg['From'] = 'sender@example.com' msg['To'] = 'receiver@example.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 S3）：

import 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, f"sentinel2/{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 查询优化技巧

1. 缩小时间窗口：分批查询比大范围查询更高效
2. 精确空间范围：避免查询不必要的大区域
3. 利用产品类型过滤：明确指定 L1C 或 L2A
4. 缓存结果：对重复查询保存中间结果

# 分月查询示例 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(days=31) month_end = min(next_month.replace(day=1) - timedelta(days=1), end) print(f"查询 {current.date()} 至 {month_end.date()}") products = api.query( aoi, date=(current.strftime('%Y%m%d'), month_end.strftime('%Y%m%d')), platformname='Sentinel-2', producttype='S2MSI2A' ) all_products.update(products) current = next_month return all_products

5.2 错误处理策略

完善的错误处理应包含：

1. API 限流处理：捕获 429 错误并等待
2. 网络异常：自动重试机制
3. 磁盘空间检查：避免因空间不足失败
4. 日志记录：详细记录错误上下文

import time import logging logging.basicConfig( filename='download.log', level=logging.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_path=download_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 False

5.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 与数据处理流程集成

典型的端到端处理流程：

1. 自动下载：本文介绍的脚本
2. 预处理：大气校正、辐射校正
3. 特征提取：NDVI、NDWI 等指数计算
4. 分类分析：土地利用/覆盖分类
5. 变化检测：时序变化分析

# 示例：下载后自动计算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 定期自动更新方案

设置定时任务保持数据最新：

1. Linux cron 作业：

0 3 * * * /usr/bin/python3 /path/to/sentinel_downloader.py --update

2. Python 定时器：

import schedule import time def daily_update(): # 获取昨天到今天的新数据 yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d') today = datetime.now().strftime('%Y%m%d') products = api.query( area_of_interest, date=(yesterday, today), platformname='Sentinel-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 不可用时，可以考虑以下替代方案：

# 使用AWS开放数据集的示例 import boto3 from botocore import UNSIGNED from botocore.config import Config s3 = boto3.client( 's3', config=Config(signature_version=UNSIGNED) ) # 列出可用数据 response = s3.list_objects_v2( Bucket='sentinel-s2-l2a', Prefix='tiles/10/S/DG/2023/1/' ) # 下载特定文件 s3.download_file( 'sentinel-s2-l2a', 'tiles/10/S/DG/2023/1/28/0/B01.jp2', 'B01.jp2' )

实际项目中，我们通常会结合多种数据源和工具构建健壮的自动化流程。关键是根据具体需求选择最适合的技术方案，并充分考虑可维护性和扩展性。