☰
Sentinel卫星数据批量下载:sentinelsat检索与IDM多线程实战
2026/10/16 1:14:43 网站建设 项目流程

1. 项目缘起与整体思路拆解

哥白尼哨兵系列卫星数据在遥感圈子里算是公认的“口粮级”素材,尤其是Sentinel-1和Sentinel-2,一个管雷达后向散射,一个管多光谱反射率,做地表覆盖分类、农作物长势监测、水体提取、城市扩张分析基本都绕不开它。但真正长期跟这批数据打交道的人都知道,最让人头疼的不是处理算法,而是下载环节——尤其是碰上那些状态标记为offline的存档数据。所谓offline,就是数据虽然还在归档目录里挂着,但实际文件已经被挪到近线存储或者冷存储上,你直接点下载链接大概率会吃一个403或者一个空文件,得先触发恢复请求,等系统把数据从冷存储拉回在线存储,才能正常拉取。

这个项目的核心目标很明确:用sentinelsat这个Python包做元数据检索和下载链接获取,再借助IDM的多线程能力把offline数据批量拉下来。为什么是这两个工具的组合?sentinelsat的优势在于它封装了哥白尼数据空间的检索API,你可以用地理范围、时间窗口、产品类型、云量阈值等条件精准筛出自己需要的那一批产品,而且它返回的下载链接是带鉴权的,能直接喂给下载器。IDM的优势则在于多连接分段下载,对于单个几百MB到几个GB的哨兵产品,能把带宽吃满,比浏览器单线程下载快好几倍。

但这里有个关键矛盾:sentinelsat本身提供的download_all()方法是单线程顺序下载,而且对offline数据的处理并不友好——它不会自动帮你触发恢复,也不会在恢复完成后自动重试。所以整个方案的设计思路就变成了:用sentinelsat做“侦察兵”,负责检索、筛选、生成下载清单;用IDM做“运输队”,负责实际的文件拉取;中间用一个轻量的调度脚本把offline数据的恢复请求和下载任务串起来。这样既保留了sentinelsat在检索端的灵活性,又发挥了IDM在传输端的效率优势。

适合谁来参考这个方案?如果你是做遥感应用开发、地理信息分析、环境监测相关工作的,手头经常要批量获取哨兵数据,而且对下载效率有要求,那这套组合拳基本可以直接抄作业。哪怕你之前没用过sentinelsat,只要会一点Python基础操作,跟着下面的步骤走也能跑通。

2. 核心细节解析与实操要点

2.1 sentinelsat的检索逻辑与offline状态识别

sentinelsat的核心价值在于它把哥白尼数据空间的OData检索接口包装成了Python对象。你创建一个SentinelAPI实例,传入账号密码,然后调用query()方法,传入各种过滤条件,它就会返回一个字典,键是产品UUID,值是一个包含产品元数据的字典。这个元数据里有一个字段叫online,布尔值,True表示数据在线可直接下载,False表示offline需要先恢复。

这里有个细节很多人会忽略:offline状态并不是永久不变的。哥白尼系统对冷存储数据的恢复有一个时间窗口,通常请求发出后几分钟到几小时不等,数据会被拉回在线存储,此时online字段会变成True。但如果你不主动触发恢复请求,它可能一直保持offline。sentinelsat提供了一个trigger_offline_retrieval()方法,传入产品UUID就能发起恢复请求。这个方法是整个流程里最关键的一环,没有它,offline数据你永远下不动。

另一个要点是检索条件的粒度。哨兵数据的检索支持按平台(Sentinel-1、Sentinel-2)、产品类型(SLC、GRD、L1C、L2A等)、时间范围、地理范围(WKT格式或GeoJSON)、云量百分比等过滤。地理范围建议用geojson参数传一个Polygon,比用area参数传WKT更直观。时间范围用date参数,格式是('20230101', '20231231')这样的元组。云量过滤用cloudcoverpercentage参数,传一个(0, 20)的元组表示只要云量在0到20之间的产品。

注意:检索返回的结果默认是按时间倒序排列的,如果你需要按其他方式排序,可以在query()里传order_by参数,比如order_by='cloudcoverpercentage'。

2.2 IDM的批量下载配置与URL清单生成

