Runs

A Catchment is a builder: its inputs are X | None until the matching read_* call has run, and that is honest. The engines need the opposite — a catchment that is finished. DistributedRun and LumpedRun are that finished form.

from_model is the single validation seam. Constructing a run is the validation: it resolves every optional input, checks the drivers, the parameter cube, the river geometry and the flow-path-length raster against the catchment grid, and refuses a combination the engines cannot run. Every engine entry point takes one of these types, so the checks are enforced by the signatures rather than by remembering to call them — which is how Calibration, going straight to Wrapper, used to skip all of them.

run = DistributedRun.from_model(model)          # checked here, once
results = Wrapper.run_muskingum(run)            # nothing left to re-check

Both are frozen. The checks happen at construction, so a mutable run would let a caller swap an input in afterwards and reach an engine with something never validated.

DistributedRun

hapi.runs.DistributedRun dataclass

Everything a distributed run needs, checked and non-optional.

Attributes:

Name	Type	Description
period	SimulationPeriod	The span the run covers, and the calendar and factors it implies.
meteo	MeteoInputs	The three driver cubes.
flow_network	FlowNetwork	The routing network and the grid it defines.
parameters	ParameterSet	The parameter array plus the (snow, maxbas) pair fixing its width.
model_setup	ConceptualModelSetup	The conceptual model instance and the state it starts from.
river_geometry	RiverGeometry | None	The five hydraulic rasters, when the flood path supplied them. None on an ordinary distributed run, which never reads them. Absent-or-complete, never half-filled -- that is what :class:~hapi.inputs.RiverGeometry guarantees.
skip_hydraulic_cells	bool	Leave river cells unrouted for a 1D hydraulic model. Needs river_geometry to identify them, checked here rather than in the routing loop.
flow_path_length	ndarray | None	Flow-path length raster, read only by :meth:~hapi.rrm.distrrm.DistributedRRM.route_maxbas_by_path_length.
keep_state_variables	bool	Whether to allocate the per-cell state array. It is (rows, cols, time, 5) -- as much memory as every other result field combined -- and nothing but results.save and results.animate reads it, so a run that will not look at it can halve its peak allocation. Defaults to True, which is what every existing caller got; Calibration turns it off, because it runs the model once per trial vector and never reads the states.

Source code in src/hapi/runs.py

