Skip to article frontmatterSkip to article content
Site not loading correctly?

This may be due to an incorrect BASE_URL configuration. See the MyST Documentation for reference.

RomsMarblInputData Class Documentation

This subsystem is driven by the forge application (cstar forge run, or the python -m cstar_forge.run module CLI it wraps), which calls ForgeExecutor.generate_inputs() (cstar_forge/forge/executor.py), which constructs a RomsMarblInputData and calls generate_all() on it. Direct construction, as shown in this document, is for developers debugging or extending input generation.

Overview

RomsMarblInputData is a dataclass (subclass of InputData, both defined in cstar_forge/forge/input_data.py) that implements ROMS-MARBL specific input data generation. It handles the creation of all input files required for a ROMS simulation, including grid, initial conditions, and all types of forcing data.

Class Definition

@dataclass
class RomsMarblInputData(InputData):
    """ROMS-MARBL specific input data generation."""

    # Inherited from InputData: domain_name, start_date, end_date, input_data_dir (kw_only)

    grid: rt.Grid
    boundaries: OpenBoundaries
    source_data: source_data.SourceData
    roms_marbl_blueprint_dir: Path
    partitioning: cstar_models.PartitioningParameterSet
    cdr_forcing: dict | None = None
    forcing_override: dict[str, Any] | None = None
    model_reference_date: datetime | None = None
    grid_parent: rt.Grid | None = None
    grid_child: rt.Grid | None = None
    metadata_child: dict[str, Any] | None = None
    settings_compile_time: dict[str, Any] | None = None  # executor-owned, bound by reference
    settings_run_time: dict[str, Any] | None = None  # executor-owned, bound by reference
    use_dask: bool = True
    dask_num_workers: int = 8
    use_pio: bool = False
    subchunk: bool = True
    verbose: bool = False
    has_bgc: bool = False  # mirrors ForgeExecutor._has_bgc (cppdefs.marbl)

    roms_marbl_blueprint_elements: RomsMarblBlueprintInputData  # Auto-initialized
    _settings_compile_time: dict  # bound to `settings_compile_time`, or {} if not given
    _settings_run_time: dict  # bound to `settings_run_time`, or {} if not given
    include_coarse_dims: bool | None = None  # Set during surface forcing generation

There is no model_spec field — the class is host-/model-spec-independent. forcing_override (injected by the caller, typically built by sources_to_forcing_override() in forge_blueprint_engine.py from a ForgeBlueprint) is what drives which inputs get generated; grid is always generated from the injected grid object regardless of forcing_override. See cstar_forge/forge/input_data.py for the full field list, including private bookkeeping fields (_subchunk_refs, _clobber, _existing_planned_outputs, _planned_output_paths) not listed above.

Initialization

Input List Derivation

During __post_init__(), the class builds input_list from forcing_override (not a model spec):

  1. Grid: Always appended as ("grid", {}) — the grid handler ignores kwargs and uses the injected grid (and, if present, grid_child/metadata_child) object directly.

  2. Initial Conditions: forcing_override["initial_conditions"], if present → ("initial_conditions", kwargs)

  3. Forcing: Iterates over forcing_override["forcing"] categories (surface, boundary, tidal, river) and items within each → ("forcing.{category}", kwargs) for each item

  4. CDR Forcing: If the cdr_forcing constructor kwarg is set → ("cdr_forcing", {"cdr_kwargs": self.cdr_forcing})

A missing forcing_override raises ValueError — it is required whenever the blueprint path is used (the resolver always fills it, from the model default or an authored selection).

Example Input List:

[
    ("grid", {}),
    ("initial_conditions", {"source": {"name": "GLORYS"}, "bgc_source": {...}}),
    ("forcing.surface", {"source": {"name": "ERA5"}, "type": "physics", ...}),
    ("forcing.surface", {"source": {"name": "UNIFIED"}, "type": "bgc", ...}),
    ("forcing.boundary", {"source": {"name": "GLORYS"}, "type": "physics", ...}),
    ("forcing.tidal", {"source": {"name": "TPXO"}, "ntides": 15}),
    ("forcing.river", {"source": {"name": "DAI"}, "include_bgc": True}),
]

