Python哨兵影像自动下载脚本:批量拉取与断点续传实战
简介:这份资源是一套面向遥感数据处理与地理信息分析人员的Python哨兵影像自动下载脚本,主要解决Sentinel卫星影像批量获取效率低、离线产品需手动触发检索、下载中断后难以恢复等实际问题,适合具备一定Python基础、需要长期批量获取哨兵数据的科研或工程用户。压缩包共3个文件,包含2个py源码文件与1个pyc编译文件,整体仅4KB,其中源码分别承担核心下载逻辑与运行入口,pyc文件为已编译的缓存模块,结构精简、便于直接调用与二次修改。脚本支持离线产品下载,可触发产品从LTA检索并在稍后通过原始URL获取;支持断点续传,运行后无需持续值守;同时提供矢量范围检索格式,便于按研究区边界筛选影像。使用前需安装sentinelsat库,并准备Copernicus Open Access Hub的账号与密码。目前已有1017人学习下载,可作为哨兵影像自动化获取的轻量工具参考。
1. 哨兵影像自动下载脚本:从手动点选到批量拉取的落地路径
做遥感项目的人大多经历过这种场景:项目要覆盖某片区域一整年的哨兵2号影像,手动在浏览器里一景一景筛选、下载,几十个产品点下来,半天就没了,还容易漏掉云量达标的时相。Python哨兵影像自动下载脚本要解决的就是这件事——把检索、筛选、下载、断点续传串成一条可复现的命令行流程,让机器替你守着下载队列。它适合做土地利用变化监测、农作物长势分析、水体提取这类需要多时相影像堆栈的从业者,也适合刚接触遥感数据获取、想用Python把重复劳动自动化掉的新手。脚本本身不神秘,核心就是调用数据源提供的检索接口,按经纬度范围、时间窗口、云量阈值过滤,再把匹配到的产品逐个拉回本地。真正决定它好不好用的,是参数怎么设、网络中断怎么办、下载下来的文件怎么校验完整性。
2. 脚本的检索逻辑与下载链路:先搞懂数据从哪来
2.1 哨兵影像的获取入口与检索参数
哨兵2号(Sentinel-2)是欧空局哥白尼计划下的多光谱成像任务,L2A级产品已经做过大气校正,地表反射率数据可以直接用于植被指数计算。公开获取渠道里,哥白尼数据空间生态门户(Copernicus Data Space Ecosystem)是目前主推的检索入口,提供STAC风格的API,支持按地理范围、时间区间、云量百分比、产品级别做过滤。脚本要做的第一件事,就是把这些过滤条件翻译成API能识别的查询参数。
常见做法是构造一个查询字典,把感兴趣区域(AOI)的边界框、起止日期、最大云量、产品类型塞进去。AOI可以用GeoJSON文件描述,也可以用四个角点坐标直接写死。时间窗口一般按ISO 8601格式传,比如2024-01-01T00:00:00Z/2024-12-31T23:59:59Z。云量阈值这个参数最容易被忽视——设成10%以下,某些多云地区可能一景都搜不到;设成80%,下载回来的影像又没法用。我一般会先跑一次检索,把返回结果的云量分布打印出来,再决定阈值卡在哪个位置。
import requests # 检索参数:AOI边界框、时间窗口、云量上限、产品级别 search_params = { "bbox": [116.0, 39.5, 117.0, 40.5], # 北京周边示例 "datetime": "2024-01-01T00:00:00Z/2024-06-30T23:59:59Z", "filter": "cloudCover<30", # 云量低于30% "limit": 100, "collections": ["sentinel-2-l2a"] } # 向STAC检索端点发起POST请求 resp = requests.post( "https://catalogue.dataspace.copernicus.eu/stac/search", json=search_params, timeout=30 ) features = resp.json().get("features", []) print(f"命中 {len(features)} 景影像") for f in features[:5]: print(f["id"], f["properties"]["cloudCover"])这段代码的逻辑很直白:把检索条件打包成JSON,POST到STAC端点,解析返回的features列表。bbox的顺序是[最小经度, 最小纬度, 最大经度, 最大纬度],写反了会搜到地球另一边的数据。filter字段的语法在不同STAC实现里略有差异,哥白尼门户支持cloudCover<30这种简写,但有些端点要求写成CQL2表达式。limit控制单次返回数量,超过100条需要翻页,翻页逻辑后面会讲。
2.2 从检索结果到下载链接的解析
检索返回的每个feature里,assets字段藏着真正的下载地址。哨兵2 L2A产品通常包含多个资产:B02、B03、B04是10米分辨率蓝绿红波段,B08是近红外,B11、B12是20米短波红外,还有thumbnail缩略图和Product完整压缩包。脚本要下载哪个,取决于后续处理流程——如果只算NDVI,下B04和B08就够了;如果要做云掩膜,还得加上SCL场景分类层。
# 解析每个feature的资产下载链接 download_list = [] for feat in features: assets = feat.get("assets", {}) # 只取10米波段和场景分类层,减少下载量 for band in ["B02", "B03", "B04", "B08", "SCL"]: if band in assets: download_list.append({ "id": feat["id"], "band": band, "url": assets[band]["href"], "size_mb": assets[band].get("file:size", 0) / 1e6 }) print(f"待下载文件数:{len(download_list)}") total_gb = sum(d["size_mb"] for d in download_list) / 1024 print(f"预估总大小:{total_gb:.2f} GB")这里有个容易翻车的点:file:size字段不是所有STAC端点都返回,拿不到的时候预估大小会显示0,不代表文件真的空。稳妥做法是先下载一个文件,用os.path.getsize量一下实际大小,再乘以文件总数做粗估。另外,assets里的href有时是重定向链接,直接丢给requests.get会跟丢,需要加allow_redirects=True,或者用requests.head先探一下最终地址。
2.3 断点续传与并发下载的实现
哨兵2单景L2A压缩包动辄700MB到1GB,网络抖动一次就前功尽弃。脚本必须支持断点续传——记录已下载的字节数,下次从断点位置继续拉。HTTP协议里的Range头就是干这个的,配合本地文件追加写入模式,能把中断成本降到最低。
import os import requests def download_with_resume(url, save_path, chunk_size=8192): """支持断点续传的下载函数""" headers = {} mode = "wb" existing_size = 0 # 如果本地已有部分文件,从断点继续 if os.path.exists(save_path): existing_size = os.path.getsize(save_path) headers["Range"] = f"bytes={existing_size}-" mode = "ab" resp = requests.get(url, headers=headers, stream=True, timeout=60) # 206表示服务器接受了Range请求 if resp.status_code == 206: total = existing_size + int(resp.headers.get("Content-Length", 0)) elif resp.status_code == 200: total = int(resp.headers.get("Content-Length", 0)) existing_size = 0 # 服务器不支持断点,从头下 mode = "wb" else: raise RuntimeError(f"下载失败,状态码 {resp.status_code}") with open(save_path, mode) as f: for chunk in resp.iter_content(chunk_size=chunk_size): if chunk: f.write(chunk) existing_size += len(chunk) # 打印进度百分比 if total > 0: print(f"\r进度:{existing_size/total*100:.1f}%", end="") print() return save_pathRange头的值是bytes=起始字节-,服务器返回206 Partial Content就说明续传生效了。如果返回200,意味着服务器不支持Range,这时候只能从头下,代码里把existing_size归零、模式切回wb就是处理这种情况。chunk_size设成8192字节是保守值,内网环境可以调到1MB以上减少IO次数。并发下载方面,concurrent.futures.ThreadPoolExecutor比多进程更适合IO密集型任务,线程数控制在4到8之间,开太多反而会因为带宽争抢导致整体变慢。
3. 认证、配额与批量任务编排:把脚本跑成流水线
3.1 访问令牌的获取与刷新
哥白尼数据空间生态门户的下载接口需要OAuth2令牌,不是匿名就能拉。令牌有有效期,通常10分钟到1小时不等,批量下载跑到一半令牌过期,后续请求全部401。脚本里得把令牌获取和刷新逻辑封装好,每次请求前检查剩余有效期,快过期就重新申请。
import time import requests class TokenManager: def __init__(self, username, password): self.username = username self.password = password self.token = None self.expire_at = 0 def get_token(self): # 提前60秒刷新,避免边界过期 if self.token and time.time() < self.expire_at - 60: return self.token resp = requests.post( "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token", data={ "grant_type": "password", "username": self.username, "password": self.password, "client_id": "cdse-public" }, timeout=30 ) resp.raise_for_status() data = resp.json() self.token = data["access_token"] self.expire_at = time.time() + data.get("expires_in", 600) return self.tokenexpires_in字段告诉你能用多少秒,代码里提前60秒刷新是留缓冲。用户名密码不要硬编码在脚本里,用环境变量或者单独的配置文件读取,避免提交到代码仓库。有些团队用服务账号,流程一样,只是client_id可能不同。
3.2 下载队列与失败重试策略
批量下载最怕的是某个文件卡住,整个队列堵死。合理的做法是给每个下载任务设超时,失败后进重试队列,重试次数用完再标记为永久失败,继续下一个。重试间隔用指数退避,第一次等5秒,第二次15秒,第三次45秒,避免短时间内反复冲击服务器。
import time from collections import deque def download_queue(tasks, max_retry=3): """带重试的下载队列""" queue = deque(tasks) failed = [] while queue: task = queue.popleft() for attempt in range(max_retry): try: download_with_resume(task["url"], task["save_path"]) print(f"完成:{task['id']}") break except Exception as e: wait = 5 * (3 ** attempt) # 5s, 15s, 45s print(f"重试 {attempt+1}/{max_retry},等待 {wait}s:{e}") time.sleep(wait) else: failed.append(task) print(f"永久失败:{task['id']}") return failedfor...else结构在这里很好用:循环正常结束(没触发break)才执行else分支,把任务记入失败列表。重试等待时间用5 * (3 ** attempt)生成指数退避序列。失败任务最后统一输出,方便人工排查是网络问题还是链接失效。
3.3 下载后的完整性校验
文件下完不等于能用。哨兵2产品在STAC元数据里通常提供checksum字段,常见的是MD5。下载完成后算一遍本地文件的MD5,跟元数据比对,不一致就说明传输过程中损坏了,得重新下。
import hashlib def verify_md5(file_path, expected_md5): """校验文件MD5""" md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): md5.update(chunk) actual = md5.hexdigest() if actual != expected_md5: raise ValueError(f"MD5不匹配:期望 {expected_md5},实际 {actual}") return Trueiter(lambda: f.read(8192), b"")是分块读取的惯用写法,避免一次性把大文件读进内存。校验失败的文件直接删掉重下,别留着占空间。有些产品只提供Content-Length没有MD5,那就退而求其次,比对文件大小是否与元数据一致。
4. 避坑与排查:那些让脚本跑不通的细节
4.1 检索返回空结果
现象:参数看着没问题,features列表就是空的。原因通常是bbox坐标顺序写反,或者时间格式少了Z后缀,再或者云量阈值卡得太死。解决:先用最小参数集(只留bbox和datetime)跑一次,确认能返回结果,再逐项加过滤条件,每加一项看结果数量变化。
4.2 下载到一半报403
现象:前几个文件正常,后面突然全部403 Forbidden。原因基本是令牌过期了,但脚本还在用旧token。解决:把TokenManager的刷新逻辑嵌到每次请求前,别只在脚本启动时取一次。另外检查系统时间是否准确,时间偏差超过令牌有效期会导致签名校验失败。
4.3 文件名冲突导致覆盖
现象:不同日期的影像下载后互相覆盖,只剩最后一景。原因:保存路径只用了产品ID的一部分,或者直接用了波段名当文件名。解决:保存路径里带上完整的产品ID和日期,比如{product_id}/{band}.jp2,产品ID本身包含时间戳和轨道号,天然唯一。
4.4 磁盘空间不足
现象:下载到第N个文件时报OSError: [Errno 28] No space left on device。原因:哨兵2单景解压后可能超过2GB,批量下载前没算总账。解决:下载前用shutil.disk_usage检查剩余空间,预估总大小乘以1.5倍安全系数,不够就先清理或分批下载。
4.5 并发数过高触发限流
现象:开了16个线程,下载速度反而比单线程还慢,部分请求返回429。原因:服务器对单IP并发连接数有限制。解决:线程数降到4到6,或者在请求头里加User-Agent标识,有些端点对带标识的客户端限流宽松些。实在不行就串行下载,稳比快重要。
5. 进阶技巧:把下载脚本接进处理流水线
脚本跑通之后,下一步是让它跟后续处理无缝衔接。我一般会在下载完成后自动触发解压和波段堆叠,把L2A产品里的单个波段文件合成一个多波段GeoTIFF,方便直接丢进ENVI或者用rasterio读取。解压用zipfile模块,堆叠用rasterio的merge或者gdal_merge.py,这些工具在Python生态里都很成熟。
import zipfile import rasterio from rasterio.merge import merge def extract_and_stack(zip_path, bands, out_path): """解压哨兵2产品并堆叠指定波段""" with zipfile.ZipFile(zip_path, "r") as z: # 只解压需要的波段文件 members = [m for m in z.namelist() if any(b in m for b in bands)] z.extractall(path="./temp", members=members) # 找到解压后的波段文件并堆叠 src_files = [] for band in bands: for root, _, files in os.walk("./temp"): for f in files: if band in f and f.endswith(".jp2"): src_files.append(rasterio.open(os.path.join(root, f))) mosaic, transform = merge(src_files) with rasterio.open(out_path, "w", driver="GTiff", height=mosaic.shape[1], width=mosaic.shape[2], count=len(src_files), dtype=mosaic.dtype, crs=src_files[0].crs, transform=transform) as dst: dst.write(mosaic) for src in src_files: src.close()merge函数把多个波段文件按地理坐标对齐后堆叠成一个数组,transform和crs从第一个源文件继承。输出格式选GeoTIFF,ENVI、QGIS、ArcGIS都能直接打开。波段顺序按bands列表里的顺序来,算NDVI的话把B04放第一个、B08放第二个,后续索引就是array[0]和array[1]。
另一个实用技巧是给下载脚本加个--dry-run模式,只检索不下载,把命中的产品列表、云量、预估大小打印出来。批量任务动辄几百GB,先dry-run确认一遍再真下,能省掉很多后悔药。我现在的习惯是:任何批量下载脚本,第一次跑必须带--dry-run,确认检索结果符合预期、磁盘空间够用、令牌有效,再摘掉参数正式跑。这个习惯帮我避开了至少三次“下了一晚上发现AOI写错”的事故。希望帮到你。
本文还有配套的精品资源,点击获取