当前位置：首页 > news >正文

python脚本下载ERA5数据详细规范和教程

news 2025/4/30 8:45:12

python脚本下载ERA5数据详细规范和教程

ECMWF官网：

https://cds.climate.copernicus.eu/

在这里插入图片描述

API请求官方教程：

https://cds.climate.copernicus.eu/api-how-to

下载数据准备工作

注册官网账户，登录生成自己的UID和key

在这里插入图片描述

配置并安装CDS API

创建一个.cdsapirc文件其实直接新建文本文档后再改也行,如下配置修改保存到C:\Users\用户下
在这里插入图片描述

安装cdsapi库

pip install cdsapi  #根据自己本地环境安装即可

下载数据规范

1.存储路径规范

/path/{产品要素}/{年}/{月}/{日}

采用分要素下载数据文件

2.文件命名规范

ERA5-{year}{month}{day}_{type_str}.grib  #type_str是数据的variable

数据格式统一采用grib或者nc格式

python下载脚本

logging.basicConfig(filename='download_log_ERA5.log',level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')
def create_folder_if_not_exists(path):if not os.path.exists(path):os.makedirs(path)logging.info(f"创建文件夹: {path}")else:logging.info(f"文件夹已存在: {path}")def is_valid_date(year, month, day):try:request_date = datetime(int(year), int(month), int(day))current_date = datetime.now()latest_era5_date = current_date - timedelta(days=5)if request_date > latest_era5_date:logging.error(f"请求日期{request_date}超过最新可用日期{latest_era5_date}")return Falsereturn Trueexcept ValueError as e:logging.error(f"无效日期: {year}-{month}-{day}, 错误: {e}")return Falsedef down(year, month, day, level, variables, type_str, path, retries=3, delay=60):if not is_valid_date(year, month, day):returnmonth = f"{int(month):02d}"day = f"{int(day):02d}"dataset = "reanalysis-era5-pressure-levels"request = {'product_type': 'reanalysis','variable': variables,'year': year,'month': month,'day': day,'time': ['00:00', '01:00', '02:00', '03:00', '04:00', '05:00','06:00', '07:00', '08:00', '09:00', '10:00', '11:00','12:00', '13:00', '14:00', '15:00', '16:00', '17:00','18:00', '19:00', '20:00', '21:00', '22:00', '23:00'],'pressure_level': level,'area': [90, -180, -90, 180], # 全球'format': 'grib'}client = cdsapi.Client()date_str = f"{year}/{month}/{day}"folder_path = os.path.join(path, date_str)file_name = f"ERA5-{year}{month}{day}_{type_str}.grib"file_path = os.path.join(folder_path, file_name)for attempt in range(retries):try:create_folder_if_not_exists(folder_path)logging.info(f"开始下载: {date_str} {type_str}")client.retrieve(dataset, request, file_path)logging.info(f"下载成功: {file_path}")returnexcept Exception as e:logging.error(f"下载失败: {e}，尝试 {attempt + 1}/{retries}")if attempt < retries - 1:time.sleep(delay)else:logging.error(f"全部重试失败: {date_str}")with open("part.json", encoding='utf-8') as f:content = json.load(f)for c in content:year = c['year']month = c['month']day = c['day']if not is_valid_date(year, month, day):continuefor type_var in c['variableUPAR']:base_path = "/ERA5/"# 处理组合变量情况/风场数据if type_var == 'uv_component_of_wind':variables = ['u_component_of_wind', 'v_component_of_wind']type_str = 'uv_component_of_wind'else:variables = [type_var]type_str = type_vardown(year, month, day, c['level'], variables, type_str, base_path)