@dataclass(frozen=True)
class DistributedRun:
    """Everything a distributed run needs, checked and non-optional.

    Attributes:
        period: The span the run covers, and the calendar and factors it implies.
        meteo: The three driver cubes.
        flow_network: The routing network and the grid it defines.
        parameters: The parameter array plus the `(snow, maxbas)` pair fixing its width.
        model_setup: The conceptual model instance and the state it starts from.
        river_geometry: The five hydraulic rasters, when the flood path supplied them. `None`
            on an ordinary distributed run, which never reads them. Absent-or-complete, never
            half-filled -- that is what :class:`~hapi.inputs.RiverGeometry` guarantees.
        skip_hydraulic_cells: Leave river cells unrouted for a 1D hydraulic model. Needs
            `river_geometry` to identify them, checked here rather than in the routing loop.
        flow_path_length: Flow-path length raster, read only by
            :meth:`~hapi.rrm.distrrm.DistributedRRM.route_maxbas_by_path_length`.
        keep_state_variables: Whether to allocate the per-cell state array. It is
            `(rows, cols, time, 5)` -- as much memory as every other result field combined --
            and nothing but `results.save` and `results.animate` reads it, so a run
            that will not look at it can halve its peak allocation. Defaults to True, which is
            what every existing caller got; `Calibration` turns it off, because it runs the
            model once per trial vector and never reads the states.
    """

    period: SimulationPeriod
    meteo: MeteoInputs
    flow_network: FlowNetwork
    parameters: ParameterSet
    model_setup: ConceptualModelSetup
    river_geometry: RiverGeometry | None = None
    skip_hydraulic_cells: bool = False
    flow_path_length: np.ndarray | None = None
    keep_state_variables: bool = True

    def __post_init__(self):
        """Check the inputs agree with each other and with the grid.

        Raises:
            ValueError: The drivers, the parameters, the river geometry or the flow-path-length
                raster do not cover the grid, or a cell skip was asked for with no geometry to
                identify the river cells.
        """
        rows, cols = self.flow_network.rows, self.flow_network.cols

        # The three cubes already agree with each other (settled when MeteoInputs was built);
        # this is the other half -- that they cover the grid, and the period.
        self.meteo.validate_against(rows, cols, self.period.date_index)

        shape = np.asarray(self.parameters.values).shape
        if shape[0] != rows:
            raise ValueError(ROWS_MISMATCH_ERROR)
        if shape[1] != cols:
            raise ValueError(COLS_MISMATCH_ERROR)

        if self.river_geometry is not None and not self.river_geometry.covers(
            rows, cols
        ):
            raise ValueError(GRID_MISMATCH_ERROR)

        # `route_maxbas_by_path_length` indexes this by the flow network's rows and cols,
        # so a raster on a different grid either raises deep inside that loop or -- if it is
        # larger -- quietly reads the wrong cells. It was carried in with a bare `getattr`
        # and was the only input this seam did not check.
        if self.flow_path_length is not None:
            shape = np.shape(self.flow_path_length)
            if shape != (rows, cols):
                raise ValueError(
                    f"the flow-path-length raster is {shape} but the catchment grid is "
                    f"({rows}, {cols}); read it from a raster aligned to the "
                    f"flow-accumulation grid"
                )

        if self.skip_hydraulic_cells and self.river_geometry is None:
            raise ValueError(
                "skipping the hydraulic cells needs the river geometry to identify them, "
                "but none is set; call read_river_geometry first"
            )

    @property
    def parameter_cube(self) -> np.ndarray:
        """np.ndarray: The parameters as the `(rows, cols, n)` cube a distributed run indexes.

        `ParameterSet.values` is a flat sequence for a lumped run and a cube for a distributed
        one, so it is typed as either. On this side of the narrowing it is always the cube --
        `__post_init__` has already read its first two axes -- and saying so here means the
        engines index a real array instead of a union.
        """
        return np.asarray(self.parameters.values)

    @property
    def routing_table(self) -> dict:
        """dict: The flow-direction table, mapping `"row,col"` to the cells draining into it.

        `FlowNetwork` carries it as optional because the MAXBAS paths never route cell to cell.
        Reaching it through here keeps that honest while giving the Muskingum routing a plain
        dict to index.

        Raises:
            ValueError: The network was built without a direction table.
        """
        table = self.flow_network.FDT
        if table is None:
            raise ValueError(
                "cell-to-cell routing needs the flow-direction table, but the flow network "
                "was built without one; pass a flow-direction raster to "
                "FlowNetwork.from_rasters"
            )
        return table

    @classmethod
    def from_model(
        cls,
        model: CatchmentLike,
        *,
        needs_flow_direction: bool = True,
        with_river_geometry: bool = False,
        skip_hydraulic_cells: bool = False,
        keep_state_variables: bool = True,
    ) -> DistributedRun:
        """Narrow a built catchment into a validated distributed run.

        The single seam every distributed execution path passes through, `Run` and
        `Calibration` alike. Everything checkable is checked here, so a caller cannot arrive at
        the engines with an unvalidated model and does not have to remember to ask.

        Args:
            model: A catchment with its inputs read.
            needs_flow_direction: Whether the flow-direction raster is required. Cell-to-cell
                routing needs it; MAXBAS sends every cell straight to the outlet and never
                reads it.
            with_river_geometry: Carry the river geometry through, for the flood path.
            skip_hydraulic_cells: Leave the river cells to a hydraulic model.
            keep_state_variables: Allocate the per-cell state array. False halves the run's
                peak memory at the cost of `results.save` / `results.animate`
                options 4 to 8.

        Returns:
            DistributedRun: The validated inputs.

        Raises:
            ValueError: A required input is unset, or the inputs disagree.
        """
        flow_network = _require(
            model,
            "flow_network",
            "assign FlowNetwork.from_rasters(...) to model.flow_network",
        )
        if needs_flow_direction:
            if flow_network.flow_dir_arr is None:
                raise ValueError(
                    "this run routes cell to cell and needs a flow-direction raster, but the "
                    "flow network was built without one; pass it to FlowNetwork.from_rasters"
                )
            # No shape check here: `FlowNetwork` guards its own invariant at construction and
            # on replacement now, so the two rasters cannot disagree. This compensated for the
            # replacement gap, and became unreachable when that closed.
            if flow_network.FDT is None:
                raise ValueError(
                    "cell-to-cell routing needs the flow-direction table; the flow network "
                    "was built without one"
                )

        geometry = None
        if with_river_geometry or skip_hydraulic_cells:
            geometry = _require(
                model, "river_geometry", "call read_river_geometry first"
            )

        return cls(
            period=model.period,
            meteo=_require(
                model, "meteo", "assign MeteoInputs.from_rasters(...) to model.meteo"
            ),
            flow_network=flow_network,
            parameters=_require(model, "parameters", "call read_parameters first"),
            model_setup=_require(model, "model_setup", "call read_lumped_model first"),
            river_geometry=geometry,
            skip_hydraulic_cells=skip_hydraulic_cells,
            flow_path_length=getattr(model, "flow_path_length_arr", None),
            keep_state_variables=keep_state_variables,
        )

