Python Reference

The pyproject.toml of this repository primarily contains requirements for building documentation. The build sequence of the nwm-rte image is dynamic/configurable and involves multiple Python virtual environments.

CLI Help Menus

These are raw outputs of the --help menu generated by Python's argparse library.

To update these, run: ./docs/update_python_cli_ref.sh

run_default.py --help

run_calibration.py --help

run_forecast.py --help

run_restart.py --help

run_tests.py --help

run_regionalization.py --help

CLI Executable Modules

run_default.py

Called by run_default.sh

run_default.py --help

ngen_rte.run_default

Command-line executable to build and run a "default" realization. Supports realtime forcing configurations, e.g. "short_range", as well as historical/retrospective sources, e.g. "aorc".

This runs inside the ngen runtime environment. The CLI structure is mimicked in part by configs.RTEDefaultConfig. For settings that are not exposed by CLI arguments, see primarily consts.py.

See run_default.sh for example calls.

run_default

run_default(rb: RealizationBuilder) -> NgenRunnerAsync

Run the provided default realization. Realization should already be built (rb.build_default_realization() already called).

Source code in bin_mounted/ngen_rte/run_default.py

def run_default(rb: RealizationBuilder) -> NgenRunnerAsync:
    """Run the provided default realization.
    Realization should already be built (rb.build_default_realization() already called).
    """
    LOG.info("Running default realization")
    # For default realization, currently postprocess needs suppress_output=True
    ngen_runner = NgenRunnerAsync(rb=rb, postprocess=True, suppress_output=True)
    ngen_runner.start()
    ngen_runner.stream_status_until_complete()
    ngen_runner.close()  # Can also let __del__ handle this.
    return ngen_runner

cli_arg_parser

cli_arg_parser() -> ArgumentParser

Build and return the CLI argument parser

Source code in bin_mounted/ngen_rte/run_default.py

def cli_arg_parser() -> argparse.ArgumentParser:
    """Build and return the CLI argument parser"""
    parser = argparse.ArgumentParser(
        description="""Script for building and running default realizations
using realtime forcing configurations or historical / retrospective forcing.
The CLI arguments mostly follow that of run_forecast.py. The exception is
that "--duration" aka "-dur" (in days) was added to this script
to support the historical / retrospective forcing use case, e.g. AORC or NWM.""",
        formatter_class=cli_args.HelpFormatter,
    )
    cli_args.add_args_for_script(parser, cli_args.Script.DEFAULT)
    return parser

run_calibration.py

Called by run_calib.sh

run_calibration.py --help

ngen_rte.run_calibration

Command-line executable to build and run a "calibration" realization.

This runs inside the ngen runtime environment. The CLI structure is mimicked in part by configs.RTECalibConfig. For settings that are not exposed by CLI arguments, see primarily consts.py.

See run_calib.sh for example calls.

get_calibration_cmd

get_calibration_cmd(
    rb: RealizationBuilder, worker_name: str, log_path: str
) -> list[str]

Get the command to run the calibration realization.

Source code in bin_mounted/ngen_rte/run_calibration.py

def get_calibration_cmd(
    rb: RealizationBuilder, worker_name: str, log_path: str
) -> list[str]:
    """Get the command to run the calibration realization."""
    cmd = [
        "calibration",
        str(rb.calib_config_file),
        "--log_path_overwrite",
        log_path,
    ]
    if worker_name:
        cmd.extend(["--worker_name", worker_name])
    return cmd

cli_arg_parser

cli_arg_parser() -> ArgumentParser

Build and return the CLI argument parser

Source code in bin_mounted/ngen_rte/run_calibration.py

def cli_arg_parser() -> argparse.ArgumentParser:
    """Build and return the CLI argument parser"""
    parser = argparse.ArgumentParser(
        description="""Script for building and running a calibration
realization using historical / retrospective forcing.""",
        formatter_class=cli_args.HelpFormatter,
    )
    cli_args.add_args_for_script(parser, cli_args.Script.CALIBRATION)
    return parser

run_forecast.py

Called by run_fcst.sh

run_forecast.py --help

ngen_rte.run_forecast

Command-line executable to build and run a "forecast" realization, optionally with a coldstart.

This runs inside the ngen runtime environment. The CLI structure is mimicked in part by configs.RTEForecastConfig. For settings that are not exposed by CLI arguments, see primarily consts.py.

See run_fcst.sh for example calls.

run_realization

run_realization(rb: RealizationBuilder) -> None

Run the realization, which can be a coldstart, forecast, or lagged ensemble.

Source code in bin_mounted/ngen_rte/run_forecast.py

def run_realization(rb: RealizationBuilder) -> None:
    """Run the realization, which can be a coldstart, forecast, or lagged ensemble."""
    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")
    else:
        ngen_runner = NgenRunnerAsync(
            rb=rb,
            postprocess=True,
            suppress_output=False,
            # timeout_secs=10,
        )
        ngen_runner.start()
        ngen_runner.stream_status_until_complete()
        ngen_runner.close()  # Can also let __del__ handle this.

cli_arg_parser

cli_arg_parser() -> ArgumentParser

Build and return the CLI argument parser

Source code in bin_mounted/ngen_rte/run_forecast.py

def cli_arg_parser() -> argparse.ArgumentParser:
    """Build and return the CLI argument parser"""
    parser = argparse.ArgumentParser(
        description="""Script for building and running a forecast realization,
optionally with a coldstart.""",
        formatter_class=cli_args.HelpFormatter,
    )
    cli_args.add_args_for_script(parser, cli_args.Script.FORECAST)
    return parser

run_restart.py

Called by run_default.sh

run_restart.py --help

ngen_rte.run_restart

Command-line executable to restart a failed run from a saved checkpoint. Currently only functional for default and regionalization runs. Calls msw-mgr chcekpoint_restart to copy and configure the run, then infers the ngen command from the destination directory. A run can only be restarted if the original run was configured to save checkpoints using the --checkpoint_interval argument.

infer_gage_id

infer_gage_id(src_path: str) -> str

Infer gage ID from source run directory path

Source code in bin_mounted/ngen_rte/run_restart.py

def infer_gage_id(src_path: str) -> str:
    """Infer gage ID from source run directory path"""
    return Path(src_path).resolve().name

infer_rb

infer_rb(
    dst_path: str, src_path: str
) -> RealizationBuilder

Infer and return a minimal RealizationBuilder from the destination directory structure

Source code in bin_mounted/ngen_rte/run_restart.py

def infer_rb(dst_path: str, src_path: str) -> RealizationBuilder:
    """Infer and return a minimal RealizationBuilder from the destination directory structure"""
    dst = Path(dst_path).resolve()
    input_dir = dst / "Input"

    ngen_bin = input_dir / "ngen"
    if not ngen_bin.exists():
        msg = f"ngen binary not found at: {ngen_bin}"
        LOG.critical(msg)
        raise FileNotFoundError(msg)

    realization_files = list(dst.rglob("*realization*.json"))
    if not realization_files:
        msg = f"No realization file found in destination run folder: {input_dir}"
        LOG.critical(msg)
        raise FileNotFoundError(msg)
    realization_file = str(realization_files[0])

    part_files = list(input_dir.rglob("*partition*.json"))
    part_file = str(part_files[0]) if part_files else None

    if part_file:
        with open(part_file) as f:
            partitions = json.load(f)
        nprocs = len(partitions.get("partitions", []))
    else:
        nprocs = 1

    # Read realization file and retrieve checkpoint save path and frequency
    with open(realization_file) as f:
        real_config = json.load(f)
    state_saving = real_config.get("state_saving", [])
    save_config = next((s for s in state_saving if s.get("direction") == "save"), None)
    if save_config is None:
        LOG.info("No state saving configuration found in realization file")

    rb = RealizationBuilder.__new__(RealizationBuilder)
    rb.realization_file = realization_file
    rb.part_file = part_file
    rb.work_dir = str(dst)
    rb.valid_yaml = None
    rb.run_type = "checkpoint"
    rb.basin = infer_gage_id(src_path)
    rb.input_configs = {"Parallel": {"nprocs": nprocs}}
    rb.checkpoint_interval = save_config.get("frequency")
    rb.save_checkpoint_to = save_config.get("path")
    return rb

run_restart

run_restart(rb: RealizationBuilder) -> None

Run the provided checkpoint restart realization.

Source code in bin_mounted/ngen_rte/run_restart.py

def run_restart(rb: RealizationBuilder) -> None:
    """
    Run the provided checkpoint restart realization.
    """

    LOG.info("Running restart realization")
    ngen_runner = NgenRunnerAsync(rb=rb, postprocess=False)
    ngen_runner.start()
    ngen_runner.stream_status_until_complete()
    ngen_runner.close()

run_tests.py

Called by run_tests.sh

run_tests.py --help

ngen_rte.tests.run_tests

Command-line executable to build and run a series of "forecast" realizations, optionally with a "calibration" realization preceding them.

