Config

The schema of a YAML run configuration. Each block below is one top-level key of the file; the rules tying them together — which blocks a given spatial_resolution requires, and which it refuses — live on RunConfig.

Build a model from a file with Catchment.from_yaml.

RunConfig

hapi.config.RunConfig

Bases: BaseModel

The full input set for one Catchment build.

Attributes:

Name	Type	Description
catchment	CatchmentConfig	Constructor arguments.
meteo	MeteoConfig	The meteorological drivers.
conceptual_model	ConceptualModelConfig	The lumped conceptual model.
parameters	ParametersConfig | None	Where the conceptual-model parameters live. Omit for a calibration, which derives them from the bounds handed to read_parameters_bound rather than reading a fitted set.
gauges	GaugesConfig | None	The observed discharge. Omit for a run that is not scored against gauges.
flow_network	FlowNetworkConfig | None	The routing network. Required for a distributed run and refused for a lumped one, which has no grid to put it on.
outputs	OutputsConfig | None	Where to write results.

Source code in src/hapi/config.py

class RunConfig(BaseModel):
    """The full input set for one `Catchment` build.

    Attributes:
        catchment: Constructor arguments.
        meteo: The meteorological drivers.
        conceptual_model: The lumped conceptual model.
        parameters: Where the conceptual-model parameters live. Omit for a calibration, which
            derives them from the bounds handed to `read_parameters_bound` rather than reading
            a fitted set.
        gauges: The observed discharge. Omit for a run that is not scored against gauges.
        flow_network: The routing network. Required for a distributed run and refused for a
            lumped one, which has no grid to put it on.
        outputs: Where to write results.
    """

    model_config = _STRICT

    catchment: CatchmentConfig
    meteo: MeteoConfig
    conceptual_model: ConceptualModelConfig
    parameters: ParametersConfig | None = None
    gauges: GaugesConfig | None = None
    flow_network: FlowNetworkConfig | None = None
    outputs: OutputsConfig | None = None

    @model_validator(mode="after")
    def _check_the_routing_method_matches_the_parameter_set(self) -> RunConfig:
        """Make `routing_method` agree with the parameter set, deriving it where it is unstated.

        The parameter-count check downstream cannot catch a disagreement: a MAXBAS set holds 11
        parameters and a Muskingum set 12, and `parameters.maxbas` is what selects which count
        is expected, so a set that contradicts the routing method still counts correctly. The
        run then completes, reading the Muskingum X as the MAXBAS value (or K and X out of a
        MAXBAS set), and produces a hydrograph that is quietly wrong.

        A lumped run picks its routing function at the call site rather than from this
        attribute, so `routing_method` is not load-bearing there -- but it is public, it is what
        `distrrm.route_muskingum` keys off, and leaving it saying `Muskingum` on a run using a
        MAXBAS parameter set would mislead the next reader. An unstated one is therefore
        derived from the parameter set rather than left at its default.

        Returns:
            RunConfig: This config, with `catchment.routing_method` filled in where it was
                unstated and the parameter set says which it is.

        Raises:
            ValueError: `catchment.routing_method` and `parameters.maxbas` disagree.
        """
        if self.parameters is None:
            return self

        if "routing_method" not in self.catchment.model_fields_set:
            self.catchment.routing_method = (
                "maxbas" if self.parameters.maxbas else "muskingum"
            )
            return self

        if (self.catchment.routing_method == "maxbas") != self.parameters.maxbas:
            raise ValueError(
                f"catchment.routing_method is {self.catchment.routing_method!r} but "
                f"parameters.maxbas is {self.parameters.maxbas}; the parameter set and the "
                f"routing method must agree, or the run reads the wrong parameter as the "
                f"routing one"
            )
        return self

    @model_validator(mode="after")
    def _check_blocks_match_the_spatial_resolution(self) -> RunConfig:
        """Enforce the fields each spatial resolution requires.

        The two resolutions share no rule -- one describes a grid and a routing network, the
        other a pair of CSVs -- so each owns a method and this one only chooses between them.

        Returns:
            RunConfig: This config, unchanged.

        Raises:
            ValueError: A block the chosen `spatial_resolution` needs is missing, or one it
                will never read is present.
        """
        if self.catchment.spatial_resolution == "distributed":
            self._check_the_distributed_blocks()
        else:
            self._check_the_lumped_blocks()
        return self

    def _check_the_distributed_blocks(self) -> None:
        """Enforce what a distributed run needs and what its `meteo.source` can read.

        Raises:
            ValueError: The routing network is missing or incomplete, the gauge table is
                absent while gauges are configured, a driver is unset, `source="netcdf"`
                names no file, or a field outside the source's own set is present.
        """
        if self.flow_network is None:
            raise ValueError(
                "catchment.spatial_resolution is 'distributed', which needs a flow_network "
                "block"
            )
        # `flow_direction` is optional on the block because MAXBAS sends every cell straight
        # to the outlet and never reads one. Muskingum routes along the network, so without
        # it the build succeeds and `Run.run_distributed` dereferences a None array after every
        # raster has been read.
        if (
            self.catchment.routing_method == "muskingum"
            and self.flow_network.flow_direction is None
        ):
            raise ValueError(
                "catchment.routing_method is 'muskingum', which routes along the network, "
                "so flow_network.flow_direction is required"
            )
        # Only when gauges are configured at all: a distributed run that is not scored
        # against observations omits the block entirely.
        if self.gauges is not None and self.gauges.table is None:
            raise ValueError(
                "catchment.spatial_resolution is 'distributed', which needs gauges.table "
                "to locate the gauges on the grid"
            )

        missing = [name for name in METEO_DRIVERS if getattr(self.meteo, name) is None]
        if missing:
            raise ValueError(missing_drivers_message(missing))
        if self.meteo.source == "netcdf" and self.meteo.path is None:
            raise ValueError(NETCDF_PATH_MESSAGE)

        _reject_fields_the_run_will_not_read(
            self.meteo,
            _METEO_FIELDS_BY_SOURCE[self.meteo.source],
            "meteo",
            f"meteo.source is {self.meteo.source!r}",
        )

    def _check_the_lumped_blocks(self) -> None:
        """Enforce what a lumped run needs, and refuse the grid it has no use for.

        `extra="forbid"` exists so a misspelled key fails rather than being dropped;
        accepting a correctly spelled but inapplicable block would be the same silence by
        another route. A lumped run has no grid, so nothing that describes one applies.

        Raises:
            ValueError: `meteo.path` is unset, a routing network or a grid `meteo.source` is
                present, or a `meteo` / `gauges` field this run will never read is set.
        """
        if self.meteo.path is None:
            raise ValueError(
                "catchment.spatial_resolution is 'lumped', which needs meteo.path -- the "
                "CSV of catchment-average drivers"
            )
        if self.flow_network is not None:
            raise ValueError(
                "catchment.spatial_resolution is 'lumped', which has no grid, so a "
                "flow_network block cannot be used"
            )
        if self.meteo.source != "rasters":
            raise ValueError(
                f"catchment.spatial_resolution is 'lumped', which reads meteo.path as a "
                f"CSV of catchment-average drivers; meteo.source "
                f"{self.meteo.source!r} does not apply"
            )

        lumped = "catchment.spatial_resolution is 'lumped'"
        _reject_fields_the_run_will_not_read(
            self.meteo,
            _LUMPED_METEO_FIELDS,
            "meteo",
            f"{lumped}, which reads meteo.path as one CSV of catchment-average drivers",
        )
        if self.gauges is not None:
            _reject_fields_the_run_will_not_read(
                self.gauges,
                _LUMPED_GAUGES_FIELDS,
                "gauges",
                f"{lumped}, which reads one discharge file and locates no gauges",
            )

    @model_validator(mode="after")
    def _check_the_dates_parse_and_are_ordered(self) -> RunConfig:
        """Check every date against the format it is written in, and that the period runs.

        Returns:
            RunConfig: This config, unchanged.

        Raises:
            ValueError: A date does not match its format, or a period ends before it starts.
        """
        for label, value, fmt in (
            ("catchment.start", self.catchment.start, self.catchment.fmt),
            ("catchment.end", self.catchment.end, self.catchment.fmt),
            ("meteo.start", self.meteo.start, self.meteo.fmt),
            ("meteo.end", self.meteo.end, self.meteo.fmt),
        ):
            if value is None:
                continue
            try:
                datetime.strptime(value, fmt)
            except ValueError as error:
                raise ValueError(
                    f"{label} {value!r} does not match its format {fmt!r}: {error}"
                ) from error

        if datetime.strptime(
            self.catchment.start, self.catchment.fmt
        ) > datetime.strptime(self.catchment.end, self.catchment.fmt):
            raise ValueError(
                f"catchment.start {self.catchment.start!r} is after catchment.end "
                f"{self.catchment.end!r}"
            )

        # The window the run actually uses, not the two literal pairs: `MeteoInputs.from_config`
        # takes each bound from `meteo` when it is stated and falls back to `catchment`
        # otherwise, so a `meteo` block stating only an end can invert the effective window
        # while neither pair is inverted on its own.
        window_start, start_fmt = (
            (self.meteo.start, self.meteo.fmt)
            if self.meteo.start is not None
            else (self.catchment.start, self.catchment.fmt)
        )
        window_end, end_fmt = (
            (self.meteo.end, self.meteo.fmt)
            if self.meteo.end is not None
            else (self.catchment.end, self.catchment.fmt)
        )
        if datetime.strptime(window_start, start_fmt) > datetime.strptime(
            window_end, end_fmt
        ):
            raise ValueError(
                f"the meteorological window runs from {window_start!r} to {window_end!r}, "
                f"which ends before it starts; each bound is taken from meteo when stated "
                f"and from catchment otherwise"
            )
        return self

