diff --git a/.vscode/launch.json b/.vscode/launch.json index d8d9f2d..0754709 100644 --- a/.vscode/launch.json +++ b/.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", diff --git a/bin_mounted/ngen_rte/configs.py b/bin_mounted/ngen_rte/configs.py index 3db8060..cfd7912 100644 --- a/bin_mounted/ngen_rte/configs.py +++ b/bin_mounted/ngen_rte/configs.py @@ -2,6 +2,7 @@ import json import os +import re import shutil from datetime import datetime, timedelta, timezone @@ -9,7 +10,6 @@ from mswm.utils import settings as mswm_settings from mswm.utils.input_configuration import ( CalibConfig, - RegionConfig, DataFileConfig, ForcingConfig, GeneralConfig, @@ -17,6 +17,7 @@ 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 diff --git a/bin_mounted/ngen_rte/execution/ngen_async.py b/bin_mounted/ngen_rte/execution/ngen_async.py index 145bb0b..fdd93f1 100644 --- a/bin_mounted/ngen_rte/execution/ngen_async.py +++ b/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( diff --git a/bin_mounted/ngen_rte/execution/ngen_logs.py b/bin_mounted/ngen_rte/execution/ngen_logs.py index df57e5f..3d02659 100644 --- a/bin_mounted/ngen_rte/execution/ngen_logs.py +++ b/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}" diff --git a/bin_mounted/ngen_rte/run_config/cli_args.py b/bin_mounted/ngen_rte/run_config/cli_args.py index e7ac1c2..3c5a366 100644 --- a/bin_mounted/ngen_rte/run_config/cli_args.py +++ b/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={ diff --git a/bin_mounted/ngen_rte/run_forecast.py b/bin_mounted/ngen_rte/run_forecast.py index b8615ee..081ac28 100644 --- a/bin_mounted/ngen_rte/run_forecast.py +++ b/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): diff --git a/docs/reference/python_cli_help__run_forecast.py.txt b/docs/reference/python_cli_help__run_forecast.py.txt index 5d73924..2ed2cda 100644 --- a/docs/reference/python_cli_help__run_forecast.py.txt +++ b/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`). diff --git a/run_fcst.sh b/run_fcst.sh index 5a389e9..865c4ea 100755 --- a/run_fcst.sh +++ b/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/ diff --git a/run_suite.sh b/run_suite.sh index d9b693b..bdcd50c 100755 --- a/run_suite.sh +++ b/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