Skip to content

Add VPU metadata support - #95

Merged
pratikshac15 merged 8 commits into
mainfrom
vpu-metadata-support
Aug 12, 2026
Merged

pratikshac15 merged 8 commits into
mainfrom
vpu-metadata-support

Conversation

@pratikshac15

@pratikshac15 pratikshac15 commented Jul 23, 2026 •

Copy link
Copy Markdown
Collaborator

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.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.csv plus catchment stats with a vpu_id column).
  • 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_east dims 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 not YYYYMMDDHH.... This will raise a ValueError in datetime.strptime. Timestamp extraction should prefer the model_output_valid_time attribute 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.

Comment on lines +148 to +151
with cf.ProcessPoolExecutor(
max_workers=nprocs,
mp_context=mp.get_context("spawn"),
) as pool:

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed. The per-process dictionaries are now merged with unique-key handling instead of dict.update(), so duplicate VPU keys will no longer silently overwrite existing catchment mappings

Comment thread src/forcingprocessor/processor.py Outdated
@JordanLaserGit JordanLaserGit added enhancement New feature or request P1 Priority level. P0: Critical, P1: High, P2: Medium, P3: Low labels Jul 24, 2026
@pratikshac15

Copy link
Copy Markdown
Collaborator Author

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.
Updated the multiprocessing logic to avoid issues with concurrent processing and preserve the expected behavior.

@quinnylee

Copy link
Copy Markdown
Collaborator

Crazy week here at AWI! I will finish my review on Monday.

@quinnylee quinnylee left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Comment thread src/forcingprocessor/processor.py Outdated
return name


def infer_vpu_id(path):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

Comment thread src/forcingprocessor/processor.py Outdated
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):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread src/forcingprocessor/processor.py Outdated
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):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Comment thread src/forcingprocessor/weights_hf2ds.py Outdated
gpd.options.io_engine = "pyogrio"


def infer_vpu_id(path):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

@pratikshac15

Copy link
Copy Markdown
Collaborator Author

Thank you for the review, Quinn. I am working on it.

@pratikshac15

Copy link
Copy Markdown
Collaborator Author

@quinnylee Thanks for the review, Quinn. I’ve addressed the design comments by moving normalize_vpu_id() to utils.py and using it in both processor.py and weights_hf2ds.py, removing the redundant infer_vpu_id() functions, simplifying the NetCDF size mapping, and removing vpu_catchment_map in favor of using jcatchment_dict directly.

I also added a test for the VPU normalization behavior. I can add additional coverage for the per-VPU metadata output as well if you think that would be useful for this PR. Let me know.

For the test_trouterestarts.py::test_restart failure, I reproduced the same one-hour offset on the current main branch, so it appears unrelated to these changes, and I left the restart logic unchanged.

@quinnylee

Copy link
Copy Markdown
Collaborator

Thanks for the update! I have visually inspected the code and am rerunning the pytests right now. A new test for the per-VPU metadata output would be good. It might belong in test_forcingprocessor.py and might take this structure:

def test_nomads_prod(download_weight_file,clean_forcings_metadata_dirs):
    nwmurl_conf['start_date'] = TODAY_START
    nwmurl_conf['end_date']   = TODAY_START    
    nwmurl_conf["urlbaseinput"] = 1
    generate_nwmfiles(nwmurl_conf)  
    conf['run']['collect_stats'] = True # test metadata generation once
    prep_ngen_data(conf)
    conf['run']['collect_stats'] = False
    assert_file=(data_dir/f"forcings/ngen.t00z.short_range.forcing.f001_f001.VPU_09.nc").resolve()
    assert assert_file.exists()
    os.remove(assert_file)   

@pratikshac15

Copy link
Copy Markdown
Collaborator Author

Thanks, Quinn! I added the per-VPU metadata output test to test_forcingprocessor.py following the given structure. I ran it locally, and it passed successfully.

@quinnylee quinnylee left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the updates!

@JordanLaserGit JordanLaserGit left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks great @pratikshac15, thanks for this work!

@ZacharyWills

Copy link
Copy Markdown

This is going to help a lot with eval and forecast changes without interrupting other testing and eval, thanks for this work!

@pratikshac15
pratikshac15 merged commit f5a9ebb into main Aug 12, 2026
45 of 47 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

enhancement New feature or request P1 Priority level. P0: Critical, P1: High, P2: Medium, P3: Low

Projects

Development

Successfully merging this pull request may close these issues.

Metadata

5 participants