Components

Catchment

class hydrobricks.Catchment(outline: str | Path | None = None, land_cover_types: list[str] | None = None, land_cover_names: list[str] | None = None, hydro_units_data: pd.DataFrame | None = None)[source]

Bases: object

Creation of catchment-related data

Parameters:

  • outline – Path to the outline of the catchment.

  • land_cover_types – The land cover types of the catchment.

  • land_cover_names – The land cover names of the catchment.

  • hydro_units_data – The hydro units data of the catchment.

area

The area of the catchment.

Type:

float

crs

The crs of the catchment outline.

Type:

str

outline

The outline of the catchment.

Type:

shapely.geometry.Polygon

dem

The DEM of the catchment [m].

Type:

rasterio.DatasetReader

dem_data

The masked DEM data of the catchment.

Type:

np.ndarray

slope

The slope map of the catchment [degrees].

Type:

np.ndarray

aspect

The aspect map of the catchment.

Type:

np.ndarray

map_unit_ids

The unit ids as a numpy array matching the DEM extent.

Type:

np.ndarray

hydro_units

The hydro units of the catchment.

Type:

HydroUnits

static calculate_cast_shadows(*args, **kwargs) → numpy.ndarray[source]

Call the calculate_cast_shadows method of the PotentialSolarRadiation class.

calculate_connectivity(*args, **kwargs) → pandas.DataFrame[source]

Call the calculate_connectivity method of the Connectivity class.

calculate_daily_potential_radiation(*args, **kwargs) → None[source]

Call the calculate_daily_potential_radiation method of the PotentialSolarRadiation class.

calculate_slope_aspect() → None[source]

Call the calculate_slope_aspect method of the Topography class.

close() → None[source]

Close all open resources (DEM dataset and MemoryFiles).

Called automatically when using the catchment as a context manager, or can be called manually to explicitly release resources.

Examples

Using as context manager (recommended):

>>> with Catchment(outline='boundary.shp') as catchment:
...     catchment.extract_dem('dem.tif')

Manual cleanup:

>>> catchment = Catchment(outline='boundary.shp')
>>> try:
...     catchment.extract_dem('dem.tif')
... finally:
...     catchment.close()
property connectivity: Any

Lazy-loaded connectivity module.

Returns:

Connectivity processor for the catchment, loaded on first access.

Return type:

CatchmentConnectivity

create_dem_pixel_geometry(i: int, j: int) → shapely.geometry.Polygon[source]

Create a shapely geometry of the DEM pixel.

Parameters:

  • i – The row of the pixel.

  • j – The column of the pixel.

Return type:

The shapely geometry of the pixel.

create_elevation_bands(*args, **kwargs) → None[source]

Call the create_elevation_bands method of the Discretization class.

property discretization: Any

Lazy-loaded discretization module.

Returns:

Discretization processor for the catchment, loaded on first access.

Return type:

CatchmentDiscretization

discretize_by(*args, **kwargs) → None[source]

Call the discretize_by method of the Discretization class.

extract_attribute_raster(raster_path: str | Path, attr_name: str, resample_to_dem_resolution: bool = True, resampling: str = 'average', replace_nans_by_zeros: bool = True) → bool[source]

Extract spatial attributes (raster) for the catchment. Does not handle change in coordinates.

Parameters:

  • raster_path – Path of the raster file containing the attribute data.

  • attr_name – Name of the attribute to store in self.attributes dictionary.

  • resample_to_dem_resolution – If True, resample the attribute raster to DEM resolution. Default: True

  • resampling – Resampling method to use when resample_to_dem_resolution is True. Options: ‘nearest’, ‘bilinear’, ‘cubic’, ‘cubic_spline’, ‘lanczos’, ‘average’, ‘mode’, ‘gauss’, ‘max’, ‘min’, ‘med’, ‘q1’, ‘q3’, ‘sum’, ‘rms’ Default: ‘average’

  • replace_nans_by_zeros – If True, replace NaN values with zero in the output raster. Default: True

Returns:

True if extraction was successful, False otherwise.

Return type:

bool

Raises:

  • FileNotFoundError – If the raster file does not exist.

  • ValueError – If the resampling method is not recognized.

extract_dem(raster_path: str | Path) → bool[source]

Extract the DEM data for the catchment. Does not handle change in coordinates.

Parameters:

raster_path – Path of the DEM raster file.

Returns:

True if extraction was successful, False otherwise.

Return type:

bool

Raises:

FileNotFoundError – If the raster file does not exist.

extract_unit_mean_lat_lon(mask_unit: numpy.ndarray) → tuple[float, float][source]

Extract the mean latitude and longitude for a hydro unit.

Calculates the mean coordinates of pixels within a unit mask and converts them from the catchment CRS to latitude/longitude (EPSG:4326).

Parameters:

mask_unit – Boolean mask array identifying the cells of the hydro unit.

