Jwade reservoir da

README.md

@@ -532,6 +532,14 @@ Parameters for configuring NWM output variables.
 | `nwm_output_variables` | bool | No | Boolean flag to activate output of full set of NWM output variables |
 | `output_format` | bool | No | Output format(s) for model outputs. Accepts 'CSV', 'NetCDF', or both |
 
+## Data Assimilation Section `[DataAssimilation]`
+Parameters for reservoir RFC and streamflow data assimilation. This section is optional and only supported for **default**, **forecast**, and **regionalization** run types.
+
+| Parameter | Type | Required | Description |
+|-----------|------|----------|-------------|
+| `reservoir_da` | bool | No | Boolean flag to enable reservoir RRC data assimilation in t-route |
+| `reservoir_rfc_dir` | str | No | Directory containing reservoir RFC forecast files. Required if `reservoir_da` is `True`. |
+
 
 ## Forcing Section: `[Forcing]`
 Parameters for forcing engine configuration. Required for all run types, including forecast.
@@ -609,6 +617,7 @@ Parameters for parallel processing configuration.
 - **Calibration runs** require all parameters in the General, Calibration, Forcing, and DataFile sections
 - **Regionalization runs** require parameters in the General, Calibration, Forcing, Regionalization, DataFile section
 - **Default runs** require all parameters in the General, Forcing, and DataFile sections
+-  The **`[DataAssimilation]` section** is optional and only applies to **default**, **forecast**, and **regionalization** run types.
 - Parameters for unused run types can be left blank
 
 ### Datetime Format

src/mswm/build_inputs.py

@@ -529,6 +529,7 @@ def _parse_config(self):
         # Retrieve input.config sections
         self.conf3 = self.input_configs.get('DataFile')
         self.forcingSec = self.input_configs.get('Forcing')
+        self.daSec = self.input_configs.get('DataAssimilation')
         self.parallelSec = self.input_configs.get('Parallel')
 
     def _load_realization(self):
@@ -1553,7 +1554,7 @@ def _update_fcst_troute(self):
         """
         Update BMI config files for t-route for forecast period
         """
-        self.real_config = gfun.update_troute(self.real_config, self.input_dir, self.basename_opt)
+        self.real_config = gfun.update_troute(self.real_config, self.input_dir, self.basename_opt, self.daSec)
         logger.info("Updated t-route file for forecast")
 
     def _create_bmi_configs(self, is_regionalization: bool = False):
@@ -1702,7 +1703,7 @@ def _create_bmi_configs(self, is_regionalization: bool = False):
                     gfun.create_topoflow_glacier_input(cat_mod, self.divides_df, mod_input_dir)
                 elif m1 == 'troute':
                     routing_config_file = os.path.join(self.work_dir + '/Input', '{}'.format(self.basin))
-                    gfun.create_troute_config(self.cat_file, self.time_period, routing_config_file, self.run_configs, self.run_type)
+                    gfun.create_troute_config(self.cat_file, self.time_period, routing_config_file, self.run_configs, self.daSec, self.run_type)
 
                     if m1 != 'troute':
                         logger.info(f'{m1}: input config files created at: {mod_input_dir}')

src/mswm/example_inputs/calibration/input_calibration.config

@@ -2,12 +2,13 @@
 # Main Input Configuration File
 #
 # This file is used to create the input config files and realization file used to run Ngen.
-# It contains five sections:
+# It contains the following sections:
 #    1. General
 #    2. Regionalization
 #    3. Calibration
 #    4. Forcing
 #    5. DataFile
+#    6. Parallel
 # Variables in the sections for unused run types (i.e. calibration section for a regionalization run) are not loaded
 # ------------------------------------------------
 

src/mswm/example_inputs/default/input_default.config

@@ -2,15 +2,16 @@
 # Main Input Configuration File
 #
 # This file is used to create the input config files and realization file used to run Ngen.
-# It contains five sections:
+# It contains the following sections:
 #    1. General
 #    2. Regionalization
 #    3. Calibration
 #    4. Forcing
 #    5. DataFile
+#    6. DataAssimilation
+#    7. Parallel
 # Variables in the sections for unused run types (i.e. calibration section for a regionalization run) are not loaded
 # ------------------------------------------------
-
 [General]
 
 # Stream gage ID at basin outlet or VPU basin
@@ -150,6 +151,9 @@ snow_17_lib = ~/ngwpc/ngen/extern/snow17/cmake_build/libsnow17bmi.so
 topmodel_lib = ~/ngwpc/ngen/extern/topmodel/cmake_build/libtopmodelbmi.so
 ueb_lib = ~/ngwpc/ngen/extern/ueb-bmi/cmake_build/src/libbmiuebcxx.so
 
