updating nexus on parquet file

src/troute-network/troute/NHDNetwork.py

@@ -66,12 +66,13 @@ def __init__(
         self._flowpath_dict = {}
         self._gl_climatology_df = pd.DataFrame()
         self._canadian_gage_link_df = pd.DataFrame(columns=['gages','link']).set_index('link')
-
+        self._nexus_dict = dict()
         super().__init__()
 
         # Create empty dataframe for coastal_boundary_depth_df. This way we can check if
         # it exists, and only read in SCHISM data during 'assemble_forcings' if it doesn't
         self._coastal_boundary_depth_df = pd.DataFrame()
+
 
     def extract_waterbody_connections(rows, target_col, waterbody_null=-9999):
         """Extract waterbody mapping from dataframe.

src/troute-nwm/src/nwm_routing/output.py

@@ -6,7 +6,7 @@
 import troute.nhd_io as nhd_io
 from build_tests import parity_check
 import logging
-
+from troute.nhd_io import updated_flowveldepth
 
 LOG = logging.getLogger('')
 
@@ -47,7 +47,7 @@ def _reindex_lake_to_link_id(target_df, crosswalk):
     return target_df
 
 
-def _parquet_output_format_converter(df, start_datetime, dt, configuration, prefix_ids):
+def _parquet_output_format_converter(df, start_datetime, dt, configuration, prefix_ids, nexus_dict):
     '''
     Utility function for convert flowveldepth dataframe
     to a timeseries and to match parquet input format
@@ -64,13 +64,24 @@ def _parquet_output_format_converter(df, start_datetime, dt, configuration, pref
     --------
     - timeseries_df (DataFrame): Converted timeseries data frame
     '''
+    nex_id = {}
+    if prefix_ids == 'nex' and nexus_dict:
+        for key, val in nexus_dict.items():
+            nex_key = int(key.split('-')[-1])
+            nex_id[nex_key] = [int(v.split('-')[-1]) for v in val] 
+
+    df = updated_flowveldepth(df, nex_id, seg_id = list(), mask_list = None)
+    df = df.reset_index().drop('Type', axis=1).set_index('featureID')
     variable_to_name_map = {"q": "streamflow", "d": "depth", "v": "velocity"}
     variable_to_units_map = {"streamflow": "m3/s", "velocity": "m/s", "depth": "m"}
 
     # Prepare the location_id with prefix
     df.index.name = 'location_id'
     df.reset_index(inplace=True)
-    location_ids = prefix_ids + '-' + df['location_id'].astype(str)
+    if nexus_dict:
+        location_ids = prefix_ids + '-' + df['location_id'].astype(str)
+    else:
+        location_ids = df['location_id'].astype(str)    
 
     # Flatten the dataframe using NumPy
     num_locations = df.shape[0]
@@ -502,7 +513,7 @@ def nwm_output_generator(
         configuration = output_parameters["parquet_output"].get("configuration")
         prefix_ids = output_parameters["parquet_output"].get("prefix_ids")
         timeseries_df = _parquet_output_format_converter(flowveldepth, restart_parameters.get("start_datetime"), dt,
-                                                         configuration, prefix_ids)
+                                                         configuration, prefix_ids, nexus_dict)
 
         parquet_output_segments_str = [prefix_ids + '-' + str(segment) for segment in parquet_output_segments]
         timeseries_df.loc[timeseries_df['location_id'].isin(parquet_output_segments_str)].to_parquet(

test/LowerColorado_TX/test_AnA_V4_NHD.yaml

@@ -105,7 +105,7 @@ compute_parameters:
                 reservoir_rfc_forecasts_lookback_hours  : 48
 
 #--------------------------------------------------------------------------------
-output_parameters:
+# output_parameters:
 #     #----------
 #     test_output: output/lcr_flowveldepth.pkl
 #     lite_restart:
@@ -124,10 +124,10 @@ output_parameters:
 #         stream_output_time: 1 #[hr]
 #         stream_output_type: '.nc' #please select only between netcdf '.nc' or '.csv' or '.pkl'
 #         stream_output_internal_frequency: 30 #[min] it should be order of 5 minutes. For instance if you want to output every hour put 60
-    parquet_output:
-        #---------
-        parquet_output_folder: output/
-        configuration: short_range
-        prefix_ids: nex
+#     parquet_output:
+#         #---------
+#         parquet_output_folder: output/
+#         configuration: short_range
+#         prefix_ids: nex