__post_init__()

Check the inputs agree with each other and with the grid.

Raises:

Type	Description
ValueError	The drivers, the parameters, the river geometry or the flow-path-length raster do not cover the grid, or a cell skip was asked for with no geometry to identify the river cells.

Source code in src/hapi/runs.py

def __post_init__(self):
    """Check the inputs agree with each other and with the grid.

    Raises:
        ValueError: The drivers, the parameters, the river geometry or the flow-path-length
            raster do not cover the grid, or a cell skip was asked for with no geometry to
            identify the river cells.
    """
    rows, cols = self.flow_network.rows, self.flow_network.cols

    # The three cubes already agree with each other (settled when MeteoInputs was built);
    # this is the other half -- that they cover the grid, and the period.
    self.meteo.validate_against(rows, cols, self.period.date_index)

    shape = np.asarray(self.parameters.values).shape
    if shape[0] != rows:
        raise ValueError(ROWS_MISMATCH_ERROR)
    if shape[1] != cols:
        raise ValueError(COLS_MISMATCH_ERROR)

    if self.river_geometry is not None and not self.river_geometry.covers(
        rows, cols
    ):
        raise ValueError(GRID_MISMATCH_ERROR)

    # `route_maxbas_by_path_length` indexes this by the flow network's rows and cols,
    # so a raster on a different grid either raises deep inside that loop or -- if it is
    # larger -- quietly reads the wrong cells. It was carried in with a bare `getattr`
    # and was the only input this seam did not check.
    if self.flow_path_length is not None:
        shape = np.shape(self.flow_path_length)
        if shape != (rows, cols):
            raise ValueError(
                f"the flow-path-length raster is {shape} but the catchment grid is "
                f"({rows}, {cols}); read it from a raster aligned to the "
                f"flow-accumulation grid"
            )

    if self.skip_hydraulic_cells and self.river_geometry is None:
        raise ValueError(
            "skipping the hydraulic cells needs the river geometry to identify them, "
            "but none is set; call read_river_geometry first"
        )

from_model(model: CatchmentLike, *, needs_flow_direction: bool = True, with_river_geometry: bool = False, skip_hydraulic_cells: bool = False, keep_state_variables: bool = True) -> DistributedRun classmethod

Narrow a built catchment into a validated distributed run.

The single seam every distributed execution path passes through, Run and Calibration alike. Everything checkable is checked here, so a caller cannot arrive at the engines with an unvalidated model and does not have to remember to ask.

Parameters:

Name	Type	Description	Default
model	CatchmentLike	A catchment with its inputs read.	required
needs_flow_direction	bool	Whether the flow-direction raster is required. Cell-to-cell routing needs it; MAXBAS sends every cell straight to the outlet and never reads it.	True
with_river_geometry	bool	Carry the river geometry through, for the flood path.	False
skip_hydraulic_cells	bool	Leave the river cells to a hydraulic model.	False
keep_state_variables	bool	Allocate the per-cell state array. False halves the run's peak memory at the cost of results.save / results.animate options 4 to 8.	True