Returns:

Tuple of (mean_latitude, mean_longitude) in degrees.

Return type:

tuple[float, float]

get_attribute_raster_x_resolution(attr_name: str = 'dem') → float[source]

Get the given attribute raster x resolution.

Parameters:

attr_name – Name of the attribute.

Return type:

The attribute raster x resolution.

get_attribute_raster_y_resolution(attr_name: str = 'dem') → float[source]

Get the given attribute raster y resolution.

Parameters:

attr_name – Name of the attribute.

Return type:

The attribute raster y resolution.

get_dem_mean_lat_lon() → tuple[float, float][source]

Get the mean latitude and longitude of the DEM extent.

Calculates the central coordinates of the catchment DEM and converts them from the catchment CRS to latitude/longitude (EPSG:4326).

Returns:

Tuple of (mean_latitude, mean_longitude) in degrees.

Return type:

tuple[float, float]

get_dem_pixel_area() → float[source]

Get the DEM pixel area.

Return type:

The DEM pixel area.

get_dem_x_resolution() → float[source]

Get the DEM x resolution.

Return type:

The DEM x resolution.

get_dem_y_resolution() → float[source]

Get the DEM y resolution.

Return type:

The DEM y resolution.

get_hillshade(*args, **kwargs) → numpy.ndarray[source]

Call the get_hillshade method of the Topography class.

get_hydro_unit_count() → int[source]

Get the number of hydro units.

Return type:

The number of hydro units.

get_hydro_units_attributes() → HydroUnits[source]

Extract the hydro units attributes.

Return type:

The hydro units attributes.

get_hydro_units_elevations() → numpy.ndarray[source]

Get the elevation of the hydro units.

Return type:

The elevation of the hydro units.

get_mean_elevation() → float[source]

Call the get_mean_elevation method of the Topography class.

static get_solar_azimuth_to_north(*args, **kwargs) → float | np.ndarray[source]

Call the get_solar_azimuth_to_north method of the PotentialSolarRadiation class.

static get_solar_azimuth_to_south(*args, **kwargs) → numpy.ndarray[source]

Call the get_solar_azimuth_to_south method of the PotentialSolarRadiation class.

static get_solar_declination_rad(*args, **kwargs) → float[source]

Call the get_solar_declination_rad method of the PotentialSolarRadiation class.

static get_solar_hour_angle_limit(*args, **kwargs) → float | np.ndarray[source]

Call the get_solar_hour_angle_limit method of the PotentialSolarRadiation class.

static get_solar_zenith(*args, **kwargs) → float | np.ndarray[source]

Call the get_solar_zenith method of the PotentialSolarRadiation class.

initialize_area_from_land_cover_change(land_cover_name: str, land_cover_change: pandas.DataFrame) → None[source]

Initialize the HydroUnits cover area from a land cover change object.

Parameters:

  • land_cover_name – The name of the land cover to initialize.

  • land_cover_change – The land cover change dataframe.

initialize_land_cover_fractions() → None[source]

Initialize land cover fractions for all hydro units.

Sets up the initial fractional areas for each land cover type within each hydro unit based on the available land cover data.

load_hydro_units_from_csv(path: str | Path) → None[source]

Load hydro units from a csv file.

Parameters:

path – Path to the csv file.

load_mean_annual_radiation_raster(*args, **kwargs) → None[source]

Call the load_mean_annual_radiation_raster method of the PotentialSolarRadiation class.

load_unit_ids_from_raster(path: str, filename: str = 'unit_ids.tif') → None[source]

Load hydro units from a raster file.

Parameters:

  • path – Path to the directory containing the raster file. If the path is a file, it will be used as the full path.

  • filename – Name of the raster file. Default is ‘unit_ids.tif’.

resample_dem_and_calculate_slope_aspect(*args, **kwargs) → tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray][source]

Call the resample_dem_and_calculate_slope_aspect method of the Topography class.

save_hydro_units_to_csv(path: str | Path) → None[source]

Save the hydro units to a csv file.

Parameters:

path – Path to the output file.

save_unit_ids_raster(output_path: str | Path, output_filename: str = 'unit_ids.tif') → None[source]

Save the unit ids raster to a file.

Parameters:

  • output_path – Path to the output file.

  • output_filename – Name of the output file. Default is ‘unit_ids.tif’.

property solar_radiation: Any

Lazy-loaded solar radiation module.

Returns:

Solar radiation processor for the catchment, loaded on first access.

Return type:

PotentialSolarRadiation

property topography: Any

Lazy-loaded topography analysis module.

Returns:

Topography processor for the catchment, loaded on first access.

Return type:

CatchmentTopography

upscale_and_save_mean_annual_radiation_rasters(*args, **kwargs) → None[source]

Call the upscale_and_save_mean_annual_radiation_rasters method of the PotentialSolarRadiation class.

HydroUnits