CatchmentConfig

hapi.config.CatchmentConfig

Bases: BaseModel

The Catchment constructor arguments.

Attributes:

Name	Type	Description
name	str	Catchment name.
start	str	Start date, parsed with fmt. Held as a string, because the constructor does the parsing; an unquoted YAML date arrives here already a date and is written back out in fmt, so both spellings work.
end	str	End date, parsed with fmt. See start.
fmt	str	strptime format for start / end.
spatial_resolution	Literal['lumped', 'distributed']	"lumped" or "distributed". Selects the shape of meteo and gauges, and whether flow_network is required.
temporal_resolution	Literal['daily', 'hourly']	"daily" or "hourly".
routing_method	Literal['muskingum', 'maxbas']	"muskingum" or "maxbas". Assigned onto model.routing_method, and constrains one other block: Muskingum routes along the network, so a distributed run needs flow_network.flow_direction. Left unwritten it is derived from parameters.maxbas, which describes the same choice from the parameter set's side; written, it must agree with it. Which Run.* entry point actually routes with it is still the caller's choice.

Source code in src/hapi/config.py

class CatchmentConfig(BaseModel):
    """The `Catchment` constructor arguments.

    Attributes:
        name: Catchment name.
        start: Start date, parsed with `fmt`. Held as a string, because the constructor does
            the parsing; an unquoted YAML date arrives here already a `date` and is written
            back out in `fmt`, so both spellings work.
        end: End date, parsed with `fmt`. See `start`.
        fmt: `strptime` format for `start` / `end`.
        spatial_resolution: `"lumped"` or `"distributed"`. Selects the shape of `meteo` and
            `gauges`, and whether `flow_network` is required.
        temporal_resolution: `"daily"` or `"hourly"`.
        routing_method: `"muskingum"` or `"maxbas"`. Assigned onto `model.routing_method`, and
            constrains one other block: Muskingum routes along the network, so a distributed
            run needs `flow_network.flow_direction`. Left unwritten it is derived from
            `parameters.maxbas`, which describes the same choice from the parameter set's
            side; written, it must agree with it. Which `Run.*` entry point actually routes
            with it is still the caller's choice.
    """

    model_config = _STRICT

    name: str
    start: str
    end: str
    fmt: str = "%Y-%m-%d"
    spatial_resolution: Literal["lumped", "distributed"] = "lumped"
    temporal_resolution: Literal["daily", "hourly"] = "daily"
    # Two of the three keys of `hapi.catchment.ROUTING_METHODS`, which is what the constructor
    # accepts. `kinematic` is left out deliberately: it selects the flood model, whose inputs
    # (`read_river_geometry`, `bankfull_depth`) this schema does not carry, so a configuration
    # naming it would validate and then build a model that cannot run. Adding a method there
    # means deciding here whether the schema can describe a run that uses it.
    routing_method: Literal["muskingum", "maxbas"] = "muskingum"

    @model_validator(mode="before")
    @classmethod
    def _accept_a_date_yaml_already_parsed(cls, values: Any) -> Any:
        """Render an unquoted YAML date back into `fmt` before the string fields see it.

        Args:
            values: The raw mapping.

        Returns:
            Any: The mapping, with `start` and `end` as strings.
        """
        return _write_dates_in_the_block_format(values, ("start", "end"))

