OneScience commited on
Commit
9333192
·
verified ·
1 Parent(s): 1cc87c4

Upload folder using huggingface_hub

Browse files
README.md ADDED
@@ -0,0 +1,140 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ ---
2
+ license: apache-2.0
3
+ tasks:
4
+ - weather-forecast
5
+ frameworks:
6
+ - pytorch
7
+ language:
8
+ - en
9
+ - zh
10
+ tags:
11
+ - OneScience
12
+ - Earth Science
13
+ - Weather Forecast
14
+ - Short-to-Medium-Range Weather Forecast
15
+ - ERA5
16
+ datasets:
17
+ - OneScience/ERA5
18
+ ---
19
+ <p align="center">
20
+ <strong>
21
+ <span style="font-size: 30px;">Pangu-Weather</span>
22
+ </strong>
23
+ </p>
24
+
25
+ # Model Introduction
26
+
27
+ Pangu-Weather is a global medium-range weather forecast model proposed by Huawei Cloud, capable of rapidly predicting surface variables and multi-pressure-level upper-air variables.
28
+
29
+ Paper: Accurate medium-range global weather forecasting with 3D neural networks
30
+ https://www.nature.com/articles/s41586-023-06185-3
31
+
32
+ # Model Description
33
+
34
+ Pangu-Weather is based on a 3D Earth-Specific Transformer architecture, trained on ERA5 data, and designed for short-to-medium-range weather forecasting.
35
+
36
+ # Use Cases
37
+
38
+ | Scenario | Description |
39
+ | :---: | :--- |
40
+ | Weather Forecast Training | Train Pangu-Weather using ERA5 HDF5 data |
41
+ | Local Quick Validation | Use synthetic data to verify data loading, model training & inference, and inference result visualization. |
42
+ | ModelScope / OneCode Execution | Download as a standalone model package, install dependencies, and run scripts directly. |
43
+ | Multi-GPU Training | Launch multi-process training via `torchrun`. |
44
+
45
+ # Usage Guide
46
+
47
+ ## 1. OneCode Usage
48
+
49
+ Experience intelligent one-click AI4S programming through the OneCode online environment:
50
+
51
+ [Click to Experience Intelligent One-Click AI4S Programming](https://web-2069360198568017922-iaaj.ksai.scnet.cn:58043/home)
52
+
53
+ ## 2. Manual Installation and Usage
54
+
55
+ **Hardware Requirements**
56
+
57
+ - A GPU or DCU is recommended.
58
+ - CPU can be used for import and small-scale connectivity verification; full training and inference will be slow.
59
+ - DCU users must install DTK in advance. DTK 25.04.2 or above, or the OneScience recommended version matching your cluster, is recommended.
60
+
61
+ ### Download the Model Package
62
+
63
+ ```bash
64
+ modelscope download --model OneScience/Pangu-Weather --local_dir ./pangu_weather
65
+ cd pangu_weather
66
+ ```
67
+
68
+ ### Install the Runtime Environment
69
+
70
+ **DCU Environment**
71
+
72
+ ```bash
73
+ # Please activate DTK and CONDA first
74
+ conda create -n onescience311 python=3.11 -y
75
+ conda activate onescience311
76
+ # uv installation is supported
77
+ pip install onescience[earth-dcu] -i http://mirrors.onescience.ai:3141/pypi/simple/ --trusted-host mirrors.onescience.ai
78
+ ```
79
+
80
+ **GPU Environment**
81
+ ```bash
82
+ # Please activate CONDA first
83
+ conda create -n onescience311 python=3.11 -y libstdcxx-ng=12 libgcc-ng=12 gcc_linux-64=12 gxx_linux-64=12
84
+ conda activate onescience311
85
+ # uv installation is supported
86
+ pip install onescience[earth-gpu] -i http://mirrors.onescience.ai:3141/pypi/simple/ --trusted-host mirrors.onescience.ai
87
+ ```
88
+
89
+ ### Training Data Introduction
90
+
91
+ The OneScience community provides ERA5 data for training (due to file size limits, the current repository contains a slice of the full dataset). Users can download it with the command below and confirm that the data path in `conf/config.yaml` is set correctly:
92
+
93
+ ```bash
94
+ modelscope download --dataset OneScience/ERA5 --local_dir ./data
95
+ ```
96
+
97
+ ### Training
98
+
99
+ Single GPU:
100
+
101
+ ```bash
102
+ python scripts/train.py
103
+ ```
104
+
105
+ Multi-GPU:
106
+
107
+ ```bash
108
+ torchrun --nproc_per_node=8 --nnodes=1 --rdzv_id=1000 --rdzv_backend=c10d --max_restarts=0 --master_addr="localhost" --master_port=29500 scripts/train.py
109
+ ```
110
+
111
+ Training will save `model_bak.pth` under `data/checkpoints/`.
112
+
113
+ ### Training Weights
114
+
115
+ This repository provides weights trained on ERA5 data from 1979 to 2025 in the `weights/` folder. At the 6-hour forecast lead time, these weights outperform the official open-source ONNX weights.
116
+
117
+ ### Inference
118
+
119
+ ```bash
120
+ python scripts/inference.py
121
+ ```
122
+
123
+ Inference results will be saved to `result/output/`.
124
+
125
+ ### Evaluation and Visualization
126
+
127
+ ```bash
128
+ python scripts/result.py
129
+ ```
130
+
131
+ # OneScience Official Information
132
+
133
+ | Platform | OneScience Main Repository | Skills Repository |
134
+ | --- | --- | --- |
135
+ | Gitee | https://gitee.com/onescience-ai/onescience | https://gitee.com/onescience-ai/oneskills |
136
+ | GitHub | https://github.com/onescience-ai/OneScience | https://github.com/onescience-ai/oneskills |
137
+
138
+ # Citation & License
139
+
140
+ - This repository is a reproduction of the original Pangu-Weather paper.
conf/config.yaml ADDED
@@ -0,0 +1,130 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ model:
2
+
3
+ start_epoch: 0
4
+ max_epoch: 1
5
+ patch_size: [2, 4, 4]
6
+ embed_dim: 192
7
+ num_heads: [6, 12, 12, 6]
8
+ window_size: [2, 6, 12]
9
+ patience: 50
10
+ checkpoint_dir: "./data/checkpoints"
11
+
12
+ datapipe:
13
+ name: "ERA5"
14
+ task: "weather_forecasting"
15
+
16
+ # dataset设定
17
+ dataset:
18
+ type: "hdf5"
19
+ data_dir: './data/'
20
+
21
+ train_time: [1979]
22
+ val_time: [2021]
23
+ test_time: [2023]
24
+
25
+ img_size: [721, 1440]
26
+ verbose: true
27
+ cache: false
28
+ # 气象变量
29
+ channels: ['mean_sea_level_pressure', '10m_u_component_of_wind', '10m_v_component_of_wind', '2m_temperature',
30
+
31
+ 'geopotential_1000', 'geopotential_925', 'geopotential_850', 'geopotential_700', 'geopotential_600', 'geopotential_500',
32
+ 'geopotential_400', 'geopotential_300', 'geopotential_250', 'geopotential_200', 'geopotential_150', 'geopotential_100', 'geopotential_50',
33
+
34
+ 'specific_humidity_1000', 'specific_humidity_925', 'specific_humidity_850', 'specific_humidity_700', 'specific_humidity_600',
35
+ 'specific_humidity_500', 'specific_humidity_400', 'specific_humidity_300', 'specific_humidity_250', 'specific_humidity_200',
36
+ 'specific_humidity_150', 'specific_humidity_100', 'specific_humidity_50',
37
+
38
+ 'temperature_1000', 'temperature_925', 'temperature_850', 'temperature_700', 'temperature_600', 'temperature_500', 'temperature_400',
39
+ 'temperature_300', 'temperature_250', 'temperature_200','temperature_150', 'temperature_100', 'temperature_50',
40
+
41
+ 'u_component_of_wind_1000', 'u_component_of_wind_925', 'u_component_of_wind_850', 'u_component_of_wind_700', 'u_component_of_wind_600',
42
+ 'u_component_of_wind_500', 'u_component_of_wind_400', 'u_component_of_wind_300', 'u_component_of_wind_250', 'u_component_of_wind_200',
43
+ 'u_component_of_wind_150', 'u_component_of_wind_100', 'u_component_of_wind_50',
44
+
45
+ 'v_component_of_wind_1000', 'v_component_of_wind_925', 'v_component_of_wind_850', 'v_component_of_wind_700', 'v_component_of_wind_600',
46
+ 'v_component_of_wind_500', 'v_component_of_wind_400', 'v_component_of_wind_300', 'v_component_of_wind_250', 'v_component_of_wind_200',
47
+ 'v_component_of_wind_150', 'v_component_of_wind_100', 'v_component_of_wind_50'
48
+ ]
49
+ weights: [ # ---- 4 surface variables ----
50
+ 1.50, # mean_sea_level_pressure
51
+ 0.77, # 10m_u_component_of_wind
52
+ 0.66, # 10m_v_component_of_wind
53
+ 3.00, # 2m_temperature
54
+ # ---- 13 geopotential_* (Z) ----
55
+ 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00, 3.00,
56
+ # ---- 13 specific_humidity_* (Q) ----
57
+ 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60, 0.60,
58
+ # ---- 13 temperature_* (T) ----
59
+ 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50, 1.50,
60
+ # ---- 13 u_component_of_wind_* (U) ----
61
+ 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77, 0.77,
62
+ # ---- 13 v_component_of_wind_* (V) ----
63
+ 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54, 0.54,
64
+ ]
65
+ variables:
66
+ - "u10" # 10m U wind component
67
+ - "v10" # 10m V wind component
68
+ - "t2m" # 2m temperature
69
+ - "msl" # Mean sea level pressure
70
+ - "z500" # Geopotential at 500 hPa
71
+ - "t850" # Temperature at 850 hPa
72
+
73
+ # 时间配置
74
+ time_range: ["2000-01-01", "2020-12-31"]
75
+ time_steps: 1
76
+ time_res: 6
77
+
78
+ # 空间配置
79
+ spatial_resolution: [0.25, 0.25]
80
+
81
+ # 采样配置
82
+ num_samples: -1 # -1 表示使用全部数据
83
+ shuffle: true
84
+ random_seed: 42
85
+
86
+ # 领域特定配置
87
+ extra:
88
+ levels: [500, 850, 1000]
89
+ lat_range: [-90, 90]
90
+ lon_range: [0, 360]
91
+
92
+ # 数据转换配置
93
+ transforms:
94
+ - type: "Normalize"
95
+ params:
96
+ mean: [0.0, 0.0, 288.0, 101325.0, 50000.0, 270.0]
97
+ std: [5.0, 5.0, 15.0, 1000.0, 5000.0, 10.0]
98
+ keys: ["input", "target"]
99
+
100
+ - type: "ToTensor"
101
+ params:
102
+ keys: null # null表示转换所有numpy数组
103
+
104
+ # 其他配置
105
+
106
+ # DataLoader配置
107
+ dataloader:
108
+ mask_dtype: "float32"
109
+ batch_size: 1
110
+ num_workers: 1
111
+ pin_memory: true
112
+ drop_last: true
113
+ shuffle: false # 使用sampler时设为false
114
+ prefetch_factor: 2
115
+ persistent_workers: true
116
+
117
+ # 分布式配置
118
+ distributed:
119
+ enabled: true
120
+ sampler: "DistributedSampler"
121
+ rank: 0
122
+ world_size: 4
123
+ shuffle: true
124
+ seed: 42
125
+ drop_last: true
126
+
127
+
128
+
129
+
130
+
configuration.json ADDED
@@ -0,0 +1 @@
 
 
1
+ {"framework":"Pytorch","task":"other"}
model/pangu.py ADDED
@@ -0,0 +1,218 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import math
2
+ import torch
3
+ import numpy as np
4
+
5
+ from torch import nn
6
+ from dataclasses import dataclass
7
+ from onescience.models.meta import ModelMetaData
8
+
9
+ from onescience.modules.embedding.panguembedding import PanguEmbedding
10
+ from onescience.modules.fuser.pangufuser import PanguFuser
11
+ from onescience.modules.recovery.pangupatchrecovery import PanguPatchRecovery
12
+ from onescience.modules.sample.pangudownsample import PanguDownSample
13
+ from onescience.modules.sample.panguupsample import PanguUpSample
14
+
15
+ @dataclass
16
+ class MetaData(ModelMetaData):
17
+ name: str = "Pangu"
18
+ # Optimization
19
+ jit: bool = False # ONNX Ops Conflict
20
+ cuda_graphs: bool = True
21
+ amp: bool = True
22
+ # Inference
23
+ onnx_cpu: bool = False # No FFT op on CPU
24
+ onnx_gpu: bool = True
25
+ onnx_runtime: bool = True
26
+ # Physics informed
27
+ var_dim: int = 1
28
+ func_torch: bool = False
29
+ auto_grad: bool = False
30
+
31
+
32
+ class Pangu(nn.Module):
33
+ """
34
+ Pangu-Weather 的主模型实现。
35
+
36
+ 该模型使用以下组件完成编码、主干特征提取与输出恢复:
37
+
38
+ - `OneEmbedding(style="PanguEmbedding")`
39
+ - 将 surface 分支与 upper-air 分支分别映射到 patch 特征空间
40
+ - `OneFuser(style="PanguFuser")`
41
+ - 在统一的三维 token 网格上完成主干特征融合
42
+ - `OneSample(style="PanguDownSample" / "PanguUpSample")`
43
+ - 在主干中完成空间尺度变换
44
+ - `OneRecovery(style="PanguPatchRecovery")`
45
+ - 将 patch 级特征恢复为目标物理场
46
+
47
+ 在当前实现中:
48
+
49
+ - 输入包含 7 个 surface 通道,其中 4 个是待预测地表变量,3 个是静态掩码
50
+ - 输入包含 5 个 upper-air 变量,每个变量对应 13 个气压层
51
+ - surface patch token 会先在 `PressureLevels` 维补成长度为 1
52
+ - 然后与 upper-air patch token 沿 `PressureLevels` 维拼接
53
+ - 之后统一送入三维 `PanguFuser` 主干
54
+
55
+ Reference:
56
+ - `Pangu-Weather: A 3D High-Resolution Model for Fast and Accurate Global Weather Forecast`
57
+ - https://arxiv.org/abs/2211.02556
58
+
59
+ Args:
60
+ img_size (tuple[int, int]):
61
+ 输入空间尺寸 `(Height, Width)`。
62
+ patch_size (tuple[int, int, int]):
63
+ patch 切分尺寸 `(PatchPressureLevels, PatchHeight, PatchWidth)`。
64
+ embed_dim (int):
65
+ patch embedding 后的特征维度。
66
+ num_heads (tuple[int, int, int, int]):
67
+ 四个主干阶段对应的注意力头数。
68
+ window_size (tuple[int, int, int]):
69
+ 三维窗口大小 `(PressureLevelsWindow, HeightWindow, WidthWindow)`。
70
+ """
71
+
72
+ def __init__(
73
+ self,
74
+ img_size=(721, 1440),
75
+ patch_size=(2, 4, 4),
76
+ embed_dim=192,
77
+ num_heads=(6, 12, 12, 6),
78
+ window_size=(2, 6, 12),
79
+ ):
80
+ super().__init__()
81
+ drop_path = np.linspace(0, 0.2, 8).tolist()
82
+ # Surface input contains 4 predicted variables and 3 static masks:
83
+ # topography mask, land-sea mask, and soil type mask.
84
+
85
+ self.patchembed2d = PanguEmbedding(
86
+ img_size=img_size,
87
+ patch_size=patch_size[1:],
88
+ Variables=7,
89
+ embed_dim=embed_dim,
90
+ )
91
+ self.patchembed3d = PanguEmbedding(
92
+ img_size=(13, *img_size),
93
+ patch_size=patch_size,
94
+ Variables=5,
95
+ embed_dim=embed_dim,
96
+ )
97
+
98
+ patched_input_shape = (
99
+ 8,
100
+ math.ceil(img_size[0] / patch_size[1]),
101
+ math.ceil(img_size[1] / patch_size[2]),
102
+ )
103
+
104
+ self.layer1 = PanguFuser(
105
+ dim=embed_dim,
106
+ input_resolution=patched_input_shape,
107
+ depth=2,
108
+ num_heads=num_heads[0],
109
+ window_size=window_size,
110
+ drop_path=drop_path[:2],
111
+ )
112
+
113
+ patched_downsampled_shape = (
114
+ 8,
115
+ math.ceil(patched_input_shape[1] / 2),
116
+ math.ceil(patched_input_shape[2] / 2),
117
+ )
118
+
119
+ self.downsample = PanguDownSample(
120
+ in_dim=embed_dim,
121
+ input_resolution=patched_input_shape,
122
+ output_resolution=patched_downsampled_shape,
123
+ )
124
+ self.layer2 = PanguFuser(
125
+ dim=embed_dim * 2,
126
+ input_resolution=patched_downsampled_shape,
127
+ depth=6,
128
+ num_heads=num_heads[1],
129
+ window_size=window_size,
130
+ drop_path=drop_path[2:],
131
+ )
132
+ self.layer3 = PanguFuser(
133
+ dim=embed_dim * 2,
134
+ input_resolution=patched_downsampled_shape,
135
+ depth=6,
136
+ num_heads=num_heads[2],
137
+ window_size=window_size,
138
+ drop_path=drop_path[2:],
139
+ )
140
+ self.upsample = PanguUpSample(
141
+ in_dim=embed_dim * 2,
142
+ out_dim=embed_dim,
143
+ input_resolution=patched_downsampled_shape,
144
+ output_resolution=patched_input_shape,
145
+ )
146
+ self.layer4 = PanguFuser(
147
+ dim=embed_dim,
148
+ input_resolution=patched_input_shape,
149
+ depth=2,
150
+ num_heads=num_heads[3],
151
+ window_size=window_size,
152
+ drop_path=drop_path[:2],
153
+ )
154
+
155
+ # The recovered surface output contains only the 4 prognostic surface variables.
156
+ # Static masks are input-only features and are not part of the prediction target.
157
+ self.patchrecovery2d = PanguPatchRecovery(
158
+ img_size=(721, 1440),
159
+ patch_size=(4, 4),
160
+ in_chans=embed_dim * 2,
161
+ out_chans=4,
162
+ )
163
+ self.patchrecovery3d = PanguPatchRecovery(
164
+ img_size=(13, 721, 1440),
165
+ patch_size=(2, 4, 4),
166
+ in_chans=embed_dim * 2,
167
+ out_chans=5,
168
+ )
169
+
170
+ def forward(self, x):
171
+ """
172
+ Args:
173
+ x (torch.Tensor):
174
+ Input tensor with shape `(Batch, 4 + 3 + 5 * 13, Height, Width)`.
175
+
176
+ Channel layout:
177
+ - first 4 channels: prognostic surface variables
178
+ - next 3 channels: static masks
179
+ - remaining `5 * 13` channels: upper-air variables flattened over pressure levels
180
+
181
+ Returns:
182
+ tuple[torch.Tensor, torch.Tensor]:
183
+ - surface output:
184
+ `(Batch, 4, Height, Width)`
185
+ - upper-air output:
186
+ `(Batch, 5, 13, Height, Width)`
187
+ """
188
+ SurfaceInput = x[:, :7, :, :]
189
+ UpperAirInput = x[:, 7:, :, :].reshape(x.shape[0], 5, 13, x.shape[2], x.shape[3])
190
+
191
+ SurfaceFeatures = self.patchembed2d(SurfaceInput)
192
+ UpperAirFeatures = self.patchembed3d(UpperAirInput)
193
+
194
+ CombinedFeatures = torch.concat(
195
+ [SurfaceFeatures.unsqueeze(2), UpperAirFeatures], dim=2
196
+ )
197
+ Batch, Channels, PressureLevels, Height, Width = CombinedFeatures.shape
198
+ Tokens = CombinedFeatures.reshape(Batch, Channels, -1).transpose(1, 2)
199
+
200
+ Tokens = self.layer1(Tokens)
201
+ SkipTokens = Tokens
202
+
203
+ Tokens = self.downsample(Tokens)
204
+ Tokens = self.layer2(Tokens)
205
+ Tokens = self.layer3(Tokens)
206
+ Tokens = self.upsample(Tokens)
207
+ Tokens = self.layer4(Tokens)
208
+
209
+ OutputFeatures = torch.concat([Tokens, SkipTokens], dim=-1)
210
+ OutputFeatures = OutputFeatures.transpose(1, 2).reshape(
211
+ Batch, -1, PressureLevels, Height, Width
212
+ )
213
+ output_surface = OutputFeatures[:, :, 0, :, :]
214
+ output_upper_air = OutputFeatures[:, :, 1:, :, :]
215
+
216
+ output_surface = self.patchrecovery2d(output_surface)
217
+ output_upper_air = self.patchrecovery3d(output_upper_air)
218
+ return output_surface, output_upper_air
scripts/fake_data.py ADDED
@@ -0,0 +1,95 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import os
2
+ import h5py
3
+ import numpy as np
4
+ import xarray as xr
5
+ from onescience.utils.YParams import YParams
6
+
7
+
8
+ # 各数据集固定的空间和时间维度
9
+ DATASET_DIMS = {"T": 10, "H": 721, "W": 1440, "time_step": 6}
10
+
11
+
12
+ def generate_fake_h5(data_dir, var_names, years, dims):
13
+ """
14
+ 为每个年份生成一个空 h5 文件。
15
+ 利用 HDF5 chunked 数据集未写入 chunk 即返回 fill_value=0 的特性,
16
+ 文件实际只含元数据,极小,但 shape 与真实数据完全一致。
17
+ 均值/标准差也作为数据集内嵌进每年的 h5,与 era5.py 新版读取方式对应。
18
+ """
19
+ os.makedirs(os.path.join(data_dir, "data"), exist_ok=True)
20
+ T, C = dims["T"], len(var_names)
21
+ H, W = dims["H"], dims["W"]
22
+
23
+ means = np.zeros((1, C, 1, 1), dtype=np.float32)
24
+ stds = np.ones((1, C, 1, 1), dtype=np.float32)
25
+
26
+ for year in years:
27
+ path = os.path.join(data_dir, "data", f"{year}.h5")
28
+ with h5py.File(path, "w") as f:
29
+ ds = f.create_dataset(
30
+ "fields",
31
+ shape=(T, C, H, W),
32
+ dtype="float32",
33
+ chunks=(1, C, H, W),
34
+ fillvalue=0.0,
35
+ )
36
+ ds.attrs["variables"] = var_names
37
+ ds.attrs["time_step"] = dims["time_step"]
38
+ f.create_dataset("global_means", data=means)
39
+ f.create_dataset("global_stds", data=stds)
40
+
41
+ size_kb = os.path.getsize(path) / 1024
42
+ print(f" {year}.h5 shape=({T},{C},{H},{W}) "
43
+ f"logical={T*C*H*W*4/1024**3:.1f}GB actual={size_kb:.1f}KB")
44
+
45
+
46
+ def get_static(data_dir, var, name):
47
+ os.makedirs(data_dir, exist_ok=True)
48
+ ds = xr.Dataset(
49
+ data_vars={
50
+ f"{var}": (("valid_time", "latitude", "longitude"),
51
+ np.random.rand(1, 721, 1440).astype(np.float32))
52
+ },
53
+ coords={
54
+ "valid_time": ["2015-12-31"],
55
+ "latitude": np.linspace(90, -90, 721, dtype=np.float64),
56
+ "longitude": np.linspace(0, 359.75, 1440, dtype=np.float64),
57
+ "number": 0,
58
+ "expver": "",
59
+ },
60
+ attrs={
61
+ "GRIB_centre": "ecmf",
62
+ "GRIB_centreDescription": "European Centre for Medium-Range Weather Forecasts",
63
+ "GRIB_subCentre": "0",
64
+ "Conventions": "CF-1.7",
65
+ "institution": "European Centre for Medium-Range Weather Forecasts",
66
+ "history": "Generated manually",
67
+ }
68
+ )
69
+
70
+ ds.to_netcdf(f"{data_dir}/{name}.nc")
71
+ arr = np.random.randn(721, 1440).astype(np.float32)
72
+ np.save(f'{data_dir}/land_mask.npy', arr)
73
+ np.save(f'{data_dir}/soil_type.npy', arr)
74
+ np.save(f'{data_dir}/topography.npy', arr)
75
+ print(f"✅ Static data: {arr.shape}, dtype: {arr.dtype}, save to {data_dir}")
76
+
77
+
78
+ if __name__ == "__main__":
79
+ cfg_datapipe = YParams("conf/config.yaml", "datapipe")
80
+
81
+ if cfg_datapipe.dataset.data_dir.startswith("/public/") or cfg_datapipe.dataset.data_dir.startswith("/work2/"):
82
+ print("请检查 config,确保各 *_dir 指向本地测试路径而非生产路径。")
83
+ exit()
84
+
85
+ years = cfg_datapipe.dataset.train_time + cfg_datapipe.dataset.val_time + cfg_datapipe.dataset.test_time
86
+ atm_vars = cfg_datapipe.dataset.channels
87
+
88
+ generate_fake_h5(cfg_datapipe.dataset.data_dir, atm_vars, years, DATASET_DIMS)
89
+
90
+ static_dir = os.path.join(cfg_datapipe.dataset.data_dir, "static")
91
+ get_static(static_dir, 'z', 'geopotential')
92
+ get_static(static_dir, 'lsm', 'land_sea_mask')
93
+
94
+
95
+ print("\n✅ Fake datasets generated.")
scripts/inference.py ADDED
@@ -0,0 +1,90 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import sys
2
+ from pathlib import Path
3
+
4
+ # 获取项目根目录(train.py上级的上级)
5
+ root_path = Path(__file__).parent.parent
6
+ sys.path.append(str(root_path))
7
+ import torch
8
+ import os
9
+ import sys
10
+ import glob
11
+ import numpy as np
12
+ import h5py
13
+ from tqdm import tqdm
14
+ from model.pangu import Pangu
15
+ from onescience.utils.YParams import YParams
16
+ from onescience.datapipes.climate import ERA5Datapipe
17
+
18
+
19
+ def get_stats(data_dir, channels):
20
+ """从新版 h5 中读取变量列表与归一化参数(均值/标准差)"""
21
+ h5_files = sorted(glob.glob(os.path.join(data_dir, "data", "*.h5")))
22
+ with h5py.File(h5_files[0], "r") as f:
23
+ ds = f["fields"]
24
+ all_variables = [v.decode() if isinstance(v, bytes) else v for v in ds.attrs["variables"]]
25
+ mu = f["global_means"][:] # [1, C, 1, 1]
26
+ std = f["global_stds"][:]
27
+
28
+ channel_indices = [all_variables.index(v) for v in channels]
29
+ means = mu[:, channel_indices, :, :]
30
+ stds = std[:, channel_indices, :, :]
31
+ return means, stds
32
+
33
+
34
+ if __name__ == "__main__":
35
+ current_path = os.getcwd()
36
+ sys.path.append(current_path)
37
+
38
+ ## Model config init
39
+ config_file_path = os.path.join(current_path, "conf/config.yaml")
40
+ cfg = YParams(config_file_path, "model")
41
+ ## DataLoader init
42
+ cfg_data = YParams(config_file_path, "datapipe")
43
+
44
+ means, stds = get_stats(cfg_data.dataset.data_dir, cfg_data.dataset.channels)
45
+
46
+ datapipe = ERA5Datapipe(
47
+ dataset_dir=cfg_data.dataset.data_dir,
48
+ used_variables=cfg_data.dataset.channels,
49
+ used_years=cfg_data.dataset.test_time,
50
+ distributed=False,
51
+ batch_size=1,
52
+ num_workers=4,
53
+ )
54
+ test_dataloader, _ = datapipe.get_dataloader("test")
55
+
56
+ static_dir = os.path.join(cfg_data.dataset.data_dir, "static")
57
+
58
+ land_mask = torch.from_numpy(np.load(os.path.join(static_dir, "land_mask.npy")).astype(np.float32))
59
+ soil_type = torch.from_numpy(np.load(os.path.join(static_dir, "soil_type.npy")).astype(np.float32))
60
+ topography = torch.from_numpy(np.load(os.path.join(static_dir, "topography.npy")).astype(np.float32))
61
+ topography = (topography - topography.mean()) / (topography.std(unbiased=False) + 1e-6)
62
+ surface_mask = torch.stack([land_mask, soil_type, topography], dim=0).to('cuda:0')
63
+ surface_mask = surface_mask.unsqueeze(0).repeat(cfg_data.dataloader.batch_size, 1, 1, 1)
64
+
65
+ ckpt = torch.load(f"{cfg.checkpoint_dir}/model_bak.pth", map_location="cuda:0")
66
+ model = Pangu(img_size=cfg_data.dataset.img_size,
67
+ patch_size=cfg.patch_size,
68
+ embed_dim=cfg.embed_dim,
69
+ num_heads=cfg.num_heads,
70
+ window_size=cfg.window_size,
71
+ ).to('cuda:0')
72
+ model.load_state_dict(ckpt["model_state_dict"])
73
+
74
+ model.eval()
75
+ os.makedirs('result/output/', exist_ok=True)
76
+ print(f"📂 samples will be generated to './result/output/'")
77
+ with torch.no_grad():
78
+ for data in tqdm(test_dataloader, desc="Inferring testset", unit="batch"):
79
+ invar = data[0]
80
+ outvar = data[1]
81
+ filename = data[4][-1][0]
82
+ invar_surface = invar[:, :4, :, :].to("cuda:0", dtype=torch.float32)
83
+ invar_upper_air = invar[:, 4:, :, :].to("cuda:0", dtype=torch.float32)
84
+ invar = torch.concat([invar_surface, surface_mask, invar_upper_air], dim=1)
85
+
86
+ out_surface, out_upper_air = model(invar)
87
+ out_upper_air = out_upper_air.reshape(invar_upper_air.shape)
88
+ pred_var = torch.concat([out_surface, out_upper_air], dim=1).cpu().numpy()
89
+ pred_var = pred_var * stds + means
90
+ np.save(f"result/output/{filename}.npy", pred_var)
scripts/result.py ADDED
@@ -0,0 +1,226 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import numpy as np
2
+ import matplotlib.pyplot as plt
3
+ import os
4
+ import sys
5
+ import glob
6
+ import h5py
7
+ from datetime import datetime
8
+ from tqdm import tqdm
9
+ from onescience.utils.fcn.YParams import YParams
10
+ from matplotlib import rcParams
11
+
12
+ # rcParams['font.family'] = 'serif'
13
+ # rcParams['font.serif'] = ['DejaVu Serif']
14
+ rcParams['mathtext.fontset'] = 'stix'
15
+ rcParams['axes.linewidth'] = 0.9
16
+ rcParams['xtick.major.width'] = 0.9
17
+ rcParams['ytick.major.width'] = 0.9
18
+
19
+
20
+ def get_metadata(data_dir, channels):
21
+ """从新版 h5 attrs 中读取变量列表和 time_step"""
22
+ h5_files = sorted(glob.glob(os.path.join(data_dir, "data", "*.h5")))
23
+ with h5py.File(h5_files[0], "r") as f:
24
+ ds = f["fields"]
25
+ all_variables = [v.decode() if isinstance(v, bytes) else v for v in ds.attrs["variables"]]
26
+ time_step = int(ds.attrs["time_step"])
27
+
28
+ channel_indices = [all_variables.index(v) for v in channels]
29
+
30
+ total_files = [f for f in os.listdir('./result/output/') if f.endswith('.npy')]
31
+ total_files.sort()
32
+ return total_files, channel_indices, time_step
33
+
34
+
35
+ def filename_to_index(filename, time_step):
36
+ """将 YYYYMMDDHH 格式的文件名转换为年度 h5 文件中的时间步索引"""
37
+ dt = datetime.strptime(filename, "%Y%m%d%H")
38
+ year_start = datetime(dt.year, 1, 1)
39
+ hours = (dt - year_start).total_seconds() / 3600
40
+ return int(hours / time_step)
41
+
42
+
43
+ def group_files_by_year(total_files, time_step):
44
+ """将输出文件按年份归组,减少重复打开年度 h5 文件的开销"""
45
+ files_by_year = {}
46
+ for file in total_files:
47
+ fname = file[:-4]
48
+ year = fname[:4]
49
+ files_by_year.setdefault(year, []).append((file, filename_to_index(fname, time_step)))
50
+ return files_by_year
51
+
52
+
53
+ def get_result(total_files, channel_indices, time_step, data_dir, clim_mean):
54
+ channel_rmse = np.zeros(len(channel_indices))
55
+ channel_acc = np.zeros(len(channel_indices))
56
+ clim_mean = clim_mean[0, :, :, :]
57
+ if not os.path.exists('./result/rmse.npy') or not os.path.exists('result/acc.npy'):
58
+ numerator = np.zeros(len(channel_indices))
59
+ pred_sq_sum = np.zeros(len(channel_indices))
60
+ label_sq_sum = np.zeros(len(channel_indices))
61
+ files_by_year = group_files_by_year(total_files, time_step)
62
+ with tqdm(total=len(total_files), unit="files") as pbar:
63
+ for year, year_files in files_by_year.items():
64
+ with h5py.File(os.path.join(data_dir, 'data', f'{year}.h5'), "r") as f:
65
+ fields = f["fields"]
66
+ for file, t_idx in year_files:
67
+ label = fields[t_idx] # [C, H, W]
68
+ label = label[channel_indices]
69
+ pred = np.load(f'result/output/{file}').squeeze()
70
+
71
+ label_anom = label - clim_mean
72
+ pred_anom = pred - clim_mean
73
+ # 累加
74
+ numerator += np.sum(pred_anom * label_anom, axis=(1, 2))
75
+ pred_sq_sum += np.sum(pred_anom ** 2, axis=(1, 2))
76
+ label_sq_sum += np.sum(label_anom ** 2, axis=(1, 2))
77
+
78
+ channel_rmse += np.sqrt(np.mean((label - pred) ** 2, axis=(1, 2)))
79
+ pbar.update(1)
80
+ channel_rmse /= len(total_files)
81
+ channel_acc = numerator / (np.sqrt(pred_sq_sum * label_sq_sum) + 1e-8)
82
+ np.save('./result/acc.npy', channel_acc)
83
+ np.save('./result/rmse.npy', channel_rmse)
84
+
85
+
86
+ def show_result():
87
+ channel_rmse = np.load('./result/rmse.npy')
88
+ channel_acc = np.load('./result/acc.npy')
89
+
90
+ channels = [cfg_data.dataset.channels[i] for i in range(len(channel_indices))]
91
+ w = 24 # 最长 channel 名宽度
92
+
93
+ # 表头
94
+ print(f"┌{'─' * (w + 2)}┬{'─' * 14}┬{'─' * 14}┐")
95
+ print(f"│ {'Channel':<{w}} │ {'RMSE':>12} │ {'ACC':>12} │")
96
+ print(f"├{'─' * (w + 2)}┼{'─' * 14}┼{'─' * 14}┤")
97
+ # 数据行
98
+ for i, ch in enumerate(channels):
99
+ print(f"│ {ch:<{w}} │ {channel_rmse[i]:>12.4f} | {channel_acc[i]:>12.4f} |")
100
+ print(f"├{'─' * (w + 2)}┼{'─' * 14}┼{'─' * 14}┤")
101
+ print(f"│ {'Average':<{w}} │ {np.mean(channel_rmse):>12.4f} │ {np.mean(channel_acc):>12.4f} │")
102
+ print(f"└{'─' * (w + 2)}┴{'─' * 14}┴{'─' * 14}┘")
103
+
104
+
105
+ def plot(label, pred, var, filename):
106
+ # 基础设置
107
+ fig, axes = plt.subplots(1, 3, figsize=(15, 4))
108
+
109
+ # 坐标轴标签
110
+ xtick_labels = ['180°W', '90°W', '0°', '90°E', '180°E']
111
+ ytick_labels = ['90°S', '45°S', '0°', '45°N', '90°N']
112
+ xticks = np.linspace(0, label.shape[-1] - 1, 5)
113
+ yticks = np.linspace(0, label.shape[-2] - 1, 5)
114
+
115
+ # 计算统一色条范围
116
+ vmin = min(label.min(), pred.min())
117
+ vmax = max(label.max(), pred.max())
118
+
119
+ # 计算差异和 RMSE
120
+ diff = label - pred
121
+ rmse = np.sqrt(np.mean(diff ** 2))
122
+ diff_abs_max = np.abs(diff).max()
123
+
124
+ # 绘图配置
125
+ plot_configs = [
126
+ {'data': label, 'title': 'Truth', 'cmap': 'viridis', 'vmin': vmin, 'vmax': vmax},
127
+ {'data': pred, 'title': 'Prediction', 'cmap': 'viridis', 'vmin': vmin, 'vmax': vmax},
128
+ {'data': diff, 'title': f'Difference (RMSE={rmse:.2f})', 'cmap': 'RdBu_r', 'vmin': -diff_abs_max, 'vmax': diff_abs_max},
129
+ ]
130
+
131
+ # 统一绘制
132
+ for ax, cfg in zip(axes, plot_configs):
133
+ im = ax.imshow(cfg['data'], cmap=cfg['cmap'], vmin=cfg['vmin'], vmax=cfg['vmax'])
134
+ ax.set_title(cfg['title'], fontsize=12, pad=4)
135
+ ax.set_xlabel('Longitude')
136
+ ax.set_ylabel('Latitude')
137
+ ax.set_xticks(xticks)
138
+ ax.set_xticklabels(xtick_labels)
139
+ ax.set_yticks(yticks)
140
+ ax.set_yticklabels(ytick_labels)
141
+ plt.colorbar(im, ax=ax, orientation='horizontal')
142
+
143
+ # 总标题
144
+ fig.suptitle(var, fontsize=14, fontweight='bold', y=0.98)
145
+
146
+ plt.savefig(filename, dpi=300, bbox_inches='tight')
147
+ plt.close()
148
+
149
+
150
+ def plot_loss(train_loss, valid_loss):
151
+
152
+ mask = ~(np.isnan(train_loss) | np.isnan(valid_loss))
153
+ train_loss = train_loss[mask]
154
+ valid_loss = valid_loss[mask]
155
+
156
+ fig, ax = plt.subplots(figsize=(5, 3.5))
157
+ # 配置
158
+ colors = {'train': '#2563EB', 'valid': '#EA580C'}
159
+ epochs = np.arange(1, len(train_loss) + 1)
160
+
161
+ # 绑定曲线
162
+ ax.plot(epochs, train_loss, color=colors['train'], linewidth=1.5, label='Train')
163
+ ax.plot(epochs, valid_loss, color=colors['valid'], linewidth=1.5, label='Valid', linestyle='--')
164
+ # 标注最小值
165
+ min_idx = np.argmin(valid_loss)
166
+ ax.scatter(epochs[min_idx], valid_loss[min_idx],
167
+ color=colors['valid'], s=40, zorder=5, edgecolors='white')
168
+ ax.annotate(f'Best: {valid_loss[min_idx]:.3f}',
169
+ xy=(epochs[min_idx], valid_loss[min_idx]),
170
+ xytext=(10, 10), textcoords='offset points', fontsize=8, color=colors['valid'],
171
+ arrowprops=dict(arrowstyle='-', color=colors['valid'], lw=0.5))
172
+
173
+ # 坐标轴
174
+ ax.set(xlabel='Epoch', ylabel='Loss', xlim=(0, len(train_loss) + 1))
175
+
176
+ # 样式
177
+ ax.legend(frameon=False, loc='upper right')
178
+ ax.grid(True, linestyle='--', alpha=0.3)
179
+ ax.spines[['top', 'right']].set_visible(False)
180
+
181
+ plt.tight_layout()
182
+ plt.savefig('./result/loss.png', dpi=300, bbox_inches='tight')
183
+ plt.close()
184
+
185
+
186
+ if __name__ == "__main__":
187
+ current_path = os.getcwd()
188
+ sys.path.append(current_path)
189
+ config_file_path = os.path.join(current_path, 'conf/config.yaml')
190
+ cfg = YParams(config_file_path, 'model')
191
+ cfg_data = YParams(config_file_path, "datapipe")
192
+
193
+ train_loss = np.load('./data/checkpoints/trloss.npy')
194
+ valid_loss = np.load('./data/checkpoints/valoss.npy')
195
+ plot_loss(train_loss, valid_loss)
196
+
197
+ data_dir = cfg_data.dataset.data_dir
198
+ total_files, channel_indices, time_step = get_metadata(data_dir, cfg_data.dataset.channels)
199
+
200
+ # Load data & Compute RMSE/ACC per channel
201
+ h5_files = sorted(glob.glob(os.path.join(data_dir, "data", "*.h5")))
202
+ with h5py.File(h5_files[0], "r") as f:
203
+ mu = f["global_means"][:]
204
+ clim_mean = mu[:, channel_indices, :, :]
205
+ get_result(total_files, channel_indices, time_step, data_dir, clim_mean)
206
+ show_result()
207
+
208
+ ##### 默认绘制 test_time 第一年的第一个时间步,用户可自行指定日期和变量 #####
209
+ test_year = cfg_data.dataset.test_time[0]
210
+ eg_files = [f'{test_year}010206']
211
+ channel_index = [cfg_data.dataset.channels.index(v) for v in ['2m_temperature', 'geopotential_500', 'temperature_500']]
212
+
213
+ selected_var = [cfg_data.dataset.channels[int(i)] for i in channel_index]
214
+ print(f"seleted date: {eg_files}")
215
+ print(f"selected channels: {selected_var}")
216
+ for file in eg_files:
217
+ year = file[:4]
218
+ t_idx = filename_to_index(file, time_step)
219
+ with h5py.File(os.path.join(data_dir, 'data', f'{year}.h5'), "r") as f:
220
+ label = f["fields"][t_idx] # [C, H, W]
221
+ label = label[channel_indices]
222
+ pred = np.load(f'result/output/{file}.npy').squeeze()
223
+ for i in range(len(selected_var)):
224
+ filename = f'./result/{file}_{selected_var[i]}.png'
225
+ plot(label[channel_index[i]], pred[channel_index[i]], selected_var[i], filename)
226
+ print(f'✅plot {filename}')
scripts/train.py ADDED
@@ -0,0 +1,248 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import sys
2
+ from pathlib import Path
3
+
4
+ # 获取项目根目录(train.py上级的上级)
5
+ root_path = Path(__file__).parent.parent
6
+ sys.path.append(str(root_path))
7
+
8
+ import torch
9
+ import os
10
+ import numpy as np
11
+ import torch.distributed as dist
12
+ import logging
13
+ import time
14
+ import torch.nn.functional as F
15
+ from torch.nn.parallel import DistributedDataParallel
16
+ from model.pangu import Pangu
17
+ from onescience.datapipes.climate import ERA5Datapipe
18
+ from onescience.utils.YParams import YParams
19
+ from onescience.memory.checkpoint import replace_function
20
+ from apex import optimizers
21
+
22
+
23
+
24
+
25
+ def loss_func(x, y, weights, level_weight=1.0):
26
+ return level_weight * (F.l1_loss(x, y, reduction='none') * weights).mean()
27
+
28
+ def main():
29
+
30
+ logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
31
+ logger = logging.getLogger()
32
+
33
+ ## Model config init
34
+ config_file_path = os.path.join(current_path, "conf/config.yaml")
35
+ cfg = YParams(config_file_path, "model")
36
+
37
+ ## Distributed config init
38
+ cfg.world_size = 1
39
+ if "WORLD_SIZE" in os.environ:
40
+ cfg.world_size = int(os.environ["WORLD_SIZE"])
41
+ world_rank = 0
42
+ local_rank = 0
43
+ if cfg.world_size > 1:
44
+ dist.init_process_group(backend="nccl", init_method="env://")
45
+ local_rank = int(os.environ["LOCAL_RANK"])
46
+ world_rank = dist.get_rank()
47
+
48
+ ## DataLoader init
49
+ cfg_data = YParams(config_file_path, "datapipe")
50
+ datapipe = ERA5Datapipe(
51
+ dataset_dir=cfg_data.dataset.data_dir,
52
+ used_variables=cfg_data.dataset.channels,
53
+ used_years=cfg_data.dataset.train_time,
54
+ distributed=dist.is_initialized(),
55
+ batch_size=cfg_data.dataloader.batch_size,
56
+ num_workers=cfg_data.dataloader.num_workers
57
+ )
58
+ train_dataloader, train_sampler = datapipe.get_dataloader("train")
59
+ datapipe = ERA5Datapipe(
60
+ dataset_dir=cfg_data.dataset.data_dir,
61
+ used_variables=cfg_data.dataset.channels,
62
+ used_years=cfg_data.dataset.val_time,
63
+ distributed=dist.is_initialized(),
64
+ batch_size=cfg_data.dataloader.batch_size,
65
+ num_workers=cfg_data.dataloader.num_workers
66
+ )
67
+ val_dataloader, val_sampler = datapipe.get_dataloader("valid")
68
+
69
+ surface_weights = torch.as_tensor(cfg_data.dataset.weights[:4], device=local_rank, dtype=torch.float32).view(1, -1, 1, 1)
70
+ pressure_weights = torch.as_tensor(cfg_data.dataset.weights[4:], device=local_rank, dtype=torch.float32).view(1, -1, 1, 1)
71
+
72
+ static_dir = os.path.join(cfg_data.dataset.data_dir, "static")
73
+
74
+ land_mask = torch.from_numpy(np.load(os.path.join(static_dir, "land_mask.npy")).astype(np.float32))
75
+ soil_type = torch.from_numpy(np.load(os.path.join(static_dir, "soil_type.npy")).astype(np.float32))
76
+ topography = torch.from_numpy(np.load(os.path.join(static_dir, "topography.npy")).astype(np.float32))
77
+ topography = (topography - topography.mean()) / (topography.std(unbiased=False) + 1e-6)
78
+ surface_mask = torch.stack([land_mask, soil_type, topography], dim=0).to(local_rank)
79
+ surface_mask = surface_mask.unsqueeze(0).repeat(cfg_data.dataloader.batch_size, 1, 1, 1)
80
+
81
+ ## Model init
82
+ model = Pangu(img_size=cfg_data.dataset.img_size,
83
+ patch_size=cfg.patch_size,
84
+ embed_dim=cfg.embed_dim,
85
+ num_heads=cfg.num_heads,
86
+ window_size=cfg.window_size,
87
+ ).to(local_rank)
88
+ optimizer = optimizers.FusedAdam(model.parameters(), betas=(0.9, 0.999), lr=5e-4, weight_decay=3e-6)
89
+ scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)
90
+
91
+ ## Train process init
92
+ os.makedirs(cfg.checkpoint_dir, exist_ok=True)
93
+ train_loss_file = f"{cfg.checkpoint_dir}/trloss.npy"
94
+ valid_loss_file = f"{cfg.checkpoint_dir}/valoss.npy"
95
+ best_valid_loss = 1.0e6
96
+ best_loss_epoch = 0
97
+ train_losses = np.empty((0,), dtype=np.float32)
98
+ valid_losses = np.empty((0,), dtype=np.float32)
99
+ current_epoch = 0
100
+
101
+ ## Get model params count
102
+ if cfg.world_size == 1:
103
+ total_params = sum(p.numel() for p in model.parameters())
104
+ print("\n\n")
105
+ print("-" * 50)
106
+ print(f"📂 now params is {total_params}, {total_params / 1e6:.2f}M, {total_params / 1e9:.2f}B")
107
+ print("-" * 50, "\n")
108
+
109
+ ## Load model weight if there exist well-trained model
110
+ if os.path.exists(f"{cfg.checkpoint_dir}/model_bak.pth"):
111
+ if world_rank == 0:
112
+ print("\n\n")
113
+ print("-" * 50)
114
+ print(f"✅ There has a model weight, load and continue training...")
115
+ print(f'If you want to train a new model, ensure there is no *.pth file in {cfg.checkpoint_dir}')
116
+ print("-" * 50, "\n")
117
+ ckpt = torch.load(f"{cfg.checkpoint_dir}/model_bak.pth", map_location=f'cuda:{local_rank}', weights_only=False)
118
+ model.load_state_dict(ckpt["model_state_dict"])
119
+ optimizer.load_state_dict(ckpt["optimizer_state_dict"])
120
+ scheduler.load_state_dict(ckpt["scheduler_state_dict"])
121
+ best_valid_loss = ckpt["best_valid_loss"]
122
+ best_loss_epoch = ckpt["best_loss_epoch"]
123
+ current_epoch = ckpt["current_epoch"]
124
+ train_losses = np.load(train_loss_file)
125
+ valid_losses = np.load(valid_loss_file)
126
+
127
+ ## Distributed model
128
+ if cfg.world_size > 1:
129
+ model = DistributedDataParallel(model, device_ids=[local_rank], output_device=local_rank)
130
+
131
+ world_rank == 0 and logger.info(f"start training ...")
132
+
133
+ for epoch in range(current_epoch, cfg.max_epoch):
134
+ if dist.is_initialized():
135
+ train_sampler.set_epoch(epoch)
136
+ val_sampler.set_epoch(epoch)
137
+
138
+ model.train()
139
+ train_loss = 0
140
+ start_time = time.time()
141
+ for j, data in enumerate(train_dataloader):
142
+ invar = data[0]
143
+ outvar = data[1]
144
+ invar_surface = invar[:, :4, :, :].to(local_rank, dtype=torch.float32)
145
+ invar_upper_air = invar[:, 4:, :, :].to(local_rank, dtype=torch.float32)
146
+ invar = torch.concat([invar_surface, surface_mask, invar_upper_air], dim=1)
147
+ tar_surface = outvar[:, :4, :, :].to(local_rank, dtype=torch.float32)
148
+ tar_upper_air = outvar[:, 4:, :, :].to(local_rank, dtype=torch.float32)
149
+
150
+ with replace_function(model,["layer2", "layer3"],cfg.world_size > 1):
151
+ out_surface, out_upper_air = model(invar)
152
+
153
+ out_upper_air = out_upper_air.reshape(tar_upper_air.shape)
154
+ loss1 = loss_func(out_surface, tar_surface, surface_weights, level_weight=0.25)
155
+ loss2 = loss_func(out_upper_air, tar_upper_air, pressure_weights, level_weight=1.0)
156
+ # 总 loss
157
+ loss = loss1 + loss2
158
+ optimizer.zero_grad()
159
+ loss.backward()
160
+ optimizer.step()
161
+ train_loss += loss.item()
162
+ if world_rank == 0:
163
+ logger.info(f'Train: Epoch {epoch}-{j+1}/{len(train_dataloader)} '
164
+ f'[cost {int((time.time()-start_time) // 60):02}:{int((time.time()-start_time) % 60):02}] '
165
+ f'[{(time.time()-start_time)/(j+1): .02f}s/{cfg_data.dataloader.batch_size}batch] '
166
+ f'loss:{train_loss / (j+1): .04f}')
167
+
168
+ train_loss /= len(train_dataloader)
169
+
170
+ model.eval()
171
+ valid_loss = 0
172
+ with torch.no_grad():
173
+ start_time = time.time()
174
+ for j, data in enumerate(val_dataloader):
175
+ invar = data[0]
176
+ outvar = data[1]
177
+ invar_surface = invar[:, :4, :, :].to(local_rank, dtype=torch.float32)
178
+ invar_upper_air = invar[:, 4:, :, :].to(local_rank, dtype=torch.float32)
179
+ invar = torch.concat([invar_surface, surface_mask, invar_upper_air], dim=1)
180
+ tar_surface = outvar[:, :4, :, :].to(local_rank, dtype=torch.float32)
181
+ tar_upper_air = outvar[:, 4:, :, :].to(local_rank, dtype=torch.float32)
182
+
183
+ with replace_function(model,["layer2", "layer3"],cfg.world_size > 1):
184
+ out_surface, out_upper_air = model(invar)
185
+
186
+ out_upper_air = out_upper_air.reshape(tar_upper_air.shape)
187
+ loss1 = loss_func(out_surface, tar_surface, surface_weights, level_weight=0.25).item()
188
+ loss2 = loss_func(out_upper_air, tar_upper_air, pressure_weights, level_weight=1.0).item()
189
+ # 总 loss
190
+ loss = loss1 + loss2
191
+
192
+ if cfg.world_size > 1:
193
+ loss_tensor = torch.tensor(loss, device=local_rank)
194
+ dist.all_reduce(loss_tensor)
195
+ loss = loss_tensor.item() / cfg.world_size
196
+ valid_loss += loss
197
+ if world_rank == 0:
198
+ logger.info(f'Valid: Epoch {epoch}-{j+1}/{len(val_dataloader)} '
199
+ f'[cost {int((time.time()-start_time) // 60):02}:{int((time.time()-start_time) % 60):02}] '
200
+ f'[{(time.time()-start_time)/(j+1): .02f}s/{cfg_data.dataloader.batch_size}batch] '
201
+ f'loss:{valid_loss / (j+1): .04f}')
202
+
203
+ valid_loss /= len(val_dataloader)
204
+ is_save_ckp = False
205
+ if valid_loss < best_valid_loss:
206
+ best_valid_loss = valid_loss
207
+ best_loss_epoch = epoch
208
+ world_rank == 0 and save_checkpoint(model, optimizer, scheduler, best_valid_loss, best_loss_epoch, cfg.checkpoint_dir, epoch)
209
+ is_save_ckp = True
210
+
211
+ scheduler.step()
212
+
213
+ if world_rank == 0:
214
+ logger.info(f"Epoch [{epoch}/{cfg.max_epoch}], "
215
+ f"Train Loss: {train_loss:.4f}, "
216
+ f"Valid Loss: {valid_loss:.4f}, "
217
+ f"Best loss at Epoch: {best_loss_epoch}"
218
+ + (", saving checkpoint" if is_save_ckp else "")
219
+ )
220
+ train_losses = np.append(train_losses, train_loss)
221
+ valid_losses = np.append(valid_losses, valid_loss)
222
+
223
+ np.save(train_loss_file, train_losses)
224
+ np.save(valid_loss_file, valid_losses)
225
+
226
+ if epoch - best_loss_epoch > cfg.patience:
227
+ print(f"Loss has not decrease in {cfg.patience} epochs, stopping training...")
228
+ exit()
229
+
230
+
231
+ def save_checkpoint(model, optimizer, scheduler, best_valid_loss, best_loss_epoch, model_path, epoch):
232
+ model_to_save = model.module if hasattr(model, "module") else model
233
+ state = {"model_state_dict": model_to_save.state_dict(),
234
+ "optimizer_state_dict": optimizer.state_dict(),
235
+ "scheduler_state_dict": scheduler.state_dict(),
236
+ "best_valid_loss": best_valid_loss,
237
+ "best_loss_epoch": best_loss_epoch,
238
+ "current_epoch": epoch
239
+ }
240
+ torch.save(state, f"{model_path}/model.pth")
241
+ ### the weight file saving may interrupted due to DCU queue limit, get a backup to ensure there at least has one model
242
+ os.system(f"mv {model_path}/model.pth {model_path}/model_bak.pth")
243
+
244
+
245
+ if __name__ == "__main__":
246
+ current_path = os.getcwd()
247
+ sys.path.append(current_path)
248
+ main()
weight/.gitkeep ADDED
@@ -0,0 +1 @@
 
 
1
+