Add VPU metadata support - #95
Conversation
There was a problem hiding this comment.
Pull request overview
Adds VPU-aware handling for multi-GeoPackage / multi-weight forcing runs, including standardized VPU identifiers and new VPU-delineated metadata outputs, while preserving existing global metadata behavior.
Changes:
- Add VPU ID inference/normalization to align naming across weights generation and forcing processing.
- Write new per-VPU metadata outputs (
metadata_by_vpu.csvplus catchment stats with avpu_idcolumn). - Adjust multiprocessing behavior in weights generation and harden weight-ingestion logging.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| src/forcingprocessor/weights_hf2ds.py | Adds VPU ID inference and uses a spawn multiprocessing context for ProcessPoolExecutor; updates naming logic used to build VPU/catchment mappings. |
| src/forcingprocessor/processor.py | Adds VPU ID normalization + config handling and writes VPU-level metadata outputs; fixes local file size accumulation and adjusts grid extraction logic. |
Comments suppressed due to low confidence (1)
src/forcingprocessor/processor.py:307
- The new condition treats any dataset with
south_north/west_eastdims like retrospective-2-1 for timestamp parsing, but many NWM files (e.g.nwm.t00z.short_range.forcing...) use those dims while the filename is notYYYYMMDDHH.... This will raise aValueErrorindatetime.strptime. Timestamp extraction should prefer themodel_output_valid_timeattribute when available, and only fall back to parsing the filename when that attribute is missing.
if "retrospective-2-1" in nwm_file or ("south_north" in nwm_data.dims and "west_east" in nwm_data.dims):
data_allvars[var_dx, :, :] = np.flip(np.squeeze(nwm_data[jvar].isel(west_east=slice(x_min, x_max+1), south_north=slice(shp[1] - (y_max+1), shp[1] - y_min)).values),axis=0)
t = datetime.strftime(datetime.strptime(nwm_file.split('/')[-1].split('.')[0],'%Y%m%d%H'),'%Y-%m-%d %H:%M:%S')
else:
data_allvars[var_dx, :, :] = np.flip(np.squeeze(nwm_data[jvar].isel(x=slice(x_min, x_max+1), y=slice(shp[1] - (y_max+1), shp[1] - y_min)).values),axis=0)
time_splt = nwm_data.attrs["model_output_valid_time"].split("_")
t = time_splt[0] + " " + time_splt[1]
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| with cf.ProcessPoolExecutor( | ||
| max_workers=nprocs, | ||
| mp_context=mp.get_context("spawn"), | ||
| ) as pool: |
|
Addressed both suggestions. Replaced the assert with a ValueError for validating the forcing.vpu_id and forcing.gpkg_file lengths so the check is always enforced. |
|
Crazy week here at AWI! I will finish my review on Monday. |
quinnylee
left a comment
There was a problem hiding this comment.
Thanks for the PR! I have a couple design comments below. Also, I ran pytests locally and noticed that tests/test_trouterestarts.py::test_restart failed with AssertionError: assert '2024-01-15_13:00:00' == '2024-01-15_12:00:00'
Would probably also be useful to include tests for this new feature
| return name | ||
|
|
||
|
|
||
| def infer_vpu_id(path): |
There was a problem hiding this comment.
This method's only function is to call another method. It would make more sense to delete this method and rename all the calls to infer_vpu_id in processor.py to normalize_vpu_id (or vice versa)
| netcdf_size_by_vpu = {} | ||
| if "netcdf" in output_file_type: | ||
| for i, jvpu in enumerate(jcatchment_dict.keys()): | ||
| if i < len(netcdf_cat_file_sizes_MB): |
There was a problem hiding this comment.
What case would lead to i >= len(netcdf_cat_file_sizes_MB)? Not sure what information is in netcdf_cat_file_sizes_MB that wouldn't be in netcdf_size_by_vpu or vice versa.
| for i, vpu_id in enumerate(vpu_ids): | ||
| if vpu_id in jcatchment_dict: | ||
| vpu_catchment_map[vpu_id] = list(jcatchment_dict[vpu_id]) | ||
| elif i < len(jcatchment_values): |
There was a problem hiding this comment.
What case would lead to i < len(jcatchment_values)? I am not sure why there is a distinction between vpu_catchment_map and jcatchment_dict
| gpd.options.io_engine = "pyogrio" | ||
|
|
||
|
|
||
| def infer_vpu_id(path): |
There was a problem hiding this comment.
This seems to be the same function that is in processor.py. Maybe it would make more sense to move it to utils.py and import it from there, so the code only has to live in one place?
This PR recreates the changes from #90 on a branch within the CIROH-UA repository so the authenticated GitHub Actions can run.
This includes the VPU metadata support changes and the latest review updates.
Replaces #90. Resolves #9.