class hydrobricks.HydroUnits(land_cover_types: list[str] | None = None, land_cover_names: list[str] | None = None, data: pd.DataFrame | None = None)[source]

Bases: object

Class for the hydro units

Parameters:

  • land_cover_types – List of land cover types. Default: [‘ground’]

  • land_cover_names – List of land cover names. Default: [‘ground’]

  • data – DataFrame containing the hydro units data.

land_cover_types

List of land cover types. Default: [‘ground’]

Type:

list[str]

land_cover_names

List of land cover names. Default: [‘ground’]

Type:

list[str]

hydro_units

Dataframe containing the hydro units data.

Type:

pd.DataFrame

FRACTION_PREFIX: ClassVar[str] = 'fraction-'
add_property(column_tuple: tuple[str, str], values: numpy.ndarray, set_first: bool = False) → None[source]

Add a property to the hydro units.

Adds a new column to the hydro_units DataFrame with the specified property name, unit, and values. Can optionally insert as the first column.

Parameters:

  • column_tuple – Tuple containing (property_name, unit_string). Example: (‘elevation’, ‘m’)

  • values – Numpy array containing the property values for each hydro unit.

  • set_first – If True, the property is added as the first column. Default: False

Raises:

ValueError – If values length doesn’t match the number of hydro units.

check_land_cover_fractions_not_empty() → None[source]

Check that the land cover fractions are not empty.

Validates that all land cover fractions have been defined. If there is a single land cover type (e.g. ‘ground’), automatically sets it to 1.0 for all hydro units.

Raises:

ValueError – If any land cover fraction contains NaN values (when multiple land cover types exist).

get_hydro_unit_count() → int[source]

Get the number of hydro units.

Return type:

Number of hydro units.

get_ids() → pandas.Series[source]

Get the hydro unit ids.

has(prop: str) → bool[source]

Check if the hydro units have a given property.

Parameters:

prop – The property name to check. Should match a column name in the hydro_units DataFrame.

Returns:

True if the property is present, False otherwise.

Return type:

bool

initialize_from_land_cover_change(land_cover_name: str, land_cover_change: pandas.DataFrame) → None[source]

Initialize the hydro units from the first values of a land cover change dataframe.

Updates the land cover fractions for specified hydro units based on a land cover change dataframe. Automatically adjusts the ‘ground’ land cover fraction to maintain conservation.

Parameters:

  • land_cover_name – The name of the land cover to initialize.

  • land_cover_change – The land cover change dataframe with columns ‘hydro_unit’ and area values.

Raises:

ValueError – If computed land cover fraction is not in the range [0, 1].

initialize_land_cover_fractions() → None[source]

Initialize land cover fractions with default values.

Sets the land cover fractions of ‘ground’ to 1.0 and all other land cover types to 0.0 for all hydro units. Used as a starting point before applying specific land cover changes.

load_from_csv(path: str | Path, column_elevation: str | None = None, column_area: str | None = None, column_fractions: dict[str, str] | None = None, columns_areas: dict[str, str] | None = None, other_columns: dict[str, str] | None = None) → None[source]

Read hydro units properties from CSV file. The file must contain two header rows. The first row contains the column names and the second row contains the units. The file must contain at minimum the units area.

Parameters:

  • path – Path to the CSV file containing hydro units data.

  • column_elevation – Column name containing the elevation values. If None, looks for ‘elevation’ column. Default: None

  • column_area – Column name containing the total area values. If None, looks for ‘area’ column. Default: None

  • column_fractions – NOT IMPLEMENTED. Dictionary mapping land cover names to area fraction column names. Default: None

  • columns_areas – Dictionary mapping land cover names to area column names. Cannot be used with column_area. Default: None

  • other_columns – Dictionary mapping property names to column names in the CSV file. Example: {‘slope’: ‘Slope’, ‘aspect’: ‘Aspect’} Default: None

Raises:

  • FileNotFoundError – If the CSV file does not exist.

  • ValueError – If required columns are missing or are inconsistent.

  • NotImplementedError – If column_fractions is provided (not yet implemented).

populate_bounded_instance() → None[source]

Populate the SettingsBasin instance from current hydro units data.

Updates the internal SettingsBasin object with current hydro unit properties, sorted by elevation in descending order. Includes land cover fractions and all custom properties.

save_as(path: str) → None[source]

Create a file containing the hydro unit properties. Such a file can be used in the command-line version of hydrobricks.

Saves hydro units and land cover information to a netCDF4 file with proper dimensions and variable attributes.

Parameters:

path – Path of the file to create.

Raises:

ImportError – If netcdf4 is not installed.

save_to_csv(path: str | Path) → None[source]

Save the hydro units to a csv file.

Exports the hydro units DataFrame to a CSV file with multi-level header containing both property names and units.

Parameters:

path – Path to the output file.

Raises:

ValueError – If no hydro units data is available.