+[DataAssimilation]
+reservoir_da = True
+reservoir_rfc_dir = ~/ngwpc/rfc_da
 
 [Parallel]
 parallel_ngen_exe = ~/ngwpc/ngen/cmake_build/ngen

src/mswm/example_inputs/forecast/input_forecast.config

@@ -2,12 +2,14 @@
 # Main Input Configuration File
 #
 # This file is used to create the input config files and realization file used to run Ngen.
-# It contains five sections:
+# It contains the following sections:
 #    1. General
 #    2. Regionalization
 #    3. Calibration
 #    4. Forcing
 #    5. DataFile
+#    6. DataAssimilation
+#    7. Parallel
 # Variables in the sections for unused run types (i.e. calibration section for a regionalization run) are not loaded
 # ------------------------------------------------
 
@@ -67,3 +69,8 @@ smp_lib = ~/ngwpc/ngen/extern/SoilMoistureProfiles/cmake_build/libsmpbmi.so
 snow_17_lib = ~/ngwpc/ngen/extern/snow17/cmake_build/libsnow17bmi.so
 topmodel_lib = ~/ngwpc/ngen/extern/topmodel/cmake_build/libtopmodelbmi.so
 ueb_lib = ~/ngwpc/ngen/extern/ueb-bmi/cmake_build/src/libbmiuebcxx.so
+
+
+[DataAssimilation]
+reservoir_da = True
+reservoir_rfc_dir = ~/ngwpc/rfc_da

src/mswm/example_inputs/regionalization/input_regionalization.config

@@ -2,12 +2,14 @@
 # Main Input Configuration File
 #
 # This file is used to create the input config files and realization file used to run Ngen.
-# It contains five sections:
+# It contains the following sections:
 #    1. General
 #    2. Regionalization
 #    3. Calibration
 #    4. Forcing
 #    5. DataFile
+#    6. DataAssimilation
+#    7. Parallel
 # Variables in the sections for unused run types (i.e. calibration section for a regionalization run) are not loaded
 # ------------------------------------------------
 
@@ -153,6 +155,9 @@ snow_17_lib = ~/ngwpc/ngen/extern/snow17/cmake_build/libsnow17bmi.so
 topmodel_lib = ~/ngwpc/ngen/extern/topmodel/cmake_build/libtopmodelbmi.so
 ueb_lib = ~/ngwpc/ngen/extern/ueb-bmi/cmake_build/src/libbmiuebcxx.so
 
+[DataAssimilation]
+reservoir_da = True
+reservoir_rfc_dir = ~/ngwpc/rfc_da
 
 [Parallel]
 parallel_ngen_exe = ~/ngwpc/ngen/cmake_build/ngen

src/mswm/utils/ginputfunc.py

