Skip to content

Commit cb754d0

Browse files
authored
Merge pull request #398 from NCAR/postblock-integration
Postblock integration
2 parents 2b3dbdf + 3dffddd commit cb754d0

29 files changed

Lines changed: 1082 additions & 382 deletions

config/gen_2/examples/example-v2026.2.yml

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -135,12 +135,12 @@ trainer:
135135
train_batch_size: 8 # per-GPU batch size; total = batch_size × n_gpus
136136
valid_batch_size: 8
137137

138-
batches_per_epoch: 0 # 0 = use full dataset each epoch
138+
batches_per_epoch: 5 # 0 = use full dataset each epoch
139139
valid_batches_per_epoch: 100
140140
stopping_patience: 999
141141

142142
start_epoch: 0
143-
num_epoch: 5 # epochs per qsub job (checkpoint after each)
143+
num_epoch: 1 # epochs per qsub job (checkpoint after each)
144144
epochs: &epochs 70 # total training target
145145

146146
use_tensorboard: True
@@ -163,6 +163,22 @@ trainer:
163163
prefetch_factor: 4
164164

165165

166+
# ---- Preblocks / Postblocks ---- #
167+
# Preblocks run before the model; postblocks run after each forward pass.
168+
# ConcatToTensor (concat) is required — it assembles the nested variable dict
169+
# into the flat (B, C, H, W) tensor the model expects and builds the channel map
170+
# used by Reconstruct. Reconstruct is required for forecast_len > 1 rollout.
171+
preblocks:
172+
per_step:
173+
concat:
174+
type: concat
175+
176+
postblocks:
177+
per_step:
178+
reconstruct:
179+
type: reconstruct
180+
181+
166182
# ---- Architecture (pre-filled for 1-degree ERA5, 181×360) ---- #
167183
# Most users should not need to change these. If you change the grid or
168184
# number of variables, update image_height/width, levels, channels, and

config/gen_2/smoke/smoke_gen2_casper.yml

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ data:
2828
diagnostic: null
2929

3030
timestep: "6h"
31-
forecast_len: 1
31+
forecast_len: 3
3232
start_datetime: "2015-01-01"
3333
end_datetime: "2016-12-31"
3434

@@ -37,13 +37,19 @@ validation_data:
3737
end_datetime: "2018-12-31"
3838

3939
preblocks:
40-
norm:
41-
type: era5_normalizer
42-
args:
43-
mean_path: "/glade/campaign/cisl/aiml/credit/static_scalers/mean_6h_1979_2018_16lev_0.25deg.nc"
44-
std_path: "/glade/campaign/cisl/aiml/credit/static_scalers/std_6h_1979_2018_16lev_0.25deg.nc"
45-
concat:
46-
type: concat
40+
per_step:
41+
# norm:
42+
# type: era5_normalizer
43+
# args:
44+
# mean_path: "/glade/campaign/cisl/aiml/credit/static_scalers/mean_6h_1979_2018_16lev_0.25deg.nc"
45+
# std_path: "/glade/campaign/cisl/aiml/credit/static_scalers/std_6h_1979_2018_16lev_0.25deg.nc"
46+
concat:
47+
type: concat
48+
49+
postblocks:
50+
per_step:
51+
reconstruct:
52+
type: reconstruct
4753

4854
model:
4955
type: "crossformer"

config/gen_2/smoke/smoke_gen2_multistep_casper.yml

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -134,11 +134,19 @@ trainer:
134134
prefetch_factor: 2
135135

136136
preblocks:
137-
norm:
138-
type: era5_normalizer
139-
args:
140-
mean_path: "/glade/campaign/cisl/aiml/ksha/CREDIT_data/ERA5_mlevel_1deg/mean_std/mean_6h_1979_2019_conserve_1deg.nc"
141-
std_path: "/glade/campaign/cisl/aiml/ksha/CREDIT_data/ERA5_mlevel_1deg/mean_std/std_6h_1979_2019_conserve_1deg.nc"
137+
per_step:
138+
norm:
139+
type: era5_normalizer
140+
args:
141+
mean_path: "/glade/campaign/cisl/aiml/ksha/CREDIT_data/ERA5_mlevel_1deg/mean_std/mean_6h_1979_2019_conserve_1deg.nc"
142+
std_path: "/glade/campaign/cisl/aiml/ksha/CREDIT_data/ERA5_mlevel_1deg/mean_std/std_6h_1979_2019_conserve_1deg.nc"
143+
concat:
144+
type: concat
145+
146+
postblocks:
147+
per_step:
148+
reconstruct:
149+
type: reconstruct
142150