set_connectivity(connectivity: pd.DataFrame | Path | str) → None[source]

Set the connectivity of the hydro units.

Configures how water flows between hydro units by setting lateral connections with specified ratios. Can accept connectivity data as a DataFrame or load from file.

Parameters:

connectivity – File or Dataframe containing the connectivity information as generated by catchment.calculate_connectivity().

Raises:

  • TypeError – If connectivity is not a DataFrame or valid file path.

  • ValueError – If connectivity DataFrame is missing required columns or has invalid values.

ParameterSet

class hydrobricks.ParameterSet[source]

Bases: object

Class for the parameter sets

add_aliases(parameter_name: str, aliases: list[str] | str) → None[source]

Add aliases to a parameter.

Parameters:

  • parameter_name – The name of the parameter with the related component (e.g., snowpack:degree_day_factor).

  • aliases – Aliases to the parameter name, such as names used in other implementations (e.g., kgl, an). Aliases must be unique.

add_data_parameter(name: str, value: float | list[float] | None = None, min_val: float | list[float] | None = None, max_val: float | list[float] | None = None, unit: str | None = None) → None[source]

Add a parameter related to the data.

Parameters:

  • name – The name of the parameter.

  • value – The parameter value.

  • min_val – Minimum value allowed for the parameter.

  • max_val – Maximum value allowed for the parameter.

  • unit – The unit of the parameter.

property allow_changing: list[str]

Get the list of parameters to assess during calibration.

Returns:

List of parameter names that are allowed to change.

Return type:

list[str]

are_valid() → bool[source]

Check if all the parameters are defined and have a value. Alias of is_valid.

Returns:

True if all parameters are defined and have a value, False otherwise.

Return type:

bool

change_range(parameter: str, min_val: float, max_val: float) → None[source]

Change the value range of a parameter.

Parameters:

  • parameter – Name (or alias) of the parameter

  • min_val – New minimum value

  • max_val – New maximum value

constraints_satisfied() → bool[source]

Check if the constraints between parameters are satisfied.

Returns:

True if constraints are satisfied, False otherwise.

Return type:

bool

define_constraint(parameter_1: str, operator: str, parameter_2: str) → None[source]

Defines a constraint between 2 parameters (e.g., paramA > paramB)

Parameters:

  • parameter_1 – The name of the first parameter.

  • operator – The operator (e.g. ‘<=’).

  • parameter_2 – The name of the second parameter.

Examples

parameter_set.define_constraint(‘paramA’, ‘>=’, ‘paramB’)

define_parameter(component: str, name: str, unit: str | None = None, aliases: str | list[str] | None = None, min_val: float | list[float] | None = None, max_val: float | list[float] | None = None, default: float | list[float] | None = None, mandatory: bool = True) → None[source]

Define a parameter by setting its properties.

Parameters:

  • component – The component (brick) name to which the parameter refer (e.g., snowpack, glacier, surface_runoff). It can be a string of a list of components when the parameter is shared between components (e.g., melt_factor in the temperature index method).

  • name – The name of the parameter in the C++ code of hydrobricks (e.g., degree_day_factor, response_factor).

  • unit – The unit of the parameter.

  • aliases – Aliases to the parameter name, such as names used in other implementations (e.g., kgl, an). Aliases must be unique.

  • min_val – Minimum value allowed for the parameter.

  • max_val – Maximum value allowed for the parameter.

  • default – The parameter default value.

  • mandatory – If the parameter needs to be defined or if it can silently use the default value.

generate_parameters(land_cover_types: list[str], land_cover_names: list[str], options: dict, structure: dict)[source]

Generate a parameters object for the provided model options and structure.

Parameters:

  • land_cover_types – The land cover types.

  • land_cover_names – The land cover names.

  • options – The model options.

  • structure – The model structure.

get(name: str) → float[source]

Get the value of a parameter by name.

Parameters:

name – The name of the parameter.

Returns:

The parameter value.

Return type:

float

get_for_spotpy() → list[spotpy.parameter][source]

Get the parameters to assess ready to be used in spotpy.

Return type:

A list of the parameters as spotpy objects.

get_model_parameters() → pandas.DataFrame[source]

Get the model-only parameters (excluding data-related parameters).

Returns:

DataFrame containing model parameters only.

Return type:

pd.DataFrame

get_undefined() → list[str][source]

Get the undefined parameters.

Returns:

List of the undefined parameter names.

Return type:

list[str]

has(name: str) → bool[source]

Check if a parameter exists.

Parameters:

name – The name of the parameter.

Returns:

True if found, False otherwise.

Return type:

bool

is_for_forcing(parameter_name: str) → bool[source]

Check if the parameter relates to forcing data.

Parameters:

parameter_name – The name of the parameter.

Returns:

True if relates to forcing data, False otherwise.

Return type:

bool

is_valid() → bool[source]