Returns:

Type	Description
DistributedRun	The validated inputs.

Raises:

Type	Description
ValueError	A required input is unset, or the inputs disagree.

Source code in src/hapi/runs.py

@classmethod
def from_model(
    cls,
    model: CatchmentLike,
    *,
    needs_flow_direction: bool = True,
    with_river_geometry: bool = False,
    skip_hydraulic_cells: bool = False,
    keep_state_variables: bool = True,
) -> DistributedRun:
    """Narrow a built catchment into a validated distributed run.

    The single seam every distributed execution path passes through, `Run` and
    `Calibration` alike. Everything checkable is checked here, so a caller cannot arrive at
    the engines with an unvalidated model and does not have to remember to ask.

    Args:
        model: A catchment with its inputs read.
        needs_flow_direction: Whether the flow-direction raster is required. Cell-to-cell
            routing needs it; MAXBAS sends every cell straight to the outlet and never
            reads it.
        with_river_geometry: Carry the river geometry through, for the flood path.
        skip_hydraulic_cells: Leave the river cells to a hydraulic model.
        keep_state_variables: Allocate the per-cell state array. False halves the run's
            peak memory at the cost of `results.save` / `results.animate`
            options 4 to 8.

    Returns:
        DistributedRun: The validated inputs.

    Raises:
        ValueError: A required input is unset, or the inputs disagree.
    """
    flow_network = _require(
        model,
        "flow_network",
        "assign FlowNetwork.from_rasters(...) to model.flow_network",
    )
    if needs_flow_direction:
        if flow_network.flow_dir_arr is None:
            raise ValueError(
                "this run routes cell to cell and needs a flow-direction raster, but the "
                "flow network was built without one; pass it to FlowNetwork.from_rasters"
            )
        # No shape check here: `FlowNetwork` guards its own invariant at construction and
        # on replacement now, so the two rasters cannot disagree. This compensated for the
        # replacement gap, and became unreachable when that closed.
        if flow_network.FDT is None:
            raise ValueError(
                "cell-to-cell routing needs the flow-direction table; the flow network "
                "was built without one"
            )

    geometry = None
    if with_river_geometry or skip_hydraulic_cells:
        geometry = _require(
            model, "river_geometry", "call read_river_geometry first"
        )

    return cls(
        period=model.period,
        meteo=_require(
            model, "meteo", "assign MeteoInputs.from_rasters(...) to model.meteo"
        ),
        flow_network=flow_network,
        parameters=_require(model, "parameters", "call read_parameters first"),
        model_setup=_require(model, "model_setup", "call read_lumped_model first"),
        river_geometry=geometry,
        skip_hydraulic_cells=skip_hydraulic_cells,
        flow_path_length=getattr(model, "flow_path_length_arr", None),
        keep_state_variables=keep_state_variables,
    )

parameter_cube: np.ndarray property

np.ndarray: The parameters as the (rows, cols, n) cube a distributed run indexes.

ParameterSet.values is a flat sequence for a lumped run and a cube for a distributed one, so it is typed as either. On this side of the narrowing it is always the cube -- __post_init__ has already read its first two axes -- and saying so here means the engines index a real array instead of a union.

routing_table: dict property

dict: The flow-direction table, mapping "row,col" to the cells draining into it.

FlowNetwork carries it as optional because the MAXBAS paths never route cell to cell. Reaching it through here keeps that honest while giving the Muskingum routing a plain dict to index.

Raises:

Type	Description
ValueError	The network was built without a direction table.

LumpedRun

hapi.runs.LumpedRun dataclass

Everything a lumped run needs, checked and non-optional.

A lumped catchment has no grid, so it carries one column per driver rather than three cubes, and no flow network at all.

Attributes:

Name	Type	Description
period	SimulationPeriod	The span the run covers.
data	ndarray	(time, 4) array of precipitation, ET, temperature and the long-term average.
parameters	ParameterSet	The parameter vector plus the (snow, maxbas) pair fixing its width.
model_setup	ConceptualModelSetup	The conceptual model instance and the state it starts from.