This runs inside the ngen runtime environment. The CLI structure is mimicked in part by configs.RTETestConfig. For settings that are not exposed by CLI arguments, see primarily consts.py.

When realizations fail, this program does not halt, but rather moves to the next configuration type in the list, with the goal of "trying" many different realization configurations in one call. The status of each configuration's build step and run step is reported and written to a json file at the end.

This includes options for stopping realizations mid-way through their run, rather than waiting for them to complete.

See run_tests.sh for example calls.

calibrations__build_and_run

calibrations__build_and_run(
    cfg: RTETestConfig, tm: TestsManager
) -> None

Build calibration realizations and run them as tests.

Source code in bin_mounted/ngen_rte/tests/run_tests.py

def calibrations__build_and_run(cfg: RTETestConfig, tm: TestsManager) -> None:
    """Build calibration realizations and run them as tests."""
    perms = cfg.get_calib_permutations()
    for obj_func, optim_algo, _ in perms:
        rte_calib_configs = get_test_configs__calibration(
            cfg,
            obj_func=obj_func,
            optim_algo=optim_algo,
        )

        for i, calib_config in enumerate(rte_calib_configs):
            fc = calib_config.forcing_configuration
            worker_name = (
                f"test_{i}_{calib_config.mswm_GeneralConfig.models.replace(',', '_')}_rootzone={calib_config.mswm_ModulePropertiesConfig.cfe_aet_rootzone}"
                if optim_algo == CalOptimizationAlgo.dds
                else None
            )
            rb_kwargs = calib_config.mswm_RealizationBuilder_kwargs
            msg_prefix = f"i={i} (ilimit={len(rte_calib_configs) - 1}) worker_name={worker_name} Calibration with forcing={repr(fc)}, models={repr(calib_config.mswm_GeneralConfig.models)}, cfe_aet_rootzone={calib_config.mswm_ModulePropertiesConfig.cfe_aet_rootzone}, obj_func={repr(obj_func.value)}, optim_algo={repr(optim_algo.value)}, obs_dir={calib_config.mswm_DataFileConfig.obs_dir}, nwmretro_file={calib_config.mswm_DataFileConfig.nwmretro_file}"

            if cfg.restart and i + 1 <= len(tm.prev_results):
                LOG.info(f"Skipping since restart={cfg.restart}: {msg_prefix}")
                continue

            LOG.info(
                f"\n\n##########\n### {msg_prefix}: setting up test with rb_kwargs = \n{json.dumps(rb_kwargs, indent=2, default=pydantic_encoder)}"
            )
            t = ForecastTest(rb_kwargs=rb_kwargs)

            # Build the realization, trapping exceptions into class attrs
            LOG.info(f"### {msg_prefix}: building realization")
            t.make_realization_builder__build_realization(
                build_method="build_calib_realization"
            )

            if t.rb_stat == TestStat.PASS:
                cfg.configure_ngen_log(t.rb)
                # Execute the realization via ngen, trapping exceptions and logs into class attrs
                LOG.info(f"### {msg_prefix}: executing calibration realization")
                t.execute_calibration(
                    cfg.quit_calibration_after_duration, worker_name=worker_name
                )

            tm.add_forecast_test(t)
            tm.evaluate_test_results(raise_if_any_failed=False, header_prefix="INTERIM")

forecasts__build_and_run

forecasts__build_and_run(
    cfg: RTETestConfig, tm: TestsManager, cs: bool
) -> None

Using ForecastTest, build and execute a list of forecast realizations. tests_manager is modified in-place, so some test results may be available if this function is interrupted. cs controls whether coldstart is used (not cfg.do_coldstart).

Source code in bin_mounted/ngen_rte/tests/run_tests.py

def forecasts__build_and_run(cfg: RTETestConfig, tm: TestsManager, cs: bool) -> None:
    """
    Using ForecastTest, build and execute a list of forecast realizations.
    tests_manager is modified in-place, so some test results may be available if this function is interrupted.
    `cs` controls whether coldstart is used (not `cfg.do_coldstart`).
    """
    for obj_func, optim_algo, test_paths in cfg.get_calib_permutations():
        test_configs = get_test_configs__forecast(cfg, use_cold_start=cs)
        for i, config_overrides in enumerate(test_configs):
            fc = config_overrides.Forcing.forcing_configuration
            msg_prefix = f"i={i} (ilimit={len(test_configs) - 1}) forecast {repr(fc)} with calib obj_func={repr(obj_func.value)}, optim_algo={repr(optim_algo.value)}"

            if cfg.restart and i + 1 <= len(tm.prev_results):
                LOG.info(f"Skipping since restart={cfg.restart}: {msg_prefix}")
                continue

            rb_kwargs = {
                # "input_path": test_paths.dir_input,
                "valid_yaml": test_paths.valid_yaml,
                "fcst_run_name": cfg._fcst_run_name_formatted,
                "config_overrides": config_overrides,
                "use_cold_start": cs,
            }
            LOG.info(
                f"\n\n##########\n### {msg_prefix}: setting up test with rb_kwargs = {rb_kwargs}"
            )

            t = ForecastTest(rb_kwargs=rb_kwargs)

            # Build the realization, trapping exceptions into class attrs
            LOG.info(f"### {msg_prefix}: building realization")
            t.make_realization_builder__build_realization(
                build_method="build_fcst_realization"
            )

            if t.rb_stat == TestStat.PASS:
                # Execute the realization via ngen, trapping exceptions and logs into class attrs
                cfg.configure_ngen_log(t.rb)
                LOG.info(f"### {msg_prefix}: executing realization via ngen")
                t.execute_forecast(
                    quit_forecast_after_duration=cfg.quit_forecast_after_duration
                )

            tm.add_forecast_test(t)
            tm.evaluate_test_results(raise_if_any_failed=False, header_prefix="INTERIM")

run_noop_mode

run_noop_mode() -> None

Run noop mode - verify imports and basic setup without executing workflows.

Source code in bin_mounted/ngen_rte/tests/run_tests.py

def run_noop_mode() -> None:
    """Run noop mode - verify imports and basic setup without executing workflows."""
    LOG.info("\nRunning in noop mode - only checking imports and basic setup.")
    LOG.info("Successfully imported all required libraries.")
    LOG.info("Noop mode complete - exiting")
    sys.exit(0)  # Exit the program directly

cli_arg_parser

cli_arg_parser() -> ArgumentParser

Build and return the CLI argument parser

Source code in bin_mounted/ngen_rte/tests/run_tests.py

def cli_arg_parser() -> argparse.ArgumentParser:
    """Build and return the CLI argument parser"""
    parser = argparse.ArgumentParser(
        description="""Script for building and running a series of test
realizations, optionally including calibration, coldstart, and forecasts,
using various forcing configurations and model formulations.""",
        formatter_class=cli_args.HelpFormatter,
    )
    cli_args.add_args_for_script(parser, cli_args.Script.TESTS)

    parser = argparse.ArgumentParser(
        description="""Script for building and running a series of test
realizations, optionally including calibration, coldstart, and forecasts,
using various forcing configurations and model formulations.""",
        formatter_class=cli_args.HelpFormatter,
    )
    parser.add_argument(
        "-nofcst",
        "--skip_forecast",
        action="store_true",
        help="""Provide to skip forecast (for testing calibrations only).
Incompatible with --do_all_forcing_configs and --do_coldstart""",
    )
    parser.add_argument(
        "-quitfcdur",
        "--quit_forecast_after_duration",
        default=None,
        type=float,
        help="""Instead of waiting for each forecast to finish,
quit after the specified elapsed processing duration in seconds.""",
    )
    parser.add_argument(
        "-calib",
        "--do_calibration",
        action="store_true",
        help="Build and run a calibration before forecasts.",
    )
    parser.add_argument(
        "-quitcaldur",
        "--quit_calibration_after_duration",
        default=None,
        type=float,
        help="""For calibrations, instead of waiting for the realization
to finish, quit after the specified processing duration. Units: seconds.""",
    )
    parser.add_argument(
        "-ofuncs",
        "--objective_functions",
        nargs="+",
        type=c.CalObjective,
        default=[c.CALIB_OBJECTIVE_FUNCTION],
        help="List of objective functions for calibration.",
    )
    parser.add_argument(
        "-allofuncs",
        "--do_all_objective_functions",
        action="store_true",
        help=f"For calibration, causes all objective functions to be executed: {cli_args.split_iter_to_chunked_str([_.value for _ in c.CalObjective])}",
    )
    parser.add_argument(
        "-optalgos",
        "--optimization_algorithms",
        nargs="+",
        type=c.CalOptimizationAlgo,
        default=[c.CALIB_OPTIMIZATION_ALGO],
        help="List of optimization algorithms for calibration.",
    )
    parser.add_argument(
        "-alloptalgos",
        "--do_all_optimization_algorithms",
        action="store_true",
        help=f"For calibration, causes all optimization algorithms to be executed: {cli_args.split_iter_to_chunked_str([_.value for _ in c.CalOptimizationAlgo])}",
    )
    parser.add_argument(
        "-allforcings",
        "--do_all_forcing_configs",
        action="store_true",
        help=f"""Run all forcing configurations rather than the default shorter default list.
For reference, the default list is: {c.FORECAST_FORCING_TYPES__TESTS}.
Incompatible with --skip_forecast.""",
    )
    parser.add_argument(
        "-mff",
        "--model_formulations_file",
        help=f"""If provided, multiple model formulations will be ran,
and this is a file path to a tsv file of the formulations list. If not provided,
then the default model formulation will be used: {c.DEFAULT_MODEL_FORMULATION_ARGS}.""",
    )
    parser.add_argument(
        "-calfsrcs",
        "--calibration_forcing_sources",
        nargs="*",
        default=c.CALIB_FORCING_TYPES,
        help=f"""Sources of forcing data for calibration runs. If not provided,
then this default will be used: {c.CALIB_FORCING_TYPES}.""",
    )
    parser.add_argument(
        "-cs",
        "--do_coldstart",
        action="store_true",
        help="Causes use_cold_start to be True for all forecasts",
    )
    parser.add_argument(
        "--noop",
        action="store_true",
        help="""Run in noop mode - only verify that the script
can import libraries and basic setup, then exit without looking
for data or running any workflows.""",
    )
    parser.add_argument(
        "--restart",
        action="store_true",
        help=f"""Run in restart mode. Read existing results json file {repr(c.TEST_RESULTS_FILE)}
if it exists, and skip indexes that already have a record in it.""",
    )
    cli_args.add_args_for_script(parser, cli_args.Script.TESTS)
    args = parser.parse_args()
    LOG.info(f"{__file__}: args: {json.dumps(vars(args), indent=2)}")

    return parser

