简介面向气象科研、气候建模与数据分析人员的CFSv2数据下载Python脚本工具适合希望自动获取长期天气预测数据、减少手动下载重复工作的用户。气候预报系统第二版CFSv2由美国国家环境预报中心NCEP开发该脚本通过HTTP请求访问大学大气研究联盟UCAR数据服务以UCAR账户完成身份认证并按用户指定的条件获取对应时间范围和参数的数据。压缩包共1个文件仅包含一个.py脚本大小约1KB轻量无额外依赖可直接放入已有项目或定时更新流程中。已有600人学习/下载。对具备基础Python知识的气象学习者而言这份代码既是可用的下载工具也展示了请求认证、分块下载、失败重试等数据采集环节的实现思路便于理解并复用自动化数据获取逻辑脚本代码量虽小但覆盖了实用的网络请求与认证处理方式适合气候预测、农业气象、水文模拟等需要定期更新CFSv2数据的研究场景。1. CFSv2 数据下载为什么需要专门脚本气象圈子里下载 NCEP 的 CFSv2 再分析数据最磨人的不是数据本身而是官方数据服务器那套基于 HTTPS 的认证和目录跳转逻辑。你手动点网页下载一两个文件还行但要补十年逐小时数据手点的效率就太低了而且文件一旦下到一半断了重来一遍非常浪费时间。download_ds094.0.py这个脚本就是用 Python 把这一整套流程固化下来构造 URL、带 UCAR 账号密码做认证、分块下载、处理网络异常最后把 GRIB2 文件完整落盘。适合需要批量拉取 ds094.0 数据集做气候模拟、季节预测校验或者驱动区域模式的工程师。对 Python 版本的要求不高3.6 以上就能跑唯一前置条件是得有 UCAR 账号然后把脚本里*替换成真实凭据即可开始。2. ds094.0 数据集结构与 UCAR 认证的底层逻辑2.1 CFSv2 再分析产品的目录组织CFSv2 在 NCEP 的数据档案里以ds094.0为数据集标识全称是 NCEP Climate Forecast System Version 2 Monthly Diurnal and 6-hourly Products。它和那种按下完就算完的普通气象数据不一样目录层级非常多典型的文件路径长这样https://www.ncei.noaa.gov/data/climate-forecast-system/version2/forecast/products/2010/20100101/grib2/06/flxf2010010106.01.2010010106.grb2这个路径的规律是年份2010、起报日期20100101、预报时次06、文件类型flxf通量文件或pgbf气压层文件最后一段里的两个日期分别表示初始时间和有效时间。ds094.0 的目录里时间粒度有 6 小时、日平均、月平均三种按你的研究需求选。搞不清目录组织直接拿网上七拼八凑的 URL 去下经常会碰到 404 或者下到重复文件。2.2 UCAR 数据接口的认证方式UCAR 对 ds094.0 的访问控制不是简单的 IP 白名单而是走标准 HTTP Basic Authentication。也就是说每次请求都会在 HTTP 头里带上一串Authorization: Basic base64(username:password)。Python 的requests库处理这个非常简单但有个坑如果你直接往 URL 里写账号密码例如https://user:passdomain/path部分代理服务器会剥掉认证头导致返回 401。正确做法是先创建Session对象再设置认证元组这样重定向时requests会自动把认证头加到新的请求上。认证失败时服务器返回的状态码通常是 401而且次数多了还可能触发风控。所以脚本里要有失败退避的机制连续 5 次 401 就应该停下来检查账号是否写错、是否该续费或重置密码。2.3 脚本里“星号替换”到底替换了什么原始工程里有个占位逻辑download_ds094.0.py里会有类似username ***和password ***的两行。你直接把***替换成 UCAR 账号和密码就能用。这里要留个心眼账号密码是敏感信息不要把写有真实凭据的脚本提交到公开仓库。我一般会改成从环境变量读取import os username os.environ.get(UCAR_USER, ***) password os.environ.get(UCAR_PASS, ***)这样脚本自身不包含任何明文敏感信息CI/CD 里通过 secrets 注入本地跑的时候用export UCAR_USERxxx设置即可。替换星号这个动作看起来简单但它实际上决定了后面所有请求能否通过认证所以最先排查的就是这里。注意不要换行、不要有多余空格密码里有特殊字符时最好先复制到文本编辑器里确认。3. download_ds094.0.py 的核心实现拆解3.1 请求构造与认证头设置脚本的基础结构是一个带认证的requests.Session示例核心部分如下import requests from requests.adapters import HTTPAdapter session requests.Session() session.auth (username, password) adapter HTTPAdapter(max_retries3, pool_connections10, pool_maxsize10) session.mount(https://, adapter) url https://www.ncei.noaa.gov/data/climate-forecast-system/version2/forecast/products/2010/20100101/grib2/06/flxf2010010106.01.2010010106.grb2 r session.get(url, streamTrue, timeout(10, 120)) r.raise_for_status()这里session.auth接收一个两元组requests会自动做 Base64 编码并放到请求头里。HTTPAdapter的max_retries3是给连接层用的重试针对的是连接建立失败或连接池无可用连接不是所有 HTTP 错误都适用。timeout(10, 120)里的 10 是连接超时120 是读超时。气象数据文件动辄几百 MB读超时不能设得太小否则长时间没有新数据块时就容易被误杀。一个常见的误用是直接把streamTrue去掉然后丢给r.content这会一次性把整个文件读进内存大文件下几百 MB 就直接把开发机的内存打爆。脚本里只要出现r.text或r.content去接 grib2 文件的地方都是需要改掉的。3.2 分块下载与断点续传ds094.0 的文件按大小从几十 MB 到 1 GB 不等分块写入用iter_content是标准姿势chunk_size 1024 * 1024 # 1MB local_file flxf2010010106.01.2010010106.grb2 with open(local_file, wb) as f: for chunk in r.iter_content(chunk_sizechunk_size): if chunk: f.write(chunk) f.flush()iter_content按指定字节数从响应流里读取返回的块不保证刚好等于chunk_size所以代码里判断if chunk是必要的。f.flush()可以把缓冲区写入磁盘但不需要每块都调用否则频繁刷盘反而降低吞吐。我一般每写 50 个块调用一次flush或者干脆用shutil.copyfileobj让 Python 内部管理缓冲。遇到网络中断时最简单的断点续传是记录已下载字节数然后利用 HTTP 的Range头继续headers {Range: fbytes{downloaded}-} r session.get(url, headersheaders, streamTrue, timeout(10, 120)) with open(local_file, ab) as f: for chunk in r.iter_content(chunk_sizechunk_size): f.write(chunk)注意服务器对Range请求返回的是206 Partial Content如果返回 200说明服务器忽略了这个头那就要把文件清空重下。写脚本时最好判断一下r.status_code。3.3 重试与异常处理机制网络层重试之外的 HTTP 层重试requests库本身不提供需要自己封装。一个实用的策略是对 500、502、503、504 做指数退避重试对 401、403、404 直接抛出异常。原因是前一类是服务器过载或临时故障后三类是账号问题或路径错误重试多少次都是同样的结果。import time from requests.exceptions import RequestException max_attempts 5 retry_status_codes {500, 502, 503, 504} for attempt in range(1, max_attempts 1): try: r session.get(url, streamTrue, timeout(10, 180)) if r.status_code in retry_status_codes: raise RequestException(fHTTP {r.status_code}) r.raise_for_status() break except (RequestException, ConnectionError) as e: if attempt max_attempts: raise RuntimeError(f下载失败: {url}) from e sleep_seconds 2 ** attempt print(f第 {attempt} 次重试, 等待 {sleep_seconds} 秒...) time.sleep(sleep_seconds)这里的网络搜索热词大多指向python环境配置和基础入门但实际跑这类下载脚本时requests和urllib3的版本不兼容也会造成异常。ConnectionError在urllib3和requests里都有定义建议捕获时写成requests.exceptions.ConnectionError避免引用混乱。另外后台被 CtrlC 中断后残留的.grb2半文件会在下次运行时造成校验错误所以重试逻辑启动前最好检查本地文件是否完整。3.4 文件名与时间序列的动态拼接原始脚本里文件名是硬编码的但实际用的时候一定得动态生成。CFSv2 的文件名有严格的范式flxf 初始时间YYYYMMDDHH.01. 有效时间YYYYMMDDHH.grb2。不同变量的前缀不同pgbf是气压层。动态拼接的示例from datetime import datetime, timedelta date_str 20100101 cycle 06 valid_hour 6 initial_time datetime.strptime(f{date_str}{cycle}, %Y%m%d%H) valid_time initial_time timedelta(hoursvalid_hour) prefix flxf filename f{prefix}{initial_time.strftime(%Y%m%d%H)}.01.{valid_time.strftime(%Y%m%d%H)}.grb2 url fhttps://www.ncei.noaa.gov/data/climate-forecast-system/version2/forecast/products/{date_str[:4]}/{date_str}/grib2/{cycle}/{filename}这个拼接逻辑要对应服务器实际目录比如products下面第一层是年份第二层是起报日期第三层写死grib2第四层是起报时次。不同时间段的数据比如 CFSR 与 CFSv2 过渡期可能需要切换到cfsr目录这里的 URL 构造就要跟着变。遇到 404 时重点看valid_hour是否超出该文件的预报时长范围CFSv2 的 45 天预报和 9 个月预报的文件结构不同后者的有效时间会跨月目录也不一样。4. 把脚本改造成你自己要的数据4.1 修改变量、层级与时间步长ds094.0 里同一时刻的文件不止一个按变量种类区分。近地面的通量文件前缀是flxf高空气压层文件前缀是pgbf还有flxh、pgbh这种逐小时诊断文件。你需要在脚本里做一个变量映射表前缀含义典型用例flxf近地面通量6 小时平均地表温度、降水、辐射flxh近地面通量逐小时日变化分析pgbf气压层预报字段位势高度、温度、风场pgbh气压层逐小时垂直速度、湿度拿到一个你需要的具体变量得先看 GRIB2 里对应的缩写名比如 2m 温度是TMP:2 m above ground降水是PRATE:surface。脚本只管下载不做变量筛选所以你在 URL 层选定文件类型即可。时间步长方面6 小时产品一天只有 4 个时次逐小时产品一天 24 个时次循环时要避免用range(0, 24, 6)去套逐小时文件否则会下重。4.2 批量循环下载多年数据改造重活在于对日期循环。下面是一个多年逐日下载的骨架限制并发为 2 个下载任务防止被服务器拒绝连接from concurrent.futures import ThreadPoolExecutor def download_one_day(date_str, cycle_list): for cycle in cycle_list: filename build_filename(date_str, cycle, prefixflxf, valid_hour6) if already_downloaded(filename): continue download_file(filename) return date_str date_list [d.strftime(%Y%m%d) for d in pd.date_range(2015-01-01, 2015-12-31, freqD)] cycle_list [00, 06, 12, 18] with ThreadPoolExecutor(max_workers2) as executor: list(executor.map(lambda dt: download_one_day(dt, cycle_list), date_list))注意pd.date_range需要 pandas如果不想引入重依赖直接用datetime的timedelta也可以。并发合成的脚本ThreadPoolExecutor里每个线程共用同一个session需要确认你的requests版本支持线程安全的 Session。如果不确定就在每个线程里新建 Session连接池开销可以接受。批量下载失败时别把整个任务重头再来写一个失败清单文件下次启动先读清单只补缺漏项。4.3 数据校验与本地存储规范下载完成后别急着做分析先验证文件的完整性与可用性。GRIB2 不是文本格式不能用os.path.getsize大于 0 来判断成功。最实用的初级校验是文件大小是否大于阈值更严格的是用wgrib2解析wgrib2 flxf2010010106.01.2010010106.grb2 /dev/nullwgrib2会先扫目录区再解析数据段如果文件在传输过程中损坏它会报出unexpected end或是 CRC 错误。没有wgrib2的环境可以用 Python 读文件头判断前四个字节是否为GRIB。目录结构我习惯这样放data/ cfsv2/ 2010/ 20100101/ flxf2010010106.01.2010010106.grb2日期目录和起报时次分开后续用xarray拼多文件时可以直接用open_mfdataset配合combine_by_coords不会因为平铺式目录把文件顺序搞错。5. 下载完成后最值得做的三件事5.1 用 xarray 快速检查 GRIB2 完整度不少人的习惯是下载完直接把文件扔给模式等到跑模式时才报错。更稳妥的做法是第一时间读进 xarray 做烟雾测试import xarray as xr ds xr.open_dataset(flxf2010010106.01.2010010106.grb2, enginecfgrib) print(ds.variables)cfgrib引擎会校验 GRIB2 消息中的迭代号文件缺块时这里直接抛异常。这一段代码不能代替wgrib2的内容校验但能确认文件在气象字段层面可读。如果xarray没有安装在跑脚本前需要先补上pip install xarray cfgrib ecmwf_codes。5.2 与 NCEP 官方校验和做核对部分 ds094.0 目录提供了.md5文件文件名和 grib2 文件一一对应。下载完成后写一个小函数做 CRC 比对import hashlib def md5_checksum(file_path, block_size8192): md5 hashlib.md5() with open(file_path, rb) as f: while chunk : f.read(block_size): md5.update(chunk) return md5.hexdigest() expected 7d3f2b... # 从服务器 .md5 文件读取 assert md5_checksum(local_file) expected, 校验和不匹配:海象运算符在 Python 3.8 才有如果你还在用老版本换成普通的while True: chunk f.read(...)即可。校验失败的删除重下别做保留因为损坏的 GRIB2 后续会反复报错而且很难定位是哪个文件搞得用户头疼。5.3 定时任务的正确姿势气象数据的发布有固定更新节奏CFSv2 的预报文件每天滚动更新你可以用系统 cron 或 Windows 任务计划定期执行脚本。但要注意NCEI 的归档服务器在北京时间比美国东部时间慢一个工作日所以定时触发时间要避开对方夜间维护窗口。另外脚本里一定要加上锁机制防止上一次还没下完、下一次任务又启动了。简单做法是用fcntl在 Linux 下锁住 pid 文件import fcntl with open(download.lock, w) as lockfile: fcntl.flock(lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB) download_all()LOCK_NB的意思是非阻塞获取锁拿不到锁说明有另一实例在跑直接退出。Windows 下没有fcntl可以用msvcrt.locking或依赖filelock库。定时任务里建议每次只下载增量文件把上次成功记录写成 JSON 存到本地脚本启动时和新任务比对避免重复下载。这一步做扎实后面做预测检验、气候统计时拿到的数据总是干净且完整的分析结果才站得住脚。本文还有配套的精品资源点击获取
` 完整指南：获取自定义篮子 Marquee 产品页链接)