Add Hindcast Workflow

.vscode/launch.json

@@ -93,6 +93,30 @@
                 "2",
             ],
         },
+        {
+            "name": "Hindcast -n 2",
+            "type": "debugpy",
+            "request": "launch",
+            "cwd": "${workspaceFolder}/bin_mounted",
+            "module": "ngen_rte.run_forecast",
+            "justMyCode": false,
+            "console": "integratedTerminal",
+            "python": "${command:python.interpreterPath}",
+            "args": [
+                "-dt",
+                "2025-09-15 00:00:00",
+                "-rname",
+                "fcst_debug_short_hindcast",
+                "-fconfig",
+                "short_range",
+                "-n",
+                "1",
+                "-hc",
+                "3",
+                "10",
+                "",
+            ],
+        },
         {
             "name": "run_default.py -n 2: aorc",
             "type": "debugpy",

bin_mounted/ngen_rte/configs.py

@@ -2,21 +2,22 @@
 
 import json
 import os
+import re
 import shutil
 from datetime import datetime, timedelta, timezone
 
 from mswm.build_inputs import RealizationBuilder
 from mswm.utils import settings as mswm_settings
 from mswm.utils.input_configuration import (
     CalibConfig,
-    RegionConfig,
     DataFileConfig,
     ForcingConfig,
     GeneralConfig,
     InputConfig,
     ModulePropertiesConfig,
     NWMOutputConfig,
     ParallelConfig,
+    RegionConfig,
 )
 from mswm.utils.settings import DEFAULT_DATETIME_FORMAT as DDF
 from mswm.utils.settings import LAGGED_ENSEMBLE_MEMBER_LAGS
@@ -148,6 +149,16 @@ class RTEBaseConfig(BaseModelStrict):
     le__closed_loop_state: str | None = Field(init=False, default=None)
     """File path for lagged ensemble closed loop state."""
 
+    # For hindcast.  Used by run_forecast.py.
+    hindcast_args: list[str] | None = Field(default=None, min_length=2, max_length=2)
+    """List of hindcast args from CLI. For details, see CLI help and ``_parse_hindcast_args()``."""
+    use_hindcast: bool | None = Field(init=False, default=False)
+    """Boolean indicating that hindcast is to be used. Passed to MSWM."""
+    hc_cycle_interval: int | None = Field(init=False, default=None)
+    """Hindcast cycle interval."""
+    hc_num_iterations: int | None = Field(init=False, default=None)
+    """Hindcast number of iterations."""
+
     def model_post_init(self, __context) -> None:
         self.time_at_init = datetime.now(tz=timezone.utc)
         self.errors = []
@@ -174,6 +185,8 @@ def model_post_init(self, __context) -> None:
         self.basin = self.vpu if self.vpu else self.gage_id
         self.subset_type = "vpu" if self.vpu else "gage"
 
+        self._parse_hindcast_args()
+
         if self.errors:
             raise RuntimeError(self.errors)
 
@@ -197,11 +210,16 @@ def configure_ngen_log(self, rb: RealizationBuilder) -> None:
         if isinstance(self, RTETestConfig):
             label = f"{label}_test"
 
-        if rb.run_type in ("default", "checkpoint", "regionalization"):
-            fallback_log_dir = str(rb.work_dir)
-        elif rb.run_type in ("forecast", "cold_start"):
-            fallback_log_dir = str(rb.input_dir)
-        elif rb.run_type == "calibration":
+        if rb.run_type in (
+            "default",
+            "checkpoint",
+            "regionalization",
+            "forecast",
+            "cold_start",
+            "hindcast",
+            "warm_start",
+            "calibration",
+        ):
             fallback_log_dir = str(rb.work_dir)
         else:
             raise RuntimeError(f"Unexpected run_type: {rb.run_type}")
@@ -270,6 +288,38 @@ def _parse_lagged_ensemble_args(self):
                 )
             )
 
+    def _parse_hindcast_args(self):
+        """Break up the multipart hindcast arg into distinct args and set them.
+        Called by child classes which define the necessary attributes."""
+        if self.hindcast_args:
+            self.use_hindcast = True
+
+            # Raw unpacking of the str args, before casting types of some of them.
+            _cycle_interval, _num_iterations = self.hindcast_args
+            if re.fullmatch(r"[0-9]+", _cycle_interval):
+                self.hc_cycle_interval = int(_cycle_interval)
+            else:
+                self.errors.append(
+                    ValueError(
+                        f"Hindcast _cycle_interval must be str representation of an integer, but got: {repr(_cycle_interval)}"
+                    )
+                )
+            if re.fullmatch(r"[0-9]+", _num_iterations):
+                self.hc_num_iterations = int(_num_iterations)
+            else:
+                self.errors.append(
+                    ValueError(
+                        f"Hindcast _num_iterations must be str representation of an integer, but got: {repr(_num_iterations)}"
+                    )
+                )
+
+            if hasattr(self, "cold_start_datetime") and self.cold_start_datetime:
+                self.errors.append(
+                    ValueError(
+                        "--hindcast_args and --cold_start_datetime were both provided, which is not allowed."
+                    )
+                )
+
     @property
     def _fcst_run_name_formatted(self) -> str:
         """Adaptive forecast run name that optionally can have a timestamped suffix appended to the end."""