MeteoConfig

hapi.config.MeteoConfig

Bases: BaseModel

The meteorological drivers: a distributed grid or a lumped CSV.

Attributes:

Name	Type	Description
source	Literal['rasters', 'netcdf', 'netcdf_files']	Which MeteoInputs loader builds the grid. Ignored for a lumped run, which always reads path as a single CSV.
precipitation	str | None	Rainfall folder ("rasters"), NetCDF path ("netcdf_files"), or the variable name holding rainfall inside path ("netcdf").
temperature	str | None	As precipitation, for temperature.
evapotranspiration	str | None	As precipitation, for evapotranspiration.
path	str | None	The combined NetCDF (source="netcdf") or the lumped meteo CSV.
variable	str | None	Which variable to take from each file, source="netcdf_files" only. None takes the single variable a file holds, which is an error if it holds several.
start	str | None	Window start; None falls back to catchment.start. Distributed only.
end	str | None	Window end; None falls back to catchment.end. Distributed only.
fmt	str	strptime format for start / end.
glob	str	Raster glob, source="rasters" only.
regex_string	str	Date regex within file names, source="rasters" only.
file_name_data_fmt	str | None	strptime format for the matched date; inferred if None.
per_variable	dict[str, dict[str, Any]] | None	Per-folder overrides of the reader arguments, source="rasters" only.
gdal_env	dict[str, str] | None	GDAL environment overrides for the raster read, source="rasters" only.