Python Constants

Currently bin_mounted/consts.py contains variables which rarely need editing.

ngen_rte.consts

Constants

FORCING_TEMPLATE_DIR module-attribute

FORCING_TEMPLATE_DIR = "/ngen-app/ngen-python/lib/python3.11/site-packages/NextGen_Forcings_Engine_BMI/BMI_NextGen_Configs/config_templates/"

Directory of forcing configuration template yaml files.

Configuration Classes

These mimic the CLI interfaces and perform some additional argument parsing and preparation of classes that are passed to other components of the system.

ngen_rte.configs

Primary configuration classes. Pydantic BaseModels directly associated with CLI executables.

RTEBaseConfig

Bases: BaseModelStrict

Base RTE configuration class to be inherited by child classes. Triggers certain setup actions, such as creation of WCOSS-path symlinks. Classes that inherit from this should call super().model_post_init(__context) inside their own model_post_init() method, if they have that method also defined in the child.

The primary usage of this class is to access property mswm_RealizationBuilder_kwargs for building (and later running) a realization using MSWM.

ATTRIBUTE	DESCRIPTION
delete_scratch_and_mesh_first	Causes scratch dir and intermediate mesh file to be deleted before building the realization. Use with caution. TYPE: bool
delete_forcing_raw_input_first	Causes forcing raw input dir to be deleted before building the realization. Use with caution. TYPE: bool
environment	Operating environment, e.g. 'test' or 'oe'. TYPE: str
nprocs	Number of processors to use for ngen execution. TYPE: int = Field(ge=1)
global_domain	Global domain, e.g. "CONUS" or "Hawaii". Must agree with the ID of the basin / gage / VPU being simulated. TYPE: str
forcing_static_dir	Directory for static forcing data. TYPE: str
basin	Basin identifier. Can be either Gage ID or VPU identifier. TYPE: str
gage_id	Gage ID. When provided, sets subset_type to 'gage'. TYPE: str
vpu	VPU identifier. When provided, sets subset_type to 'vpu'. TYPE: str
subset_type	Type of basin identifier. Accepts 'gage' or 'vpu'. TYPE: str
model_formulation_cli_csv	Comma-separated string of model names comprising the formulation. TYPE: str | None = Field(default=None)
model_formulation_cli_rootzone	Boolean for model formulation rootzone parameter. TYPE: str | None = Field(default=None)
add_timestamp_to_run_name	Boolean causing the final forecast run name to contain a timestamp suffix. TYPE: bool = Field(default=False)
nwm_output_vars	Boolean causing NWM output variables to be included in the output. Passed to MSWM NWMOutputConfig. Does not apply to calibration workflow. TYPE: bool = Field(default=False)
output_format	Output format(s) for output variables. Accepts 'CSV', 'NetCDF', or both. Defaults to ["CSV"] in MSWM. TYPE: list[str] | None = Field(default=None)
hydrofab_file	Optional hydrofabric file path. If provided, bypasses MSWM Icefabric server API call. TYPE: str | None = Field(default=None)
fcst_run_name	Forecast run name. TYPE: str | None = Field(default=None)
cycle_datetime	Cycle datetime for forecast TYPE: datetime | None = Field(default=None)
checkpoint_interval	Integer number of timesteps for interval of checkpoint output TYPE: int = Field(default=None)
checkpoint_dir	Optional directory to save checkpoint states. Defaults to work_dir>/checkpoint/ in msw-mgr. TYPE: str | None = Field(default=None)
load_state_from	Path to existing state file to load at start of run TYPE: str | None = Field(default=None)
save_state	Boolean to activate state save at end of run TYPE: bool = Field(default=False)
save_state_directory	Optional directory to save model state at end of run. Defaults in /state_save/ in msw-mgr. TYPE: str | None = Field(default=None)
lookback	Optional override of the forcing template LookBack (minutes), controlling the AnA simulation window. None means use the template value. TYPE: int | None = Field(default=None)

Source code in bin_mounted/ngen_rte/configs.py

