批量下载区域平均数据

需求：下载山西省（adcode 140000）2025 年逐日各要素的区域平均值。

历史数据模型：era5_seamless
以 Python 代码为例，分四步完成：

---

什么是区域平均

气象格点数据的分辨率约为 25km，一个区县范围内通常覆盖多个格点。区域平均（area_average: true）表示将该行政区划边界内所有格点的数值取空间平均，输出为一行一个区划的时序数据，而非逐格点数据。

相比逐格点输出，区域平均的优势：

• 数据量大幅减少，下载和处理更高效
• 代表整个行政区的综合气象状况，适合统计分析

区域平均仅支持以行政区划（adcodes）指定范围，不支持自定义经纬度点位或网格范围。

---

第一步：创建批量下载任务

直接指定山西省的 adcode（140000）。行政区划 adcode 可从平台提供的 CSV 文件中查询：省/直辖市使用 province.csv，地级市使用 city.csv，区县使用 district.csv。

与普通点位下载相比，只需将 area_average 设为 true。

import time
import zipfile
import requests

API_KEY = "your_api_key"
BASE_URL = "https://api.mirror-earth.com"

# 山西省 adcode（可从 province.csv 中查询）
adcodes = ["140000"]

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json",
}

payload = {
    "domain": "era5_seamless",
    "hourly": [],
    "daily": ["temperature_2m_mean", "rain_sum"],
    "monthly": [],
    "time_range": {
        "type": "date_range",
        "start": "2025-01-01",
        "end": "2025-12-31",
    },
    "area": {
        "type": "point",
        "adcodes": adcodes,
    },
    "timezone": "Asia/Shanghai",
    "area_average": True,   # 开启区域平均
}

res = requests.post(f"{BASE_URL}/api/tasks/export", json=payload, headers=headers)
res.raise_for_status()
task_id = res.json()["data"]["task_id"]
print(f"任务已创建，task_id: {task_id}")

若需要下载多个省份，将 adcodes 改为包含多个 adcode 的列表即可，如 adcodes = ["140000", "150000"]。

---

第二步：轮询等待任务完成

output_file = None

while True:
    res = requests.get(f"{BASE_URL}/api/tasks/{task_id}/status", headers=headers)
    res.raise_for_status()
    data = res.json()["data"]

    status = data["status"]
    progress = data.get("progress_percent", 0)
    print(f"任务状态：{status}，进度：{progress}%")

    if status == "completed":
        output_file = data["output_file"]
        break
    elif status in ("failed", "cancelled"):
        raise RuntimeError(f"任务异常，状态：{status}")

    time.sleep(10)

---

第三步：下载并解压任务结果

download_url = f"{BASE_URL}/user_downloads/{output_file}"
res = requests.get(download_url, headers=headers, stream=True)
res.raise_for_status()

zip_path = "shanxi_avg_2025.zip"
with open(zip_path, "wb") as f:
    for chunk in res.iter_content(chunk_size=8192):
        f.write(chunk)

with zipfile.ZipFile(zip_path, "r") as z:
    z.extractall("shanxi_avg_2025")

print("下载完成，数据已解压到 shanxi_avg_2025/")

---

完整代码

import time
import zipfile
import requests

API_KEY = "your_api_key"
BASE_URL = "https://api.mirror-earth.com"

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json",
}

# ── 1. 指定山西省 adcode（可从 province.csv 中查询）────────────
adcodes = ["140000"]

# ── 2. 创建任务（area_average=True）────────────────────────────
payload = {
    "domain": "era5_seamless",
    "hourly": [],
    "daily": ["temperature_2m_mean", "rain_sum"],
    "monthly": [],
    "time_range": {
        "type": "date_range",
        "start": "2025-01-01",
        "end": "2025-12-31",
    },
    "area": {
        "type": "point",
        "adcodes": adcodes,
    },
    "timezone": "Asia/Shanghai",
    "area_average": True,
}

res = requests.post(f"{BASE_URL}/api/tasks/export", json=payload, headers=headers)
res.raise_for_status()
task_id = res.json()["data"]["task_id"]
print(f"任务已创建，task_id: {task_id}")

# ── 3. 轮询任务进度 ─────────────────────────────────────────────
output_file = None
while True:
    res = requests.get(f"{BASE_URL}/api/tasks/{task_id}/status", headers=headers)
    res.raise_for_status()
    data = res.json()["data"]
    status = data["status"]
    progress = data.get("progress_percent", 0)
    print(f"任务状态：{status}，进度：{progress}%")

    if status == "completed":
        output_file = data["output_file"]
        break
    elif status in ("failed", "cancelled"):
        raise RuntimeError(f"任务异常，状态：{status}")
    time.sleep(10)

# ── 4. 下载并解压结果 ───────────────────────────────────────────
download_url = f"{BASE_URL}/user_downloads/{output_file}"
res = requests.get(download_url, headers=headers, stream=True)
res.raise_for_status()

zip_path = "shanxi_avg_2025.zip"
with open(zip_path, "wb") as f:
    for chunk in res.iter_content(chunk_size=8192):
        f.write(chunk)

with zipfile.ZipFile(zip_path, "r") as z:
    z.extractall("shanxi_avg_2025")

print("下载完成，数据已解压到 shanxi_avg_2025/")

模型列表

只有这些模型支持区域平均数据

预报模型

历史观测

---

要素列表

逐小时要素

逐日要素

逐月要素

逐月要素名称与逐日完全相同，聚合方式一致。