routing-only datastream inputs

[quinnylee]

It would be useful to produce the inputs for the routing-only datastream (see this issue in ngen-datastream) using the forcingprocessor.

Here's what it would need to do:

• download (or read) the channel routing forecast files from the NWM (can be done using NWMURL)
• These files link lateral runoff values to the NHD IDs. However, for NRDS, since we're using the NextGen hydrofabric, we'll have to do a remapping using this map I generated update: use the NEXUS box link in that issue for the most accurate mapping
• Write the lateral runoff values into a directory system and files readable by t-route

[JordanLaserGit]

Hi @quinnylee, here is what I think would need to happen in order for this to work.

1. We build a conf_fp like this:

{
    "forcing"  : {
        "nwm_file"   : "/home/user/data/filenamelist.txt",    // Assume a file exists on nwm channel routing files
        "gpkg_file"  : ["ngenhydrofabric_NHD_mapping.json"]   // This typically is a gpkg or weight file, but we could provide the map file via this field. 
    },

    "storage":{
        "storage_type"      : "s3",
        "output_path"       : "ciroh-community-ngen-datastream/outputs/nwm_troute/test",   
        "output_file_type" : ["parquet"]            
    },    

    "run" : {
        "verbose"       : true,
        "collect_stats" : false,
        "nprocs"        : 8
    }
}

---

2. Then inside processor.py,

forcingprocessor/src/forcingprocessor/processor.py

Lines 712 to 732 in cf12d6d

	log_time("READWEIGHTS_START", log_file)
	tw = time.perf_counter()
	if ii_verbose: print(f'Obtaining weights\n',flush=True)
	global weights_df
	weights_df, jcatchment_dict = multiprocess_hf2ds(gpkg_files,nwm_forcing_files[0],nprocs)
	log_time("READWEIGHTS_END", log_file)
	# # conus hack
	# x = {}
	# x_list = []
	# for jdict in jcatchment_dict:
	# [x_list.append(x) for x in jcatchment_dict[jdict]]
	# x['conus'] = x_list
	# jcatchment_dict = x
	log_time("CALC_WINDOW_START", log_file)
	ncatchments = len(weights_df)
	global x_min, x_max, y_min, y_max
	x_min, x_max, y_min, y_max = get_window(weights_df)
	weight_time = time.perf_counter() - tw
	log_time("CALC_WINDOW_END", log_file)

A check can exist to skip the weight generation step and windowing step and instead read your mappings into mappings_df. This file will be saved in the metadata, so a check will need to exist here to write the mappings to metadata.

forcingprocessor/src/forcingprocessor/processor.py

Line 776 in cf12d6d

weights_df.to_parquet(buf, index=False)

forcingprocessor/src/forcingprocessor/processor.py

Lines 774 to 778 in cf12d6d

	buf = BytesIO()
	filename = metaf_path + f"/weights.parquet"
	weights_df.to_parquet(buf, index=False)
	buf.seek(0)
	s3.put_object(Bucket=bucket, Key="/".join(filename.split('/')[3:]), Body=buf.getvalue())

---

3. Multiprocessing step

This line is where the main function prep_ngen_data executes the core processing

forcingprocessor/src/forcingprocessor/processor.py

Line 830 in cf12d6d

data_array, t_ax, nwm_data, nwm_file_sizes_MB = multiprocess_data_extract(nwm_forcing_files,nprocs,weights_df,fs)

Will need to pass your mappings instead of the weights. For some curious reason I made the weights_df variable global. Each process will get it's own copy anyway, so I'm not sure why I did this originally. It's better practice that the variables be passed explicitly to the multiprocess_data_extract() function and then on to forcing_grid2_catchment().

---

4. Core regridding function forcing_grid2catchment()

This line is where data is read from the netcdf

forcingprocessor/src/forcingprocessor/processor.py

Line 195 in cf12d6d

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)

And this line is where the the nextgen catchment averaged forcings are calculated.

forcingprocessor/src/forcingprocessor/processor.py

Line 208 in cf12d6d

for row in weights_df.itertuples():

These steps may not be easily generalized to your application. The channel routing data is not gridded and no weight calculation needs to be performed.

---

5. Write the data out in parquet. This can be done with

forcingprocessor/src/forcingprocessor/processor.py

Line 337 in cf12d6d

def write_data(

with little changes to the function.

---

@quinnylee after writing this all out, I'm thinking you may want to write a function specifically for reading the channel routing files and doing the remapping. A substitute for forcing_grid2catchment(). Something like channelrouting_nwm2ngen(). If you don't want to mess with any of the metadata, we can leave that option off.