Skip to content

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

run_regionalization_standalone.py --help

run_output_postprocess.py --help

run_output_mosaic.py --help

make_coastal_forcing.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, cfg: RTEDefaultConfig
) -> NgenRunnerAsync

Run the provided default realization.

Realization should already be built (rb.build_default_realization() already called).

Args: rb: The RealizationBuilder instance containing the realization. cfg: The RTEDefaultConfig instance, used to access ecf_iface and ecf_ctx.

Source code in bin_mounted/ngen_rte/run_default.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def run_default(
    rb: RealizationBuilder,
    cfg: RTEDefaultConfig,
) -> NgenRunnerAsync:
    """Run the provided default realization.

    Realization should already be built (rb.build_default_realization() already called).

    Args:
        rb: The RealizationBuilder instance containing the realization.
        cfg: The RTEDefaultConfig instance, used to access ecf_iface and ecf_ctx.
    """
    LOG.info("Running default realization")
    # For default realization, currently postprocess needs suppress_output=True
    ngen_runner = NgenRunnerAsync(
        rb=rb,
        postprocess=True,
        suppress_output=True,
        ecf_iface=cfg.ecf_iface,
        ecf_ctx=cfg.ecf_ctx,
    )
    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 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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,
        allow_abbrev=False,
    )
    cli_args.add_args_for_script(parser, cli_args.Script.DEFAULT)
    parser.add_argument(
        "--noop",
        action="store_true",
        help="""Run in noop mode - only verify that the script can import libraries and do initial pydantic config validation, and then exit after transmitting status.""",
    )
    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 
37
38
39
40
41
42
43
44
45
46
47
48
49
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 
102
103
104
105
106
107
108
109
110
111
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,
        allow_abbrev=False,
    )
    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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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.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()
        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 
117
118
119
120
121
122
123
124
125
126
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,
        allow_abbrev=False,
    )
    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 
28
29
30
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 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
82
83
84
85
86
87
88
89
90
91
def run_restart(rb: RealizationBuilder) -> None:
    """
    Run the provided checkpoint restart realization.
    """

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

run_tests.py#

Called by run_test_formulations.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_test_formulations.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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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 
140
141
142
143
144
145
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 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
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,
        allow_abbrev=False,
    )
    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

run_regionalization_standalone.py#

Called by run_region_standalone.sh

run_regionalization_standalone.py --help

ngen_rte.run_regionalization_standalone #

Command-line executable to build and run a "regionalization" 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.RTERegionConfig. For settings that are not exposed by CLI arguments, see primarily consts.py.

See run_default.sh for example calls.

run_region #

run_region(rb: RealizationBuilder) -> NgenRunnerAsync

Run the provided regionalization realization. Realization should already be built (rb.build_region_realization() already called).