Registry Validation

The class validates that all keys in input_list have registered handlers in INPUT_REGISTRY. Missing handlers raise a ValueError.

ROMS-MARBL Blueprint Elements Initialization

Creates RomsMarblBlueprintInputData instance with empty datasets:

Validation:

Settings Initialization

_settings_compile_time/_settings_run_time are bound directly to the settings_compile_time/ settings_run_time constructor args (no copy) -- in the normal ForgeExecutor path these ARE the executor’s own live settings dicts, so generation steps mutate the executor’s dicts in place and there is no merge-back step. When either arg is omitted (standalone/test use), a fresh empty dict {} is created instead.

Registry Framework

Input Registry

The INPUT_REGISTRY dictionary maps input keys to InputStep instances:

INPUT_REGISTRY: Dict[str, InputStep] = {
    "grid": InputStep(name="grid", order=10, label="Writing ROMS grid", handler=_generate_grid),
    "initial_conditions": InputStep(name="initial_conditions", order=20, label="Generating initial conditions", handler=_generate_initial_conditions),
    "forcing.surface": InputStep(name="forcing.surface", order=30, label="Generating surface forcing", handler=_generate_surface_forcing),
    "forcing.boundary": InputStep(name="forcing.boundary", order=40, label="Generating boundary forcing", handler=_generate_boundary_forcing),
    "forcing.tidal": InputStep(name="forcing.tidal", order=50, label="Generating tidal forcing", handler=_generate_tidal_forcing),
    "forcing.river": InputStep(name="forcing.river", order=60, label="Generating river forcing", handler=_generate_river_forcing),
    "cdr_forcing": InputStep(name="cdr_forcing", order=80, label="Generating CDR forcing", handler=_generate_cdr_forcing),
    "forcing.corrections": InputStep(name="forcing.corrections", order=90, label="Generating corrections forcing", handler=_generate_corrections),
}

Registration Decorator

@register_input(name: str, order: int, label: str | None = None)

Parameters:

Example:

@register_input(name="forcing.surface", order=30, label="Generating surface forcing")
def _generate_surface_forcing(self, key: str = "forcing.surface", **kwargs):
    """Generate surface forcing input files."""
    # Implementation...

Input Generation Process

generate_all() Method

Main entry point for generating all input files:

def generate_all(
    self,
    clobber: bool = False,
    partition_files: bool = False,
    test: bool = False,
    only: set[str] | None = None,
) -> RomsMarblBlueprintInputData | None:
    """
    Generate all ROMS input files.

    Returns
    -------
    RomsMarblBlueprintInputData | None
        Blueprint subset with generated input file paths, or None if the input
        directory is non-empty and clobber is False. Settings are NOT returned:
        `_settings_compile_time`/`_settings_run_time` are the executor-owned dicts
        passed in via the `settings_compile_time`/`settings_run_time` constructor
        args and mutated in place by generation steps -- the caller already holds
        the up-to-date dicts through its own reference.
    """

only restricts generation to a subset of canonical INPUT_REGISTRY keys (see resolve_input_selection(), which maps user-facing aliases like "ic"/"bry"/"tides" onto them); the grid step always runs regardless, since every other step depends on the in-memory grid object.

Process:

  1. Clobber Check: With clobber=False, existing .nc files in input_data_dir are left in place (an informational count is printed) and reused per-step per the planned-output list; with clobber=True, all existing .nc files are deleted first (_ensure_empty_or_clobber())

  2. Build Step List: Creates list of (step, kwargs) tuples from input_list, sorted by order

  3. Plan Outputs: Computes the planned NetCDF outputs for the run up front (_planned_netcdf_outputs) and records which already exist on disk, so each step can decide whether to reuse an existing file instead of regenerating it

  4. Dask/Thread Guards: When use_dask is True, caps dask’s worker count (dask_num_workers) and pins BLAS/OpenMP to 1 thread for the duration of the loop, to avoid thread oversubscription on high-core HPC nodes

  5. Execute Handlers: For each step (skipping boundary forcing when all open boundaries are disabled, and skipping any step not in only when only is given), calls the handler with key and kwargs

  6. Partitioning: Optionally partitions files across tiles if partition_files=True

  7. Return: Returns roms_marbl_blueprint_elements (settings dicts are mutated in place, not returned -- see generate_all() above)