Source code in src/hapi/config.py

class MeteoConfig(BaseModel):
    """The meteorological drivers: a distributed grid or a lumped CSV.

    Attributes:
        source: Which `MeteoInputs` loader builds the grid. Ignored for a lumped run, which
            always reads `path` as a single CSV.
        precipitation: Rainfall folder (`"rasters"`), NetCDF path (`"netcdf_files"`), or the
            variable name holding rainfall inside `path` (`"netcdf"`).
        temperature: As `precipitation`, for temperature.
        evapotranspiration: As `precipitation`, for evapotranspiration.
        path: The combined NetCDF (`source="netcdf"`) or the lumped meteo CSV.
        variable: Which variable to take from each file, `source="netcdf_files"` only. `None`
            takes the single variable a file holds, which is an error if it holds several.
        start: Window start; `None` falls back to `catchment.start`. Distributed only.
        end: Window end; `None` falls back to `catchment.end`. Distributed only.
        fmt: `strptime` format for `start` / `end`.
        glob: Raster glob, `source="rasters"` only.
        regex_string: Date regex within file names, `source="rasters"` only.
        file_name_data_fmt: `strptime` format for the matched date; inferred if `None`.
        per_variable: Per-folder overrides of the reader arguments, `source="rasters"` only.
        gdal_env: GDAL environment overrides for the raster read, `source="rasters"` only.
    """

    model_config = _STRICT

    source: Literal["rasters", "netcdf", "netcdf_files"] = "rasters"
    precipitation: str | None = None
    temperature: str | None = None
    evapotranspiration: str | None = None
    path: str | None = None
    variable: str | None = None
    start: str | None = None
    end: str | None = None
    fmt: str = "%Y-%m-%d"
    glob: str = "*.tif"
    regex_string: str = r"\d{4}.\d{2}.\d{2}"
    file_name_data_fmt: str | None = None
    per_variable: dict[str, dict[str, Any]] | None = None
    gdal_env: dict[str, str] | None = None

    @model_validator(mode="before")
    @classmethod
    def _accept_a_date_yaml_already_parsed(cls, values: Any) -> Any:
        """Render an unquoted YAML date back into `fmt` before the string fields see it.

        Args:
            values: The raw mapping.

        Returns:
            Any: The mapping, with `start` and `end` as strings.
        """
        return _write_dates_in_the_block_format(values, ("start", "end"))

FlowNetworkConfig

hapi.config.FlowNetworkConfig

Bases: BaseModel

The routing network. Distributed runs only.

Attributes:

Name	Type	Description
flow_accumulation	str	Path to the flow-accumulation raster.
flow_direction	str | None	Path to the flow-direction raster. Muskingum needs it; MAXBAS sends every cell straight to the outlet and never reads one, so it may be omitted.

Source code in src/hapi/config.py

class FlowNetworkConfig(BaseModel):
    """The routing network. Distributed runs only.

    Attributes:
        flow_accumulation: Path to the flow-accumulation raster.
        flow_direction: Path to the flow-direction raster. Muskingum needs it; MAXBAS sends
            every cell straight to the outlet and never reads one, so it may be omitted.
    """

    model_config = _STRICT

    flow_accumulation: str
    flow_direction: str | None = None

ParametersConfig

hapi.config.ParametersConfig

Bases: BaseModel

Where the conceptual-model parameters live.

Attributes:

Name	Type	Description
path	str	Folder of parameter rasters (distributed) or a single file (lumped).
snow	bool	Whether the parameter set includes the snow routine (15 parameters against 10).
maxbas	bool	Whether the set carries the triangular-routing parameter. It describes the parameter set rather than the run, but RunConfig requires it to agree with catchment.routing_method: the two counts differ (11 against 12) and maxbas is what selects which is expected, so a disagreeing pair still passes the count check and then reads the wrong parameter as the routing one.

Source code in src/hapi/config.py