Source code in bin_mounted/ngen_rte/run_regionalization_standalone.py 
31
32
33
34
35
36
37
38
39
40
41
def run_region(rb: RealizationBuilder) -> NgenRunnerAsync:
    """Run the provided regionalization realization.
    Realization should already be built (rb.build_region_realization() already called).
    """
    LOG.info("Running regionalization 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_regionalization_standalone.py 
75
76
77
78
79
80
81
82
83
84
85
def cli_arg_parser() -> argparse.ArgumentParser:
    """Build and return the CLI argument parser"""
    parser = argparse.ArgumentParser(
        description="""Script for building and running regionalization realizations
using realtime forcing configurations or historical / retrospective forcing.
The CLI arguments mostly follow that of run_default.py.""",
        formatter_class=cli_args.HelpFormatter,
        allow_abbrev=False,
    )
    cli_args.add_args_for_script(parser, cli_args.Script.REGIONALIZATION)
    return parser

run_output_postprocess.py#

Called by run_output_postprocess.sh

run_output_postprocess.py --help

ngen_rte.run_output_postprocess #

Command-line executable to run the overall NetCDF production workflow from nwm-data-assimilation. Calls data-assimilation's overall_netcdf_workflow to convert ngen catchment output into NWM-formatted gridded NetCDF products.

run_output_mosaic.py#

Called by run_output_mosaic.sh

run_output_mosaic.py --help

ngen_rte.run_output_mosaic #

Command-line executable to run the output mosaicing workflow from nwm-data-assimilation. Calls data-assimilation's combine_basin_grids to merge per-timestep NetCDF gridded outputs into a single NetCDF product.

make_coastal_forcing.py#

Example calls are provided by make_coastal_forcing.sh. See the nwm-coastal repository for complete coastal workflow documentation.

make_coastal_forcing.py --help

ngen_rte.coastal.make_coastal_forcing #

CLI script to generate gridded atmospheric forcing files for coastal models.

NOTE: This includes a T0 output needed for coastal, in addition to the normal T1+ outputs. NOTE: The T0 output has identical forcing values as T1. See comments in ngen-forcing model.py function write_t0_output. An independent T0 state cannot be reliably computed because several forecast products omit required fields at hour zero and substitute hour one. Examples: https://github.com/NGWPC/ngen-forcing/blob/27e03ba138478dd449ce957b1c3ba4c36fc33d8f/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/time_handling.py#L1202-L1206 https://github.com/NGWPC/ngen-forcing/blob/27e03ba138478dd449ce957b1c3ba4c36fc33d8f/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/time_handling.py#L1444-L1448 https://github.com/NGWPC/ngen-forcing/blob/27e03ba138478dd449ce957b1c3ba4c36fc33d8f/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/time_handling.py#L2043-L2047 https://github.com/NGWPC/ngen-forcing/blob/27e03ba138478dd449ce957b1c3ba4c36fc33d8f/NextGen_Forcings_Engine_BMI/NextGen_Forcings_Engine/core/time_handling.py#L4129-L4135

NOTE: See repository nwm-coastal (file nwm-coastal/scripts/setup_data_coastal.sh) for the scripts needed to download / set up input data required to run coastal forcing workflows:

These files are masked versions of the full CONUS nc files, masked to an area around the test gage for testing CONUS workflows
on a machine that may not have enough RAM for an actual full CONUS run:
    run_ngen/data/esmf_mesh/NWM/domain/GEOGRID_LDASOUT_Spatial_Metadata_CONUS_debug_gauge_01123000.nc
    run_ngen/data/esmf_mesh/NWM/domain/geo_em_CONUS_debug_gauge_01123000.nc

Usage: Choose exactly one timing mode:

* ``--cycle_datetime`` (``-dt``) runs an operational forecast cycle. The
    forcing engine determines the time window from the selected configuration.
* ``--start_time`` (``-start``) together with ``--end_time`` (``-end``) runs
    an explic`it time window, such as an AORC retrospective run.

Examples: python -um ngen_rte.coastal.make_coastal_forcing -dt "2025-09-15 00:00:00" -fconfig standard_ana_hawaii -gdomain Hawaii python -um ngen_rte.coastal.make_coastal_forcing -start "2013-07-25 00:00:00" -end "2013-07-26 00:00:00" -fconfig aorc -gdomain Hawaii

make_coastal_forcing #

make_coastal_forcing(
    forcing_configuration: str,
    global_domain: str,
    run_name: str = DEFAULT_RUN_NAME,
    cycle_datetime: datetime | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    lookback: int | None = None,
    forecast_input_horizons: int | None = None,
    debug_conus_subset: bool = False,
) -> Path

Create one gridded coastal forcing file and return its path.

Source code in bin_mounted/ngen_rte/coastal/make_coastal_forcing.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def make_coastal_forcing(
    forcing_configuration: str,
    global_domain: str,
    run_name: str = consts.DEFAULT_RUN_NAME,
    cycle_datetime: datetime | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    lookback: int | None = None,
    forecast_input_horizons: int | None = None,
    debug_conus_subset: bool = False,
) -> Path:
    """Create one gridded coastal forcing file and return its path."""
    if cycle_datetime is not None and (start_time is not None or end_time is not None):
        raise ValueError("cycle_datetime cannot be combined with start_time/end_time")
    if cycle_datetime is None and (start_time is None or end_time is None):
        raise ValueError("Provide cycle_datetime or both start_time and end_time")
    if (
        lookback is not None or forecast_input_horizons is not None
    ) and cycle_datetime is None:
        raise ValueError("lookback and forecast_input_horizons require cycle_datetime")

    config_manager = CoastalConfigManager(
        forcing_configuration,
        global_domain,
        cycle_datetime=cycle_datetime,
        start_time=start_time,
        lookback=lookback,
        forecast_input_horizons=forecast_input_horizons,
        forcing_root_dir=rte_consts.FORCING_ROOT_DIR,
        forcing_template_dir=rte_consts.FORCING_TEMPLATE_DIR,
        geogrid_dir=consts.GEOGRID_DIR,
        regrid_weights_dir=consts.REGRID_WEIGHTS_DIR,
        spatial_metadata_suffix=consts.SPATIAL_META_SUFFIX,
        debug_subset_name=consts.DEBUG_CONUS_SUBSET,
        debug_conus_subset=debug_conus_subset,
    )
    output_path = config_manager.output_path
    partial_output_path = output_path.with_suffix(f"{output_path.suffix}.partial")
    config_path = config_manager.write_config()
    num_ana_updates = analysis_num_updates(config_path)
    is_ana = num_ana_updates is not None

    LOG.info(
        "Running coastal forcing %s: configuration=%s domain=%s output=%s",
        run_name,
        forcing_configuration,
        global_domain,
        output_path,
    )
    run_arguments = {
        "start_time": (
            start_time.strftime("%Y-%m-%d %H:%M:%S") if start_time else None
        ),
        "end_time": end_time.strftime("%Y-%m-%d %H:%M:%S") if end_time else None,
        "config_path": config_path,
        "cycle_datetime": cycle_datetime,
        "geogrid": str(config_manager.geogrid),
        "output_path": partial_output_path,
        "output_t0": not is_ana,
    }
    if is_ana:
        run_arguments["num_updates"] = num_ana_updates
    run_bmi(**run_arguments)
    partial_output_path.replace(output_path)

    final_msg = f"Coastal forcing written to {output_path}"
    print(final_msg)
    LOG.info(final_msg)
    return output_path

Python Constants#

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

ngen_rte.consts #

Constants

FORCING_TEMPLATE_DIR module-attribute #

FORCING_TEMPLATE_DIR = f"{getsitepackages()[0]}/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

edfs_url

Base URL for the Icefabric EDFS API used to retrieve hydrofabric geopackage

TYPE: str | None = Field(default=None)

environment

Operating environment used to set edfs_url, e.g. 'test' or 'oe'. Overriden if specific edfs_url is provided.

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)

reservoir_rfc_dir

Optional path to reservoir RFC forecast time-series directory. If provided, reservoir_da is set to True and passed to DataAssimilationConfig

TYPE: str | None = Field(default=None)

usgs_timeslice_dir

Optional path to USGS timeslice time-series directory. If provided, streamflow_da is set to True and passed to DataAssimilationConfig

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 /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_dir

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 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
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.
    edfs_url: str | None = Field(default=None)
        Base URL for the Icefabric EDFS API used to retrieve hydrofabric geopackage
    environment: str
        Operating environment used to set edfs_url, e.g. 'test' or 'oe'. Overriden if specific edfs_url is provided.
    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.
    reservoir_rfc_dir: str | None = Field(default=None)
        Optional path to reservoir RFC forecast time-series directory. If provided, reservoir_da is set to True and passed to DataAssimilationConfig
    usgs_timeslice_dir: str | None = Field(default=None)
        Optional path to USGS timeslice time-series directory. If provided, streamflow_da is set to True and passed to DataAssimilationConfig
    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_dir: 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
    edfs_url: str | None = Field(default=None)
    environment: Literal["test", "oe"] | None = Field(default=None)
    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)
    reservoir_rfc_dir: str | None = Field(default=None)
    usgs_timeslice_dir: 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."""

    # 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."""

    # For ecFlow integration
    ecf_task: str | None = None
    """ECFlow task path, e.g. '/nwm/hourly/nwm_analysis_assim/jnwm_conus_analysis_assim'."""
    ecf_subtask: str | None = None
    """ECFlow subtask base identifier. Example: 'no_subtask_type__20260615_120000_000000__vpu__03S'."""
    ecf_iface: EcflowInterface | None = Field(init=False, default=None)
    """EcflowInterface instance created if ecf_task and ecf_subtask are provided."""
    ecf_ctx: SubtaskCallbackContext | None = Field(init=False, default=None)
    """SubtaskCallbackContext instance created if ecf_task and ecf_subtask are provided."""

    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

        # Resolve edfs_url, an explicit edfs_url value wins over environment
        # environment maps to the corresponding base URL set in consts.py, otherwise use default
        if not self.edfs_url:
            if self.environment == "oe":
                self.edfs_url = c.OE_EDFS_URL
            elif self.environment == "test":
                self.edfs_url = c.TEST_EDFS_URL
            elif self.environment is None:
                self.edfs_url = c.DEFAULT_EDFS_URL
            else:
                raise ValueError(
                    f"Invalid value for variable related to EDFS server, self.environment: {self.environment!r}. When EDFS URL not provided explicitly, then we expect self.environment to be in ('oe', 'test', None), where None triggers a default URL."
                )

        # 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"

        self._parse_hindcast_args()
        self._ecflow_connect()

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

    def _ecflow_connect(self):
        # Initialize ecFlow interface if task and subtask are provided
        if self.ecf_task or self.ecf_subtask:
            if not ECFLOW_AVAILABLE:
                raise RuntimeError(
                    "ecf_task_mgr is required when --ecf-task and --ecf-subtask are provided."
                )

            LOG.info("ecFlow task metadata provided, connecting to ecFlow server...")
            if not self.ecf_task and self.ecf_subtask:
                raise ValueError(
                    "When ecf_task or ecf_subtask are provided, both must be provided."
                )
            conn = EcflowConnection(settings_path=c.ECFLOW_SETTINGS)
            self.ecf_iface = EcflowInterface(conn)
            LOG.info("Connected to ecFlow server.")
            self.ecf_ctx = SubtaskCallbackContext(
                task_path=self.ecf_task,
                subtask_var_base=self.ecf_subtask,
            )
            LOG.info(
                f"Checking that ecFlow task, subtask exists: {self.ecf_task}, {self.ecf_subtask}"
            )
            self.ecf_iface.get_node(self.ecf_ctx.task)
            if not self.ecf_iface.var_exists(
                self.ecf_ctx.task, self.ecf_ctx.var_status
            ):
                self.errors.append(
                    RuntimeError(
                        f"ecFlow var {self.ecf_ctx.task}, {self.ecf_ctx.var_status} not found."
                    )
                )
            if not self.ecf_iface.var_exists(self.ecf_ctx.task, self.ecf_ctx.var_info):
                self.errors.append(
                    RuntimeError(
                        f"ecFlow var {self.ecf_ctx.task}, {self.ecf_ctx.var_info} not found."
                    )
                )
        else:
            LOG.info("ecFlow task *not* provided, *not* connecting to ecFlow server...")
            self.ecf_iface = None
            self.ecf_ctx = None

    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",
            "forecast",
            "cold_start",
            "hindcast",
            "warm_start",
            "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)"
                )
            )

    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."""
        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,
            edfs_url=self.edfs_url,
            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, RTEDefaultConfig, RTERegionConfig))
            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_DataAssimilationConfig(self) -> DataAssimilationConfig:
        """MSWM DataAssimilationConfig instance"""
        dac = DataAssimilationConfig(
            reservoir_da=bool(self.reservoir_rfc_dir),
            reservoir_rfc_dir=self.reservoir_rfc_dir,
            streamflow_da=bool(self.usgs_timeslice_dir),
            usgs_timeslice_dir=self.usgs_timeslice_dir,
        )
        return dac

    @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
        data_assimilation = self.mswm_DataAssimilationConfig
        parallel = self.mswm_ParallelConfig
        ic = InputConfig(
            General=general,
            ModuleProperties=module_properties,
            NWMOutput=nwm_output,
            Regionalization=regionalization,
            Calibration=calibration,
            Forcing=forcing,
            DataFile=data_file,
            DataAssimilation=data_assimilation,
            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_hindcast": self.use_hindcast,
            "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.

hindcast_args class-attribute instance-attribute #

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 class-attribute instance-attribute #

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

Boolean indicating that hindcast is to be used. Passed to MSWM.

hc_cycle_interval class-attribute instance-attribute #

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

Hindcast cycle interval.

hc_num_iterations class-attribute instance-attribute #

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

Hindcast number of iterations.

ecf_task class-attribute instance-attribute #

ecf_task: str | None = None

ECFlow task path, e.g. '/nwm/hourly/nwm_analysis_assim/jnwm_conus_analysis_assim'.

ecf_subtask class-attribute instance-attribute #

ecf_subtask: str | None = None

ECFlow subtask base identifier. Example: 'no_subtask_type__20260615_120000_000000__vpu__03S'.

ecf_iface class-attribute instance-attribute #

ecf_iface: EcflowInterface | None = Field(
    init=False, default=None
)

EcflowInterface instance created if ecf_task and ecf_subtask are provided.

ecf_ctx class-attribute instance-attribute #

ecf_ctx: SubtaskCallbackContext | None = Field(
    init=False, default=None
)

SubtaskCallbackContext instance created if ecf_task and ecf_subtask are provided.

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_DataAssimilationConfig property #

mswm_DataAssimilationConfig: DataAssimilationConfig

MSWM DataAssimilationConfig 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 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
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",
        "forecast",
        "cold_start",
        "hindcast",
        "warm_start",
        "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)

cold_start_datetime

Start time of the coldstart realization. If None, coldstart is not performed.

TYPE: datetime | None

noop

Causes a noop to occur (for confirming that Python packages are importable).

TYPE: bool

Source code in bin_mounted/ngen_rte/configs.py 
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
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.
    cold_start_datetime: datetime | None
        Start time of the coldstart realization. If None, coldstart is not performed.
    noop: bool
        Causes a noop to occur (for confirming that Python packages are importable).
    """

    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)
    cold_start_datetime: datetime | None
    noop: bool = False

    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_regionalization_standalone.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

cold_start_datetime

Start time of the coldstart realization. If None, coldstart is not performed.

TYPE: datetime | None

Source code in bin_mounted/ngen_rte/configs.py 
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
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_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
        File containing catchment groupings for regionalization
    cold_start_datetime: datetime | None
        Start time of the coldstart realization. If None, coldstart is not performed.
    """

    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
    cold_start_datetime: datetime | None

    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 
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
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 
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
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 
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
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
    edfs_url: 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 
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
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 
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
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 
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
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.