[2pt] PR: Changes to pass optional arg for using the precalb_discharge_cms for inundate scripts

docs/CHANGELOG.md

@@ -1,6 +1,18 @@
 All notable changes to this project will be documented in this file.
 We follow the [Semantic Versioning 2.0.0](http://semver.org/) format.
 
+## v4.8.X.X - 2025-09-02 - [PR#1606](https://github.com/NOAA-OWP/inundation-mapping/pull/1606)
+
+Added new optional input argument to inundate scripts and `synthesize_test_cases.py` to use the `precalb_discharge_cms` values in the hydrotable instead of the default `discharge_cms`.
+
+### Changes
+- `tools/inundate_gms.py`: added "precalb_option", specify "precalb_discharge_cms" as an req input column and force the hydrotable input to be the csv file ("precalb_discharge_cms" is not currently one of the outputs in the feather file). Note that there is logic to fill the "precalb_discharge_cms" with "discharge_cms" when the precalb_discharge is null (this happens for hucs/branches where the calibration routines do not run - not benchmark data to calibrate to)
+- `tools/inundate_mosaic_wrapper.py`: added the "precalb_option" argument for passing to downstream functions
+- `tools/inundation.py`: added the "precalb_option" argument and added logic to use the "precalb_discharge_cms" data for interpolating the stages.
+- `tools/run_test_case.py`: added the "precalb_option" argument for passing to downstream functions
+- `tools/synthesize_test_cases.py`: added the "precalb_option" as an optional input argument and pass to the appropriate downstream functions
+<br/>
+
 ## v4.8.10.3 - 2025-08-29 - [PR#1627](https://github.com/NOAA-OWP/inundation-mapping/pull/1627)
 
 Adds gcsfs dependency to allow retrieval of NWM output from the Google Cloud Service.

tools/inundate_gms.py

@@ -26,6 +26,7 @@ def Inundate_gms(
     verbose: Optional[bool] = False,
     log_file: Optional[str] = None,
     output_fileNames: Optional[str] = None,
+    precalb_option: Optional[bool] = False,
     windowed: Optional[bool] = False,
     multi_process: Optional[bool] = False,
 ) -> pd.DataFrame:
@@ -54,6 +55,8 @@ def Inundate_gms(
         Name of file to log output
     output_fileNames: Optional[str], default = None
         Name of file to output filenames from gms inundation routine
+    precalb_option: Optional[bool], default = False
+        Whether to use precalb discharge in hydrotable
     windowed: Optional[bool], default = False
         Whether to use window memory optimization
     multi_process: Optional[bool], default = False
@@ -109,6 +112,7 @@ def Inundate_gms(
         hydro_table_df,
         verbose=False,
         windowed=windowed,
+        precalb_option=precalb_option,
     )
 
     # start up process pool
@@ -205,6 +209,7 @@ def __inundate_gms_generator(
     forecast: Union[str, pd.DataFrame],
     hydro_table_df: Union[str, pd.DataFrame],
     verbose: Optional[bool] = False,
+    precalb_option: Optional[bool] = False,
     windowed: Optional[bool] = False,
 ) -> Tuple[dict, List[str]]:
     """
@@ -226,6 +231,8 @@ def __inundate_gms_generator(
         Hydrotable DataFrame.
     verbose: Optional[bool], default = False
         Whether to silence output or not
+    precalb_option: Optional[bool], default = False
+        Whether to use precalb discharge in hydrotable
     windowed: Optional[bool], default = False
         Whether to use window memory optimization
 
@@ -249,9 +256,6 @@ def __inundate_gms_generator(
         catchments_file_name = f"gw_catchments_reaches_filtered_addedAttributes_{branch_id}.tif"
         catchments_branch = os.path.join(branch_dir, catchments_file_name)
 
-        # FIM versions > 4.3.5 use an aggregated hydrotable file rather than individual branch hydrotables
-        htable_req_cols = ["HUC", "branch_id", "feature_id", "HydroID", "stage", "discharge_cms", "LakeID"]
-
         if isinstance(hydro_table_df, pd.DataFrame):
             hydro_table_all = hydro_table_df.set_index(["HUC", "feature_id", "HydroID"], inplace=False)
             hydro_table_branch = hydro_table_all.loc[hydro_table_all["branch_id"] == int(branch_id)]
@@ -265,19 +269,43 @@ def __inundate_gms_generator(
                 "feature_id": str,
                 "HydroID": str,
                 "stage": float,
+                "precalb_discharge_cms": float,
                 "discharge_cms": float,
                 "LakeID": int,
             }
 
-            if s3_or_local_path_exists(os.path.join(huc_dir, "hydrotable.feather")):  # Quicker reads
+            if (
+                s3_or_local_path_exists(os.path.join(huc_dir, "hydrotable.feather"))
+                and precalb_option == False
+            ):  # Quicker reads
                 hydro_table_huc = os.path.join(huc_dir, "hydrotable.feather")
                 hydro_table_all = pd.read_feather(hydro_table_huc)
             elif s3_or_local_path_exists(os.path.join(huc_dir, "hydrotable.csv")):
                 hydro_table_huc = os.path.join(huc_dir, "hydrotable.csv")
+                # FIM versions > 4.3.5 use an aggregated hydrotable file rather than individual branch hydrotables
+                htable_req_cols = [
+                    "HUC",
+                    "branch_id",
+                    "feature_id",
+                    "HydroID",
+                    "stage",
+                    "precalb_discharge_cms",
+                    "discharge_cms",
+                    "LakeID",
+                ]
                 hydro_table_all = pd.read_csv(hydro_table_huc, dtype=dtype, usecols=htable_req_cols)
             else:
                 hydro_table_huc = None
 
+            if precalb_option:
+                if "precalb_discharge_cms" not in hydro_table_all.columns:
+                    raise ValueError("Missing expected column 'precalb_discharge_cms' in hydrotable.")
+                missing_count = hydro_table_all["precalb_discharge_cms"].isna().sum()
+                if missing_count > 0:
+                    hydro_table_all["precalb_discharge_cms"].fillna(
+                        hydro_table_all["discharge_cms"], inplace=True
+                    )
+
             if hydro_table_huc is not None and s3_or_local_isfile(hydro_table_huc):
                 hydro_table_all.set_index(["HUC", "feature_id", "HydroID"], inplace=True)
                 hydro_table_branch = hydro_table_all.loc[hydro_table_all["branch_id"] == int(branch_id)]
@@ -319,6 +347,7 @@ def __inundate_gms_generator(
             "inundation_raster": inundation_branch_raster,
             "depths": depths_branch_raster,
             "quiet": not verbose,
+            "precalb_option": precalb_option,
             "windowed": windowed,
         }
 

tools/inundate_mosaic_wrapper.py

@@ -29,6 +29,7 @@ def produce_mosaicked_inundation(
     verbose: Optional[bool] = False,
     is_mosaic_for_branches: Optional[bool] = False,
     num_threads: Optional[int] = 1,
+    precalb_option: Optional[bool] = False,
     windowed: Optional[bool] = False,
     log_file: Optional[str] = None,
     nodata: Optional[int] = elev_raster_ndv,
@@ -73,6 +74,8 @@ def produce_mosaicked_inundation(
         Whether the mosaic routine is for branches
     num_threads : Optional[int], default=1
         Number of threads to process
+    precalb_option : Optional[bool], default=False
+        Whether to use precalb discharge in hydrotable. If True, will use precalb_discharge_cms column
     windowed : Optional[bool], default=False
         Memory conscious creation of inundation and depth datasets
     log_file : Optional[str], default=None
@@ -139,6 +142,7 @@ def produce_mosaicked_inundation(
         inundation_raster=inundation_raster,
         depths_raster=depths_raster,
         verbose=verbose,
+        precalb_option=precalb_option,
         windowed=windowed,
         log_file=log_file,
         multi_process=gms_multi_process,

tools/inundation.py

@@ -47,6 +47,7 @@ def inundate(
     depths: Optional[str] = None,
     src_table: Optional[str] = None,
     quiet: Optional[bool] = False,
+    precalb_option: Optional[bool] = False,
     windowed: Optional[bool] = False,
 ) -> Tuple[List[str], List[str], List[str]]:
     """
@@ -93,6 +94,8 @@ def inundate(
         Table to subset main hydrotable.
     quiet : Optional[bool], default=False
         Quiet output.
+    precalb_option : Optional[bool], default=False
+        Whether to use precalb discharge in hydrotable. If True, will use precalb_discharge_cms column
     windowed : Optional[bool], default=False
         Memory efficient operation to process inundation
 
@@ -175,7 +178,9 @@ def inundate(
 
     # catchment stages dictionary
     if hydro_table is not None:
-        catchmentStagesDict, hucSet = __subset_hydroTable_to_forecast(hydro_table, forecast, subset_hucs)
+        catchmentStagesDict, hucSet = __subset_hydroTable_to_forecast(
+            hydro_table, forecast, subset_hucs, precalb_option
+        )
     else:
         raise TypeError("Pass hydro table csv")
 
@@ -567,7 +572,10 @@ def __append_huc_code_to_file_name(fileName: str, hucCode: str) -> str:
 
 
 def __subset_hydroTable_to_forecast(
-    hydroTable: Union[str, pd.DataFrame], forecast: Union[str, pd.DataFrame], subset_hucs=None
+    hydroTable: Union[str, pd.DataFrame],
+    forecast: Union[str, pd.DataFrame],
+    subset_hucs=None,
+    precalb_option: bool = False,
 ) -> Tuple[typed.Dict, List[str]]:
     """
     Subset hydrotable with forecast
@@ -588,7 +596,15 @@ def __subset_hydroTable_to_forecast(
 
     """
     if isinstance(hydroTable, str):
-        htable_req_cols = ['HUC', 'feature_id', 'HydroID', 'stage', 'discharge_cms', 'LakeID']
+        htable_req_cols = [
+            'HUC',
+            'feature_id',
+            'HydroID',
+            'stage',
+            'precalb_discharge_cms',
+            'discharge_cms',
+            'LakeID',
+        ]
         file_ext = hydroTable.split('.')[-1]
         if file_ext == 'csv':
             hydroTable = pd.read_csv(
@@ -598,6 +614,7 @@ def __subset_hydroTable_to_forecast(
                     'feature_id': str,
                     'HydroID': str,
                     'stage': float,
+                    'precalb_discharge_cms': float,
                     'discharge_cms': float,
                     'LakeID': int,
                     'last_updated': object,
@@ -683,11 +700,18 @@ def __subset_hydroTable_to_forecast(
 
         # interpolate stages
         for hid, sub_table in hydroTable.groupby(level='HydroID'):
-            interpolated_stage = np.interp(
-                sub_table.loc[:, 'discharge'].unique(),
-                sub_table.loc[:, 'discharge_cms'],
-                sub_table.loc[:, 'stage'],
-            )
+            if precalb_option:
+                interpolated_stage = np.interp(
+                    sub_table.loc[:, 'discharge'].unique(),
+                    sub_table.loc[:, 'precalb_discharge_cms'],
+                    sub_table.loc[:, 'stage'],
+                )
+            else:
+                interpolated_stage = np.interp(
+                    sub_table.loc[:, 'discharge'].unique(),
+                    sub_table.loc[:, 'discharge_cms'],
+                    sub_table.loc[:, 'stage'],
+                )
 
             # add this interpolated stage to catchment stages dict
             h = round(interpolated_stage[0], 4)

tools/run_test_case.py

@@ -192,6 +192,7 @@ def alpha_test(
         overwrite=True,
         verbose=False,
         gms_workers=1,
+        precalb_option=False,
         threads=8,
     ):
         '''Compares a FIM directory with benchmark data from a variety of sources.
@@ -285,6 +286,7 @@ def alpha_test(
                         model=model,
                         verbose=verbose,
                         gms_workers=gms_workers,
+                        precalb_option=precalb_option,
                         threads=threads,
                     )
 
@@ -305,7 +307,15 @@ def alpha_test(
             sys.exit(1)
 
     def _inundate_and_compute(
-        self, magnitude, lid, compute_only=False, model='', verbose=False, gms_workers=1, threads=8
+        self,
+        magnitude,
+        lid,
+        precalb_option,
+        compute_only=False,
+        model='',
+        verbose=False,
+        gms_workers=1,
+        threads=8,
     ):
         '''Method for inundating and computing contingency rasters as part of the alpha_test.
         Used by both the alpha_test() and composite() methods.
@@ -367,6 +377,7 @@ def _inundate_and_compute(
                     verbose=verbose,
                     num_threads=threads,
                     num_workers=gms_workers,
+                    precalb_option=precalb_option,
                     windowed=True,
                     gms_multi_process=True,
                 )
@@ -417,6 +428,7 @@ def run_alpha_test(
         magnitude,
         calibrated,
         model,
+        precalb_option=False,
         archive_results=False,
         mask_type='huc',
         inclusion_area='',
@@ -439,6 +451,7 @@ def run_alpha_test(
             overwrite,
             verbose,
             gms_workers,
+            precalb_option,
             threads,
         )
 

tools/synthesize_test_cases.py

@@ -374,6 +374,14 @@ def progress_bar_handler(executor_dict, verbose, desc):
         default=False,
         action='store_true',
     )
+    parser.add_argument(
+        '-p',
+        '--precalb-option',
+        help='Using this argument will use the precalb_discharge_cms in hydrotable. ',
+        required=False,
+        default=False,
+        action='store_true',
+    )
     parser.add_argument(
         '-e',
         '--model',
@@ -508,6 +516,7 @@ def progress_bar_handler(executor_dict, verbose, desc):
     master_metrics_csv = args['master_metrics_csv']
     fr_run_dir = args['fr_run_dir']
     calibrated = args['calibrated']
+    precalb_option = args['precalb_option']
     model = args['model']
     verbose = bool(args['verbose'])
     gms_verbose = bool(args['gms_verbose'])
@@ -628,6 +637,7 @@ def progress_bar_handler(executor_dict, verbose, desc):
                     'overwrite': overwrite,
                     'verbose': gms_verbose if model == 'GMS' else verbose,
                     'gms_workers': job_number_branch,
+                    'precalb_option': precalb_option,
                     'threads': thread_number_branch,
                 }
 
@@ -667,6 +677,7 @@ def progress_bar_handler(executor_dict, verbose, desc):
                         'mask_type': 'huc',
                         'verbose': verbose,
                         'overwrite': overwrite,
+                        'precalb_option': precalb_option,
                     }
                     try:
                         future = executor.submit(test_case_class.alpha_test, **alpha_test_args)