IDM本身是一个Windows平台上的下载工具,但它支持通过命令行或者导入URL列表的方式批量添加任务。我们的做法是:用sentinelsat生成一个包含所有下载链接的文本文件,每行一个URL,然后用IDM的“从文件导入”功能批量添加。但这里有个坑:sentinelsat返回的下载链接是带时效性的,通常有效期只有几分钟到几十分钟,如果你先生成清单再慢慢导入IDM,链接可能已经过期了。

所以更稳妥的做法是:在Python脚本里直接调用IDM的命令行接口。IDM提供了一个IDMan.exe可执行文件,支持/d参数传URL、/p参数传保存路径、/f参数传文件名、/a参数表示添加到队列、/q参数表示下载完成后退出。你可以在Python里用subprocess模块调用它,每获取一个产品的下载链接就立即推给IDM,这样链接时效性问题就解决了。

但IDM的命令行接口有个限制:它不支持直接传鉴权头。sentinelsat返回的下载链接里其实已经包含了鉴权信息(通常是URL里的token参数),所以直接把这个完整URL传给IDM是可行的。不过要注意,有些产品的下载链接会重定向到另一个域名,IDM默认会跟随重定向,这个行为是符合预期的。

提示:IDM的命令行参数在不同版本间可能有细微差异,建议先用一个测试链接验证一下IDMan.exe的调用方式,确认能正常拉起下载窗口再批量操作。

2.3 恢复请求与下载任务的调度策略

offline数据的恢复请求不是即时的,你发出请求后需要等待一段时间。如果只是少量数据,手动等一等再重新检索确认online状态变成True后再下载就行。但如果是批量操作,比如几百个产品里有几十个是offline的,手动等就不现实了。这时候需要一个简单的调度逻辑:先对所有检索结果做一次分类,把online的直接推给IDM,offline的触发恢复请求并记录UUID;然后每隔一段时间(比如10分钟)重新查询这些UUID的状态,一旦变成online就推给IDM。

这个调度逻辑可以用一个简单的循环实现,不需要引入复杂的任务队列框架。关键是要设置一个合理的轮询间隔和最大等待时间。轮询间隔太短会给服务器造成不必要的压力,太长则影响整体效率。根据实际经验,10到15分钟的轮询间隔比较合适,最大等待时间建议设为2到4小时,超过这个时间还没恢复的数据可以考虑放弃或者换一个时间窗口重新检索。

还有一个细节:恢复请求本身是有配额限制的。哥白尼系统对每个账号的恢复请求数量有每日上限,具体数值可能会调整,但一般来说不要一次性触发太多恢复请求,建议分批触发,比如每批不超过20个,触发完一批等一段时间再触发下一批。

3. 实操过程与核心环节实现

3.1 环境准备与依赖安装

先确保你的Python环境是3.8以上版本,然后安装sentinelsat:

pip install sentinelsat

这个包依赖requests、geojson、click等库,pip会自动处理。IDM需要你提前在Windows上安装好,并确保IDMan.exe的路径已经添加到系统环境变量里,或者在脚本里写绝对路径。

哥白尼数据空间的账号需要提前注册好,注册地址在哥白尼官网,这里不展开。拿到账号密码后,建议不要直接写在脚本里,而是用环境变量或者配置文件管理,避免泄露。

3.2 检索与分类脚本编写

下面是一个完整的检索与分类脚本示例,我把它拆成几个函数,方便你按需修改:

import os import time import subprocess from sentinelsat import SentinelAPI, read_geojson, geojson_to_wkt # 配置区 COPERNICUS_USER = os.environ.get('COPERNICUS_USER') COPERNICUS_PASSWORD = os.environ.get('COPERNICUS_PASSWORD') IDM_PATH = r'C:\Program Files (x86)\Internet Download Manager\IDMan.exe' DOWNLOAD_DIR = r'D:\SentinelData' GEOJSON_PATH = r'D:\aoi.geojson' START_DATE = '20230101' END_DATE = '20231231' CLOUD_COVER = (0, 20) PLATFORM = 'Sentinel-2' PRODUCT_TYPE = 'S2MSI2A' # 初始化API api = SentinelAPI(COPERNICUS_USER, COPERNICUS_PASSWORD, 'https://apihub.copernicus.eu/apihub') # 读取AOI footprint = geojson_to_wkt(read_geojson(GEOJSON_PATH)) # 检索 products = api.query( footprint, date=(START_DATE, END_DATE), platformname=PLATFORM, producttype=PRODUCT_TYPE, cloudcoverpercentage=CLOUD_COVER ) print(f'共检索到 {len(products)} 个产品') # 分类 online_products = {} offline_products = {} for uuid, info in products.items(): if info['online']: online_products[uuid] = info else: offline_products[uuid] = info print(f'在线产品: {len(online_products)} 个') print(f'离线产品: {len(offline_products)} 个')