143151
pbs:
144152
conda: "credit-main-casper"

credit/applications/train_gen2.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -129,7 +129,8 @@ def main_cli():
129129
inject_flat_var_keys(conf)
130130
if "post_conf" in conf["model"]:
131131
warnings.warn(
132-
"Gen 2 training does not support Gen 1 postblocks. Any postblocks included in the conf will be ignored."
132+
"Gen 2 training does not support Gen 1 postblocks (conf['model']['post_conf']). "
133+
"These will be ignored. Gen 2 postblocks (conf['postblocks']) are still applied normally."
133134
)
134135
conf["model"].pop("post_conf", None)
135136
m = load_model(conf)

credit/datasets/base_dataset.py

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -656,6 +656,7 @@ def _get_file_source(
656656
field_config (dict[str, Any]): Validated field-type config dict.
657657
658658
Raises:
659+
FileNotFoundError: If ``self.mode == "local"`` and the glob pattern matches no files.
659660
ValueError: If ``self.mode`` is not a recognised mode.
660661
661662
Returns:
@@ -665,9 +666,14 @@ def _get_file_source(
665666
The expected return type should be consistent within a dataset class.
666667
"""
667668
if self.mode == "local":
668-
files = sorted(glob(field_config.get("path", "")))
669+
path = field_config.get("path", "")
670+
files = sorted(glob(path))
671+
if not files:
672+
raise FileNotFoundError(
673+
f"No files found matching '{path}'. Check that the path exists and is accessible from this machine."
674+
)
669675
time_fmt: str = field_config.get("filename_time_format", "%Y")
670-
return _map_files(files, time_fmt) if files else None
676+
return _map_files(files, time_fmt)
671677
elif self.mode == "remote":
672678
return True
673679
else:

credit/datasets/era5.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -396,6 +396,7 @@ def __init__(
396396
# Initialised lazily on the first __getitem__ call (worker-safe).
397397
self._fs = None
398398
self.store = None
399+
self.mode = "remote"
399400
super().init_register_all_fields()
400401

401402
# ------------------------------------------------------------------

credit/postblock/__init__.py

Lines changed: 66 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -3,77 +3,94 @@
33
from credit.postblock.reconstruct import Reconstruct
44
from credit.postblock.wet_mask_samudra import WetMaskBlock
55
from credit.postblock.scaler import BridgeScalerTransformer
6-
6+
from credit.postblock.gen1 import TracerFixer, GlobalMassFixer, GlobalWaterFixer, GlobalEnergyFixer
7+
from credit.postblock.geopotential import GeopotentialDiagnostic
78

89
POSTBLOCK_REGISTRY = {
910
"reconstruct": Reconstruct,
1011
"bridgescaler_transform": BridgeScalerTransformer,
1112
"wet_mask_samudra": WetMaskBlock,
13+
"tracer_fixer": TracerFixer,
14+
"global_mass_fixer": GlobalMassFixer,
15+
"global_water_fixer": GlobalWaterFixer,
16+
"global_energy_fixer": GlobalEnergyFixer,
17+
"geopotential_diagnostic": GeopotentialDiagnostic,
1218
}
1319

20+
_VALID_SECTIONS = {"per_step", "post_rollout"}
1421

15-
def build_postblocks(postblock_cfg: dict | None = None) -> nn.ModuleDict:
16-
"""Instantiates all postblocks from the config's ``postblocks`` section.
17-
18-
``per_step`` defaults to ``False`` for every block and can be set per-block
19-
in the config as a signal to the trainer about call timing.
2022

21-
Args:
22-
postblock_cfg: the full postblocks dict from the config, e.g.::
23-
24-
postblocks:
25-
reconstruct:
26-
type: reconstruct
27-
inverse_scale:
28-
type: bridgescaler_transform
29-
args:
30-
method: inverse_transform
31-
scaler_path: /path/to/scaler.json
32-
per_step: false
33-
mass_fixer:
34-
type: global_mass_fixer
35-
args: ...
36-
37-
Returns:
38-
``nn.ModuleDict`` of instantiated postblocks, ordered as in config.
39-
"""
23+
def _build_postblock_section(section_cfg: dict) -> nn.ModuleDict:
4024
modules = {}
41-
for name, block_cfg in (postblock_cfg or {}).items():
42-
block_type = block_cfg["type"]
43-
args = block_cfg.get("args") or {}
44-
instance = POSTBLOCK_REGISTRY[block_type](**args)
45-
instance.per_step = block_cfg.get("per_step", False)
46-
modules[name] = instance
25+
for name, block_cfg in section_cfg.items():
26+
modules[name] = POSTBLOCK_REGISTRY[block_cfg["type"]](**(block_cfg.get("args") or {}))
4727
return nn.ModuleDict(modules)
4828

4929

50-
def apply_postblocks(postblocks: nn.ModuleDict, batch_dict: dict) -> dict:
51-
"""Applies all postblocks sequentially on a shared batch dict.
30+
def build_postblocks(postblock_cfg: dict | None = None, phase: str = "per_step") -> nn.ModuleDict:
31+
"""Instantiate postblocks for a single phase from a two-section config.
32+
33+
Config format::
34+
35+
postblocks:
36+
per_step: # run after every forward pass in the rollout loop
37+
reconstruct:
38+
type: reconstruct
39+
inverse_scale:
40+
type: bridgescaler_transform
41+
args:
42+
method: inverse_transform
43+
scaler_path: /path/to/scaler.json
44+
post_rollout: # run once after all rollout steps complete
45+
mass_fixer:
46+
type: global_mass_fixer
47+
args: ...
48+
49+
Typical usage — build once per phase, store separately::
5250
53-
The caller is responsible for adding ``"prediction"`` (flat model output
54-
tensor) and ``"meta"`` (metadata from ``apply_preblocks``) to ``batch_dict``
55-
before calling. Any additional data needed by postblocks (e.g. ``"input"``,
56-
``"target"``, ``"_raw"``) should also be added by the caller beforehand.
51+
step_postblocks = build_postblocks(cfg, phase="per_step")
52+
rollout_postblocks = build_postblocks(cfg, phase="post_rollout")
5753
58-
``Reconstruct`` must be the first registered postblock — it converts
59-
``batch_dict["prediction"]`` from a flat tensor into a nested variable dict.
60-
Subsequent postblocks operate on that nested dict via their ``key=`` arg.
54+
# inside rollout loop, after each forward pass:
55+
full_data_dict = apply_postblocks(step_postblocks, full_data_dict)
56+
57+
# once after rollout loop completes:
58+
apply_postblocks(rollout_postblocks, full_data_dict)
6159
6260
Args:
63-
postblocks: ``nn.ModuleDict`` built by ``build_postblocks``.
64-
batch_dict: Dict containing at minimum ``"prediction"`` (flat tensor)
65-
and ``"meta"`` (with ``_channel_map["output"]``).
61+
postblock_cfg: the full ``postblocks`` config dict (both sections).
62+
phase: which section to build — ``"per_step"`` or ``"post_rollout"``.
6663
6764
Returns:
68-
The same ``batch_dict`` after all postblocks have run. ``"prediction"``
69-
will be a nested variable dict after ``Reconstruct`` runs.
65+
``nn.ModuleDict`` of instantiated blocks for the requested phase.
7066
7167
Raises:
72-
RuntimeError: if ``postblocks`` is empty or ``Reconstruct`` is not first.
68+
ValueError: if the config contains keys other than ``"per_step"`` / ``"post_rollout"``,
69+
or if ``phase`` is not one of those values.
7370
"""
74-
blocks = list(postblocks.values())
71+
cfg = postblock_cfg or {}
72+
unknown = set(cfg) - _VALID_SECTIONS
73+
if unknown:
74+
raise ValueError(
75+
f"build_postblocks: unexpected top-level keys {sorted(unknown)}. "
76+
"Expected only 'per_step' and/or 'post_rollout'. "
77+
"If you are using the old flat postblock format, migrate to the two-section layout."
78+
)
79+
if phase not in _VALID_SECTIONS:
80+
raise ValueError(f"build_postblocks: phase must be one of {sorted(_VALID_SECTIONS)}, got {phase!r}.")
81+
return _build_postblock_section(cfg.get(phase) or {})
7582

76-
for block in blocks:
77-
batch_dict = block(batch_dict)
7883

84+
def apply_postblocks(postblocks: nn.ModuleDict, batch_dict: dict) -> dict:
85+
"""Apply a postblock group built by ``build_postblocks``.
86+
87+
Args:
88+
postblocks: ``nn.ModuleDict`` built by ``build_postblocks`` for a single phase.
89+
batch_dict: dict containing at minimum ``"y_pred"`` and ``"metadata"``.
90+
91+
Returns:
92+
The same ``batch_dict`` after all blocks in the group have run.
93+
"""
94+
for block in postblocks.values():
95+
batch_dict = block(batch_dict)
7996
return batch_dict

credit/postblock/base.py

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,6 @@
44
class BasePostblock(nn.Module):
55
"""Base class for all postblocks.
66
7-
Subclasses that need to run at every rollout step (e.g. conservation fixers)
8-
should set ``per_step = True``. All others leave it at the default ``False``
9-
and will be called once after the full rollout, on the reconstructed dict.
10-
117
Forward signature for all postblocks::
128
139
forward(batch: dict) -> dict
@@ -20,8 +16,6 @@ class BasePostblock(nn.Module):
2016
}
2117
"""
2218

23-
per_step: bool = True
24-
2519
def forward(self, batch: dict) -> dict:
2620
pass
2721

credit/postblock/geopotential.py

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,8 @@ def __init__(
132132
level_info_file: str = "ERA5_Lev_Info.nc",
133133
model_a_half_var: str = "a_half",
134134
model_b_half_var: str = "b_half",
135+
static_source_key: str = "ic_raw",
136+
levels: list[int] | None = None,
135137
):
136138
super().__init__()
137139
self.output_name = output_name
@@ -146,9 +148,18 @@ def __init__(
146148
self.level_info_file = get_meta_file_path(level_info_file)
147149
self.model_a_half_var = model_a_half_var
148150
self.model_b_half_var = model_b_half_var
151+
self.static_source_key = static_source_key
152+
self.levels = levels
149153
with xr.open_dataset(self.level_info_file) as level_info:
150-
self.model_a_half = torch.Tensor(level_info[self.model_a_half_var].values)
151-
self.model_b_half = torch.Tensor(level_info[self.model_b_half_var].values)
154+
a_all = torch.Tensor(level_info[self.model_a_half_var].values)
155+
b_all = torch.Tensor(level_info[self.model_b_half_var].values)
156+
if levels is not None:
157+
half_idx = [lv - 1 for lv in levels] + [levels[-1]]
158+
self.model_a_half = a_all[half_idx]
159+
self.model_b_half = b_all[half_idx]
160+
else:
161+
self.model_a_half = a_all
162+
self.model_b_half = b_all
152163
return
153164

154165
def forward(self, data_dict: dict):
@@ -175,16 +186,20 @@ def forward(self, data_dict: dict):
175186
pred_shape = list(data[self.dataset_name][self.temperature_var].shape) # (B, n_levels, n_time, H, W)
176187
pred_flat = {}
177188
dsn = self.dataset_name
189+
static_data = data_dict[self.static_source_key]
178190
for input_var in [
179191
self.surface_geopotential_var,
180192
self.surface_pressure_var,
181193
self.temperature_var,
182194
self.specific_humidity_var,
183195
]:
184-
new_dim_order = tuple([0] + list(range(2, len(data[dsn][input_var].shape))) + [1])
185-
pred_per = torch.permute(data[dsn][input_var], new_dim_order) # (B, n_time, H, W, n_levels)
196+
src = static_data[dsn] if input_var == self.surface_geopotential_var else data[dsn]
197+
new_dim_order = tuple([0] + list(range(2, len(src[input_var].shape))) + [1])
198+
pred_per = torch.permute(src[input_var], new_dim_order) # (B, n_time, H, W, n_levels)
186199
total_shape = int(np.prod(pred_per.shape[:-1]))
187200
pred_flat[input_var] = pred_per.reshape(total_shape, pred_per.shape[-1])
201+
device = pred_flat[self.surface_pressure_var].device
202+
pred_flat = {k: v.to(device) for k, v in pred_flat.items()}
188203
vgeo = torch.vmap(
189204
partial(geopotential, flip_vertical=self.flip_vertical),
190205
(0, 0, 0, 0, None, None),
@@ -195,8 +210,8 @@ def forward(self, data_dict: dict):
195210
pred_flat[self.surface_pressure_var],
196211
pred_flat[self.temperature_var],
197212
pred_flat[self.specific_humidity_var],
198-
self.model_a_half,
199-
self.model_b_half,
213+
self.model_a_half.to(device),
214+
self.model_b_half.to(device),
200215
).reshape(*[pred_shape[0]] + pred_shape[2:] + [pred_shape[1]]) # (B, n_time, H, W, n_levels)
201216
final_dim_order = tuple([0] + [len(pred_shape) - 1] + list(range(1, len(pred_shape) - 1)))
202217
data[dsn][self.output_name] = torch.permute(geo_out, final_dim_order)

0 commit comments

Comments
 (0)