Source code in src/hapi/runs.py

@dataclass(frozen=True)
class LumpedRun:
    """Everything a lumped run needs, checked and non-optional.

    A lumped catchment has no grid, so it carries one column per driver rather than three cubes,
    and no flow network at all.

    Attributes:
        period: The span the run covers.
        data: `(time, 4)` array of precipitation, ET, temperature and the long-term average.
        parameters: The parameter vector plus the `(snow, maxbas)` pair fixing its width.
        model_setup: The conceptual model instance and the state it starts from.
    """

    period: SimulationPeriod
    data: np.ndarray
    parameters: ParameterSet
    model_setup: ConceptualModelSetup

    def __post_init__(self):
        """Check the driver record covers the period the model was built for.

        Raises:
            ValueError: The record is not four columns wide, or does not span the period.
        """
        if np.ndim(self.data) != 2 or np.shape(self.data)[1] != 4:
            raise ValueError(
                "the lumped drivers must be a (time, 4) array of precipitation, ET, "
                f"temperature and the long-term average, got shape {np.shape(self.data)}"
            )
        steps = np.shape(self.data)[0]
        if steps != len(self.period):
            raise ValueError(
                f"the lumped drivers hold {steps} steps but the model spans "
                f"{len(self.period)} ({self.period.start:%Y-%m-%d} to "
                f"{self.period.end:%Y-%m-%d}); the run is positional, so a mismatch silently "
                "pairs each step with the wrong date"
            )

    @classmethod
    def from_model(cls, model: CatchmentLike) -> LumpedRun:
        """Narrow a built catchment into a validated lumped run.

        Args:
            model: A catchment with its inputs read.

        Returns:
            LumpedRun: The validated inputs.

        Raises:
            ValueError: A required input is unset, or the record does not span the period.
        """
        return cls(
            period=model.period,
            data=_require(model, "data", "call read_lumped_inputs first"),
            parameters=_require(model, "parameters", "call read_parameters first"),
            model_setup=_require(model, "model_setup", "call read_lumped_model first"),
        )

__post_init__()

Check the driver record covers the period the model was built for.

Raises:

Type	Description
ValueError	The record is not four columns wide, or does not span the period.

Source code in src/hapi/runs.py

def __post_init__(self):
    """Check the driver record covers the period the model was built for.

    Raises:
        ValueError: The record is not four columns wide, or does not span the period.
    """
    if np.ndim(self.data) != 2 or np.shape(self.data)[1] != 4:
        raise ValueError(
            "the lumped drivers must be a (time, 4) array of precipitation, ET, "
            f"temperature and the long-term average, got shape {np.shape(self.data)}"
        )
    steps = np.shape(self.data)[0]
    if steps != len(self.period):
        raise ValueError(
            f"the lumped drivers hold {steps} steps but the model spans "
            f"{len(self.period)} ({self.period.start:%Y-%m-%d} to "
            f"{self.period.end:%Y-%m-%d}); the run is positional, so a mismatch silently "
            "pairs each step with the wrong date"
        )

from_model(model: CatchmentLike) -> LumpedRun classmethod

Narrow a built catchment into a validated lumped run.

Parameters:

Name	Type	Description	Default
model	CatchmentLike	A catchment with its inputs read.	required

Returns:

Type	Description
LumpedRun	The validated inputs.

Raises:

Type	Description
ValueError	A required input is unset, or the record does not span the period.

Source code in src/hapi/runs.py

@classmethod
def from_model(cls, model: CatchmentLike) -> LumpedRun:
    """Narrow a built catchment into a validated lumped run.

    Args:
        model: A catchment with its inputs read.

    Returns:
        LumpedRun: The validated inputs.

    Raises:
        ValueError: A required input is unset, or the record does not span the period.
    """
    return cls(
        period=model.period,
        data=_require(model, "data", "call read_lumped_inputs first"),
        parameters=_require(model, "parameters", "call read_parameters first"),
        model_setup=_require(model, "model_setup", "call read_lumped_model first"),
    )