资讯详情

资讯详情

哥白尼哨兵数据下载工具全攻略:从手动点选到Python批量下载与断点续传

简介哥白尼哨兵数据下载工具是一款面向GIS从业者、遥感科研人员及地理信息相关专业学生的实用软件用于便捷获取欧洲空间局哥白尼计划提供的哨兵卫星数据解决手动检索与批量下载效率低的问题。压缩包共4个文件约7KB包含2个Python脚本、1个txt说明文档和1个geojson地理数据文件Python脚本承担与数据服务器交互、参数指定及自动化下载的核心逻辑geojson用于存储卫星覆盖区域等地理信息txt文档则提供配置与使用指引。目前已有578人学习下载。通过该工具读者可快速指定下载范围与时间参数实现哨兵数据的自动化获取并借助地理数据文件直观了解数据来源与覆盖范围为环境监测、气候变化研究、土地利用分析等场景提供数据支撑适合具备一定Python基础的用户参考使用。1. 哥白尼哨兵数据下载工具从手动点选到脚本批量的分水岭如果你做过遥感相关的活儿大概率经历过这样的场景打开浏览器在地图上一块一块框选区域设置时间范围点下载等压缩包解压发现云量太高不能用再回去重新框。一个中等规模的时序分析任务光下载就能耗掉一整天。哥白尼哨兵数据下载工具要解决的正是这个「数据获取」环节的效率瓶颈——它把哨兵系列卫星影像的检索、筛选、下载从浏览器交互变成可编程、可批量、可复现的流程。这篇文章面向两类人一是刚接触哨兵数据、还在用网页端一张张下的新手二是已经写过一些下载脚本但总在认证、断点、限流上翻车的老手。我会把工具选型、认证配置、批量下载脚本、断点续传和常见报错拆开讲每一步都落到能直接跑的代码和参数上。哨兵数据本身是免费的但免费不等于好拿下载工具选不对时间成本比数据本身贵得多。2. 哨兵数据下载工具怎么选四种主流路径的适用边界2.1 先搞清楚你要下的是哪颗星、哪级产品哨兵系列不是一颗卫星是一组。做下载工具选型之前必须先明确目标数据否则工具选错后面全是白费功夫。常见的几类卫星传感器常用产品级别典型分辨率重访周期主要用途Sentinel-1C波段SARSLC / GRD5m×20m / 10m6~12天形变监测、洪水、海冰Sentinel-2多光谱MSIL1C / L2A10/20/60m5天植被、水体、城市变化Sentinel-3OLCI/SLSTRL1B / L2300m~1km1~2天大尺度海洋、大气Sentinel-5PTROPOMIL23.5km×5.5km1天大气成分、气溶胶选工具的第一步不是看工具本身而是看你要的产品级别。L1C是大气表观反射率L2A是地表反射率后者已经过了大气校正做植被指数一般直接用L2A。SAR的GRD是地距探测产品已经过多视和投影SLC是单视复数做干涉才需要。级别选错下载工具再顺手也白搭。2.2 四种下载路径的对比网页、API、命令行、Python库目前主流的下载方式有四类各有明确的适用场景网页端Copernicus Browser适合零星几张、快速预览。缺点是批量操作几乎不可行框选区域大了会卡下载队列管理很弱。新手用它熟悉数据没问题但一旦超过20景就该换方式。OData API哥白尼数据空间提供的标准接口支持按属性过滤、分页、排序。适合需要精细控制查询条件的场景比如「2024年6月到8月云量小于10%覆盖某流域的L2A」。返回的是JSON需要自己解析下载链接。认证走OAuth2或token。命令行工具如某下载客户端适合服务器环境、定时任务。配置好之后一条命令拉一批可以配合cron做每日增量。缺点是配置文件格式需要熟悉报错信息有时不够直观。Python库如sentinelsat、cdsapi等灵活度最高能和pandas、geopandas、rasterio串起来。适合做自动化流水线查询→筛选→下载→预处理一条龙。缺点是依赖版本偶尔打架认证环节容易出玄学问题。我一般会这样选一次性小批量用网页固定区域的周期性任务用命令行需要和GIS分析串联的用Python库查询条件复杂但不需要后续处理的用OData API直接拉。2.3 用Python库跑通第一次认证与查询以sentinelsat为例这是社区里用得比较多的一个库。先装pip install sentinelsat认证和查询的最小代码from sentinelsat import SentinelAPI, read_geojson, geojson_to_wkt from datetime import date # 认证用户名密码走哥白尼数据空间账号 api SentinelAPI(your_username, your_password, https://apihub.copernicus.eu/apihub) # 定义查询区域用GeoJSON文件描述AOI footprint geojson_to_wkt(read_geojson(aoi.geojson)) # 查询指定产品类型、时间、云量、区域 products api.query( footprint, date(20240601, 20240831), # 时间范围 platformnameSentinel-2, # 卫星平台 producttypeS2MSI2A, # L2A产品 cloudcoverpercentage(0, 10) # 云量0~10% ) # 转成DataFrame方便筛选 df api.to_dataframe(products) print(df[[title, cloudcoverpercentage, size]].head())这段代码的逻辑先建立认证会话然后把AOI的GeoJSON转成WKT格式API只认WKT接着用query方法传过滤条件。producttypeS2MSI2A对应L2Acloudcoverpercentage是云量百分比区间。返回的products是字典用to_dataframe转成表格后就能按需排序筛选。参数说明date格式是(YYYYMMDD, YYYYMMDD)闭区间platformname可选Sentinel-1/2/3/5Pcloudcoverpercentage只对光学卫星有效SAR没有这个字段。如果查询返回空先检查AOI是不是太小、时间范围是不是太窄、产品类型拼写对不对。2.4 下载环节并发、限流与断点续传查询到产品列表后下载才是真正考验工具的地方。sentinelsat的download_all支持并发from sentinelsat import SentinelAPI api SentinelAPI(your_username, your_password, https://apihub.copernicus.eu/apihub) # 下载全部查询结果设置并发数和断点续传 api.download_all( products, directory_path./sentinel_data, max_attempts3, # 单文件最多重试3次 checksumTrue, # 校验MD5 n_concurrent_dl2 # 并发下载数 )max_attempts控制单个文件失败后的重试次数网络不稳时调到3~5。checksumTrue会在下载完成后校验MD5防止文件损坏。n_concurrent_dl是并发数官方API对并发有限制一般设2~4比较稳设太高会被限流甚至临时封禁。断点续传方面sentinelsat本身对单个文件的续传支持有限如果下载大文件比如Sentinel-1的SLC动辄几个GB中途断了重新跑download_all会跳过已完成的但未完成的会从头下。更稳的做法是用api.download()逐个下配合自己的重试逻辑import time for pid, product in products.items(): for attempt in range(3): try: api.download(pid, directory_path./sentinel_data) break except Exception as e: print(f第{attempt1}次失败: {e}) time.sleep(30) # 等30秒再试 else: print(f{product[title]} 三次均失败跳过)这段逻辑对每个产品ID单独下载失败后等30秒重试三次都失败就跳过并记录。比一次性download_all更可控适合网络环境不稳定的情况。3. 批量下载脚本的工程化从能跑到跑得稳3.1 把查询条件外置成配置文件硬编码查询条件在脚本里改一次就要动代码不是好习惯。把条件抽到YAML# config.yaml aoi: ./aoi.geojson date_start: 20240601 date_end: 20240831 platform: Sentinel-2 product_type: S2MSI2A cloud_max: 10 download_dir: ./sentinel_data max_concurrent: 2 max_attempts: 3读取配置的代码import yaml from sentinelsat import SentinelAPI, read_geojson, geojson_to_wkt with open(config.yaml, r, encodingutf-8) as f: cfg yaml.safe_load(f) api SentinelAPI(your_username, your_password, https://apihub.copernicus.eu/apihub) footprint geojson_to_wkt(read_geojson(cfg[aoi])) products api.query( footprint, date(cfg[date_start], cfg[date_end]), platformnamecfg[platform], producttypecfg[product_type], cloudcoverpercentage(0, cfg[cloud_max]) )这样换区域、换时间、换云量阈值都只改YAML脚本本身不动。做时序分析时经常要按季度或月份分批下配置文件一改就能复用。3.2 按时间分片下载避免一次性拉太多一次性查询半年甚至一年的数据返回列表可能几百上千条下载队列太长容易中途出问题。按月份分片from datetime import date, timedelta import calendar def month_ranges(start, end): 把日期范围拆成按月分片 ranges [] cur date(int(start[:4]), int(start[4:6]), 1) end_d date(int(end[:4]), int(end[4:6]), 1) while cur end_d: last_day calendar.monthrange(cur.year, cur.month)[1] ranges.append(( cur.strftime(%Y%m%d), cur.replace(daylast_day).strftime(%Y%m%d) )) # 移到下个月 if cur.month 12: cur cur.replace(yearcur.year1, month1) else: cur cur.replace(monthcur.month1) return ranges for d_start, d_end in month_ranges(cfg[date_start], cfg[date_end]): products api.query(footprint, date(d_start, d_end), platformnamecfg[platform], producttypecfg[product_type], cloudcoverpercentage(0, cfg[cloud_max])) if products: api.download_all(products, directory_pathcfg[download_dir], max_attemptscfg[max_attempts], n_concurrent_dlcfg[max_concurrent]) print(f{d_start}~{d_end} 完成共{len(products)}景)分片的好处每批数量可控失败影响面小进度可见知道下到哪个月了中途断了重跑只需从断的月份开始。3.3 下载日志与去重别重复下同一景同一个区域不同时间查询可能返回重叠的产品。下载前先检查本地已有文件import os import pandas as pd def filter_downloaded(products, download_dir): 过滤掉本地已存在的产品 existing set() for f in os.listdir(download_dir): if f.endswith(.zip): # 文件名通常包含产品ID existing.add(f.replace(.zip, )) df api.to_dataframe(products) # 假设title字段能匹配文件名 df df[~df[title].isin(existing)] return df # 使用 df filter_downloaded(products, cfg[download_dir]) print(f过滤后待下载: {len(df)}景)更严谨的做法是用产品UUID做匹配因为title可能因为命名规则变化而不一致。下载日志建议单独记一个CSV记录产品ID、下载时间、文件大小、MD5方便后续核对。3.4 定时增量下载配合cron做每日更新如果做近实时监测需要每天拉最新数据。写一个增量脚本只查最近N天from datetime import date, timedelta def incremental_download(days_back3): end date.today() start end - timedelta(daysdays_back) products api.query( footprint, date(start.strftime(%Y%m%d), end.strftime(%Y%m%d)), platformnamecfg[platform], producttypecfg[product_type], cloudcoverpercentage(0, cfg[cloud_max]) ) if products: api.download_all(products, directory_pathcfg[download_dir], max_attempts3, n_concurrent_dl2) print(f增量下载完成: {len(products)}景) incremental_download()配合cron每天跑一次# 每天凌晨2点执行 0 2 * * * /usr/bin/python3 /path/to/incremental_download.py /var/log/sentinel_dl.log 21days_back3是留缓冲因为哨兵数据从拍摄到可下载有延迟L2A通常要几小时到一天。设3天能覆盖大部分延迟情况同时避免重复下载太多。4. 避坑与排查下载工具最容易翻车的五个地方4.1 认证失败密码对了也报401现象用户名密码确认没错但API返回401 Unauthorized。原因哥白尼数据空间的认证体系改过版旧版apihub和新版数据空间的token机制不同。另外如果账号长时间未登录网页端可能被临时锁定。解决先确认用的API endpoint和账号体系匹配。旧版用https://apihub.copernicus.eu/apihub新版用https://catalogue.dataspace.copernicus.eu/odata/v1。如果持续401去网页端登录一次激活账号再重试。另外注意密码里如果有特殊字符在代码里要正确转义。4.2 查询返回空条件都对但就是没结果现象AOI、时间、产品类型都设了返回空列表。原因最常见的是AOI的GeoJSON坐标系不对。API期望WGS84经纬度如果AOI是投影坐标比如UTM转成WKT后坐标值巨大查询自然为空。其次是产品类型拼写S2MSI2A和S2MSI1C差一个字符结果完全不同。解决用geopandas检查AOI的CRSimport geopandas as gpd gdf gpd.read_file(aoi.geojson) print(gdf.crs) # 应该是EPSG:4326 if gdf.crs ! EPSG:4326: gdf gdf.to_crs(EPSG:4326) gdf.to_file(aoi_wgs84.geojson, driverGeoJSON)产品类型去官方文档核对别凭记忆写。4.3 下载中途断流大文件下到一半停了现象Sentinel-1的SLC文件几个GB下到70%卡住等很久也不动。原因官方下载服务对单连接有时长限制或者本地网络抖动。sentinelsat默认的超时设置可能不够。解决调大超时并改用逐个下载重试api SentinelAPI(user, pass, https://apihub.copernicus.eu/apihub, timeout60) # 超时调到60秒如果还是断用api.download()逐个下每个文件独立会话断了只影响当前文件。另外检查磁盘空间空间不足时下载会静默失败。4.4 并发过高被限流429 Too Many Requests现象设了n_concurrent_dl8开始很快几分钟后全部报429。原因官方API对并发连接数和请求频率有限制超了会临时封禁IP或账号。解决并发降到2~4请求之间加延迟。如果批量很大分时段跑比如白天查好列表晚上下载。被限流后等15~30分钟再试别频繁重试加重封禁。4.5 磁盘写满下载目录没空间了现象下载突然全部失败报错信息含糊。原因哨兵数据体积大Sentinel-2的L2A一景约500MB~1GBSentinel-1的SLC一景可能4~8GB。下几十景就是几十上百GB。解决下载前检查可用空间脚本里加判断import shutil def check_disk(path, required_gb50): total, used, free shutil.disk_usage(path) free_gb free // (2**30) if free_gb required_gb: raise RuntimeError(f磁盘剩余{free_gb}GB不足{required_gb}GB) print(f磁盘剩余{free_gb}GB充足) check_disk(cfg[download_dir])另外下载完的zip如果不再需要及时解压后删掉zip能省不少空间。5. 进阶技巧用OData API做精细过滤与元数据预取5.1 为什么有时该绕过Python库直接用ODatasentinelsat这类库封装得好但封装也意味着灵活性受限。比如你想按轨道号、相对轨道号、或者特定的处理基线过滤库的query方法可能不支持这些字段。OData API直接暴露了所有可过滤属性适合做精细筛选。OData的查询URL结构https://catalogue.dataspace.copernicus.eu/odata/v1/Products? $filterCollection/Name eq SENTINEL-2 and OData.CSC.Intersects(areageographySRID4326;POLYGON((...))) and ContentDate/Start gt 2024-06-01T00:00:00.000Z and ContentDate/Start lt 2024-08-31T23:59:59.999Z and Attributes/OData.CSC.DoubleAttribute/any(att:att/Name eq cloudCover and att/OData.CSC.DoubleAttribute/Value lt 10) $top100 $orderbyContentDate/Start desc这个查询做了几件事限定集合为SENTINEL-2空间相交判断时间范围云量小于10按时间倒序取前100条。5.2 用Python发OData请求并解析结果import requests import pandas as pd BASE https://catalogue.dataspace.copernicus.eu/odata/v1/Products def odata_query(aoi_wkt, date_start, date_end, cloud_max10, top100): 直接调OData API查询 filter_str ( fCollection/Name eq SENTINEL-2 fand OData.CSC.Intersects(areageographySRID4326;{aoi_wkt}) fand ContentDate/Start gt {date_start}T00:00:00.000Z fand ContentDate/Start lt {date_end}T23:59:59.999Z fand Attributes/OData.CSC.DoubleAttribute/any( fatt:att/Name eq cloudCover fand att/OData.CSC.DoubleAttribute/Value lt {cloud_max}) ) params { $filter: filter_str, $top: top, $orderby: ContentDate/Start desc } r requests.get(BASE, paramsparams, timeout30) r.raise_for_status() return r.json() # 使用 result odata_query(footprint, 2024-06-01, 2024-08-31) df pd.json_normalize(result[value]) print(df[[Name, ContentDate.Start, Online]].head())$filter是OData的过滤语法$top限制返回条数$orderby排序。返回的JSON里value是产品列表每个产品有Id、Name、ContentDate、Online等字段。Online为True表示可以直接下载False可能需要等待归档恢复。5.3 从OData结果直接构造下载链接拿到产品ID后下载链接的构造方式def get_download_url(product_id): 构造下载URL return fhttps://catalogue.dataspace.copernicus.eu/odata/v1/Products({product_id})/$value # 批量下载 import os for _, row in df.iterrows(): pid row[Id] name row[Name] url get_download_url(pid) # 用requests流式下载 with requests.get(url, streamTrue, timeout60) as r: r.raise_for_status() out_path os.path.join(cfg[download_dir], f{name}.zip) with open(out_path, wb) as f: for chunk in r.iter_content(chunk_size8192): f.write(chunk) print(f完成: {name})流式下载streamTrue避免把整个文件读进内存适合大文件。iter_content按块写盘8KB一块是常用值网络好可以调到64KB或128KB。5.4 元数据预取先看属性再决定下不下OData的一个好处是可以在下载前拿到完整元数据包括云量、轨道号、处理基线、文件大小。先拉元数据存成CSV人工或脚本筛选后再下载能省很多流量和时间# 提取关键属性 def extract_attrs(product): attrs {} for a in product.get(Attributes, []): attrs[a[Name]] a.get(Value) return attrs rows [] for p in result[value]: attrs extract_attrs(p) rows.append({ Id: p[Id], Name: p[Name], Start: p[ContentDate][Start], Cloud: attrs.get(cloudCover), Size: p.get(ContentLength), Online: p.get(Online) }) meta_df pd.DataFrame(rows) meta_df.to_csv(metadata.csv, indexFalse) print(f元数据已保存共{len(meta_df)}条)拿到metadata.csv后可以按云量排序、按大小过滤、按时间筛选确定最终下载列表再动手。这个习惯我坚持了很久比盲目下载再删要省心得多。5.5 一个我踩过的坑Online字段的误导有次批量下载脚本跑了一晚上第二天发现大部分文件是0字节。排查后发现OData返回的Online字段为True但实际文件还在归档中下载请求返回的是空内容。后来改成先检查ContentLength为0或异常小的跳过并且下载后校验文件大小和MD5。这个教训让我养成了一个习惯任何批量下载下载完必须做一次完整性校验别信「下载成功」的日志。import hashlib def verify_md5(filepath, expected_md5): 校验文件MD5 h hashlib.md5() with open(filepath, rb) as f: for chunk in iter(lambda: f.read(8192), b): h.update(chunk) return h.hexdigest() expected_md5MD5值可以从OData的Checksum字段拿到下载后逐个比对。对不上就重下别将就。遥感数据一旦损坏后面分析出的结果全是错的而且很难定位是数据问题还是算法问题。希望帮到你。本文还有配套的精品资源点击获取
觉得有用,分享给同行:

为您的企业打造数字门面

稳重轻奢商务风格,端正雅致视觉,长效耐看不易过时。

立即咨询 →