Python自动化获取Sentinel-2卫星数据的完整指南
遥感数据处理的第一步往往是从数据获取开始的。对于Sentinel-2这样的开源卫星数据，虽然欧空局提供了免费访问，但手动下载过程繁琐耗时——需要反复登录网站、设置筛选条件、等待下载队列，一个不小心还可能漏掉关键时间段的数据。作为长期从事遥感分析的从业者，我深知这种低效操作对研究进度的影响。本文将分享如何用Python脚本彻底解决这一痛点，实现Sentinel-2数据的自动化获取全流程。
1. 环境配置与API准备
1.1 必备工具安装
开始前需要配置Python环境并安装关键库：
pip install sentinelsat sentinelhub asf_search geopandas- sentinelsat：欧空局官方推荐的Python客户端，支持查询和下载Sentinel数据
- sentinelhub：提供更高级的API功能和数据处理工具
- asf_search：NASA ASF DAAC的搜索接口，可作为备用数据源
- geopandas：处理地理空间数据
提示：建议使用conda创建独立环境，避免库版本冲突。特别是gdal库的安装，conda能自动解决依赖问题。
1.2 欧空局账户配置
访问 Copernicus Open Access Hub 注册账号，获取API密钥。然后在home目录创建配置文件：
# ~/.sentinelsat.conf [general] username = your_username password = your_password或者在代码中直接认证：
from sentinelsat import SentinelAPI api = SentinelAPI('your_username', 'your_password', 'https://scihub.copernicus.eu/dhus')2. 智能查询与筛选策略
2.1 基础查询参数
核心查询参数包括：
| 参数 | 说明 | 示例值 |
|---|---|---|
| platformname | 卫星名称 | 'Sentinel-2' |
| date | 时间范围 | ('20230101', '20230131') |
| cloudcoverpercentage | 云量阈值 | (0, 30) |
| producttype | 产品级别 | 'S2MSI2A' (L2A级) |
| area | 感兴趣区域(WKT格式) | POLYGON((...)) |
典型查询示例：
# 定义AOI区域（上海周边） shanghai_aoi = '''POLYGON((121.1 30.8, 121.9 30.8, 121.9 31.5, 121.1 31.5, 121.1 30.8))''' products = api.query( area=shanghai_aoi, date=('20230101', '20230110'), platformname='Sentinel-2', producttype='S2MSI2A', cloudcoverpercentage=(0, 10) )2.2 高级筛选技巧
时间序列分析场景下，我们需要获取连续时间段的数据：
from datetime import datetime, timedelta def get_date_ranges(start_date, end_date, interval_days=5): date_ranges = [] current = datetime.strptime(start_date, '%Y%m%d') end = datetime.strptime(end_date, '%Y%m%d') while current < end: next_date = current + timedelta(days=interval_days) date_ranges.append(( current.strftime('%Y%m%d'), min(next_date, end).strftime('%Y%m%d') )) current = next_date return date_ranges for date_range in get_date_ranges('20230101', '20230331'): products.update(api.query( area=shanghai_aoi, date=date_range, platformname='Sentinel-2', cloudcoverpercentage=(0, 30) ))多条件复合筛选示例：
# 筛选特定轨道且处理基线号较高的数据 filtered = { uuid: meta for uuid, meta in products.items() if (meta['orbitnumber'] == 132 and float(meta['processinglevel'].split('-')[-1]) >= 4.0) }3. 批量下载与断点续传
3.1 基础下载方法
最简单的下载方式是直接调用API：
api.download_all(products, directory_path='./sentinel_data')但这种方法缺乏灵活性，且网络中断后需要重新开始。更健壮的方案是：
3.2 增强型下载管理器
from pathlib import Path from tqdm import tqdm import requests def enhanced_download(product_id, metadata, save_dir='./data'): save_path = Path(save_dir) / f"{metadata['title']}.zip" if save_path.exists(): print(f"文件已存在: {save_path}") return True url = api.get_product_odata(product_id)['url'] headers = {'Authorization': f'Basic {api.session.auth}'} try: with requests.get(url, headers=headers, stream=True) as r: r.raise_for_status() total_size = int(r.headers.get('content-length', 0)) with open(save_path, 'wb') as f, tqdm( unit='B', unit_scale=True, total=total_size, desc=metadata['title'] ) as pbar: for chunk in r.iter_content(chunk_size=8192): f.write(chunk) pbar.update(len(chunk)) return True except Exception as e: print(f"下载失败 {product_id}: {str(e)}") if save_path.exists(): save_path.unlink() # 删除不完整文件 return False # 批量执行 for product_id, metadata in products.items(): success = enhanced_download(product_id, metadata) if not success: print(f"重试下载 {metadata['title']}") enhanced_download(product_id, metadata) # 简单重试机制3.3 下载队列管理
对于大规模下载任务，建议实现优先级队列：
from queue import PriorityQueue class DownloadTask: def __init__(self, product_id, metadata, priority=5): self.product_id = product_id self.metadata = metadata self.priority = priority def __lt__(self, other): return self.priority < other.priority def download_worker(task_queue): while not task_queue.empty(): task = task_queue.get() if not enhanced_download(task.product_id, task.metadata): # 失败任务降低优先级重新入队 new_task = DownloadTask( task.product_id, task.metadata, task.priority - 1 ) task_queue.put(new_task) task_queue.task_done() # 初始化优先级队列 dl_queue = PriorityQueue() for pid, meta in products.items(): # 云量低的优先下载 priority = 10 - meta['cloudcoverpercentage'] // 10 dl_queue.put(DownloadTask(pid, meta, priority)) # 启动多个下载线程 from threading import Thread for _ in range(4): # 4个并发下载 Thread(target=download_worker, args=(dl_queue,)).start() dl_queue.join()4. 数据处理与质量检查
4.1 自动解压与组织
下载的zip文件需要解压并合理组织：
import zipfile from datetime import datetime def organize_sentinel_data(raw_dir='./data', output_dir='./processed'): raw_dir = Path(raw_dir) output_dir = Path(output_dir) output_dir.mkdir(exist_ok=True) for zip_file in raw_dir.glob('*.zip'): with zipfile.ZipFile(zip_file) as zf: # 提取获取日期 fname = zip_file.stem acq_date = datetime.strptime( fname.split('_')[2], '%Y%m%dT%H%M%S' ).strftime('%Y%m%d') # 创建日期目录 date_dir = output_dir / acq_date date_dir.mkdir(exist_ok=True) # 解压到日期目录 zf.extractall(date_dir) print(f"解压完成: {zip_file} -> {date_dir}") # 移动原始zip文件到存档目录 archive_dir = raw_dir / 'archived' archive_dir.mkdir(exist_ok=True) zip_file.rename(archive_dir / zip_file.name) organize_sentinel_data()4.2 数据质量验证
自动化检查下载数据的完整性：
def validate_sentinel_product(product_path): product_path = Path(product_path) required_files = [ 'MTD_MSIL2A.xml', 'GRANULE/*/MTD_TL.xml', 'GRANULE/*/IMG_DATA/*_B02_10m.jp2', 'GRANULE/*/IMG_DATA/*_B03_10m.jp2', 'GRANULE/*/IMG_DATA/*_B04_10m.jp2', 'GRANULE/*/IMG_DATA/*_B08_10m.jp2' ] missing_files = [] for pattern in required_files: if not list(product_path.glob(pattern)): missing_files.append(pattern) if missing_files: print(f"数据不完整，缺失文件: {missing_files}") return False # 检查文件大小是否异常 b02_size = next(product_path.glob('**/*_B02_10m.jp2')).stat().st_size if b02_size < 1024: # 小于1KB视为异常 print(f"波段文件大小异常: {b02_size} bytes") return False return True # 批量验证 for date_dir in Path('./processed').iterdir(): for product_dir in date_dir.glob('*.SAFE'): if not validate_sentinel_product(product_dir): print(f"数据验证失败: {product_dir}") # 可加入自动重新下载逻辑5. 实战技巧与经验分享
5.1 下载策略优化
- 时间段选择：欧空局服务器在欧洲工作时间(UTC 6:00-18:00)负载较高，尽量避开这些时段安排批量下载
- 区域分割：当需要大范围数据时，将AOI分割为多个小区域并行下载，速度更快
- 产品级别选择：
- L1C：原始数据，文件较大但包含全部信息
- L2A：经过大气校正，适合直接分析但处理需要时间
5.2 常见问题解决
问题1：下载速度慢或不稳定
解决方案：
# 在requests.get中添加超时和重试逻辑 from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retries = Retry( total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504] ) session.mount('https://', HTTPAdapter(max_retries=retries))问题2：API查询限制
欧空局对API调用有频率限制，解决方法：
import time for i, product_id in enumerate(products): if i > 0 and i % 50 == 0: time.sleep(60) # 每50次查询暂停1分钟 # 正常查询逻辑5.3 扩展应用
自动生成缩略图：
from osgeo import gdal import numpy as np import matplotlib.pyplot as plt def generate_thumbnail(safe_path, output_path=None): # 查找10m分辨率波段 b02 = next(Path(safe_path).glob('**/*_B02_10m.jp2')) b03 = next(Path(safe_path).glob('**/*_B03_10m.jp2')) b04 = next(Path(safe_path).glob('**/*_B04_10m.jp2')) # 读取波段数据 def read_band(band_path): ds = gdal.Open(str(band_path)) return ds.ReadAsArray() rgb = np.stack([ read_band(b04), # Red read_band(b03), # Green read_band(b02) # Blue ], axis=-1) # 简单拉伸增强 rgb = np.clip(rgb * 3.5 / 10000, 0, 1) # 生成缩略图 plt.figure(figsize=(10, 10)) plt.imshow(rgb) plt.axis('off') if output_path: plt.savefig(output_path, bbox_inches='tight', pad_inches=0) else: plt.show() plt.close()在实际项目中，这套自动化系统将原本需要数天的手动下载工作缩短到几小时，特别是对于长期时间序列分析，优势更加明显。一个实用的建议是建立本地数据索引数据库，记录已下载数据的元信息，避免重复下载。