Check if all the parameters are defined and have a value.

Returns:

True if all parameters are defined and have a value, False otherwise.

Return type:

bool

list_constraints() → None[source]

List the constraints currently defined.

Prints all defined parameter constraints to the console.

needs_random_forcing() → bool[source]

Check if one of the parameters to assess involves the meteorological data.

Return type:

True if one of the parameters to assess involves the meteorological data.

range_satisfied() → bool[source]

Check if the parameter value ranges are satisfied.

Returns:

True if ranges are satisfied, False otherwise.

Return type:

bool

remove_constraint(parameter_1: str, operator: str, parameter_2: str) → None[source]

Removes a constraint between 2 parameters (e.g., paramA > paramB)

Parameters:

  • parameter_1 – The name of the first parameter.

  • operator – The operator (e.g. ‘<=’).

  • parameter_2 – The name of the second parameter.

Examples

parameter_set.remove_constraint(‘paramA’, ‘>=’, ‘paramB’)

save_as(directory: str, name: str, file_type: str = 'both')[source]

Create a configuration file containing the parameter values.

Such a file can be used when using the command-line version of hydrobricks. It contains the model parameter values.

Parameters:

  • directory – The directory to write the file.

  • name – The name of the generated file.

  • file_type – The type of file to generate: ‘json’, ‘yaml’, or ‘both’.

set_prior(parameter: str, prior: spotpy.parameter) → None[source]

Set a prior distribution for a parameter.

Assigns a prior probability distribution to a parameter for use in Bayesian calibration methods.

Parameters:

  • parameter – Name (or alias) of the parameter

  • prior – The prior distribution (instance of spotpy.parameter)

Raises:

ImportError – If spotpy is not installed.

set_random_values(parameters: list[str]) → pandas.DataFrame[source]

Set the provided parameter to random values.

Randomly assigns values to specified parameters within their defined ranges. Iterates until all constraints are satisfied.

Parameters:

parameters – The name or alias of the parameters to set to random values. Example: [‘kr’, ‘A’]

Returns:

A dataframe with the assigned parameter values.

Return type:

pd.DataFrame

Raises:

ValueError – If parameter constraints cannot be satisfied after 1000 iterations.

set_values(values: dict, check_range: bool = True, allow_adapt: bool = False) → None[source]

Set the parameter values.

Parameters:

  • values – The values must be provided as a dictionary with the parameter name with the related component or one of its aliases as the key. Example: {‘k’: 32, ‘A’: 300} or {‘slow_reservoir:capacity’: 300}

  • check_range – Check that the parameter value falls into the allowed range.

  • allow_adapt – Allow the parameter values to be adapted to enforce defined constraints (e.g.: min, max).

Forcing

class hydrobricks.Forcing(spatial_entity: HydroUnits | Catchment)[source]

Bases: object

Class for managing forcing (meteorological) data for hydrological models.

class Variable(value)[source]

Bases: StrEnum

Enumeration of supported meteorological variables.

P = 'p'
PET = 'pet'
PRESSURE = 'pressure'
RH = 'rh'
RH_MAX = 'rh_max'
RH_MIN = 'rh_min'
R_NET = 'r_net'
R_SOLAR = 'r_solar'
SD = 'sd'
T = 't'
T_DEW_POINT = 't_dew_point'
T_MAX = 't_max'
T_MIN = 't_min'
WIND = 'wind'
apply_operations(parameters: ParameterSet | None = None, apply_to_all: bool = True) → None[source]

Apply the pre-defined operations.

Executes all spatialization, correction, and PET computation operations that were previously defined. Operations are applied in a fixed order: 1. Prior corrections 2. Station data spatialization 3. Gridded data spatialization 4. PET computation

Parameters:

  • parameters – The parameter object instance. Required if operations reference parameters using the ‘param:’ prefix.

  • apply_to_all – If True, the operations will be applied to all variables. If False, the operations will only be applied to the variables related to parameters defined in the parameters.allow_changing list. This is useful to avoid re-applying, during the calibration phase, operations that have already been applied previously. Default: True

Raises:

ValueError – If operations reference parameters but no parameter object is provided.

compute_pet(**kwargs: Any) → None[source]

Define the PET computation operations using the pyet library. The PET will be computed for all hydro units.