这段脚本的核心就是api.query()返回的字典,每个产品的online字段决定了后续处理路径。注意producttype参数要根据你的实际需求填,Sentinel-2的L2A产品是S2MSI2A,L1C是S2MSI1C,Sentinel-1的GRD是GRD,SLC是SLC。

3.3 触发恢复请求与轮询等待

对于offline产品,逐个触发恢复请求:

def trigger_retrieval(api, offline_products, batch_size=20): uuids = list(offline_products.keys()) for i in range(0, len(uuids), batch_size): batch = uuids[i:i+batch_size] for uuid in batch: try: api.trigger_offline_retrieval(uuid) print(f'已触发恢复请求: {uuid}') except Exception as e: print(f'触发失败 {uuid}: {e}') if i + batch_size < len(uuids): print(f'等待60秒后继续下一批...') time.sleep(60) trigger_retrieval(api, offline_products)

触发完恢复请求后,进入轮询循环:

def wait_for_online(api, offline_products, max_wait_hours=4, poll_interval_minutes=10): uuids = list(offline_products.keys()) start_time = time.time() max_wait_seconds = max_wait_hours * 3600 poll_interval_seconds = poll_interval_minutes * 60 while uuids and (time.time() - start_time) < max_wait_seconds: print(f'等待 {poll_interval_minutes} 分钟后重新检查...') time.sleep(poll_interval_seconds) still_offline = [] for uuid in uuids: try: info = api.get_product_odata(uuid) if info['online']: print(f'产品已上线: {uuid}') yield uuid, info else: still_offline.append(uuid) except Exception as e: print(f'查询失败 {uuid}: {e}') still_offline.append(uuid) uuids = still_offline print(f'剩余离线产品: {len(uuids)} 个') if uuids: print(f'以下产品在 {max_wait_hours} 小时内未上线,建议稍后重试:') for uuid in uuids: print(uuid)

这个生成器函数会在产品上线时逐个yield出来,方便后续直接推给IDM。

3.4 调用IDM批量下载

把上线产品的下载链接推给IDM:

def download_with_idm(api, uuid, info): download_url = api.get_product_odata(uuid)['url'] filename = info['title'] + '.zip' save_path = os.path.join(DOWNLOAD_DIR, filename) cmd = [ IDM_PATH, '/d', download_url, '/p', DOWNLOAD_DIR, '/f', filename, '/a', '/q' ] try: subprocess.run(cmd, check=True) print(f'已添加到IDM: {filename}') except subprocess.CalledProcessError as e: print(f'IDM调用失败 {filename}: {e}') # 先下载在线产品 for uuid, info in online_products.items(): download_with_idm(api, uuid, info) # 再处理离线产品 for uuid, info in wait_for_online(api, offline_products): download_with_idm(api, uuid, info)

这里有个细节:api.get_product_odata(uuid)['url']返回的下载链接是带鉴权的完整URL,直接传给IDM即可。/a参数表示添加到队列而不是立即开始,/q表示下载完成后退出IDM。如果你希望IDM下载完成后自动关机或者执行其他操作,可以去掉/q参数,在IDM的设置里配置。

注意:IDM的/p参数指定的是保存目录,/f参数指定的是文件名。如果文件名包含特殊字符,建议先做一次清洗,把空格、括号等替换成下划线,避免IDM解析出错。

4. 常见问题与排查技巧实录

4.1 检索返回空结果或报错

最常见的原因是AOI的GeoJSON格式不对。sentinelsat要求GeoJSON是一个FeatureCollection,里面至少有一个Feature,Feature的geometry类型是Polygon。如果你从GIS软件导出的GeoJSON是MultiPolygon,需要先转成Polygon,或者用geojson_to_wkt之前手动处理一下。另一个原因是时间范围格式不对,必须是('YYYYMMDD', 'YYYYMMDD')这样的元组,字符串之间不能有空格。