class RTEBaseConfig(BaseModelStrict):
    """Base RTE configuration class to be inherited by child classes.
    Triggers certain setup actions, such as creation of WCOSS-path symlinks.
    Classes that inherit from this should call super().model_post_init(__context) inside their own
    model_post_init() method, if they have that method also defined in the child.

    The primary usage of this class is to access property mswm_RealizationBuilder_kwargs
    for building (and later running) a realization using MSWM.

    Attributes
    ----------
    delete_scratch_and_mesh_first: bool
        Causes scratch dir and intermediate mesh file to be deleted before building the realization. Use with caution.
    delete_forcing_raw_input_first: bool
        Causes forcing raw input dir to be deleted before building the realization. Use with caution.
    environment: str
        Operating environment, e.g. 'test' or 'oe'.
    nprocs: int = Field(ge=1)
        Number of processors to use for ngen execution.
    global_domain: str
        Global domain, e.g. "CONUS" or "Hawaii". Must agree with the ID of the basin / gage / VPU being simulated.
    forcing_static_dir: str
        Directory for static forcing data.
    basin: str
        Basin identifier. Can be either Gage ID or VPU identifier.
    gage_id: str
        Gage ID. When provided, sets subset_type to 'gage'.
    vpu: str
        VPU identifier.  When provided, sets subset_type to 'vpu'.
    subset_type: str
        Type of basin identifier. Accepts 'gage' or 'vpu'.
    model_formulation_cli_csv: str | None = Field(default=None)
        Comma-separated string of model names comprising the formulation.
    model_formulation_cli_rootzone: str | None = Field(default=None)
        Boolean for model formulation rootzone parameter.
    add_timestamp_to_run_name: bool = Field(default=False)
        Boolean causing the final forecast run name to contain a timestamp suffix.
    nwm_output_vars: bool = Field(default=False)
        Boolean causing NWM output variables to be included in the output. Passed to MSWM NWMOutputConfig. Does not apply to calibration workflow.
    output_format: list[str] | None = Field(default=None)
        Output format(s) for output variables. Accepts 'CSV', 'NetCDF', or both. Defaults to ["CSV"] in MSWM.
    hydrofab_file: str | None = Field(default=None)
        Optional hydrofabric file path. If provided, bypasses MSWM Icefabric server API call.
    fcst_run_name: str | None = Field(default=None)
        Forecast run name.
    cycle_datetime: datetime | None = Field(default=None)
        Cycle datetime for forecast
    checkpoint_interval: int = Field(default=None)
        Integer number of timesteps for interval of checkpoint output
    checkpoint_dir: str | None = Field(default=None)
        Optional directory to save checkpoint states. Defaults to work_dir>/checkpoint/ in msw-mgr.
    load_state_from: str | None = Field(default=None)
        Path to existing state file to load at start of run
    save_state: bool = Field(default=False)
        Boolean to activate state save at end of run
    save_state_directory: str | None = Field(default=None)
        Optional directory to save model state at end of run. Defaults in <work_dir>/state_save/ in msw-mgr.
    lookback: int | None = Field(default=None)
        Optional override of the forcing template `LookBack` (minutes), controlling
        the AnA simulation window. None means use the template value.
    """

    # Set during init
    delete_scratch_and_mesh_first: bool
    delete_forcing_raw_input_first: bool
    environment: str
    nprocs: int = Field(ge=1)
    global_domain: str
    forcing_static_dir: str
    gage_id: str | None = Field(default=None)
    vpu: str | None = Field(default=None)
    model_formulation_cli_csv: str | None = Field(default=None)
    model_formulation_cli_rootzone: str | None = Field(default=None)
    add_timestamp_to_run_name: bool = Field(default=False)
    nwm_output_vars: bool = Field(default=False)
    output_format: list[str] | None = Field(default=None)
    hydrofab_file: str | None = Field(default=None)
    fcst_run_name: str | None = Field(default=None)
    cycle_datetime: datetime | None = Field(default=None)
    lookback: int | None = Field(default=None)
    checkpoint_interval: int | None = Field(default=None)
    checkpoint_dir: str | None = Field(default=None)
    load_state_from: str | None = Field(default=None)
    save_state: bool = Field(default=False)
    save_state_dir: str | None = Field(default=None)

    # Set after init (not provided as args)
    time_at_init: datetime | None = Field(init=False, default=None)
    """Time at class instantiation."""
    errors: list | None = Field(init=False, default=None)
    """List of exceptions encountered during init."""
    subset_type: str | None = Field(init=False, default=None)
    """Subset_type set depending on vpu and gage_id args"""
    basin: str | None = Field(init=False, default=None)
    """Basin id set depending on vpu and gage_id args"""

    # For lagged ensemble.  Used by run_forecast.py and run_default.py.  See CLI args for those scripts and see _parse_lagged_ensemble_args() for details.
    use_lagged_ensemble: bool | None = Field(init=False, default=False)
    """Boolean indicating that lagged ensemble is to be used. Passed to MSWM."""
    lagged_ens_mem: str | None = Field(init=False, default=None)
    """Lagged ensemble member name. Passed to MSWM."""
    forcing_lag: int | None = Field(init=False, default=None)
    """Lagged ensemble forcing lag. Looked up based on lagged ensemble member name. Passed to MSWM."""
    le__open_loop_state: str | None = Field(init=False, default=None)
    """File path for lagged ensemble open loop state."""
    le__closed_loop_state: str | None = Field(init=False, default=None)
    """File path for lagged ensemble closed loop state."""

    def model_post_init(self, __context) -> None:
        self.time_at_init = datetime.now(tz=timezone.utc)
        self.errors = []
        make_wcoss_path_symlinks()

        if self.vpu and self.gage_id:
            self.errors.append(
                ValueError(
                    "--vpu and --gage_id are mutually exclusive, only one can be passed."
                )
            )

        if self.hydrofab_file and not (self.vpu or self.gage_id):
            self.errors.append(
                ValueError(
                    "--hydrofab_file requires --gage_id or --vpu to be explicitly provded."
                )
            )

        if not self.gage_id:
            self.gage_id = c.DEFAULT_GAGE_ID

        # Set basin from vpu if provided else gage_id
        self.basin = self.vpu if self.vpu else self.gage_id
        self.subset_type = "vpu" if self.vpu else "gage"

        if self.errors:
            raise RuntimeError(self.errors)

    def configure_ngen_log(self, rb: RealizationBuilder) -> None:
        """Configure the ngen logging, by setting the associated OS env variable for the directory to hold the logs,
        and copying the associated json file into that directory.

        ``fallback_log_dir`` is ignored when the RTE OS env var key NGEN_LOG_TO_RTE is true.
        It is used to emulate behavior of nwm-cal-mgr and nwm-fcst-mgr (what they would use without RTE).

        Parameters
        ----------
        rb : RealizationBuilder
            An already built realization.
        """
        now_str = datetime.now(timezone.utc).strftime(r"%Y%m%d_%H%M%S_%f")

        label = rb.run_type
        if getattr(rb, "use_cold_start", False):
            label = f"{label}_cs"
        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":
            fallback_log_dir = str(rb.work_dir)
        else:
            raise RuntimeError(f"Unexpected run_type: {rb.run_type}")

        # Confirm that it's valid json content
        LOG.debug(f"Reading: {c.SRC_LOG_CONFIG_JSON}")
        with open(c.SRC_LOG_CONFIG_JSON) as f:
            try:
                _ = json.load(f)
            except Exception as e:
                raise RuntimeError(
                    f"Could not read or parse as json: {c.SRC_LOG_CONFIG_JSON}: {e}"
                ) from e

        # Decide the dir
        setting_val = os.environ.get(c.RTE_NGEN_LOG_BEHAVIOR_KEY, "").lower().strip()
        if setting_val in ("yes", "true"):
            log_dir = os.path.join(c.CONTAINER_LOGS_DIR, "ngen", f"{now_str}_{label}")
        elif setting_val in ("no", "false", ""):
            log_dir = fallback_log_dir
        else:
            raise ValueError(
                f"Invalid value for key {repr(c.RTE_NGEN_LOG_BEHAVIOR_KEY)}: {repr(setting_val)} (expected YES or NO, defaulting to NO if not provided)"
            )

        # Make the dir, copy the log json config into it, and set the OS env var for ngen to be able to find it.
        LOG.debug(f"Making directory: {log_dir}")
        os.makedirs(log_dir, exist_ok=True)
        LOG.info(f"Copying: {c.SRC_LOG_CONFIG_JSON} -> {log_dir}/")
        shutil.copy2(c.SRC_LOG_CONFIG_JSON, log_dir)
        LOG.info(f"Setting OS env var {c.NGEN_LOG_DIR_KEY} to {log_dir}")
        os.environ[c.NGEN_LOG_DIR_KEY] = log_dir

    def _parse_lagged_ensemble_args(self):
        """Break up the multipart lagged ensemble arg into distinct args and set them.
        Called by child classes which define the necessary attributes."""
        if self.lagged_ensemble_args:
            if self.forcing_configuration != "medium_range":
                self.errors.append(
                    ValueError(
                        f"lagged ensemble only supported for medium_range, but forcing configuration {repr(self.forcing_configuration)} was provided"
                    )
                )
                return

            self.use_lagged_ensemble = True

            member_name, open_ls, closed_ls = self.lagged_ensemble_args

            self.lagged_ens_mem = member_name if member_name.strip() else None
            self.forcing_lag = LAGGED_ENSEMBLE_MEMBER_LAGS[self.lagged_ens_mem]
            self.le__open_loop_state = open_ls if open_ls.strip() else None
            self.le__closed_loop_state = closed_ls if closed_ls.strip() else None

            if self.lagged_ens_mem not in LAGGED_ENSEMBLE_MEMBER_LAGS:
                self.errors.append(
                    KeyError(
                        f"Invalid lagged ensemble member {repr(self.lagged_ens_mem)} (choose from: {list(LAGGED_ENSEMBLE_MEMBER_LAGS)})"
                    )
                )

        if self.le__open_loop_state or self.le__closed_loop_state:
            self.errors.append(
                NotImplementedError(
                    "Lagged ensemble args for Open Loop State and Closed Loop State are not yet implemented in nwm-rte (should be provided as empty strings for now)"
                )
            )

    @property
    def _fcst_run_name_formatted(self) -> str:
        """Adaptive forecast run name that optionally can have a timestamped suffix appended to the end."""
        if self.add_timestamp_to_run_name:
            if not self.fcst_run_name:
                raise ValueError(
                    "Must provide fcst_run_name when using timestamp_run_name"
                )
            return f"{self.fcst_run_name}_{self.time_at_init.strftime(c.RUN_NAME_TIMESTAMP_SUFFIX_FORMAT)}"
        else:
            return f"{self.fcst_run_name}"

    @property
    def realtime_mode(self) -> bool:
        """Realtime mode boolean, leveraged by default realization which can run both realtime and historical/retrospective configurations."""
        if self.forcing_configuration in c.FORECAST_FORCING_TYPES + ["medium_range"]:
            return True
        else:
            return False

    @property
    def forcing_provider_paths(self) -> ForcingProviderPaths:
        """Helper class for managing forcing provider paths."""
        fpp = ForcingProviderPaths(
            global_domain=self.global_domain,
            forcing_static_dir=self.forcing_static_dir,
        )
        return fpp

    @property
    def calib_windows(self) -> CalibTimeWindows:
        """Class of various calibration time windows."""
        windows = CalibTimeWindows(
            calib_sim_start=self.calib_sim_start
            if self.calib_sim_start
            else c.CALIB_SIM_START_DEFAULT,
            calib_sim_duration=self.duration
            if self.duration
            else c.CALIB_SIM_DURATION_DEFAULT,
            calib_eval_delayment=c.CALIB_EVAL_DELAYMENT_DEFAULT,
            valid_sim_advancement=c.VALID_SIM_ADVANCEMENT_DEFAULT,
            valid_eval_curtailment=c.VALID_EVAL_CURTAILMENT_DEFAULT,
        )
        return windows

    @property
    def start_period__end_period(self) -> tuple[str | None, str | None]:
        """Tuple of start period and end period for calibration and default realizations."""
        if isinstance(self, RTECalibConfig):
            start_period = self.calib_windows.calib_eval_start.strftime(DDF)
            end_period = self.calib_windows.calib_eval_end.strftime(DDF)
        elif isinstance(self, RTEDefaultConfig) and not self.realtime_mode:
            start_period = self.cycle_datetime.strftime(DDF)
            end_period = (self.cycle_datetime + self.duration).strftime(DDF)
        else:
            start_period = None
            end_period = None
        return start_period, end_period

    @property
    def model_formulation(self) -> ModelFormulation:
        """Model formulation helper class for MSWM models list and rootzone parameter."""
        mf = ModelFormulation(
            self.model_formulation_cli_csv,
            self.model_formulation_cli_rootzone,
        )
        return mf

    @property
    def run_type(self) -> str:
        """Run type for MSWM GeneralConfig"""
        if isinstance(self, RTECalibConfig):
            rt = "calibration"
        elif isinstance(self, RTEDefaultConfig):
            rt = "default"
        elif isinstance(self, RTEForecastConfig):
            rt = "default"
        elif isinstance(self, RTERegionConfig):
            rt = "regionalization"
        else:
            raise ValueError(
                f"Unexpected config class {type(self)}. Expected one of RTEForecastConfig, RTECalibConfig, RTERegionConfig, or RTEDefaultConfig."
            )
        return rt

    @property
    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:
            formulation = self.fcst_run_name
        else:
            formulation = self.forcing_provider_paths.formulation_name

        return GeneralConfig(
            basin=self.basin,
            environment=self.environment,
            run_type=self.run_type,
            models=self.model_formulation.models_csv,
            formulation=formulation,
            main_dir=c.DEFAULT_MAIN_DIR,
            start_period=start_period,
            end_period=end_period,
            output_precip=True,
            output_swe=True,
            output_sm=True,
            domain=self.global_domain.lower(),
            subset_type=self.subset_type,
        )

    @property
    def mswm_ModulePropertiesConfig(self) -> ModulePropertiesConfig:
        """MSWM ModulePropertiesConfig instance"""
        mpc = ModulePropertiesConfig(
            cfe_aet_rootzone=self.model_formulation.cfe_aet_rootzone,
        )
        return mpc

    @property
    def mswm_NWMOutputConfig(self) -> NWMOutputConfig:
        """MSWM NWMOutputConfig instance"""
        oc = NWMOutputConfig(
            nwm_output_variables=self.nwm_output_vars,
            output_format=self.output_format,
        )
        return oc

    @property
    def mswm_RegionalizationConfig(self) -> RegionConfig | None:
        """MSWM RegionalizationConfig instance"""
        if not isinstance(self, RTERegionConfig):
            return None
        rc = RegionConfig(
            form_assign_file=self.form_assign_file,
            cat_grp_file=self.cat_grp_file,
        )
        return rc

    @property
    def mswm_CalibConfig(self) -> CalibConfig | None:
        """MSWM CalibConfig instance"""
        if not isinstance(self, RTECalibConfig):
            return None
        cc = CalibConfig(
            optimization_algorithm=self.optimization_algorithm,
            swarm_size=c.CALIB_SWARM_SIZE,
            c1=c.CALIB_PSO_C1,
            c2=c.CALIB_PSO_C2,
            w=c.CALIB_PSO_W,
            objective_function=self.objective_function,
            start_iteration=c.CALIB_ITER_START,
            number_iteration=c.CALIB_ITER_COUNT,
            calib_output_vars=True,
            valid_output_vars=True,
            calib_start_period=self.calib_windows.calib_sim_start.strftime(DDF),
            calib_end_period=self.calib_windows.calib_sim_end.strftime(DDF),
            calib_eval_start_period=self.calib_windows.calib_eval_start.strftime(DDF),
            calib_eval_end_period=self.calib_windows.calib_eval_end.strftime(DDF),
            valid_start_period=self.calib_windows.valid_sim_start.strftime(DDF),
            valid_end_period=self.calib_windows.valid_sim_end.strftime(DDF),
            valid_eval_start_period=self.calib_windows.valid_eval_start.strftime(DDF),
            valid_eval_end_period=self.calib_windows.valid_eval_end.strftime(DDF),
            full_eval_start_period=self.calib_windows.full_eval_start.strftime(DDF),
            full_eval_end_period=self.calib_windows.full_eval_end.strftime(DDF),
            save_plot_iter_freq=c.CALIB_SAVE_PLOT_ITER_FREQ,
            ngen_cerf=False,
            calib_parameter_file=c.CALIB_PARAMETERS_DIR,
        )
        return cc

    @property
    def mswm_ForcingConfig(self) -> ForcingConfig:
        """MSWM ForcingConfig instance.
        Contains dynamic logic for handling different types of child classes of RTEBaseConfig."""
        if isinstance(self, RTECalibConfig):
            cdt = self.calib_windows.calib_sim_start.strftime(
                mswm_settings.DEFAULT_DATETIME_FORMAT
            )
        elif isinstance(self, (RTEForecastConfig, RTEDefaultConfig, RTERegionConfig)):
            cdt = (
                self.cycle_datetime.strftime(mswm_settings.DEFAULT_DATETIME_FORMAT)
                if self.cycle_datetime
                else None
            )
        else:
            raise ValueError(
                f"Unexpected config class {type(self)}. Expected one of RTEForecastConfig, RTECalibConfig, RTERegionConfig, or RTEDefaultConfig."
            )
        cold_start_datetime = (
            self.cold_start_datetime.strftime(mswm_settings.DEFAULT_DATETIME_FORMAT)
            if isinstance(self, RTEForecastConfig) and self.cold_start_datetime
            else None
        )
        fc = ForcingConfig(
            forcing_provider=c.FORCING_PROVIDER,
            forcing_dir=self.forcing_static_dir,
            forcing_template_dir=c.FORCING_TEMPLATE_DIR,
            root_dir=c.FORCING_ROOT_DIR,
            forcing_configuration=self.forcing_configuration,
            cycle_datetime=cdt,
            cold_start_datetime=cold_start_datetime,
            forcing_static_dir=self.forcing_static_dir,
            scratch_dir_override=c.SCRATCH_DIR_OVERRIDE,
            forcing_product_versions=c.FORCING_PRODUCT_VERSIONS_DICT,
            lookback=self.lookback,
        )
        return fc

    @property
    def mswm_DataFileConfig(self) -> DataFileConfig:
        """MSWM DataFileConfig instance."""
        obs_dir, nwmretro_file, errors = get_paths_for_observed_and_retro_data(
            self.global_domain,
            self.gage_id,
            models_csv=self.model_formulation.models_csv,
        )
        if errors:
            raise RuntimeError(errors)
        dfc = DataFileConfig(
            **(
                c.DATAFILE_LIBS
                | {
                    "obs_dir": obs_dir,
                    "nwmretro_file": nwmretro_file,
                    "hydrofab_file": self.hydrofab_file,
                }
            )
        )
        return dfc

    @property
    def mswm_ParallelConfig(self) -> ParallelConfig:
        """MSWM ParallelConfig instance."""
        pc = make_parallel_config(self.nprocs)
        return pc

    @property
    def mswm_InputConfig(self) -> InputConfig:
        """MSWM InputConfig instance. This is the composite MSWM class that contains the other sub-config classes."""
        general = self.mswm_GeneralConfig
        module_properties = self.mswm_ModulePropertiesConfig
        nwm_output = self.mswm_NWMOutputConfig
        regionalization = self.mswm_RegionalizationConfig
        calibration = self.mswm_CalibConfig
        forcing = self.mswm_ForcingConfig
        data_file = self.mswm_DataFileConfig
        parallel = self.mswm_ParallelConfig
        ic = InputConfig(
            General=general,
            ModuleProperties=module_properties,
            NWMOutput=nwm_output,
            Regionalization=regionalization,
            Calibration=calibration,
            Forcing=forcing,
            DataFile=data_file,
            Parallel=parallel,
        )
        return ic

    @property
    def mswm_RealizationBuilder_kwargs(self) -> dict:
        """Full set of kwargs that are passed to MSWM RealizationBuilder's constructor for bulding a realization."""
        kwargs = {
            # "input_path": forecast_vars.forecast_input_config,
            "valid_yaml": self.valid_best_yaml,
            "fcst_run_name": self._fcst_run_name_formatted,
            "config_overrides": self.mswm_InputConfig,
            "use_lagged_ens": self.use_lagged_ensemble,
            "lagged_ens_mem": self.lagged_ens_mem,
            "forcing_lag": self.forcing_lag,
            "checkpoint_interval": self.checkpoint_interval,
            "checkpoint_dir": self.checkpoint_dir,
            "load_state_from": self.load_state_from,
            "save_state": self.save_state,
            "save_state_dir": self.save_state_dir,
        }
        if self.errors:
            raise RuntimeError(self.errors)
        return kwargs

    @property
    def run_dir_base(self) -> str:
        """Run directory root"""
        ret = f"{c.DEFAULT_MAIN_DIR}/{self.objective_function.value}_{self.optimization_algorithm.value}/test_{c.FORCING_PROVIDER}/{self.gage_id}"
        if not os.path.isdir(ret):
            msg = f"Not a directory: {repr(ret)}. Please review choices for objective function, optimization algorithm, and gage, which affect this path."
            raise NotADirectoryError(msg)
        return ret

    @property
    def run_dir_input(self) -> str:
        """Input run directory"""
        return f"{self.run_dir_base}/Input"

    @property
    def run_dir_output(self) -> str:
        """Output run directory"""
        return f"{self.run_dir_base}/Output"

    @property
    def ngen_log_file(self) -> str:
        """ngen stdout + stderr stream log file"""
        return f"{self.run_dir_base}/logs/ngen.log"

    @property
    def valid_best_yaml(self) -> str:
        """Validation yaml file (output from previously-ran calibration realization)"""
        return (
            f"{self.run_dir_output}/Validation_Run/{self.gage_id}_config_valid_best.yaml"
            if isinstance(self, RTEForecastConfig)
            else None
        )