@@ -360,7 +410,10 @@ def mswm_GeneralConfig(self) -> GeneralConfig:
         """MSWM GeneralConfig instance"""
         start_period, end_period = self.start_period__end_period
 
-        if isinstance(self, (RTEDefaultConfig, RTERegionConfig)) and self.fcst_run_name != c.DEFAULT_FORECAST_RUN_NAME:
+        if (
+            isinstance(self, (RTEDefaultConfig, RTERegionConfig))
+            and self.fcst_run_name != c.DEFAULT_FORECAST_RUN_NAME
+        ):
             formulation = self.fcst_run_name
         else:
             formulation = self.forcing_provider_paths.formulation_name
@@ -538,6 +591,7 @@ def mswm_RealizationBuilder_kwargs(self) -> dict:
             "valid_yaml": self.valid_best_yaml,
             "fcst_run_name": self._fcst_run_name_formatted,
             "config_overrides": self.mswm_InputConfig,
+            "use_hindcast": self.use_hindcast,
             "use_lagged_ens": self.use_lagged_ensemble,
             "lagged_ens_mem": self.lagged_ens_mem,
             "forcing_lag": self.forcing_lag,
@@ -632,7 +686,7 @@ class RTERegionConfig(RTEBaseConfig):
     fcst_run_name: str
         Name of the forecast realization run. Affects a directory name.
     lagged_ensemble_args: list[str] | None = Field(min_length=3, max_length=3)
-        See CLI help menu for [`run_default.py`](python_cli_help__run_default.py.txt) for details.
+        See CLI help menu for [`run_regionalization_standalone.py`](python_cli_help__run_regionalization_standalone.py.txt) for details.
     form_assign_file: str
         File containing formulation assignments for catchments
     cat_grp_file: str

bin_mounted/ngen_rte/execution/ngen_async.py

@@ -55,6 +55,9 @@ class NgenRunnerAsync(BaseModelStrict):
     suppress_output: bool = False
     """Passed to ForecastExecutionManager.postprocess()"""
     timeout_secs: float | None = None
+    """Timeout limit on ngen execution"""
+    do_override_log_file_prefix : bool = False
+    """Passed to ForecastExecutionManager.preprocess()"""
 
     fem: ForecastExecutionManager | None = Field(default=None, init=False)
     log_parsers: list[_LogParserBase] = Field(default_factory=list, init=False)
@@ -106,6 +109,8 @@ def start(self) -> None:
             "cold_start",
             "checkpoint",
             "regionalization",
+            "hindcast",
+            "warm_start",
         ):
             self.fem = ForecastExecutionManager(
                 real_path=str(self.rb.realization_file),
@@ -122,7 +127,7 @@ def start(self) -> None:
                     tolerant=True,
                 )
             )