Parameters:

  • method (str) – Name of the method to use. Possible values are those provided in the table from the pyet documentation: https://pypi.org/project/pyet/. The method name or the pyet function name can be used.

  • use (list) – List of the meteorological variables to use to compute the PET. Only the variables listed here will be used. The variables must be named according to pyet naming convention (see pyet API documentation: https://pyet.readthedocs.io/en/latest/api/index.html). These variables must be available (data loaded in forcing) and spatialized. Example: use=[‘t’, ‘tmin’, ‘tmax’, ‘lat’, ‘elevation’]

  • lat (float, optional) – Latitude of the catchment. If not provided, the latitude computed for each hydro unit will be used.

  • options (other) – options will be passed to the pyet function.

Raises:

  • DependencyError – If pyet is not installed.

  • ValueError – If required variables are not available.

correct_station_data(**kwargs: Any) → None[source]

Define prior correction operations to apply to forcing data.

Parameters:

**kwargs –

Keyword arguments defining the correction operation. Required keys:

  • variablestr

    Name of the variable to correct.

  • methodstr

    Correction method: ‘additive’ or ‘multiplicative’

  • correction_factorfloat

    Value of the correction factor (to add or multiply).

Examples

>>> forcing.correct_station_data(
...     variable='temperature',
...     method='additive',
...     correction_factor=0.5  # Add 0.5 degrees
... )
get_total_precipitation() → float[source]

Calculate the catchment-average total precipitation.

Computes the weighted average precipitation across all hydro units based on their areas.

Returns:

Total precipitation in mm (or original units if data is in different units).

Return type:

float

get_variable_enum(variable: str) → Variable[source]

Match a variable name string to the corresponding Variable enum value.

Parameters:

variable – Variable name or alias (e.g., ‘precipitation’, ‘precip’, ‘p’, ‘P’).

Returns:

The corresponding Variable enum value.

Return type:

Variable

Raises:

ValueError – If the variable name is not recognized.

Examples

>>> forcing = Forcing(hydro_units)
>>> var = forcing.get_variable_enum('precip')
>>> var == Forcing.Variable.P
True
is_initialized() → bool[source]

Check if the forcing has been initialized.

Returns:

True if the forcing has been initialized, False otherwise.

Return type:

bool

load_from(path: str | Path) → None[source]

Load data from a netCDF file created using save_as().

Reads a previously saved netCDF file containing spatialized forcing data and loads it into the Forcing object’s data2D structure.

Parameters:

path – Path of the file to read.

Raises:

  • ImportError – If netcdf4 is not installed.

  • ValueError – If the hydro units in the file don’t match the Forcing object’s hydro units.

Notes

The loaded data will have the same structure as if it was created through spatialization operations. The Forcing object will be marked as initialized after successful loading.

load_station_data_from_csv(path: str | Path, column_time: str, time_format: str, content: dict[str, str] | None = None) → None[source]

Read 1D time series data from CSV file for a single station.

Parameters:

  • path – Path to the CSV file containing station data.

  • column_time – Column name containing the time values.

  • time_format – Format string for parsing time values (e.g., ‘%Y-%m-%d’).

  • content – Dictionary mapping variable names/aliases to CSV column names. Example: {‘precipitation’: ‘Precipitation (mm)’, ‘temperature’: ‘Temp (C)’} Default: None

Raises:

  • FileNotFoundError – If the CSV file does not exist.

  • KeyError – If required columns are not found in the CSV file.

Examples

>>> forcing.load_station_data_from_csv(
...     'weather.csv',
...     'Date',
...     '%Y-%m-%d',
...     {'precipitation': 'P (mm)', 'temperature': 'T (C)'}
... )
save_as(path: str | Path, max_compression: bool = False) → None[source]

Create a netCDF file with the forcing data.

Saves the 2D spatialized forcing data to a netCDF4 file with the structure suitable for later loading with load_from().

Parameters:

  • path – Path of the file to create.

  • max_compression – Option to allow maximum compression for data in file. When True, uses compression with least_significant_digit=3 for better storage efficiency. Default: False

Raises:

ImportError – If netcdf4 is not installed.

Notes

If apply_operations() has not been called, it will be called automatically before saving to ensure data is properly spatialized.

spatialize_from_gridded_data(**kwargs: Any) → None[source]

Define the spatialization operations from gridded data to all hydro units.

Parameters:

  • variable (str) – Name of the variable to spatialize.

  • method (str) – Name of the method to use. Can be: * regrid_from_netcdf: regrid data from a single or multiple netCDF files.

  • path (str|Path) – Path to the file containing the data or to a folder containing multiple files.

  • file_pattern (str, optional) – Pattern of the files to read. If None, the path is considered to be a single file.

  • data_crs (int, optional) – CRS (as EPSG id) of the data file. If None, the CRS is read from the file.

  • var_name (str) – Name of the variable to read.

  • dim_time (str) – Name of the time dimension.

  • dim_x (str) – Name of the x dimension.

  • dim_y (str) – Name of the y dimension.

  • raster_hydro_units (str|Path) – Path to a raster containing the hydro unit ids to use for the spatialization.

  • apply_data_gradient (bool, optional) – If True, elevation-based gradients will be retrieved from the data and applied to the hydro units (e.g., for temperature and precipitation). If False, the data will be regridded without applying any gradient. Default is True for temperature and precipitation variables, and False for other variables.

spatialize_from_station_data(**kwargs: Any) → None[source]

Define the spatialization operations from station data to all hydro units.

Parameters:

  • variable (str) – Name of the variable to spatialize.

  • method (str) –

    Name of the method to use. Can be:

    • constant: the same value will be used

    • additive_elevation_gradient: use of an additive elevation gradient that is either constant or changes for every month. Parameters: ‘ref_elevation’, ‘gradient’.

    • multiplicative_elevation_gradient: use of a multiplicative elevation gradient that is either constant or changes for every month. Parameters: ‘ref_elevation’, ‘gradient’.

    • multiplicative_elevation_threshold_gradients: same as multiplicative_elevation_gradient, but with an elevation threshold with a gradient below and a gradient above. Parameters: ‘ref_elevation’, ‘gradient’, ‘gradient_2’, ‘elevation_threshold’

  • ref_elevation (float) – Reference (station) elevation. For method(s): ‘elevation_gradient’

  • gradient (float/list) – Gradient of the variable to apply per 100m (e.g., °C/100m). Can be a unique value or a list providing a value for every month. For method(s): ‘elevation_gradient’, ‘elevation_multi_gradients’

  • gradient_1 (float/list) – Same as parameter ‘gradient’

  • gradient_2 (float/list) – Gradient of the variable to apply per 100m (e.g., °C/100m) for the units above the elevation threshold defined by ‘elevation_threshold’. For method(s): ‘elevation_multi_gradients’

  • elevation_threshold (int/float) – Threshold elevation to switch from gradient to gradient_2. For method(s): ‘elevation_multi_gradients’

Observations

class hydrobricks.Observations[source]

Bases: TimeSeries1D

Class for observation time series data

compute_reference_metric(metric: str, start_date: str | None = None, end_date: str | None = None, with_exclusion: bool = False, mean_discharge: bool = False, all_combinations: bool = False, n_evals: int = 100) → float[source]

Compute a reference for the provided metric (goodness of fit) by block bootstrapping the observed series n_evals times (100 times by default), evaluating the bootstrapped series using the provided metric and computing the mean of the results.

Parameters:

  • metric – The abbreviation of the function as defined in HydroErr (https://hydroerr.readthedocs.io/en/stable/list_of_metrics.html) Examples: ‘nse’, ‘kge_2012’, ‘rmse’, etc.

  • start_date – Start date string for period of interest (format: ‘YYYY-MM-DD’). If None, uses full time series. Default: None

  • end_date – End date string for period of interest (format: ‘YYYY-MM-DD’). If None, uses full time series. Default: None

  • with_exclusion – If True, avoid using the same year’s data for the same position in the bootstrapped sample, ensuring no self-selection for specific years. Default: False

  • mean_discharge – If True, computes the average on the discharge directly rather than on the result of the HydroErr function. Default: False

  • all_combinations – If True uses all combinations possible for the bootstrapping. If False, randomly samples n_evals combinations. Default: False

  • n_evals – Number of random evaluations to perform (ignored if all_combinations=True). Default: 100

Returns:

The mean value of n_evals realizations of the selected metric.

Return type:

float

Raises:

  • DataError – If there is only one year of data (insufficient for block bootstrapping).

  • ValueError – If metric is not recognized or if time series setup is invalid.

Examples

>>> obs = Observations()
>>> obs.load_from_csv('data.csv', 'date', '%Y-%m-%d', {'discharge': 'Q'})
>>> ref_metric = obs.compute_reference_metric('nse', n_evals=100)

Results

class hydrobricks.Results(filename: str)[source]

Bases: object

Class for the detailed results of a model run. This class is used to read the results of a model run (from a netCDF file) and to provide methods to extract the results.

close() → None[source]

Close the netCDF dataset and release the file handle.

get_hydro_units_values(component: str, start_date: str | None = None, end_date: str | None = None) → numpy.ndarray[source]

Get the values of a component at the hydro units.

Retrieves time series or snapshot data for a specific model component distributed across hydro units. Supports optional temporal slicing.

Parameters:

  • component – The name of the component (e.g., ‘snowpack’, ‘soil_moisture’). Use list_hydro_units_components() to see available options.

  • start_date – The start date of the period to extract (format: ‘YYYY-MM-DD’). If None, returns full time series from the beginning. Default: None

  • end_date – The end date of the period to extract (format: ‘YYYY-MM-DD’). If None, returns up to end of time series. Default: None

Returns:

Values of the component at the hydro units. Shape: (n_timesteps, n_hydro_units) for time series, or (n_hydro_units,) for single date if only start_date provided.

Return type:

np.ndarray

Raises:

  • ValueError – If the component is not found in the results.

  • KeyError – If date selection fails or dates are not in the time series.

get_land_cover_areas(land_cover: str) → numpy.ndarray[source]

Get the areas of a land cover across the hydro units.

Calculates the spatial distribution of a specific land cover type across hydro units over time by multiplying the land cover fractions with the hydro unit areas.

Parameters:

land_cover – The name of the land cover type (e.g., ‘glacier’, ‘ground’, ‘forest’).

Returns:

Areas of the land cover across the hydro units (2D array: time × units). Units match the hydro unit area units (typically m² or km²).

Return type:

np.ndarray

Raises:

  • ValueError – If the land cover is not found in the results.

  • IndexError – If labels_land_cover is None or empty.

get_mean_hydro_units_values(land_cover: str, component: str, start_date: str | None = None, end_date: str | None = None) → numpy.ndarray[source]

Get the mean values of a component across the hydro units weighted by land cover area.

Computes area-weighted average of a component for a specific land cover type, accounting for spatial variation in land cover distribution across hydro units.

Parameters:

  • land_cover – The name of the land cover type to weight by (e.g., ‘glacier’, ‘ground’, ‘forest’).

  • component – The name of the component (e.g., ‘snowpack’, ‘soil_moisture’). Use list_hydro_units_components() to see available options.

  • start_date – The start date of the period to extract (format: ‘YYYY-MM-DD’). If None, returns full time series. Default: None

  • end_date – The end date of the period to extract (format: ‘YYYY-MM-DD’). If None, returns up to end of time series. Default: None

Returns:

Weighted mean values of the component across the hydro units (1D time series). Weights are based on the land cover area in each hydro unit.

Return type:

np.ndarray

Raises:

ValueError – If the land cover or component is not found in the results.

get_mean_swe(start_date: str | None = None, end_date: str | None = None) → numpy.ndarray[source]

Get the mean snow water equivalent (SWE) across the hydro units weighted by land cover.

Computes the catchment-wide average snow water equivalent by aggregating SWE values across all land cover types and hydro units, weighted by their respective areas.

Parameters:

  • start_date – The start date of the period to extract (format: ‘YYYY-MM-DD’). If None, returns full time series. Default: None

  • end_date – The end date of the period to extract (format: ‘YYYY-MM-DD’). If None, returns up to end of time series. Default: None

Returns:

Mean SWE across the hydro units (1D time series, units: mm water equivalent).

Return type:

np.ndarray

Raises:

  • ValueError – If SWE is not found in the component labels.

  • KeyError – If snowpack component data is not available for any land cover.

get_time_array(start_date: str, end_date: str) → numpy.ndarray[source]

Get the time array.

Extracts the time coordinates from the results dataset for a specified date range. Useful for creating time-aligned arrays for plotting or analysis.

Parameters:

  • start_date – The start date of the period to extract (format: ‘YYYY-MM-DD’).

  • end_date – The end date of the period to extract (format: ‘YYYY-MM-DD’).

Returns:

Array of time values (typically datetime64) for the specified period.

Return type:

np.ndarray

Raises:

KeyError – If dates are not found in the results time coordinates.

list_hydro_units_components() → None[source]

Print a list of the distributed (hydro unit level) components.

Displays all component names that have values distributed across individual hydro units. These are typically state variables like snowpack, soil moisture, or groundwater storage.

list_sub_basin_components() → None[source]

Print a list of the aggregated (sub-basin level) components.

Displays all component names that have aggregated values at the sub-basin level. These are typically fluxes or flows that are summed across the entire catchment (e.g., total runoff, evapotranspiration).

TimeSeries

class hydrobricks.TimeSeries[source]

Bases: object

Class for generic time series data

get_dates_as_mjd() → float | np.ndarray[source]

Convert time series dates to modified Julian dates.

Returns:

Modified Julian dates. Returns float if single date, array if multiple dates.

Return type:

float | np.ndarray

Trainer

class hydrobricks.trainer.SpotpySetup(model: Model | list[Model], params: ParameterSet, forcing: Forcing | list[Forcing], obs: Observations | list[Observations], warmup: int = 365, obj_func: str | Callable[[np.ndarray, np.ndarray], float] | None = None, invert_obj_func: bool = False, dump_outputs: bool = False, dump_forcing: bool = False, dump_dir: str = '')[source]

Bases: object

Setup class for SPOTPY optimization framework integration.

evaluation() → list[numpy.ndarray][source]
objectivefunction(simulation: list[np.ndarray], evaluation: list[np.ndarray], params: spotpy.parameter | None = None) → float[source]
parameters() → Any[source]

Generate random parameter sets that satisfy constraints.

Returns:

SPOTPY parameter object with valid random parameter values.

Return type:

Any

Raises:

RuntimeError – If unable to generate valid parameters after 1000 attempts.

simulation(x: spotpy.parameter) → list[np.ndarray] | None[source]

Run model simulation with given parameter set.

Parameters:

x – SPOTPY parameter object containing parameter values.

Returns:

List of simulated discharge time series (one per model), or None if simulation failed or constraints were violated.

Return type:

list[np.ndarray] | None