time_at_init class-attribute instance-attribute

time_at_init: datetime | None = Field(
    init=False, default=None
)

Time at class instantiation.

errors class-attribute instance-attribute

errors: list | None = Field(init=False, default=None)

List of exceptions encountered during init.

subset_type class-attribute instance-attribute

subset_type: str | None = Field(init=False, default=None)

Subset_type set depending on vpu and gage_id args

basin class-attribute instance-attribute

basin: str | None = Field(init=False, default=None)

Basin id set depending on vpu and gage_id args

use_lagged_ensemble class-attribute instance-attribute

use_lagged_ensemble: bool | None = Field(
    init=False, default=False
)

Boolean indicating that lagged ensemble is to be used. Passed to MSWM.

lagged_ens_mem class-attribute instance-attribute

lagged_ens_mem: str | None = Field(init=False, default=None)

Lagged ensemble member name. Passed to MSWM.

forcing_lag class-attribute instance-attribute

forcing_lag: int | None = Field(init=False, default=None)

Lagged ensemble forcing lag. Looked up based on lagged ensemble member name. Passed to MSWM.

le__open_loop_state class-attribute instance-attribute

le__open_loop_state: str | None = Field(
    init=False, default=None
)

File path for lagged ensemble open loop state.

le__closed_loop_state class-attribute instance-attribute