-            self.fem.preprocess()
+            self.fem.preprocess(self.do_override_log_file_prefix)
             self.fem.execute(wait=False, log_file_open_mode="w")
         elif self.rb.run_type == "calibration":
             raise NotImplementedError(
@@ -165,8 +170,10 @@ def stream_status_until_complete(self) -> None:
             raise RuntimeError(errors)
 
     def _make_config_cache(self) -> ConfigCache | None:
-        """Make and return a ConfigCache based on the type of realization."""
-        if self.rb.run_type in ("forecast", "cold_start"):
+        """Make and return a ConfigCache based on the type of realization.
+        For ConfigCache for hindcast and warmstart, mimic pattern from nwm-fcst-mgr's forecast."""
+
+        if self.rb.run_type in ("forecast", "cold_start", "hindcast", "warm_start"):
             config_cache = ConfigCache(valid_yaml=self.rb.valid_yaml, no_valid=False)
         elif self.rb.run_type in ("default", "checkpoint", "regionalization"):
             config_cache = ConfigCache(

bin_mounted/ngen_rte/execution/ngen_logs.py

@@ -186,6 +186,10 @@ def ngen_log_basename(self, mpi_rank: int, payload_bool: bool):
             bn_prefix = "calib"
         elif self.rb.run_type in ("forecast", "cold_start"):
             bn_prefix = self.rb.fcst_run_name
+        elif self.rb.run_type == "hindcast":
+            bn_prefix = f"hindcast_{self.rb.hind_cycle}"
+        elif self.rb.run_type == "warm_start":
+            bn_prefix = f"warm_start_{self.rb.hind_cycle}"
         else:
             raise NotImplementedError(
                 f"Unsupported realization type: {self.rb.run_type}"

bin_mounted/ngen_rte/run_config/cli_args.py

@@ -261,6 +261,31 @@ def add_arg(parser: argparse.ArgumentParser, arg: ArgsKwargs) -> None:
     scripts=[Script.FORECAST, Script.DEFAULT, Script.REGIONALIZATION],
 )
 
+
+HINDCAST = ArgsKwargs(
+    args=["-hc", "--hindcast"],
+    kwargs={
+        "dest": "hindcast_args",
+        "type": str,
+        "nargs": 2,
+        "required": False,
+        "help": """Provide this multi-part argument to run a hindcast sequence.
+
+Available only for forecasts that are based on an existing calibration/validation (at a gage),
+e.g. not available for a "default" realization or a forecast based on a regionalization.
+
+This argument has 2 parts:
+    1. cycle_interval : int (required when -hc provided)
+        Cycle interval (in hours) between hindcast runs.
+    2. num_iterations : int (required when -hc provided)
+        Number of hindcast cycles to perform.
+
+For additional information, see nwm-fcst-mgr's forecast.py and README.md.""",
+    },
+    scripts=[Script.FORECAST],
+)
+
+
 OBJECTIVE_FUNCTION = ArgsKwargs(
     args=["-ofunc", "--objective_function"],
     kwargs={

bin_mounted/ngen_rte/run_forecast.py

@@ -9,9 +9,9 @@
 """
 
 import argparse
-import os
 
 from mswm.build_inputs import RealizationBuilder
+from nwm_fcst_mgr.forecast import run_hindcast
 
 from ngen_rte.configs import RTEForecastConfig
 from ngen_rte.execution.ngen_async import NgenRunnerAsync
@@ -34,16 +34,21 @@ def run_realization(rb: RealizationBuilder) -> None:
     LOG.info(
         f"Running realization with Forcing configuration: {rb.input_configs['Forcing']}"
     )
-    if rb.use_hindcast:
-        raise NotImplementedError("use_hindcast not yet implemented in nwm-rte")
-    elif rb.use_warm_start:
-        raise NotImplementedError("use_warm_start not yet implemented in nwm-rte")
+    if rb.run_type in ("hindcast", "warm_start"):
+        do_override_log_file_prefix = True
+    else:
+        do_override_log_file_prefix = False
+    if False:
+        raise NotImplementedError(
+            "This is a placeholder exception for unallowed execution paths"
+        )
     else:
         ngen_runner = NgenRunnerAsync(
             rb=rb,
             postprocess=True,
             suppress_output=False,
             # timeout_secs=10,
+            do_override_log_file_prefix=do_override_log_file_prefix,
         )
         ngen_runner.start()
         ngen_runner.stream_status_until_complete()
@@ -68,12 +73,33 @@ def _main(cfg: RTEForecastConfig):
         run_realization(rb_cs)
 
     elif cfg.cycle_datetime:
-        rb_fcst = build_realization(
-            cfg.mswm_RealizationBuilder_kwargs | {"use_cold_start": False},
-            "build_fcst_realization",
+        if cfg.use_hindcast:
+            rb_generator = run_hindcast(
+                config=cfg.mswm_InputConfig,
+                valid_yaml=cfg.valid_best_yaml,
+                fcst_run_name=cfg._fcst_run_name_formatted,
+                cycle_interval=cfg.hc_cycle_interval,
+                num_iterations=cfg.hc_num_iterations,
+                cold_start_state=cfg.load_state_from,
+                # This causes run_hindcast to be a generator of (yield) RealizationBuilder instances instead of running each realization itself.
+                yield_realizations=True,
+            )
+            for i, rb in enumerate(rb_generator):
+                LOG.info(f"About to run ngen for iteration {i} of hindcast workflow")
+                cfg.configure_ngen_log(rb)
+                run_realization(rb)
+        else:
+            rb_fcst = build_realization(
+                cfg.mswm_RealizationBuilder_kwargs | {"use_cold_start": False},
+                "build_fcst_realization",
+            )
+            cfg.configure_ngen_log(rb_fcst)
+            run_realization(rb_fcst)
+
+    else:
+        raise ValueError(
+            "Neither --cold_start_datetime nor --cycle_datetime were provided."
         )
-        cfg.configure_ngen_log(rb_fcst)
-        run_realization(rb_fcst)
 
 
 def main(cfg: RTEForecastConfig):

docs/reference/python_cli_help__run_forecast.py.txt

@@ -3,6 +3,7 @@ usage: run_forecast.py [-h] [-csdt COLD_START_DATETIME] -dt CYCLE_DATETIME
                        [-rname FCST_RUN_NAME] [-fconfig FORCING_CONFIGURATION]
                        [-fstatic FORCING_STATIC_DIR] [-g GAGE_ID]
                        [-gdomain {CONUS,Alaska,Hawaii,Puerto_Rico,GL}]
+                       [-hc HINDCAST_ARGS HINDCAST_ARGS]
                        [-le LAGGED_ENSEMBLE_ARGS LAGGED_ENSEMBLE_ARGS LAGGED_ENSEMBLE_ARGS]
                        [-lsf LOAD_STATE_FROM] [-nwmout] [-n NPROCS]
                        [-ofunc OBJECTIVE_FUNCTION]
@@ -74,6 +75,20 @@ options:
   -gdomain, --global_domain : type=str default='CONUS'
                         Global domain/region of forcing data.
 
+  -hc, --hindcast : type=str default=None
+                        Provide this multi-part argument to run a hindcast sequence.
+
+                        Available only for forecasts that are based on an existing calibration/validation (at a gage),
+                        e.g. not available for a "default" realization or a forecast based on a regionalization.
+
+                        This argument has 2 parts:
+                            1. cycle_interval : int (required when -hc provided)
+                                Cycle interval (in hours) between hindcast runs.
+                            2. num_iterations : int (required when -hc provided)
+                                Number of hindcast cycles to perform.
+
+                        For additional information, see nwm-fcst-mgr's forecast.py and README.md.
+
   -le, --lagged-ensemble : type=str default=None
                         Provide this multi-part argument to run one member
                         of a lagged ensemble (see nwm-fcst-mgr function `run_lagged_ensemble`).

run_fcst.sh

@@ -33,12 +33,15 @@ fcst_run_name=${1:-"fcst_run1"}
 
 set -x
 
-
+TEST_SAVED_STATE="/ngwpc/run_ngen/kge_dds/test_bmi/${TEST_GAGE}/Output/Model_State_Run/Cold_Start_Run/fcst_run1_cs_short/state_save/"
 # docker_run python -um "ngen_rte.run_forecast" --help
 
 # State saving and loading
 # docker_run python -um "ngen_rte.run_forecast" -csdt "2025-09-12 00:00:00" -dt "2025-09-15 00:00:00" -rname "${fcst_run_name}_cs_short" -fconfig short_range --save_state
-# docker_run python -um "ngen_rte.run_forecast" -dt "2025-09-15 00:00:00" -rname "${fcst_run_name}_short" -fconfig short_range --load_state_from /ngwpc/run_ngen/kge_dds/test_bmi/${TEST_GAGE}/Output/Model_State_Run/Cold_Start_Run/fcst_run1_cs_short/state_save/
+## Regular warmstart
+# docker_run python -um "ngen_rte.run_forecast" -dt "2025-09-15 00:00:00" -rname "${fcst_run_name}_short" -fconfig short_range --load_state_from "${TEST_SAVED_STATE}"
+## Hindcast warmstart
+# docker_run python -um "ngen_rte.run_forecast" -n 2 -fconfig "short_range" -dt "2025-09-15 00:00:00" -rname "fcst_run1_short_range_warm_start_hindcast" -hc 3 10 --load_state_from "${TEST_SAVED_STATE}"
 
 # docker_run python -um "ngen_rte.run_forecast" -csdt "2025-09-12 00:00:00" -dt "2025-09-15 00:00:00" -rname "${fcst_run_name}_cs_short" -fconfig short_range --save_state --state_save_dir /ngwpc/run_ngen/test_bmi/${TEST_GAGE}/Output/Model_State_Run/Cold_Start_Run/fcst_run1_cs_short/state_save_directory/
 

run_suite.sh

@@ -38,6 +38,8 @@ docker_run python -um "ngen_rte.run_forecast" -fconfig "short_range" -dt "2025-0
 # docker_run python -um "ngen_rte.run_forecast" -fconfig "short_range" -dt "2025-07-10 04:00:00" -rname "fcst_run1_short_range_n2" -n 2
 docker_run python -um "ngen_rte.run_forecast" -fconfig "standard_ana" -dt "2025-07-10 10:00:00" -rname "fcst_run1_standard_ana"
 docker_run python -um "ngen_rte.run_forecast" -fconfig "medium_range_blend" -dt "2025-07-10 00:00:00" -rname "fcst_run1_medium_range_blend"
+# Hindcast short_range
+docker_run python -um "ngen_rte.run_forecast" -n 2 -fconfig "short_range" -dt "2025-07-10 04:00:00" -rname "fcst_run1_short_range_hindcast" -hc 3 10
 
 # PR sample calibration during Hurricane Maria
 # sudo rm -rf ~/ngwpc/run_ngen/kge_dds/test_bmi/50027000