还有一种情况是账号权限问题。哥白尼数据空间的账号需要确认已经同意了数据使用条款,否则检索会返回401。登录网页版确认一下账号状态,如果提示需要同意条款,点一下同意即可。

4.2 恢复请求触发失败

trigger_offline_retrieval()报错通常有几个原因:一是产品UUID不存在或者已经被删除;二是恢复请求配额已满;三是网络超时。建议在触发请求时加一个try-except,把失败的UUID记录下来,等一段时间后重试。如果配额满了,只能等第二天再试。

4.3 IDM下载速度慢或断连

IDM的多线程下载对服务器端是有压力的,有些服务器会限制单IP的并发连接数。如果发现下载速度上不去或者频繁断连,可以在IDM的设置里把最大连接数从默认的8调到4或者2。另外,IDM的“站点管理”里可以针对特定域名设置连接数,把哥白尼的下载域名单独配置一下。

还有一个坑是下载链接过期。sentinelsat返回的下载链接有效期通常只有几分钟,如果你在脚本里先生成了一大堆链接再慢慢推给IDM,后面的链接可能已经失效了。解决办法就是前面说的,边获取边推送,不要攒着。

4.4 下载的文件不完整或损坏

哨兵产品的压缩包通常有几百MB到几个GB,下载过程中如果网络不稳定,文件可能不完整。IDM本身有断点续传功能,但如果服务器不支持Range请求,断点续传就会失败。建议下载完成后用zipfile模块做一次完整性校验:

import zipfile def verify_zip(filepath): try: with zipfile.ZipFile(filepath, 'r') as z: bad_file = z.testzip() if bad_file: print(f'文件损坏: {bad_file}') return False return True except zipfile.BadZipFile: print(f'不是有效的ZIP文件: {filepath}') return False

如果校验失败,删掉重新下载即可。

4.5 常见问题速查表

问题现象可能原因排查方法解决措施
检索返回空AOI格式不对检查GeoJSON是否为Polygon转换MultiPolygon为Polygon
检索报401账号未同意条款登录网页版查看同意数据使用条款
恢复请求失败配额已满查看返回错误信息等次日重试或分批触发
IDM下载慢并发连接过多查看IDM连接数设置降低最大连接数至4
下载链接失效链接过期检查URL中的token时间戳边获取边推送,不攒链接
文件损坏下载中断用zipfile校验删除重新下载

实操心得:我一般会在脚本里加一个日志记录,把每个产品的UUID、标题、online状态、恢复请求时间、下载完成时间都写到一个CSV文件里。这样万一中间断了,重新跑的时候可以先读日志,跳过已经下载完成的产品,避免重复劳动。这个习惯在批量处理几百个产品的时候特别有用。

5. 效率优化与扩展思路

5.1 并发检索与异步恢复请求

sentinelsat的检索是同步的,如果你要检索多个时间窗口或者多个AOI,可以考虑用concurrent.futures的ThreadPoolExecutor做并发检索。但要注意哥白尼API对并发请求有限制,建议并发数不要超过3。恢复请求同理,虽然可以并发触发,但配额限制在那里,并发也快不了多少。

5.2 与本地数据管理流程的衔接

下载下来的哨兵产品通常是ZIP压缩包,里面包含JPEG2000格式的影像和XML元数据。如果你后续要用GDAL或者rasterio处理,建议下载完成后自动解压并整理目录结构。我通常按平台/产品类型/年份/月份的层级组织目录,方便后续批量处理。

5.3 扩展到其他数据源

这套“sentinelsat检索+IDM下载”的思路其实可以扩展到其他数据源。比如Landsat数据可以用landsatxplore包做检索,下载链接同样可以推给IDM。再比如一些气象数据、海洋数据,只要提供了HTTP下载链接和鉴权机制,都可以用类似的模式处理。核心逻辑就是:用专门的检索库做元数据筛选,用通用下载器做文件传输,中间用脚本做调度。

最后再分享一个小技巧:如果你经常需要下载同一区域的数据,可以把检索条件写成一个配置文件,每次只需要改时间范围就行。这样脚本的复用性会高很多,不用每次都改代码。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询