le__closed_loop_state: str | None = Field(
    init=False, default=None
)

File path for lagged ensemble closed loop state.

realtime_mode property

realtime_mode: bool

Realtime mode boolean, leveraged by default realization which can run both realtime and historical/retrospective configurations.

forcing_provider_paths property

forcing_provider_paths: ForcingProviderPaths

Helper class for managing forcing provider paths.

calib_windows property

calib_windows: CalibTimeWindows

Class of various calibration time windows.

start_period__end_period property

start_period__end_period: tuple[str | None, str | None]

Tuple of start period and end period for calibration and default realizations.

model_formulation property

model_formulation: ModelFormulation

Model formulation helper class for MSWM models list and rootzone parameter.

run_type property

run_type: str

Run type for MSWM GeneralConfig

mswm_GeneralConfig property

mswm_GeneralConfig: GeneralConfig

MSWM GeneralConfig instance

mswm_ModulePropertiesConfig property

mswm_ModulePropertiesConfig: ModulePropertiesConfig

MSWM ModulePropertiesConfig instance

mswm_NWMOutputConfig property

mswm_NWMOutputConfig: NWMOutputConfig

MSWM NWMOutputConfig instance

mswm_RegionalizationConfig property

mswm_RegionalizationConfig: RegionConfig | None

MSWM RegionalizationConfig instance

mswm_CalibConfig property

mswm_CalibConfig: CalibConfig | None

MSWM CalibConfig instance

mswm_ForcingConfig property

mswm_ForcingConfig: ForcingConfig

MSWM ForcingConfig instance. Contains dynamic logic for handling different types of child classes of RTEBaseConfig.

mswm_DataFileConfig property

mswm_DataFileConfig: DataFileConfig

MSWM DataFileConfig instance.

mswm_ParallelConfig property

mswm_ParallelConfig: ParallelConfig

MSWM ParallelConfig instance.

mswm_InputConfig property

mswm_InputConfig: InputConfig

MSWM InputConfig instance. This is the composite MSWM class that contains the other sub-config classes.

mswm_RealizationBuilder_kwargs property

mswm_RealizationBuilder_kwargs: dict

Full set of kwargs that are passed to MSWM RealizationBuilder's constructor for bulding a realization.

run_dir_base property

run_dir_base: str

Run directory root

run_dir_input property

run_dir_input: str

Input run directory

run_dir_output property

run_dir_output: str

Output run directory

ngen_log_file property

ngen_log_file: str

ngen stdout + stderr stream log file

valid_best_yaml property

valid_best_yaml: str

Validation yaml file (output from previously-ran calibration realization)

configure_ngen_log

configure_ngen_log(rb: RealizationBuilder) -> None

Configure the ngen logging, by setting the associated OS env variable for the directory to hold the logs, and copying the associated json file into that directory.

fallback_log_dir is ignored when the RTE OS env var key NGEN_LOG_TO_RTE is true. It is used to emulate behavior of nwm-cal-mgr and nwm-fcst-mgr (what they would use without RTE).

PARAMETER	DESCRIPTION
rb	An already built realization. TYPE: RealizationBuilder

Source code in bin_mounted/ngen_rte/configs.py

def configure_ngen_log(self, rb: RealizationBuilder) -> None:
    """Configure the ngen logging, by setting the associated OS env variable for the directory to hold the logs,
    and copying the associated json file into that directory.

    ``fallback_log_dir`` is ignored when the RTE OS env var key NGEN_LOG_TO_RTE is true.
    It is used to emulate behavior of nwm-cal-mgr and nwm-fcst-mgr (what they would use without RTE).

    Parameters
    ----------
    rb : RealizationBuilder
        An already built realization.
    """
    now_str = datetime.now(timezone.utc).strftime(r"%Y%m%d_%H%M%S_%f")

    label = rb.run_type
    if getattr(rb, "use_cold_start", False):
        label = f"{label}_cs"
    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":
        fallback_log_dir = str(rb.work_dir)
    else:
        raise RuntimeError(f"Unexpected run_type: {rb.run_type}")

    # Confirm that it's valid json content
    LOG.debug(f"Reading: {c.SRC_LOG_CONFIG_JSON}")
    with open(c.SRC_LOG_CONFIG_JSON) as f:
        try:
            _ = json.load(f)
        except Exception as e:
            raise RuntimeError(
                f"Could not read or parse as json: {c.SRC_LOG_CONFIG_JSON}: {e}"
            ) from e

    # Decide the dir
    setting_val = os.environ.get(c.RTE_NGEN_LOG_BEHAVIOR_KEY, "").lower().strip()
    if setting_val in ("yes", "true"):
        log_dir = os.path.join(c.CONTAINER_LOGS_DIR, "ngen", f"{now_str}_{label}")
    elif setting_val in ("no", "false", ""):
        log_dir = fallback_log_dir
    else:
        raise ValueError(
            f"Invalid value for key {repr(c.RTE_NGEN_LOG_BEHAVIOR_KEY)}: {repr(setting_val)} (expected YES or NO, defaulting to NO if not provided)"
        )

    # Make the dir, copy the log json config into it, and set the OS env var for ngen to be able to find it.
    LOG.debug(f"Making directory: {log_dir}")
    os.makedirs(log_dir, exist_ok=True)
    LOG.info(f"Copying: {c.SRC_LOG_CONFIG_JSON} -> {log_dir}/")
    shutil.copy2(c.SRC_LOG_CONFIG_JSON, log_dir)
    LOG.info(f"Setting OS env var {c.NGEN_LOG_DIR_KEY} to {log_dir}")
    os.environ[c.NGEN_LOG_DIR_KEY] = log_dir

RTEDefaultConfig

Bases: RTEBaseConfig

Configuration class for building and running one default realization (realtime forcing configuration or historical / retrospective forcing configuration).

ATTRIBUTE	DESCRIPTION
cycle_datetime	Start time of the realization TYPE: datetime
duration	Duration of the simulation (only used for historical / retrospective forcing configurations) TYPE: timedelta | None
forcing_configuration	Forcing configuration, e.g. "aorc" or "short_range" TYPE: str
fcst_run_name	Name of the forecast realization run. Affects a directory name. TYPE: str
lagged_ensemble_args	See CLI help menu for run_default.py for details. TYPE: list[str] | None = Field(min_length=3, max_length=3)

Source code in bin_mounted/ngen_rte/configs.py