@@ -1578,7 +1578,8 @@ def create_topmodel_input(
 def update_troute(
         real_config: dict,
         run_dir: Path,
-        basename_opt: str
+        basename_opt: str,
+        da_sec: dict,
 ) -> dict:
     """
     For t-route, create new BMI config file with adjusted start/end times, and then
@@ -1589,6 +1590,7 @@ def update_troute(
     real_config: dictionary containing the realization configuration
     run_dir: folder for the new troute output file
     basename_opt: new file basename for forecast or cold start
+    da_sec: dictionary containing data assimilation inputs
 
     Returns
     -------
@@ -1631,6 +1633,29 @@ def update_troute(
     rt_config['compute_parameters']['forcing_parameters']['max_loop_size'] = max_loop_size
     rt_config['output_parameters']['stream_output']['stream_output_time'] = max_loop_size
 
+    # update reservoir data assimilation parameters if supplied
+    reservoir_da = da_sec.get('reservoir_da', False) if da_sec else False
+    reservoir_rfc_dir = da_sec.get('reservoir_rfc_dir') if da_sec else None
+
+    if reservoir_da:
+        rt_config.setdefault('compute_parameters', {}).setdefault('data_assimilation_parameters', {})
+        rt_config['compute_parameters']['data_assimilation_parameters']['reservoir_da'] = {
+            "reservoir_persistence_da": {
+                "reservoir_persistence_greatLake": False,
+                "reservoir_persistence_usace": False,
+                "reservoir_persistence_usbr": False,
+                "reservoir_persistence_usgs": False,
+            },
+            "reservoir_rfc_da": {
+                "reservoir_rfc_forecast_persist_days": 11,
+                "reservoir_rfc_forecasts": True,
+                "reservoir_rfc_forecasts_lookback_hours": 28,
+                "reservoir_rfc_forecasts_offset_hours": 0,
+                "reservoir_rfc_forecasts_time_series_path": str(reservoir_rfc_dir),
+            }
+        }
+        logger.info("RFC reservoir data assimilation activated.")
+
     # write to new t-route config file
     new_basename = os.path.basename(src).replace("valid_best", basename_opt)
 
@@ -1659,6 +1684,7 @@ def create_troute_config(
         time_period: dict,
         rt_cfg_file: Union[str, Path],
         run_configs: List[str],
+        da_sec: dict,
         run_type: str
 ) -> None:
     """ Create routing configuration YAML file
@@ -1669,6 +1695,7 @@ def create_troute_config(
     time_period: simulation time period
     rt_cfg_file : t-route configuration YAML file
     run_configs: list of file name suffixes for varying run types
+    da_sec: dictionary containing data assimilation inputs
     run_type: type of run (calib, regionalization, or default)
 
     Returns
@@ -1684,6 +1711,10 @@ def create_troute_config(
     }
     run_names = run_type_map.get(run_type)
 
+    # Retrieve reservoir da parameters
+    reservoir_da = da_sec.get('reservoir_da', False) if da_sec else False
+    reservoir_rfc_dir = da_sec.get('reservoir_rfc_dir') if da_sec else None
+
     # Set base log parameters
     log_param = {
         "showtiming": True,
@@ -1707,14 +1738,32 @@ def create_troute_config(
         "diffusive_streamflow_nudging": False,
     }
 
-    res_da = {
-        "reservoir_persistence_da": {
-            "reservoir_persistence_usgs": False,
-        },
-        "reservoir_rfc_da": {
-            "reservoir_rfc_forecasts": False,
-        },
-    }
+    if reservoir_da and run_type != "calibration":
+        res_da = {
+            "reservoir_persistence_da": {
+                "reservoir_persistence_greatLake": False,
+                "reservoir_persistence_usace": False,
+                "reservoir_persistence_usbr": False,
+                "reservoir_persistence_usgs": False,
+            },
+            "reservoir_rfc_da": {
+                "reservoir_rfc_forecast_persist_days": 11,
+                "reservoir_rfc_forecasts": True,
+                "reservoir_rfc_forecasts_lookback_hours": 28,
+                "reservoir_rfc_forecasts_offset_hours": 0,
+                "reservoir_rfc_forecasts_time_series_path": str(reservoir_rfc_dir),
+            }
+        }
+        logger.info("RFC reservoir data assimilation activated.")
+    else:
+        res_da = {
+            "reservoir_persistence_da": {
+                "reservoir_persistence_usgs": False,
+            },
+            "reservoir_rfc_da": {
+                "reservoir_rfc_forecasts": False,
+            },
+        }
 
     for file_name, run_name in zip(run_configs, run_names):
         if not len(time_period['run_time_period'][run_name][0]) != 0 & len(time_period['run_time_period'][run_name][0]):
@@ -1751,6 +1800,7 @@ def create_troute_config(
 
         # Set output_parameters
         output_param = {
+            'lakeout_output': ".",
             'stream_output': {
                 'stream_output_directory': ".",
                 'stream_output_time': max_loop_size,

src/mswm/utils/input_configuration.py

@@ -357,6 +357,23 @@ class DataFileConfig(StrictBaseModel):
     ueb_lib: Optional[str] = None
 
 
+class DataAssimilationConfig(StrictBaseModel):
+    """
+    Input.config DataAssimilation section requirement
+    """
+    reservoir_da: bool = False
+    reservoir_rfc_dir: Optional[str] = None
+
+    @model_validator(mode="after")
+    def validate_reservoir_rfc_dir(self):
+        if self.reservoir_da:
+            if not self.reservoir_rfc_dir:
+                raise ValueError("reservoir_da is True, but reservoir_rfc_dir was not provided")
+            if not Path(self.reservoir_rfc_dir).exists():
+                raise ValueError(f"reservoir_rfc_dir does not exist: {self.reservoir_rfc_dir}")
+        return self
+
+
 class ParallelConfig(StrictBaseModel):
     """
     Input.config Parallel section requirement
@@ -377,6 +394,7 @@ class InputConfig(StrictBaseModel):
     Calibration: Optional[CalibConfig] = None
     Forcing: Optional[ForcingConfig] = None
     DataFile: Optional[DataFileConfig] = None
+    DataAssimilation: Optional[DataAssimilationConfig] = None
     Parallel: Optional[ParallelConfig] = None
 
     # Check optional sections are present