class ParametersConfig(BaseModel):
    """Where the conceptual-model parameters live.

    Attributes:
        path: Folder of parameter rasters (distributed) or a single file (lumped).
        snow: Whether the parameter set includes the snow routine (15 parameters against 10).
        maxbas: Whether the set carries the triangular-routing parameter. It describes the
            parameter set rather than the run, but `RunConfig` requires it to agree with
            `catchment.routing_method`: the two counts differ (11 against 12) and `maxbas`
            is what selects which is expected, so a disagreeing pair still passes the count
            check and then reads the wrong parameter as the routing one.
    """

    model_config = _STRICT

    path: str
    snow: bool = False
    maxbas: bool = False

ConceptualModelConfig

hapi.config.ConceptualModelConfig

Bases: BaseModel

The lumped conceptual model, run per cell (distributed) or per catchment (lumped).

Attributes:

Name	Type	Description
model_class	str	Name of the conceptual model, e.g. "HBVBergestrom92". Resolved to a class by the builder, which owns the registry of available models.
catchment_area	float	Catchment area, km2.
initial_condition	list[float]	[sp, sm, uz, lz, wc], exactly five values.
q_init	float | None	Initial discharge; None derives it from the initial condition.

Source code in src/hapi/config.py

class ConceptualModelConfig(BaseModel):
    """The lumped conceptual model, run per cell (distributed) or per catchment (lumped).

    Attributes:
        model_class: Name of the conceptual model, e.g. `"HBVBergestrom92"`. Resolved to a class
            by the builder, which owns the registry of available models.
        catchment_area: Catchment area, km2.
        initial_condition: `[sp, sm, uz, lz, wc]`, exactly five values.
        q_init: Initial discharge; `None` derives it from the initial condition.
    """

    # `model_class` would collide with pydantic's protected `model_` namespace, so the namespace
    # is cleared rather than renaming a field the YAML already uses.
    model_config = ConfigDict(**_STRICT, protected_namespaces=())

    model_class: str
    catchment_area: float = Field(gt=0)
    initial_condition: list[float] = Field(min_length=5, max_length=5)
    q_init: float | None = None

GaugesConfig

hapi.config.GaugesConfig

Bases: BaseModel

The observed discharge the run is scored against.

Attributes:

Name	Type	Description
discharge	str	Folder of one CSV per gauge id (distributed) or a single CSV (lumped).
table	str | None	Gauge locations and properties. Distributed only; a lumped run has no grid to locate gauges on.
column	str	Gauge-table column naming the columns of the resulting hydrograph frame. It does not select the discharge file names -- read_discharge_gauges reads <id>.csv regardless -- so a table can label its hydrographs with human-readable names while the files stay named after the ids.
delimiter	str	Discharge CSV delimiter.
fmt	str	strptime format for the discharge CSV's date column.
table_fmt	str | None	strptime format for the gauge table's optional start / end columns, which bound each gauge's validity period. A separate field because the table is a separate file that a separate hand may have written; None falls back to fmt, which is right whenever the two were written together.

Source code in src/hapi/config.py

class GaugesConfig(BaseModel):
    """The observed discharge the run is scored against.

    Attributes:
        discharge: Folder of one CSV per gauge id (distributed) or a single CSV (lumped).
        table: Gauge locations and properties. Distributed only; a lumped run has no grid to
            locate gauges on.
        column: Gauge-table column naming the columns of the resulting hydrograph frame. It
            does not select the discharge file names -- `read_discharge_gauges` reads
            `<id>.csv` regardless -- so a table can label its hydrographs with human-readable
            names while the files stay named after the ids.
        delimiter: Discharge CSV delimiter.
        fmt: `strptime` format for the discharge CSV's date column.
        table_fmt: `strptime` format for the gauge table's optional `start` / `end` columns,
            which bound each gauge's validity period. A separate field because the table is a
            separate file that a separate hand may have written; `None` falls back to `fmt`,
            which is right whenever the two were written together.
    """

    model_config = _STRICT

    discharge: str
    table: str | None = None
    column: str = "id"
    delimiter: str = ","
    fmt: str = "%Y-%m-%d"
    table_fmt: str | None = None

OutputsConfig

hapi.config.OutputsConfig

Bases: BaseModel

Where to write results after the run.

Attributes:

Name	Type	Description
results_dir	str | None	Folder SimulationResults.save writes into.

Source code in src/hapi/config.py

class OutputsConfig(BaseModel):
    """Where to write results after the run.

    Attributes:
        results_dir: Folder `SimulationResults.save` writes into.
    """

    model_config = _STRICT

    results_dir: str | None = None