class RTEDefaultConfig(RTEBaseConfig):
    """Configuration class for building and running one default realization
    (realtime forcing configuration or historical / retrospective forcing configuration).

    Attributes
    ----------
    cycle_datetime: datetime
        Start time of the realization
    duration: timedelta | None
        Duration of the simulation (only used for historical / retrospective forcing configurations)
    forcing_configuration: str
        Forcing configuration, e.g. "aorc" or "short_range"
    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.
    """

    cycle_datetime: datetime
    duration: timedelta | None
    forcing_configuration: str
    fcst_run_name: str
    # For medium-range lagged ensemble
    lagged_ensemble_args: list[str] | None = Field(min_length=3, max_length=3)

    def model_post_init(self, __context) -> None:
        super().model_post_init(__context)  # Call RTEBaseConfig's post init
        super()._parse_lagged_ensemble_args()
        if self.errors:
            raise RuntimeError(self.errors)

RTERegionConfig

Bases: RTEBaseConfig

Configuration class for building and running one regionalization realization (realtime forcing configuration or historical / retrospective forcing configuration).

ATTRIBUTE	DESCRIPTION
cycle_datetime	Start time of the realization TYPE: datetime
duration	Duration of the simulation (only used for historical / retrospective forcing configurations) TYPE: timedelta | None
forcing_configuration	Forcing configuration, e.g. "aorc" or "short_range" TYPE: str
fcst_run_name	Name of the forecast realization run. Affects a directory name. TYPE: str
lagged_ensemble_args	See CLI help menu for run_default.py for details. TYPE: list[str] | None = Field(min_length=3, max_length=3)
form_assign_file	File containing formulation assignments for catchments TYPE: str
cat_grp_file	File containing catchment groupings for regionalization TYPE: str

Source code in bin_mounted/ngen_rte/configs.py

class RTERegionConfig(RTEBaseConfig):
    """Configuration class for building and running one regionalization realization
    (realtime forcing configuration or historical / retrospective forcing configuration).

    Attributes
    ----------
    cycle_datetime: datetime
        Start time of the realization
    duration: timedelta | None
        Duration of the simulation (only used for historical / retrospective forcing configurations)
    forcing_configuration: str
        Forcing configuration, e.g. "aorc" or "short_range"
    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.
    form_assign_file: str
        File containing formulation assignments for catchments
    cat_grp_file: str
        File containing catchment groupings for regionalization
    """

    cycle_datetime: datetime
    duration: timedelta | None
    forcing_configuration: str
    fcst_run_name: str
    # For medium-range lagged ensemble
    lagged_ensemble_args: list[str] | None = Field(min_length=3, max_length=3)
    form_assign_file: str
    cat_grp_file: str

    def model_post_init(self, __context) -> None:
        super().model_post_init(__context)  # Call RTEBaseConfig's post init
        super()._parse_lagged_ensemble_args()
        if self.errors:
            raise RuntimeError(self.errors)

RTECalibConfig

Bases: RTEBaseConfig

Configuration class for building and running one calibration realization.

ATTRIBUTE	DESCRIPTION
objective_function	Objective function, e.g. "kge" TYPE: CalObjective
optimization_algorithm	Optimization algorithm, e.g. "dds" TYPE: CalOptimizationAlgo
calib_sim_start	Calibration start time TYPE: datetime
duration	Calibration simulation duration TYPE: timedelta
calib_eval_delayment	Used for evaluation / validation time windowing TYPE: timedelta
valid_sim_advancement	Used for evaluation / validation time windowing TYPE: timedelta
valid_eval_curtailment	Used for evaluation / validation time windowing TYPE: timedelta
forcing_configuration	Source of forcing data, e.g. "aorc" or "nwm" TYPE: str
worker_name	Name of the ngen worker (used to build a directory name) TYPE: str | None

Source code in bin_mounted/ngen_rte/configs.py

class RTECalibConfig(RTEBaseConfig):
    """Configuration class for building and running one calibration realization.

    Attributes
    ----------
    objective_function: c.CalObjective
        Objective function, e.g. "kge"
    optimization_algorithm: c.CalOptimizationAlgo
        Optimization algorithm, e.g. "dds"
    calib_sim_start: datetime
        Calibration start time
    duration: timedelta
        Calibration simulation duration
    calib_eval_delayment: timedelta
        Used for evaluation / validation time windowing
    valid_sim_advancement: timedelta
        Used for evaluation / validation time windowing
    valid_eval_curtailment: timedelta
        Used for evaluation / validation time windowing
    forcing_configuration: str
        Source of forcing data, e.g. "aorc" or "nwm"
    worker_name: str | None
        Name of the ngen worker (used to build a directory name)
    """

    objective_function: c.CalObjective
    optimization_algorithm: c.CalOptimizationAlgo
    calib_sim_start: datetime
    duration: timedelta
    calib_eval_delayment: timedelta
    valid_sim_advancement: timedelta
    valid_eval_curtailment: timedelta
    forcing_configuration: str
    worker_name: str | None

    def model_post_init(self, __context) -> None:
        super().model_post_init(__context)  # Call RTEBaseConfig's post init
        if self.forcing_configuration not in c.CALIB_FORCING_TYPES:
            self.errors.append(
                ValueError(
                    f"Unexpected forcing_configuration: {self.forcing_configuration} (for calibration, choose from: {c.CALIB_FORCING_TYPES})"
                )
            )

        if self.nwm_output_vars:
            self.errors.append(
                ValueError("nwm_output_vars not supported for calibration workflow.")
            )

        if self.errors:
            raise RuntimeError(self.errors)

RTEForecastConfig

Bases: RTEBaseConfig

Configuration class for building and running one forecast realization.

ATTRIBUTE	DESCRIPTION
objective_function	Affects input realization path. Objective function of previously-ran calibration realization, e.g. "kge" TYPE: CalObjective
optimization_algorithm	Affects input realization path. Optimization algorithm of previously-ran calibration realization, e.g. "dds" TYPE: CalOptimizationAlgo
cycle_datetime	Start time of the realization (or end time for coldstart, if cold_start_datetime is provided) TYPE: datetime | None
cold_start_datetime	Start time of the coldstart realization. If None, coldstart is not performed. TYPE: datetime | None
forcing_configuration	Forcing configuration, e.g. "aorc" or "short_range" TYPE: str
fcst_run_name	Name of the forecast realization run TYPE: str
lagged_ensemble_args	See CLI help menu for run_forecast.py for details. TYPE: list[str] | None = Field(min_length=3, max_length=3)

Source code in bin_mounted/ngen_rte/configs.py

class RTEForecastConfig(RTEBaseConfig):
    """Configuration class for building and running one forecast realization.

    Attributes
    ----------
    objective_function: c.CalObjective
        Affects input realization path. Objective function of previously-ran calibration realization, e.g. "kge"
    optimization_algorithm: c.CalOptimizationAlgo
        Affects input realization path. Optimization algorithm of previously-ran calibration realization, e.g. "dds"
    cycle_datetime: datetime | None
        Start time of the realization (or end time for coldstart, if `cold_start_datetime` is provided)
    cold_start_datetime: datetime | None
        Start time of the coldstart realization. If None, coldstart is not performed.
    forcing_configuration: str
        Forcing configuration, e.g. "aorc" or "short_range"
    fcst_run_name: str
        Name of the forecast realization run
    lagged_ensemble_args: list[str] | None = Field(min_length=3, max_length=3)
        See CLI help menu for [`run_forecast.py`](python_cli_help__run_forecast.py.txt) for details.
    """

    # These calibration parameters affect directory path
    objective_function: c.CalObjective
    optimization_algorithm: c.CalOptimizationAlgo
    cycle_datetime: datetime | None
    cold_start_datetime: datetime | None
    forcing_configuration: str
    fcst_run_name: str
    # For medium-range lagged ensemble
    lagged_ensemble_args: list[str] | None = Field(min_length=3, max_length=3)

    def model_post_init(self, __context) -> None:
        super().model_post_init(__context)  # Call RTEBaseConfig's post init
        super()._parse_lagged_ensemble_args()
        self._check_time_config()
        if self.errors:
            raise RuntimeError(self.errors)

    def _check_time_config(self) -> None:
        """Validate the configuration"""
        if not (self.cold_start_datetime or self.cycle_datetime):
            self.errors.append(
                ValueError(
                    "Must provide cold_start_datetime or cycle_datetime (or both), but neither were provided."
                )
            )

RTEAsyncConfig

Bases: RTEBaseConfig

Minimal configuration class for run using NgenRunnerAsync Set RTEBaseConfigs variables to defaults, not used by NgenRunnerAsync

Source code in bin_mounted/ngen_rte/configs.py