Handler Function Signature

All registered handlers follow this pattern:

@register_input(name="input_key", order=ORDER, label="Label")
def _generate_input(self, key: str = "input_key", **kwargs):
    """
    Generate input file(s) for this input type.
    
    Parameters
    ----------
    key : str
        Input key (matches registered name)
    **kwargs
        Input-specific arguments from input_list
        
    Side Effects
    ------------
    - Creates NetCDF file(s) in input_data_dir
    - Creates YAML metadata file in roms_marbl_blueprint_dir
    - Appends Resource(s) to roms_marbl_blueprint_elements
    - Updates _settings_compile_time and/or _settings_run_time
    """

Registered Input Handlers

Grid (grid, order=10)

Handler: _generate_grid()

Generates:

Updates ROMS-MARBL Blueprint:

Populates Settings:

Initial Conditions (initial_conditions, order=20)

Handler: _generate_initial_conditions()

Generates:

Source Resolution:

Updates ROMS-MARBL Blueprint:

Populates Settings:

Surface Forcing (forcing.surface, order=30)

Handler: _generate_surface_forcing()

Generates:

Key Features:

Source Resolution:

Updates ROMS-MARBL Blueprint:

Populates Settings:

Boundary Forcing (forcing.boundary, order=40)

Handler: _generate_boundary_forcing()

Generates:

Key Features:

Source Resolution:

Updates ROMS-MARBL Blueprint:

Populates Settings:

Note: Compile-time settings are not populated by the boundary handler.

Tidal Forcing (forcing.tidal, order=50)

Handler: _generate_tidal_forcing()

Generates:

Key Features:

Source Resolution:

Updates ROMS-MARBL Blueprint:

Populates Settings:

River Forcing (forcing.river, order=60)

Handler: _generate_river_forcing()

Generates:

Key Features:

Source Resolution:

Updates ROMS-MARBL Blueprint:

Populates Settings: note this is run-time, not compile-time, despite the river_frc section name

CDR Forcing (cdr_forcing, order=80)

Handler: _generate_cdr_forcing()

Generates:

Key Features:

Updates ROMS-MARBL Blueprint:

Populates Settings:

Corrections Forcing (forcing.corrections, order=90)

Handler: _generate_corrections()

Status: Registered but unwired — the resolver never emits a corrections category, so the step never enters input_list in production; the handler raises NotImplementedError

Source Resolution

_resolve_source_block() Method

Normalizes source blocks and injects file paths:

def _resolve_source_block(
    self,
    block: str | dict[str, Any],
    time_window: tuple[datetime, datetime] | None = None,
) -> dict[str, Any]:
    """
    Normalize a "source"/"bgc_source" block and inject a 'path' based on SourceData.

    Parameters
    ----------
    block : str or dict
        Source specification (e.g., "GLORYS" or {"name": "GLORYS", "climatology": True})
    time_window : tuple[datetime, datetime], optional
        When given and the resolved path is a per-day file list, trims it to the files
        covering that window (e.g. initial conditions only need `ini_time`'s day).

    Returns
    -------
    dict
        Source block with 'name' and optional 'path' fields
    """

Process:

  1. Normalize to dict: If string, convert to {"name": str}

  2. Extract name: Get name field from dict (raises if a dict block has no name)

  3. Check streamability: SourceData.streamable_for_source(name, glorys_layout=...) — if streamable (e.g. ERA5), don’t add a path unless one was explicitly provided

  4. Get path: SourceData.path_for_source(name, glorys_layout=...) for non-streamable sources

  5. Either time-window trim (when time_window is given and the path is a multi-file list, via filter_paths_by_time_window()) or subchunking (when subchunk is on and the source is a multi-file GLORYS path: the path is replaced with a memoized kerchunk-subchunked reference, _subchunked_glorys_path()) — never both. A trimmed list deliberately skips the subchunk branch so the memoized reference is only ever built from the full file list

  6. Return: Dict with name and optional path