class RTEAsyncConfig(RTEBaseConfig):
    """Minimal configuration class for run using NgenRunnerAsync
    Set RTEBaseConfigs variables to defaults, not used by NgenRunnerAsync
    """

    delete_scratch_and_mesh_first: bool = False
    delete_forcing_raw_input_first: bool = False
    environment: str = ""
    global_domain: str = ""
    forcing_static_dir: str = ""
    gage_id: str = ""

    def model_post_init(self, __context) -> None:
        super().model_post_init(__context)  # Call RTEBaseConfig's post init
        if self.errors:
            raise RuntimeError(self.errors)

RTETestConfig

Bases: RTEBaseConfig

Configuration class for building and running a set of test realizations.

ATTRIBUTE	DESCRIPTION
skip_forecast	Causes forecast to be skipped (only do calibration) TYPE: bool
quit_forecast_after_duration	Causes forecasts to be stopped midway after a set duration (seconds of processing time) TYPE: float | None = Field(ge=0)
do_calibration	Causes calibration to be ran, before forecasts (needed if a calibration has not yet been ran for the gage) TYPE: bool
quit_calibration_after_duration	Causes calibrations to be stopped midway after a set duration (seconds of processing time) TYPE: float | None = Field(ge=0)
objective_functions	For calibration, list of objective functions to run, e.g. "kge". Replaced with full list when do_all_objective_functions = True TYPE: list[CalObjective]
do_all_objective_functions	For calibration, causes all objective functions to be used. TYPE: bool
optimization_algorithms	For calibration, list of optimization algorithms to run, e.g. "dds". Replaced with full list when do_all_optimization_algorithms = True TYPE: list[CalOptimizationAlgo]
model_formulations_file	File containing model formulations to iterate over TYPE: str | None
calibration_forcing_sources	Calibration forcing configurations, e.g. "aorc" "nwm" TYPE: list[str]
do_all_optimization_algorithms	For calibration, causes all optimization algorithms to be used. TYPE: bool
do_all_forcing_configs	Causes all forcing configurations to be used, e.g. "short_range", "standard_ana", "medium_range_blend", "extended_ana", "short_range_hawaii", etc. TYPE: bool
do_coldstart	Causes coldstart to be ran before forecast. TYPE: bool
fcst_run_name	Name of the forecast realization run. Affects a directory name. TYPE: str
noop	Causes a noop to occur (for confirming that Python packages are importable). TYPE: bool
restart	Causes the test to be restarted by reading the existing c.TEST_RESULTS_FILE and skipping configurations which had completed earlier. Can be used when long-running tests are interrupted. Use with caution. TYPE: bool

Source code in bin_mounted/ngen_rte/configs.py

class RTETestConfig(RTEBaseConfig):
    """Configuration class for building and running a set of test realizations.

    Attributes
    ----------
    skip_forecast: bool
        Causes forecast to be skipped (only do calibration)
    quit_forecast_after_duration: float | None = Field(ge=0)
        Causes forecasts to be stopped midway after a set duration (seconds of processing time)
    do_calibration: bool
        Causes calibration to be ran, before forecasts (needed if a calibration has not yet been ran for the gage)
    quit_calibration_after_duration: float | None = Field(ge=0)
        Causes calibrations to be stopped midway after a set duration (seconds of processing time)
    objective_functions: list[c.CalObjective]
        For calibration, list of objective functions to run, e.g. "kge". Replaced with full list when do_all_objective_functions = True
    do_all_objective_functions: bool
        For calibration, causes all objective functions to be used.
    optimization_algorithms: list[c.CalOptimizationAlgo]
        For calibration, list of optimization algorithms to run, e.g. "dds". Replaced with full list when do_all_optimization_algorithms = True
    model_formulations_file: str | None
        File containing model formulations to iterate over
    calibration_forcing_sources: list[str]
        Calibration forcing configurations, e.g. "aorc" "nwm"
    do_all_optimization_algorithms: bool
        For calibration, causes all optimization algorithms to be used.
    do_all_forcing_configs: bool
        Causes all forcing configurations to be used, e.g. "short_range", "standard_ana", "medium_range_blend", "extended_ana", "short_range_hawaii", etc.
    do_coldstart: bool
        Causes coldstart to be ran before forecast.
    fcst_run_name: str
        Name of the forecast realization run. Affects a directory name.
    noop: bool
        Causes a noop to occur (for confirming that Python packages are importable).
    restart: bool
        Causes the test to be restarted by reading the existing c.TEST_RESULTS_FILE and skipping configurations which had completed earlier. Can be used when long-running tests are interrupted. Use with caution.
    """

    skip_forecast: bool
    quit_forecast_after_duration: float | None = Field(ge=0)
    do_calibration: bool
    quit_calibration_after_duration: float | None = Field(ge=0)
    # Replaced with full list when do_all_objective_functions = True
    objective_functions: list[c.CalObjective]
    do_all_objective_functions: bool
    # Replaced with full list when do_all_optimization_algorithms = True
    optimization_algorithms: list[c.CalOptimizationAlgo]
    model_formulations_file: str | None
    calibration_forcing_sources: list[str]
    do_all_optimization_algorithms: bool
    do_all_forcing_configs: bool
    do_coldstart: bool
    fcst_run_name: str
    noop: bool
    restart: bool

    def model_post_init(self, __context) -> None:
        super().model_post_init(__context)  # Call RTEBaseConfig's post init

        errors_extend = parse_fcst_run_name(self._fcst_run_name_formatted)
        self.errors.extend(errors_extend)

        if self.do_all_objective_functions:
            self.objective_functions = list(c.CalObjective)
        if self.do_all_optimization_algorithms:
            self.optimization_algorithms = list(c.CalOptimizationAlgo)

        if self.do_all_forcing_configs:
            if self.skip_forecast and (not self.do_coldstart):
                self.errors.append(
                    ValueError(
                        f"When do_all_forcing_configs={self.do_all_forcing_configs}, must have coldstart and/or forecast enabled."
                    )
                )

        if self.errors:
            raise RuntimeError(self.errors)

    def get_calib_permutations(
        self,
    ) -> list[tuple[c.CalObjective, c.CalOptimizationAlgo, TestPaths]]:
        """Returns the permutations of objective function and optimization algorithm specified in the config, as well as a TestPaths instance for each.
        If only_first, then only the first permutation will be returned. Else all permutations will be returned."""
        ret = []
        for obj_func in self.objective_functions:
            if obj_func == c.CalOptimizationAlgo.none:
                # TODO enable objective function "none" for supported circumstances
                continue
            for optim_algo in self.optimization_algorithms:
                if optim_algo == c.CalOptimizationAlgo.none:
                    # TODO enable optimization algo "none" for supported circumstances
                    continue
                ret.append(
                    (
                        obj_func,
                        optim_algo,
                        TestPaths(
                            self.gage_id,
                            obj_func,
                            optim_algo,
                            self.global_domain,
                            self.forcing_static_dir,
                        ),
                    )
                )
        return ret

get_calib_permutations

get_calib_permutations() -> (
    list[
        tuple[CalObjective, CalOptimizationAlgo, TestPaths]
    ]
)

Returns the permutations of objective function and optimization algorithm specified in the config, as well as a TestPaths instance for each. If only_first, then only the first permutation will be returned. Else all permutations will be returned.

Source code in bin_mounted/ngen_rte/configs.py

def get_calib_permutations(
    self,
) -> list[tuple[c.CalObjective, c.CalOptimizationAlgo, TestPaths]]:
    """Returns the permutations of objective function and optimization algorithm specified in the config, as well as a TestPaths instance for each.
    If only_first, then only the first permutation will be returned. Else all permutations will be returned."""
    ret = []
    for obj_func in self.objective_functions:
        if obj_func == c.CalOptimizationAlgo.none:
            # TODO enable objective function "none" for supported circumstances
            continue
        for optim_algo in self.optimization_algorithms:
            if optim_algo == c.CalOptimizationAlgo.none:
                # TODO enable optimization algo "none" for supported circumstances
                continue
            ret.append(
                (
                    obj_func,
                    optim_algo,
                    TestPaths(
                        self.gage_id,
                        obj_func,
                        optim_algo,
                        self.global_domain,
                        self.forcing_static_dir,
                    ),
                )
            )
    return ret

make_parallel_config

make_parallel_config(nprocs: int) -> ParallelConfig

MSWM ParallelConfig instance.

Source code in bin_mounted/ngen_rte/configs.py

def make_parallel_config(nprocs: int) -> ParallelConfig:
    """MSWM ParallelConfig instance."""
    if nprocs and nprocs > 1:
        parallel = ParallelConfig(
            parallel_ngen_exe=c.NGEN_BIN__LINK,
            partition_generator_exe=c.PARTITION_GENERATOR_BIN__LINK,
            nprocs=nprocs,
        )
    else:
        parallel = ParallelConfig(nprocs=nprocs)
    return parallel

Regionalization Workflows

The regionalization workflows run via command-line interface (CLI) Scripts.

Here is the argparse help menu of run_regionalization.py:

run_regionalization.py --help

run_regionalization is called by run_region.sh

For more information, see the nwm-region-mgr repository.

ngen_rte.run_regionalization

RTE regionalization workflow runner script.