SourceData.dataset_key_for_source() is used elsewhere (subchunk-reference memoization), not inside this method.

Examples:

# String input
"GLORYS" → {"name": "GLORYS", "path": Path("/path/to/GLORYS_REGIONAL_file.nc")}

# Dict input
{"name": "UNIFIED", "climatology": True} → {"name": "UNIFIED", "climatology": True, "path": Path("/path/to/UNIFIED_BGC_file.nc")}

# Streamable source
"ERA5" → {"name": "ERA5"}  # No path (streamable)

_build_input_args() Method

Merges default arguments with runtime overrides:

def _build_input_args(
    self,
    key: str,
    extra: dict[str, Any] | None = None,
    base_kwargs: dict[str, Any] | None = None,
    time_window: tuple[datetime, datetime] | None = None,
) -> dict[str, Any]:
    """
    Merge per-input defaults with runtime arguments.

    Uses base_kwargs (always provided from input_list).
    Resolves "source" and "bgc_source" through SourceData.
    Merges with extra, where extra overrides defaults.
    """

Process:

  1. Get base config: base_kwargs, always supplied from an input_list entry (there is no model-spec fallback)

  2. Resolve source blocks: Convert source and bgc_source Pydantic models to dicts with paths via _resolve_source_block()

  3. Unpack options: an optional options passthrough dict in the item config is popped out and forwarded verbatim to the roms-tools constructor

  4. Merge: cfg (base kwargs) < item_options < extra (extra always wins — it carries runtime injections like dates)

  5. If subchunking swapped a source path for a kerchunk reference and the merged args don’t already specify chunks, sets chunks={} so xarray/dask honors the reference’s native layout

Settings Population for Forcing

Compile-Time Settings

Surface Forcing:

CDR Forcing:

Boundary Forcing:

Tidal/River Forcing:

Run-Time Settings

Surface Forcing:

Boundary Forcing:

Tidal Forcing:

River Forcing:

CDR Forcing:

ROMS-MARBL Blueprint Element Updates

Each handler appends Resource objects to the appropriate roms_marbl_blueprint element:

Resource Creation:

resource = Resource(              # from cstar.orchestration.models
    location=str(out_path),       # path to generated NetCDF file
    partitioned=False,            # set to True after partitioning
)

Blueprint Updates:

File Partitioning

_partition_files() Method

Partitions whole-field input files across tiles:

def _partition_files(self, **kwargs):
    """
    Partition whole input files across tiles using roms_tools.partition_netcdf.
    
    Uses the paths stored in roms_marbl_blueprint_elements to build the list of whole-field files,
    and records the partitioned paths in the Resource objects.
    """

Process:

  1. Iterate over input_list: For each input key, get corresponding dataset from roms_marbl_blueprint_elements

  2. Partition each Resource: Call rt.partition_netcdf() for each Resource.location

  3. Create partitioned Resources: Replace original resources with partitioned ones

  4. Update partitioned flag: Set partitioned=True on new resources

Partitioning Arguments:

input_args = dict(
    np_eta=self.partitioning.n_procs_y,
    np_xi=self.partitioning.n_procs_x,
    output_dir=self.input_data_dir,
    include_coarse_dims=self.include_coarse_dims,  # set during surface forcing generation
)

Result:

Return Value

generate_all() returns just roms_marbl_blueprint_elements: RomsMarblBlueprintInputData (or None; see generate_all() above). The settings dicts are not returned -- they are the executor-owned _settings_compile_time/_settings_run_time, mutated in place by generation:

Usage:

ForgeExecutor.generate_inputs() (cstar_forge/forge/executor.py) passes its own self._settings_compile_time/self._settings_run_time in by reference, so it already holds the up-to-date settings after generate_all() returns -- no merge-back step.

Developer Guide
Input Data Generation Overview
Developer Guide
Model Specification