User API

Execution

Main module.

class ravenpy.ravenpy.Emulator(config: Config, workdir: str | PathLike | None = None, modelname: str | None = None, overwrite: bool = False)

property config: Config

Read-only model configuration.

property modelname: str

File name stem of configuration files.

property output: OutputReader

Simulation output object.

property output_path: Path | None

Path to model outputs.

resume(timestamp: bool = True) → Config

Return new model configuration using state variables from the end of the run.

Parameters:

timestamp (bool) – If False, ignore time stamp information in the solution. If True, the solution will set StartDate to the solution’s timestamp.

run(overwrite: bool = False) → OutputReader

Run the model.

This will write RV files if not already done.

Parameters:

overwrite (bool) – If True, overwrite existing files.

property workdir: Path

Path to RV files and output subdirectory.

class ravenpy.ravenpy.EnsembleReader(*, run_name: str | None = None, paths: list[str | PathLike] | None = None, runs: list[OutputReader] | None = None, dim: str = 'member')

property files

property hydrograph

property storage

class ravenpy.ravenpy.OutputReader(run_name: str | None = None, path: str | Path | None = None)

property diagnostics: dict | None

Model diagnostics.

property files: dict

Report paths to output files.

property hydrograph: Dataset

Generate the hydrograph.

property messages: str | None

property path: Path

Path to output directory.

property solution: dict | None

Solution file content.

property storage: Dataset

Storage variables.

exception ravenpy.ravenpy.RavenError

RavenError exception class.

An error that is meant to be raised whenever a message of type “ERROR” is found in the Raven_errors.txt file resulting from a Raven (i.e. the C program) run.

exception ravenpy.ravenpy.RavenWarning

RavenWarning warning class.

A warning corresponding to a message of type “WARNING” in the Raven_errors.txt file resulting from a Raven (i.e. the C program) run.

ravenpy.ravenpy.run(modelname: str, configdir: str | Path, outputdir: str | Path | None = None, overwrite: bool = True, verbose: bool = False) → Path

Run Raven given the path to an existing model configuration.

Parameters:

• modelname (str) – Configuration files stem, i.e. the file name without extension.

• configdir (Path or str) – Path to configuration files directory.

• outputdir (Path or str, optional) – Path to model simulation output. If None, will write to configdir/output.

• overwrite (bool) – If True, overwrite existing files.

• verbose (bool) – If True, always display Raven warnings. If False, warnings will only be printed if an error occurs.

Returns:

The path to the model outputs.

Return type:

Path

Configuration

class ravenpy.config.commands.AssimilateStreamflow(*, sb_id: str)

Subbasin ID to assimilate streamflow for.

model_config = {'arbitrary_types_allowed': True, 'coerce_numbers_to_str': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sb_id: str

class ravenpy.config.commands.AssimilatedState(*, state: Literal['ATMOSPHERE', 'ATMOS_PRECIP', 'CANOPY', 'CANOPY_SNOW', 'CANOPY_TEMP', 'COLD_CONTENT', 'CONSTITUENT', 'CONSTITUENT_SINK', 'CONSTITUENT_SRC', 'CONSTITUENT_SW', 'CONVOLUTION', 'CONV_STOR', 'CROP_HEAT_UNITS', 'CUM_INFIL', 'CUM_SNOWMELT', 'DEPRESSION', 'GLACIER', 'GLACIER_CC', 'GLACIER_ICE', 'GROUNDWATER', 'LAKE_STORAGE', 'MULTIPLE', 'PERMAFROST_DEPTH', 'PONDED_WATER', 'ROOT', 'SNOW', 'SNOW_AGE', 'SNOW_ALBEDO', 'SNOW_COVER', 'SNOW_DEPTH', 'SNOW_LIQ', 'SNOW_TEMP', 'SOIL', 'SOIL[0]', 'SOIL[1]', 'SOIL[2]', 'SOIL_TEMP', 'SURFACE_WATER', 'SURFACE_WATER_TEMP', 'TOTAL_SWE', 'TRUNK', 'WETLAND', 'STREAMFLOW'], group: str)

group: str

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

state: Literal['ATMOSPHERE', 'ATMOS_PRECIP', 'CANOPY', 'CANOPY_SNOW', 'CANOPY_TEMP', 'COLD_CONTENT', 'CONSTITUENT', 'CONSTITUENT_SINK', 'CONSTITUENT_SRC', 'CONSTITUENT_SW', 'CONVOLUTION', 'CONV_STOR', 'CROP_HEAT_UNITS', 'CUM_INFIL', 'CUM_SNOWMELT', 'DEPRESSION', 'GLACIER', 'GLACIER_CC', 'GLACIER_ICE', 'GROUNDWATER', 'LAKE_STORAGE', 'MULTIPLE', 'PERMAFROST_DEPTH', 'PONDED_WATER', 'ROOT', 'SNOW', 'SNOW_AGE', 'SNOW_ALBEDO', 'SNOW_COVER', 'SNOW_DEPTH', 'SNOW_LIQ', 'SNOW_TEMP', 'SOIL', 'SOIL[0]', 'SOIL[1]', 'SOIL[2]', 'SOIL_TEMP', 'SURFACE_WATER', 'SURFACE_WATER_TEMP', 'TOTAL_SWE', 'TRUNK', 'WETLAND', 'STREAMFLOW']

class ravenpy.config.commands.BasinIndex(*, sb_id: int = 1, name: str = 'watershed', ChannelStorage: float = 0.0, RivuletStorage: float = 0.0, Qout: Sequence[float] = (1.0, 0.0, 0.0), Qlat: Sequence[float] | None = None, Qin: Sequence[float] | None = None)

Initial conditions for a flow segment.

channel_storage: float

model_config = {'arbitrary_types_allowed': True, 'coerce_numbers_to_str': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

classmethod parse(s)

qin: Sequence[float] | None

qlat: Sequence[float] | None

qout: Sequence[float]

rivulet_storage: float

sb_id: int

class ravenpy.config.commands.BasinStateVariables(root: RootModelRootType = PydanticUndefined)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod parse(sol)

root: Sequence[BasinIndex]

class ravenpy.config.commands.ChannelProfile(*, name: str = 'chn_XXX', Bedslope: float = 0, SurveyPoints: SurveyPoints = SurveyPoints(root=(SurveyPoint(root=()),)), RoughnessZones: RoughnessZones = RoughnessZones(root=(RoughnessZone(root=()),)))

ChannelProfile command (RVP).

class RoughnessZones(root: RootModelRootType = PydanticUndefined)

RoughnessZones record.

[x_zone, mannings_n] x number of roughness zones.

class RoughnessZone(root: RootModelRootType = PydanticUndefined)

RoughnessZone record.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: tuple[float, float]

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Sequence[RoughnessZone]

class SurveyPoints

[x, bed_elevation] x number of survey points.

class SurveyPoint(root: RootModelRootType = PydanticUndefined)

SurveyPoint record.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: tuple[float, float]

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Sequence[SurveyPoint]

bed_slope: float

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

classmethod parse(s) → list

Parse ChannelProfile commands and return a list of ChannelProfile records.

roughness_zones: RoughnessZones

survey_points: SurveyPoints

class ravenpy.config.commands.CustomOutput(*, time_per: Literal['DAILY', 'MONTHLY', 'YEARLY', 'WATER_YEARLY', 'CONTINUOUS'], stat: Literal['AVERAGE', 'MAXIMUM', 'MINIMUM', 'RANGE', 'MEDIAN', 'QUARTILES'], variable: str, space_agg: Literal['BY_BASIN', 'BY_HRU', 'BY_HRU_GROUP', 'BY_SB_GROUP', 'ENTIRE_WATERSHED'], filename: str = '')

Create custom output file to track a single variable, parameter, or forcing function over time at a number of basins, HRUs, or across watershed.

:CustomOutput DAILY AVERAGE AET BY_HRU

filename: str

Output file name.

Defaults to something approximately like: <run name>_<variable>_<time_per>_<stat>_<space_agg>.nc.

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

space_agg: Literal['BY_BASIN', 'BY_HRU', 'BY_HRU_GROUP', 'BY_SB_GROUP', 'ENTIRE_WATERSHED']

Spatial evaluation domain.

stat: Literal['AVERAGE', 'MAXIMUM', 'MINIMUM', 'RANGE', 'MEDIAN', 'QUARTILES']

Statistic reported for each time interval.

time_per: Literal['DAILY', 'MONTHLY', 'YEARLY', 'WATER_YEARLY', 'CONTINUOUS']

Time period.

variable: str

Variable or parameter name.

Consult the Raven documentation for the list of allowed names.

class ravenpy.config.commands.Data(*, data_type: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL'] = '', units: str = '', ReadFromNetCDF: ReadFromNetCDF)

data_type: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL']

classmethod from_nc(fn, data_type, station_idx=1, alt_names=(), **kwds)

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

read_from_netcdf: ReadFromNetCDF

units: str

class ravenpy.config.commands.EnsembleMode(*, mode: Literal['ENSEMBLE_ENKF'] = 'ENSEMBLE_ENKF', n: int)

mode: Literal['ENSEMBLE_ENKF']

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

n: int

class ravenpy.config.commands.EvaluationPeriod(*, name: str, start: date, end: date)

:EvaluationPeriod [period_name] [start yyyy-mm-dd] [end yyyy-mm-dd]

end: date

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

start: date

class ravenpy.config.commands.ForcingPerturbation(*, forcing: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL'], dist: Literal['DIST_UNIFORM', 'DIST_NORMAL', 'DIST_GAMMA'], p1: float, p2: float, adj: Literal['ADDITIVE', 'MULTIPLICATIVE'], hru_grp: str = '')

adj: Literal['ADDITIVE', 'MULTIPLICATIVE']

dist: Literal['DIST_UNIFORM', 'DIST_NORMAL', 'DIST_GAMMA']

forcing: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL']

hru_grp: str

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p1: float

p2: float

class ravenpy.config.commands.Gauge(*, name: str = 'default', Latitude: float, Longitude: float, Elevation: float | None = None, RainCorrection: Variable | ExpressionNode | float | None = None, SnowCorrection: Variable | ExpressionNode | float | None = None, MonthlyAveEvaporation: Sequence | None = None, MonthlyAveTemperature: Sequence | None = None, MonthlyMinTemperature: Sequence | None = None, MonthlyMaxTemperature: Sequence | None = None, Data: Sequence[Data] | None = None)

classmethod confirm_monthly(v)

data: Sequence[Data] | None

property ds: Dataset

Create a Dataset with forcing variables keyed by Raven forcing names.

elevation: float | None

classmethod from_nc(fn: str | Path | Sequence[Path], data_type: Sequence[str] | None = None, station_idx: int = 1, alt_names: dict[str, str] | None = None, mon_ave: bool = False, data_kwds: dict[str, Any] | None = None, engine: str = 'h5netcdf', **kwds) → Gauge

Return Gauge instance with configuration options inferred from the netCDF itself.

Parameters:

• fn (str or Path or Sequence[Path]) – NetCDF file path or paths.

• data_type (Sequence[str], optional) – Raven data types to extract from netCDF files, e.g. ‘PRECIP’, ‘AVE_TEMP’. The algorithm tries to find all forcings in each file until one is found, then it stops searching for it in the following files.

• station_idx (int) – Index along station dimension. Starts at 1. Should be the same for all netCDF files.

• alt_names (dict) – Alternative variable names keyed by data type. Use this if variables do not correspond to CF standard defaults.

• mon_ave (bool) – If True, compute the monthly average.

• data_kwds (dict[options.Forcings, dict[str, str]]) – Additional :Data parameters keyed by forcing type and station id. Overrides inferred parameters. Use keyword “ALL” to pass parameters to all variables.

• engine ({"h5netcdf", "netcdf4", "pydap"}) – The engine used to open the dataset. Default is ‘h5netcdf’.

• **kwds (dict) – Additional arguments for Gauge.

Returns:

Gauge instance.

Return type:

Gauge

latitude: float

longitude: float

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

monthly_ave_evaporation: Sequence | None

monthly_ave_temperature: Sequence | None

monthly_max_temperature: Sequence | None

monthly_min_temperature: Sequence | None

name: str

rain_correction: Variable | ExpressionNode | float | None

snow_correction: Variable | ExpressionNode | float | None

class ravenpy.config.commands.GridWeights(*, NumberHRUs: int = 1, NumberGridCells: int = 1, data: Sequence[GWRecord] = (GWRecord(root=(1, 0, 1.0)),))

GridWeights command.

Notes

command can be embedded in both a GriddedForcing or a StationForcing.

The default is to have a single cell that covers an entire single HRU, with a weight of 1.

class GWRecord(root: RootModelRootType = PydanticUndefined)

GridWeights record.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: tuple[int, int, float]

data: Sequence[GWRecord]

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

number_grid_cells: int

number_hrus: int

classmethod parse(s)

class ravenpy.config.commands.GriddedForcing(*, FileNameNC: HttpUrl | Path, VarNameNC: str, DimNamesNC: Sequence[str], station_idx: int | None = None, TimeShift: float | None = None, LinearTransform: LinearTransform | None = None, Deaccumulate: bool | None = None, LatitudeVarNameNC: str | None = None, LongitudeVarNameNC: str | None = None, ElevationVarNameNC: str | None = None, name: str = '', ForcingType: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL'] | None = None, GridWeights: GridWeights | RedirectToFile = GridWeights(number_hrus=1, number_grid_cells=1, data=(GWRecord(root=(1, 0, 1.0)),)))

GriddedForcing command (RVT).

classmethod check_dims(v)

forcing_type: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL'] | None

grid_weights: GridWeights | RedirectToFile

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

station_idx: int | None

class ravenpy.config.commands.HRU(*, hru_id: Annotated[int, Gt(gt=0)] = 1, area: Variable | ExpressionNode | float | None = 0, elevation: float = 0, latitude: float = 0, longitude: float = 0, subbasin_id: Annotated[int, Gt(gt=0)] = 1, land_use_class: str = '[NONE]', veg_class: str = '[NONE]', soil_profile: str = '[NONE]', aquifer_profile: str = '[NONE]', terrain_class: str = '[NONE]', slope: Annotated[float, Ge(ge=0)] = 0.0, aspect: Annotated[float, Ge(ge=0), Le(le=360)] = 0.0, hru_type: str | None = None)

Record to populate :HRUs command internal table (RVH).

aquifer_profile: str

area: Variable | ExpressionNode | float | None

aspect: Annotated[float, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=0), Le(le=360)])]

elevation: float

hru_id: Annotated[int, Gt(gt=0)]

hru_type: str | None

land_use_class: str

latitude: float

longitude: float

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

slope: Annotated[float, Ge(ge=0)]

soil_profile: str

subbasin_id: Annotated[int, Gt(gt=0)]

terrain_class: str

veg_class: str

class ravenpy.config.commands.HRUGroup(*, name: str, groups: _Rec)

groups: _Rec

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

class ravenpy.config.commands.HRUState(*, hru_id: int = 1, data: dict[str, ~pymbolic.primitives.Variable | ~pymbolic.primitives.ExpressionNode | float | None]=<factory>)

data: dict[str, Variable | ExpressionNode | float | None]

hru_id: int

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod parse(sol, names=None)

class ravenpy.config.commands.HRUStateVariableTable(root: RootModelRootType = PydanticUndefined)

Table of HRU state variables.

If the HRUState include different attributes, the states will be modified to include all attributes.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

classmethod parse(sol: str)

root: Sequence[HRUState]

set_attributes()

class ravenpy.config.commands.HRUs(root: RootModelRootType = PydanticUndefined)

HRUs command (RVH).

classmethod ignore_unrecognized_hrus(values)

Ignore HRUs with unrecognized hru_type.

HRUs are ignored only if all allowed HRU classes define hru_type, and if the values passed include it.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

classmethod parse(s)

Parse a HRUs command.

rename(mapping)

Rename HRU attributes.

Parameters:

mapping (dict) – Nested dictionary keyed by HRU attributes, with keys as old names and values as new names.

root: Sequence[HRU]

ravenpy.config.commands.LOGGER = <Logger RavenPy (WARNING)>

Raven commands

The syntax of those commands match as closely as possible the Raven documentation.

ravenpy.config.commands.LU

alias of LandUseClass

class ravenpy.config.commands.LandUseClass(*, name: str = '', impermeable_frac: Variable | ExpressionNode | float | None = 0.0, forest_coverage: Variable | ExpressionNode | float | None = 0.0)

forest_coverage: Variable | ExpressionNode | float | None

impermeable_frac: Variable | ExpressionNode | float | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

class ravenpy.config.commands.LandUseClasses(root: RootModelRootType = PydanticUndefined)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

root: Sequence[LandUseClass]

class ravenpy.config.commands.LandUseParameterList(*, Parameters: Sequence[Literal['ABST_PERCENT', 'AET_COEFF', 'BF_LOSS_FRACTION', 'B_EXP', 'CC_DECAY_COEFF', 'DD_AGGRADATION', 'DD_MELT_TEMP', 'DD_REFREEZE_TEMP', 'DEP_K', 'DEP_MAX', 'DEP_MAX_FLOW', 'DEP_N', 'DEP_SEEP_K', 'DEP_THRESHOLD', 'FOREST_COVERAGE', 'FOREST_PET_CORR', 'FOREST_SPARSENESS', 'GAMMA_SCALE', 'GAMMA_SCALE2', 'GAMMA_SHAPE', 'GAMMA_SHAPE2', 'GLAC_STORAGE_COEFF', 'GR4J_X4', 'HBV_GLACIER_AG', 'HBV_GLACIER_KMIN', 'HBV_MELT_ASP_CORR', 'HBV_MELT_FOR_CORR', 'HBV_MELT_GLACIER_CORR', 'HMETS_RUNOFF_COEFF', 'IMPERMEABLE_FRAC', 'LAKE_PET_CORR', 'LAKE_REL_COEFF', 'MAX_DEP_AREA_FRAC', 'MAX_MELT_FACTOR', 'MAX_SAT_AREA_FRAC', 'MELT_FACTOR', 'MIN_MELT_FACTOR', 'OW_PET_CORR', 'PARTITION_COEFF', 'PDMROF_B', 'PONDED_EXP', 'REFREEZE_EXP', 'REFREEZE_FACTOR', 'ROUGHNESS', 'SCS_CN', 'SCS_IA_FRACTION', 'SNOW_PATCH_LIMIT', 'STREAM_FRACTION', 'UBC_ICEPT_FACTOR']] | None = None, Units: Sequence[str] | None = None, pl: Sequence[ParameterList])

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: Sequence[Literal['ABST_PERCENT', 'AET_COEFF', 'BF_LOSS_FRACTION', 'B_EXP', 'CC_DECAY_COEFF', 'DD_AGGRADATION', 'DD_MELT_TEMP', 'DD_REFREEZE_TEMP', 'DEP_K', 'DEP_MAX', 'DEP_MAX_FLOW', 'DEP_N', 'DEP_SEEP_K', 'DEP_THRESHOLD', 'FOREST_COVERAGE', 'FOREST_PET_CORR', 'FOREST_SPARSENESS', 'GAMMA_SCALE', 'GAMMA_SCALE2', 'GAMMA_SHAPE', 'GAMMA_SHAPE2', 'GLAC_STORAGE_COEFF', 'GR4J_X4', 'HBV_GLACIER_AG', 'HBV_GLACIER_KMIN', 'HBV_MELT_ASP_CORR', 'HBV_MELT_FOR_CORR', 'HBV_MELT_GLACIER_CORR', 'HMETS_RUNOFF_COEFF', 'IMPERMEABLE_FRAC', 'LAKE_PET_CORR', 'LAKE_REL_COEFF', 'MAX_DEP_AREA_FRAC', 'MAX_MELT_FACTOR', 'MAX_SAT_AREA_FRAC', 'MELT_FACTOR', 'MIN_MELT_FACTOR', 'OW_PET_CORR', 'PARTITION_COEFF', 'PDMROF_B', 'PONDED_EXP', 'REFREEZE_EXP', 'REFREEZE_FACTOR', 'ROUGHNESS', 'SCS_CN', 'SCS_IA_FRACTION', 'SNOW_PATCH_LIMIT', 'STREAM_FRACTION', 'UBC_ICEPT_FACTOR']] | None

class ravenpy.config.commands.LinearTransform(*, scale: float = 1, offset: float = 0)

:LinearTransform 1.0 -273.15

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

offset: float

scale: float

to_rv()

Return Raven configuration string.

class ravenpy.config.commands.ObservationData(*, data_type: Literal['HYDROGRAPH'] = 'HYDROGRAPH', units: str = '', ReadFromNetCDF: ReadFromNetCDF, uid: str = '1')

data_type: Literal['HYDROGRAPH']

classmethod from_nc(fn, station_idx: int = 1, alt_names=(), engine='h5netcdf', **kwds)

model_config = {'arbitrary_types_allowed': True, 'coerce_numbers_to_str': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

uid: str

class ravenpy.config.commands.ObservationErrorModel(*, state: Literal['STREAMFLOW'], dist: Literal['DIST_UNIFORM', 'DIST_NORMAL', 'DIST_GAMMA'], p1: float, p2: float, adj: Literal['ADDITIVE', 'MULTIPLICATIVE'])

adj: Literal['ADDITIVE', 'MULTIPLICATIVE']

dist: Literal['DIST_UNIFORM', 'DIST_NORMAL', 'DIST_GAMMA']

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p1: float

p2: float

state: Literal['STREAMFLOW']

class ravenpy.config.commands.Process(*, algo: str = 'RAVEN_DEFAULT', source: str | None = None, to: Sequence[str] = ())

Process type embedded in HydrologicProcesses command.

See processes.py for list of processes.

algo: str

classmethod is_list(v)

classmethod is_source_state_variable(v: str)

classmethod is_to_state_variable(v: Sequence)

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

source: str | None

to: Sequence[str]

to_rv()

Return Raven configuration string.

class ravenpy.config.commands.RainSnowTransition(*, temp: Variable | ExpressionNode | float | None, delta: Variable | ExpressionNode | float | None)

Specify the range of temperatures over which there will be a rain/snow mix when partitioning total precipitation into rain/snow components.

:RainSnowTransition [temp] [delta]

# equivalent to (the preferred option) :GlobalParameter RAINSNOW_TEMP [rainsnow_temp] :GlobalParameter RAINSNOW_DELTA [rainsnow_delta]

delta: Variable | ExpressionNode | float | None

Range [C].

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

temp: Variable | ExpressionNode | float | None

Midpoint of the temperature range [C].

to_rv()

Return Raven configuration string.

class ravenpy.config.commands.ReadFromNetCDF(*, FileNameNC: HttpUrl | Path, VarNameNC: str, DimNamesNC: Sequence[str], StationIdx: int = 1, TimeShift: float | None = None, LinearTransform: LinearTransform | None = None, Deaccumulate: bool | None = None, LatitudeVarNameNC: str | None = None, LongitudeVarNameNC: str | None = None, ElevationVarNameNC: str | None = None)

property da: DataArray

Create a DataArray from configuration.

deaccumulate: bool | None

dim_names_nc: Sequence[str]

elevation_var_name_nc: str | None

file_name_nc: HttpUrl | Path

classmethod from_nc(fn, data_type, station_idx=None, alt_names=(), engine='h5netcdf', **kwds)

Instantiate class from netCDF dataset.

latitude_var_name_nc: str | None

linear_transform: LinearTransform | None

longitude_var_name_nc: str | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod reorder_time(v)

Return dimensions as x, y, t.

This is a fail safe because if input files are CF-compliant, dimensions should already have been ordered by nc_specs.

station_idx: int

time_shift: float | None

var_name_nc: str

class ravenpy.config.commands.RedirectToFile(root: RootModelRootType = PydanticUndefined)

RedirectToFile command (RVT).

Notes

For the moment, this command can only be used in the context of a GriddedForcingCommand or a StationForcingCommand, as a grid_weights field replacement when inlining is not desired.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Annotated[Path, PathType(path_type=file)]

to_rv()

Return Raven configuration string.

class ravenpy.config.commands.Reservoir(*, name: str = 'Lake_XXX', SubBasinID: int = 0, HRUID: int = 0, Type: str = 'RESROUTE_STANDARD', WeirCoefficient: float = 0, CrestWidth: float = 0, MaxDepth: float = 0, LakeArea: float = 0, AbsoluteCrestHeight: float | None = None, MaxCapacity: float | None = None, SeepageParameters: SeepageParameters | None = None, StageRelations: StageRelations | None = None)

Reservoir command (RVH).

class OutflowControlStructure(*, TargetSubBasin: int | None = None, DownstreamReferenceElevation: float | None = None, StageDischargeTable: Sequence[StageDischargeTable] | None = None, BasicWeir: BasicWeir = None, OperatingRegime: OperatingRegime = None)

Outflow control structure for the reservoir.

class BasicWeir(*, curve_name: str, elev: float, crest_width: float, coeff: float)

Basic weir for the outflow control structure.

:BasicWeir [curve_name3] [elev] [crestwidth] [coeff]

coeff: float

crest_width: float

curve_name: str

elev: float

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OperatingRegime(*, name: str, UseCurve: str = None, Condition: str = None, Constraint: str = None)

Operating regime for the outflow control structure.

:OperatingRegime [curve_name] [elev] [coeff]

condition: str

constraint: str

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

use_curve: str

class StageDischargeTable(root: RootModelRootType = PydanticUndefined)

Stage discharge table for the outflow control structure.

Example:

:StageDischargeTable C1 #one gate open
      N
      [h,Q]xN
:EndStageDischargeTable

class StageDischargeRecord(root: RootModelRootType = PydanticUndefined)

Stage discharge record.

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: tuple[float, float]

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

root: Sequence[StageDischargeRecord]

basic_weir: BasicWeir

downstream_reference_elevation: float | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

operating_regime: OperatingRegime

stage_discharge_table: Sequence[StageDischargeTable] | None

target_subbasin_id: int | None

class SeepageParameters(*, k_seep: float, h_ref: float)

:SeepageParameters [K_seep] [href]

h_ref: float

k_seep: float

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class StageRelations(root: RootModelRootType = PydanticUndefined)

Stage relations for the reservoir.

class StageRelation(root: RootModelRootType = PydanticUndefined)

Stage relation record.

h, q, v, a, [u]

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: tuple[float, float, float, float] | tuple[float, float, float, float, float]

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Sequence[StageRelation]

absolute_crest_height: float | None

crest_width: float

hru_id: int

lake_area: float

max_capacity: float | None

max_depth: float

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

classmethod parse(s) → list

Return a list of Reservoir records parsed from an RV file.

seepage_parameters: SeepageParameters | None

stage_relations: StageRelations | None

subbasin_id: int

type: str

weir_coefficient: float

ravenpy.config.commands.SB

alias of SubBasin

class ravenpy.config.commands.SBGroupPropertyMultiplier(*, group_name: str, parameter_name: str, mult: Variable | ExpressionNode | float | None)

:SBGroupPropertyMultiplier [group_name] [parameter_name] [mult]

group_name: str

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

mult: Variable | ExpressionNode | float | None

parameter_name: str

ravenpy.config.commands.SC

alias of SoilClass

ravenpy.config.commands.SP

alias of SoilProfile

class ravenpy.config.commands.SeasonalRelativeHeight(root: RootModelRootType = PydanticUndefined)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Sequence[_MonthlyRecord]

class ravenpy.config.commands.SeasonalRelativeLAI(root: RootModelRootType = PydanticUndefined)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Sequence[_MonthlyRecord]

class ravenpy.config.commands.SoilClasses(root: RootModelRootType = PydanticUndefined)

SoilClasses command.

Example:

:SoilClasses
  :Attributes, %SAND, %CLAY, %SILT, %ORGANIC
  :Units, none, none, none, none
  SAND, 1, 0, 0, 0
  LOAM, 0.5, 0.1, 0.4, 0.4
:EndSoilClasses

class SoilClass(*, name: str, mineral: tuple[float, float, float] | None = None, organic: float | None = None)

SoilClass.

mineral: tuple[float, float, float] | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

organic: float | None

classmethod validate_mineral(v)

Assert sum of mineral fraction is 1.

classmethod validate_mineral_pct(v: tuple)

classmethod validate_organic_pct(v: float)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

root: Sequence[SoilClass]

class ravenpy.config.commands.SoilModel(root: RootModelRootType = PydanticUndefined)

:SoilModel SOIL_MULTILAYER 6

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: int

to_rv()

Return Raven configuration string.

class ravenpy.config.commands.SoilParameterList(*, Parameters: Sequence[Literal['AIR_ENTRY_PRESSURE', 'ALBEDO_DRY', 'ALBEDO_WET', 'BASEFLOW_COEFF', 'BASEFLOW_COEFF2', 'BASEFLOW_N', 'BASEFLOW_THRESH', 'BF_LOSS_FRACTION', 'BULK_DENSITY', 'B_EXP', 'CLAPP N,CLAPP M', 'CLAPP_B', 'CLAY_CON', 'EVAP_RES_FC', 'FIELD_CAPACITY', 'GR4J_X2', 'GR4J_X3', 'HBV_BETA', 'HEAT_CAPACITY', 'HYDRAUL_COND', 'INTERFLOW_COEF', 'MAX_BASEFLOW_RATE', 'MAX_CAP_RISE_RATE', 'MAX_INTERFLOW_RATE', 'MAX_PERC_RATE', 'ORG_CON', 'PERC_COEFF', 'PERC_N', 'PET_CORRECTION', 'POROSITY', 'SAC_PERC_ALPHA', 'SAC_PERC_EXPON', 'SAC_PERC_PFREE', 'SAND_CON', 'SAT_RES', 'SAT_WILT', 'SHUTTLEWORTH_B', 'SILT_CON', 'STONE_FRAC', 'STORAGE_THRESHOLD', 'THERMAL_COND', 'UBC_EVAL_SOIL_DEF', 'UBC_INFIL_SOIL_DEF', 'UNAVAIL_FRAC', 'VIC ALPHA', 'VIC_EVAP_GAMMA', 'VIC_ZMAX', 'VIZ_ZMIN', 'WETTING_FRONT_PSI', 'WILTING_PRESSURE']] | None = None, Units: Sequence[str] | None = None, pl: Sequence[ParameterList])

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: Sequence[Literal['AIR_ENTRY_PRESSURE', 'ALBEDO_DRY', 'ALBEDO_WET', 'BASEFLOW_COEFF', 'BASEFLOW_COEFF2', 'BASEFLOW_N', 'BASEFLOW_THRESH', 'BF_LOSS_FRACTION', 'BULK_DENSITY', 'B_EXP', 'CLAPP N,CLAPP M', 'CLAPP_B', 'CLAY_CON', 'EVAP_RES_FC', 'FIELD_CAPACITY', 'GR4J_X2', 'GR4J_X3', 'HBV_BETA', 'HEAT_CAPACITY', 'HYDRAUL_COND', 'INTERFLOW_COEF', 'MAX_BASEFLOW_RATE', 'MAX_CAP_RISE_RATE', 'MAX_INTERFLOW_RATE', 'MAX_PERC_RATE', 'ORG_CON', 'PERC_COEFF', 'PERC_N', 'PET_CORRECTION', 'POROSITY', 'SAC_PERC_ALPHA', 'SAC_PERC_EXPON', 'SAC_PERC_PFREE', 'SAND_CON', 'SAT_RES', 'SAT_WILT', 'SHUTTLEWORTH_B', 'SILT_CON', 'STONE_FRAC', 'STORAGE_THRESHOLD', 'THERMAL_COND', 'UBC_EVAL_SOIL_DEF', 'UBC_INFIL_SOIL_DEF', 'UNAVAIL_FRAC', 'VIC ALPHA', 'VIC_EVAP_GAMMA', 'VIC_ZMAX', 'VIZ_ZMIN', 'WETTING_FRONT_PSI', 'WILTING_PRESSURE']] | None

class ravenpy.config.commands.SoilProfile(*, name: str = '', soil_classes: Sequence[str] = (), thicknesses: Sequence[Variable | ExpressionNode | float | None] = ())

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

soil_classes: Sequence[str]

thicknesses: Sequence[Variable | ExpressionNode | float | None]

class ravenpy.config.commands.SoilProfiles(root: RootModelRootType = PydanticUndefined)

SoilProfiles command.

Example:

:SoilProfiles
  # name, #horizons, hor1, th1, hor2, th2
  LAKE, 0
  GLACIER, 0
  LOAM_SEQ, 2, LOAM, 0.5, SAND, 1.5
  ALL_SAND, 2, SAND, 0.5, SAND, 1.5
:EndSoilProfiles

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: Sequence[SoilProfile]

class ravenpy.config.commands.StationForcing(*, FileNameNC: HttpUrl | Path, VarNameNC: str, DimNamesNC: Sequence[str], station_idx: int | None = None, TimeShift: float | None = None, LinearTransform: LinearTransform | None = None, Deaccumulate: bool | None = None, LatitudeVarNameNC: str | None = None, LongitudeVarNameNC: str | None = None, ElevationVarNameNC: str | None = None, name: str = '', ForcingType: Literal['AIR_DENS', 'AIR_PRES', 'CLOUD_COVER', 'DAY_ANGLE', 'DAY_LENGTH', 'ET_RADIA', 'LW_INCOMING', 'LW_RADIA_NET', 'OW_PET', 'PET', 'PET_MONTH_AVE', 'POTENTIAL_MELT', 'PRECIP', 'PRECIP_5DAY', 'PRECIP_DAILY_AVE', 'RAINFALL', 'RECHARGE', 'REL_HUMIDITY', 'SHORTWAVE', 'SNOWFALL', 'SNOW_FRAC', 'SUBDAILY_CORR', 'SW_RADIA', 'SW_RADIA_NET', 'TEMP_AVE', 'TEMP_AVE_UNC', 'TEMP_DAILY_AVE', 'TEMP_DAILY_MAX', 'TEMP_DAILY_MIN', 'TEMP_MAX', 'TEMP_MAX_UNC', 'TEMP_MIN', 'TEMP_MIN_UNC', 'TEMP_MONTH_AVE', 'TEMP_MONTH_MAX', 'TEMP_MONTH_MIN', 'WIND_VEL'] | None = None, GridWeights: GridWeights | RedirectToFile = GridWeights(number_hrus=1, number_grid_cells=1, data=(GWRecord(root=(1, 0, 1.0)),)))

StationForcing command (RVT).

classmethod check_dims(v)

classmethod from_nc(fn, data_type, alt_names=(), **kwds)

Instantiate class from netCDF dataset.

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ravenpy.config.commands.SubBasin(*, subbasin_id: Annotated[int, Gt(gt=0)] = 1, name: str = 'sub_001', downstream_id: int = -1, profile: str = 'NONE', reach_length: float | str = 0, gauged: bool = True, gauge_id: str | None = '')

Record to populate RVH :SubBasins command internal table.

classmethod check_reach_length(v)

downstream_id: int

gauge_id: str | None

gauged: bool

model_config = {'arbitrary_types_allowed': True, 'coerce_numbers_to_str': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

profile: str

reach_length: float | str

subbasin_id: Annotated[int, Gt(gt=0)]

class ravenpy.config.commands.SubBasinGroup(*, name: str = '', sb_ids: Sequence[int] = ())

SubBasinGroup command (RVH).

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

classmethod parse(s) → list

Parse a SubBasinGroup command and return a list of SubBasinGroup records.

sb_ids: Sequence[int]

to_rv()

Return Raven configuration string.

class ravenpy.config.commands.SubBasinProperties(*, Parameters: Sequence[Literal['CELERITY', 'DIFFUSIVITY', 'GAMMA_SCALE', 'GAMMA_SHAPE', 'MANNINGS_N', 'NUM_RESERVOIRS', 'Q_REFERENCE', 'RAIN_CORR', 'RES_CONSTANT', 'SLOPE', 'SNOW_CORR', 'TIME_CONC', 'TIME_LAG', 'TIME_TO_PEAK']] | None = None, records: Sequence[SubBasinProperty] | None = None)

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: Sequence[Literal['CELERITY', 'DIFFUSIVITY', 'GAMMA_SCALE', 'GAMMA_SHAPE', 'MANNINGS_N', 'NUM_RESERVOIRS', 'Q_REFERENCE', 'RAIN_CORR', 'RES_CONSTANT', 'SLOPE', 'SNOW_CORR', 'TIME_CONC', 'TIME_LAG', 'TIME_TO_PEAK']] | None

records: Sequence[SubBasinProperty] | None

class ravenpy.config.commands.SubBasinProperty(*, sb_id: str, values: Sequence[Variable | ExpressionNode | float | None])

model_config = {'arbitrary_types_allowed': True, 'coerce_numbers_to_str': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

sb_id: str

values: Sequence[Variable | ExpressionNode | float | None]

class ravenpy.config.commands.SubBasins(root: RootModelRootType = PydanticUndefined)

SubBasins command (RVH)

Example:

:SubBasins
  :Attributes, NAME, DOWNSTREAM_ID, PROFILE, REACH_LENGTH, GAUGED
  :Units, none, none, none, km, none
  1, Downstream, -1, DEFAULT, 3.0, 1
  2, Upstream, 1, DEFAULT, 3.0, 0
:EndSubBasins

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

classmethod parse(s)

Parse a SubBasins command.

root: Sequence[SubBasin]

ravenpy.config.commands.TC

alias of TerrainClass

class ravenpy.config.commands.TerrainClass(*, name: str, hillslope_length: Variable | ExpressionNode | float | None, drainage_density: Variable | ExpressionNode | float | None, topmodel_lambda: Variable | ExpressionNode | float | None = None)

drainage_density: Variable | ExpressionNode | float | None

hillslope_length: Variable | ExpressionNode | float | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

topmodel_lambda: Variable | ExpressionNode | float | None

class ravenpy.config.commands.TerrainClasses(root: RootModelRootType = PydanticUndefined)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

root: Sequence[TerrainClass]

ravenpy.config.commands.VC

alias of VegetationClass

class ravenpy.config.commands.VegetationClass(*, name: str = '', max_ht: Variable | ExpressionNode | float | None = 0.0, max_lai: Variable | ExpressionNode | float | None = 0.0, max_leaf_cond: Variable | ExpressionNode | float | None = 0.0)

max_ht: Variable | ExpressionNode | float | None

max_lai: Variable | ExpressionNode | float | None

max_leaf_cond: Variable | ExpressionNode | float | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

class ravenpy.config.commands.VegetationClasses(root: RootModelRootType = PydanticUndefined)

model_config = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

root: Sequence[VegetationClass]

class ravenpy.config.commands.VegetationParameterList(*, Parameters: Sequence[Literal['ALBEDO', 'ALBEDO_WET', 'CAP_LAI_RATIO', 'CHU_MATURITY', 'DRIP_PROPORTION', 'MAX_CAPACITY', 'MAX_HEIGHT', 'MAX_INTERCEPT_RATE', 'MAX_LAI', 'MAX_LEAF_COND', 'MAX_ROOT_LENGTH', 'MAX_SNOW_CAPACITY', 'MIN_RESISTIVITY', 'PET_VEG_CORR', 'PSI_CRITICAL', 'RAIN_ICEPT_FACT', 'RAIN_ICEPT_PCT', 'RELATIVE_HT', 'RELATIVE_LAI', 'ROOTRADIUS', 'ROOT_EXTINCT', 'SAI_HT_RATIO', 'SNOCAP_LAI_RATIO', 'SNOW_ICEPT_FACT', 'SNOW_ICEPT_PCT', 'STEMFLOW_FRAC', 'SVF_EXTINCTION', 'TFRAIN', 'TFSNOW', 'TRUNK_FRACTION', 'VEG_DENS', 'VEG_DIAM', 'VEG_MBETA', 'XYLEM_FRAC']] | None = None, Units: Sequence[str] | None = None, pl: Sequence[ParameterList])

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameters: Sequence[Literal['ALBEDO', 'ALBEDO_WET', 'CAP_LAI_RATIO', 'CHU_MATURITY', 'DRIP_PROPORTION', 'MAX_CAPACITY', 'MAX_HEIGHT', 'MAX_INTERCEPT_RATE', 'MAX_LAI', 'MAX_LEAF_COND', 'MAX_ROOT_LENGTH', 'MAX_SNOW_CAPACITY', 'MIN_RESISTIVITY', 'PET_VEG_CORR', 'PSI_CRITICAL', 'RAIN_ICEPT_FACT', 'RAIN_ICEPT_PCT', 'RELATIVE_HT', 'RELATIVE_LAI', 'ROOTRADIUS', 'ROOT_EXTINCT', 'SAI_HT_RATIO', 'SNOCAP_LAI_RATIO', 'SNOW_ICEPT_FACT', 'SNOW_ICEPT_PCT', 'STEMFLOW_FRAC', 'SVF_EXTINCTION', 'TFRAIN', 'TFSNOW', 'TRUNK_FRACTION', 'VEG_DENS', 'VEG_DIAM', 'VEG_MBETA', 'XYLEM_FRAC']] | None

class ravenpy.config.rvs.Config(*, EnKFMode: EnKFMode | None = <factory>, WindowSize: int | None = <factory>, SolutionRunName: str | None = <factory>, ExtraRVTFilename: str | None = <factory>, OutputDirectoryFormat: str | Path | None = <factory>, ForecastRVTFilename: str | None = <factory>, TruncateHindcasts: bool | None = <factory>, ForcingPerturbation: Sequence[ForcingPerturbation] | None = <factory>, AssimilatedState: Sequence[AssimilatedState] | None = <factory>, AssimilateStreamflow: Sequence[AssimilateStreamflow] | None = <factory>, ObservationErrorModel: Sequence[ObservationErrorModel] | None = <factory>, params: Any = None, SoilClasses: SoilClasses | None = <factory>, SoilProfiles: SoilProfiles | None = <factory>, VegetationClasses: VegetationClasses | None = <factory>, LandUseClasses: LandUseClasses | None = <factory>, TerrainClasses: TerrainClasses | None = <factory>, SoilParameterList: SoilParameterList | None = <factory>, LandUseParameterList: LandUseParameterList | None = <factory>, VegetationParameterList: VegetationParameterList | None = <factory>, ChannelProfile: Sequence[ChannelProfile] | None = <factory>, GlobalParameter: dict[str, ~pymbolic.primitives.Variable | ~pymbolic.primitives.ExpressionNode | float | None] | None={}, RainSnowTransition: RainSnowTransition | None = <factory>, SeasonalRelativeLAI: SeasonalRelativeLAI | None = <factory>, SeasonalRelativeHeight: SeasonalRelativeHeight | None = <factory>, Gauge: Sequence[Gauge] | None = <factory>, StationForcing: Sequence[StationForcing] | None = <factory>, GriddedForcing: Sequence[GriddedForcing] | None = <factory>, ObservationData: Sequence[ObservationData] | None = <factory>, SubBasins: SubBasins | None = <factory>, SubBasinGroup: Sequence[SubBasinGroup] | None = <factory>, SubBasinProperties: SubBasinProperties | None = <factory>, SBGroupPropertyMultiplier: Sequence[SBGroupPropertyMultiplier] | None = <factory>, GaugedSubBasinGroup: str | None = <factory>, HRUs: HRUs | None = <factory>, HRUGroup: Sequence[HRUGroup] | None = <factory>, Reservoirs: Sequence[Reservoir] | None = <factory>, HRUStateVariableTable: HRUStateVariableTable | None = <factory>, BasinStateVariables: BasinStateVariables | None = <factory>, UniformInitialConditions: dict[str, ~pymbolic.primitives.Variable | ~pymbolic.primitives.ExpressionNode | float | None] | None=<factory>, SilentMode: bool | None = <factory>, NoisyMode: bool | None = <factory>, RunName: str | None = <factory>, Calendar: Calendar | None = <factory>, StartDate: str | date | datetime | datetime | None = <factory>, AssimilationStartTime: date | datetime | datetime | None = <factory>, EndDate: str | date | datetime | datetime | None = <factory>, Duration: float | None = <factory>, TimeStep: float | str | None = <factory>, Interpolation: Interpolation | None = <factory>, Routing: Routing | None = <factory>, CatchmentRoute: CatchmentRoute | None = <factory>, Evaporation: Evaporation | None = <factory>, OW_Evaporation: Evaporation | None = <factory>, SWRadiationMethod: SWRadiationMethod | None = <factory>, SWCloudCorrect: SWCloudCorrect | None = <factory>, SWCanopyCorrect: SWCanopyCorrect | None = <factory>, LWRadiationMethod: LWRadiationMethod | None = <factory>, WindspeedMethod: WindspeedMethod | None = <factory>, RainSnowFraction: RainSnowFraction | None = <factory>, PotentialMeltMethod: PotentialMeltMethod | None = <factory>, OroTempCorrect: OroTempCorrect | None = <factory>, OroPrecipCorrect: OroPrecipCorrect | None = <factory>, OroPETCorrect: OroPETCorrect | None = <factory>, CloudCoverMethod: CloudCoverMethod | None = <factory>, PrecipIceptFract: PrecipIceptFract | None = <factory>, SubdailyMethod: SubdailyMethod | None = <factory>, MonthlyInterpolationMethod: MonthlyInterpolationMethod | None = <factory>, SoilModel: SoilModel | None = <factory>, TemperatureCorrection: bool | None = <factory>, LakeStorage: Literal['ATMOSPHERE', 'ATMOS_PRECIP', 'CANOPY', 'CANOPY_SNOW', 'CANOPY_TEMP', 'COLD_CONTENT', 'CONSTITUENT', 'CONSTITUENT_SINK', 'CONSTITUENT_SRC', 'CONSTITUENT_SW', 'CONVOLUTION', 'CONV_STOR', 'CROP_HEAT_UNITS', 'CUM_INFIL', 'CUM_SNOWMELT', 'DEPRESSION', 'GLACIER', 'GLACIER_CC', 'GLACIER_ICE', 'GROUNDWATER', 'LAKE_STORAGE', 'MULTIPLE', 'PERMAFROST_DEPTH', 'PONDED_WATER', 'ROOT', 'SNOW', 'SNOW_AGE', 'SNOW_ALBEDO', 'SNOW_COVER', 'SNOW_DEPTH', 'SNOW_LIQ', 'SNOW_TEMP', 'SOIL', 'SOIL[0]', 'SOIL[1]', 'SOIL[2]', 'SOIL_TEMP', 'SURFACE_WATER', 'SURFACE_WATER_TEMP', 'TOTAL_SWE', 'TRUNK', 'WETLAND'] | None=<factory>, RelativeHumidityMethod: RelativeHumidityMethod | None = <factory>, DefineHRUGroups: Sequence[str] | None = <factory>, HydrologicProcesses: Sequence[Process | Conditional | ProcessGroup] | None = <factory>, EvaluationMetrics: Sequence[EvaluationMetrics] | None = <factory>, EvaluationPeriod: Sequence[EvaluationPeriod] | None = <factory>, EnsembleMode: EnsembleMode | None = <factory>, WriteNetcdfFormat: bool | None = <factory>, NetCDFAttribute: dict[str, str] | None={}, CustomOutput: Sequence[CustomOutput] | None = <factory>, DirectEvaporation: bool | None = <factory>, DeltaresFEWSMode: bool | None = <factory>, DebugMode: bool | None = <factory>, DontWriteWatershedStorage: bool | None = <factory>, PavicsMode: bool | None = <factory>, SuppressOutput: bool | None = <factory>, WriteForcingFunctions: bool | None = <factory>, WriteSubbasinFile: bool | None = <factory>, WriteLocalFlows: bool | None = <factory>)

duplicate(**kwds)

Duplicate this model, changing the values given in the keywords.

static header(rv)

Return the header to print at the top of each RV file.

property is_symbolic

If configuration contains symbolic expressions, returns True.

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property rvc

property rve

property rvh

property rvi

property rvp

property rvt

set_params(params: dict | Sequence) → Config

Return a new instance of Config with params frozen to their numerical values.

set_solution(fn: Path, timestamp: bool = True) → Config

Return a new instance of Config with hru, basin states and start date set from an existing solution.

Parameters:

• fn (Path) – Path to solution file.

• timestamp (bool) – If False, ignore time stamp information in the solution. If True, the solution will set StartDate to the solution’s timestamp.

Returns:

Config with internal state set from the solution file.

Return type:

Config

write_rv(workdir: str | Path, modelname: str | None = None, overwrite: bool = False, header: bool = True)

Write configuration files to disk.

Parameters:

• workdir (str, Path) – A directory where rv files will be written to disk.

• modelname (str) – File name stem for rv files. If not given, defaults to RunName if set, otherwise raven.

• overwrite (bool) – If True, overwrite existing configuration files.

• header (bool) – If True, write a header at the top of each RV file.

zip(workdir: str | Path, modelname: str | None = None, overwrite: bool = False)

Write configuration to zip file.

Parameters:

• workdir (Path, str) – Path to zip archive storing RV files.

• modelname (str, optional) – File name stem for rv files. If not given, defaults to RunName if set, otherwise raven.

• overwrite (bool) – If True, overwrite existing configuration zip file.

class ravenpy.config.rvs.RVC(*, HRUStateVariableTable: HRUStateVariableTable | None = <factory>, BasinStateVariables: BasinStateVariables | None = <factory>, UniformInitialConditions: dict[str, ~pymbolic.primitives.Variable | ~pymbolic.primitives.ExpressionNode | float | None] | None=<factory>)

basin_state_variables: BasinStateVariables | None

hru_state_variable_table: HRUStateVariableTable | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

uniform_initial_conditions: dict[str, Variable | ExpressionNode | float | None] | None

class ravenpy.config.rvs.RVE(*, EnKFMode: EnKFMode | None = <factory>, WindowSize: int | None = <factory>, SolutionRunName: str | None = <factory>, ExtraRVTFilename: str | None = <factory>, OutputDirectoryFormat: str | Path | None = <factory>, ForecastRVTFilename: str | None = <factory>, TruncateHindcasts: bool | None = <factory>, ForcingPerturbation: Sequence[ForcingPerturbation] | None = <factory>, AssimilatedState: Sequence[AssimilatedState] | None = <factory>, AssimilateStreamflow: Sequence[AssimilateStreamflow] | None = <factory>, ObservationErrorModel: Sequence[ObservationErrorModel] | None = <factory>)

assimilate_streamflow: Sequence[AssimilateStreamflow] | None

assimilated_state: Sequence[AssimilatedState] | None

enkf_mode: EnKFMode | None

extra_rvt_filename: str | None

forcing_perturbation: Sequence[ForcingPerturbation] | None

forecast_rvt_filename: str | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

observation_error_model: Sequence[ObservationErrorModel] | None

output_directory_format: str | Path | None

solution_run_name: str | None

truncate_hindcasts: bool | None

window_size: int | None

class ravenpy.config.rvs.RVH(*, SubBasins: SubBasins | None = <factory>, SubBasinGroup: Sequence[SubBasinGroup] | None = <factory>, SubBasinProperties: SubBasinProperties | None = <factory>, SBGroupPropertyMultiplier: Sequence[SBGroupPropertyMultiplier] | None = <factory>, GaugedSubBasinGroup: str | None = <factory>, HRUs: HRUs | None = <factory>, HRUGroup: Sequence[HRUGroup] | None = <factory>, Reservoirs: Sequence[Reservoir] | None = <factory>)

gauged_sub_basin_group: str | None

hru_group: Sequence[HRUGroup] | None

hrus: HRUs | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reservoirs: Sequence[Reservoir] | None

sb_group_property_multiplier: Sequence[SBGroupPropertyMultiplier] | None

sub_basin_group: Sequence[SubBasinGroup] | None

sub_basin_properties: SubBasinProperties | None

sub_basins: SubBasins | None

class ravenpy.config.rvs.RVI(*, SilentMode: bool | None = <factory>, NoisyMode: bool | None = <factory>, RunName: str | None = <factory>, Calendar: Calendar | None = <factory>, StartDate: str | date | datetime | datetime | None = <factory>, AssimilationStartTime: date | datetime | datetime | None = <factory>, EndDate: str | date | datetime | datetime | None = <factory>, Duration: float | None = <factory>, TimeStep: float | str | None = <factory>, Interpolation: Interpolation | None = <factory>, Routing: Routing | None = <factory>, CatchmentRoute: CatchmentRoute | None = <factory>, Evaporation: Evaporation | None = <factory>, OW_Evaporation: Evaporation | None = <factory>, SWRadiationMethod: SWRadiationMethod | None = <factory>, SWCloudCorrect: SWCloudCorrect | None = <factory>, SWCanopyCorrect: SWCanopyCorrect | None = <factory>, LWRadiationMethod: LWRadiationMethod | None = <factory>, WindspeedMethod: WindspeedMethod | None = <factory>, RainSnowFraction: RainSnowFraction | None = <factory>, PotentialMeltMethod: PotentialMeltMethod | None = <factory>, OroTempCorrect: OroTempCorrect | None = <factory>, OroPrecipCorrect: OroPrecipCorrect | None = <factory>, OroPETCorrect: OroPETCorrect | None = <factory>, CloudCoverMethod: CloudCoverMethod | None = <factory>, PrecipIceptFract: PrecipIceptFract | None = <factory>, SubdailyMethod: SubdailyMethod | None = <factory>, MonthlyInterpolationMethod: MonthlyInterpolationMethod | None = <factory>, SoilModel: SoilModel | None = <factory>, TemperatureCorrection: bool | None = <factory>, LakeStorage: Literal['ATMOSPHERE', 'ATMOS_PRECIP', 'CANOPY', 'CANOPY_SNOW', 'CANOPY_TEMP', 'COLD_CONTENT', 'CONSTITUENT', 'CONSTITUENT_SINK', 'CONSTITUENT_SRC', 'CONSTITUENT_SW', 'CONVOLUTION', 'CONV_STOR', 'CROP_HEAT_UNITS', 'CUM_INFIL', 'CUM_SNOWMELT', 'DEPRESSION', 'GLACIER', 'GLACIER_CC', 'GLACIER_ICE', 'GROUNDWATER', 'LAKE_STORAGE', 'MULTIPLE', 'PERMAFROST_DEPTH', 'PONDED_WATER', 'ROOT', 'SNOW', 'SNOW_AGE', 'SNOW_ALBEDO', 'SNOW_COVER', 'SNOW_DEPTH', 'SNOW_LIQ', 'SNOW_TEMP', 'SOIL', 'SOIL[0]', 'SOIL[1]', 'SOIL[2]', 'SOIL_TEMP', 'SURFACE_WATER', 'SURFACE_WATER_TEMP', 'TOTAL_SWE', 'TRUNK', 'WETLAND'] | None=<factory>, RelativeHumidityMethod: RelativeHumidityMethod | None = <factory>, DefineHRUGroups: Sequence[str] | None = <factory>, HydrologicProcesses: Sequence[Process | Conditional | ProcessGroup] | None = <factory>, EvaluationMetrics: Sequence[EvaluationMetrics] | None = <factory>, EvaluationPeriod: Sequence[EvaluationPeriod] | None = <factory>, EnsembleMode: EnsembleMode | None = <factory>, WriteNetcdfFormat: bool | None = <factory>, NetCDFAttribute: dict[str, str] | None={}, CustomOutput: Sequence[CustomOutput] | None = <factory>, DirectEvaporation: bool | None = <factory>, DeltaresFEWSMode: bool | None = <factory>, DebugMode: bool | None = <factory>, DontWriteWatershedStorage: bool | None = <factory>, PavicsMode: bool | None = <factory>, SuppressOutput: bool | None = <factory>, WriteForcingFunctions: bool | None = <factory>, WriteSubbasinFile: bool | None = <factory>, WriteLocalFlows: bool | None = <factory>)

assimilation_start_time: date | datetime | datetime | None

calendar: Calendar | None

catchment_route: CatchmentRoute | None

cloud_cover_method: CloudCoverMethod | None

custom_output: Sequence[CustomOutput] | None

classmethod dates2cf(v, info)

Convert dates to cftime dates.

debug_mode: bool | None

define_hru_groups: Sequence[str] | None

deltares_fews_mode: bool | None

direct_evaporation: bool | None

dont_write_watershed_storage: bool | None

duration: float | None

end_date: str | date | datetime | datetime | None

ensemble_mode: EnsembleMode | None

evaluation_metrics: Sequence[EvaluationMetrics] | None

evaluation_period: Sequence[EvaluationPeriod] | None

evaporation: Evaporation | None

hydrologic_processes: Sequence[Process | Conditional | ProcessGroup] | None

classmethod init_soil_model(v)

interpolation: Interpolation | None

lake_storage: Literal['ATMOSPHERE', 'ATMOS_PRECIP', 'CANOPY', 'CANOPY_SNOW', 'CANOPY_TEMP', 'COLD_CONTENT', 'CONSTITUENT', 'CONSTITUENT_SINK', 'CONSTITUENT_SRC', 'CONSTITUENT_SW', 'CONVOLUTION', 'CONV_STOR', 'CROP_HEAT_UNITS', 'CUM_INFIL', 'CUM_SNOWMELT', 'DEPRESSION', 'GLACIER', 'GLACIER_CC', 'GLACIER_ICE', 'GROUNDWATER', 'LAKE_STORAGE', 'MULTIPLE', 'PERMAFROST_DEPTH', 'PONDED_WATER', 'ROOT', 'SNOW', 'SNOW_AGE', 'SNOW_ALBEDO', 'SNOW_COVER', 'SNOW_DEPTH', 'SNOW_LIQ', 'SNOW_TEMP', 'SOIL', 'SOIL[0]', 'SOIL[1]', 'SOIL[2]', 'SOIL_TEMP', 'SURFACE_WATER', 'SURFACE_WATER_TEMP', 'TOTAL_SWE', 'TRUNK', 'WETLAND'] | None

lw_radiation_method: LWRadiationMethod | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

monthly_interpolation_method: MonthlyInterpolationMethod | None

netcdf_attribute: dict[str, str] | None

noisy_mode: bool | None

oro_pet_correct: OroPETCorrect | None

oro_precip_correct: OroPrecipCorrect | None

oro_temp_correct: OroTempCorrect | None

ow_evaporation: Evaporation | None

pavics_mode: bool | None

potential_melt_method: PotentialMeltMethod | None

precip_icept_frac: PrecipIceptFract | None

rain_snow_fraction: RainSnowFraction | None

relative_humidity_method: RelativeHumidityMethod | None

routing: Routing | None

run_name: str | None

silent_mode: bool | None

soil_model: SoilModel | None

start_date: str | date | datetime | datetime | None

subdaily_method: SubdailyMethod | None

suppress_output: bool | None

sw_canopy_correct: SWCanopyCorrect | None

sw_cloud_correct: SWCloudCorrect | None

sw_radiation_method: SWRadiationMethod | None

temperature_correction: bool | None

time_step: float | str | None

windspeed_method: WindspeedMethod | None

write_forcing_functions: bool | None

write_local_flows: bool | None

write_netcdf_format: bool | None

write_subbasin_file: bool | None

class ravenpy.config.rvs.RVP(*, params: Any = None, SoilClasses: SoilClasses | None = <factory>, SoilProfiles: SoilProfiles | None = <factory>, VegetationClasses: VegetationClasses | None = <factory>, LandUseClasses: LandUseClasses | None = <factory>, TerrainClasses: TerrainClasses | None = <factory>, SoilParameterList: SoilParameterList | None = <factory>, LandUseParameterList: LandUseParameterList | None = <factory>, VegetationParameterList: VegetationParameterList | None = <factory>, ChannelProfile: Sequence[ChannelProfile] | None = <factory>, GlobalParameter: dict[str, ~pymbolic.primitives.Variable | ~pymbolic.primitives.ExpressionNode | float | None] | None={}, RainSnowTransition: RainSnowTransition | None = <factory>, SeasonalRelativeLAI: SeasonalRelativeLAI | None = <factory>, SeasonalRelativeHeight: SeasonalRelativeHeight | None = <factory>)

channel_profile: Sequence[ChannelProfile] | None

global_parameter: dict[str, Variable | ExpressionNode | float | None] | None

land_use_classes: LandUseClasses | None

land_use_parameter_list: LandUseParameterList | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

params: Any

rain_snow_transition: RainSnowTransition | None

seasonal_relative_height: SeasonalRelativeHeight | None

seasonal_relative_lai: SeasonalRelativeLAI | None

soil_classes: SoilClasses | None

soil_parameter_list: SoilParameterList | None

soil_profiles: SoilProfiles | None

terrain_classes: TerrainClasses | None

vegetation_classes: VegetationClasses | None

vegetation_parameter_list: VegetationParameterList | None

class ravenpy.config.rvs.RVT(*, Gauge: Sequence[Gauge] | None = <factory>, StationForcing: Sequence[StationForcing] | None = <factory>, GriddedForcing: Sequence[GriddedForcing] | None = <factory>, ObservationData: Sequence[ObservationData] | None = <factory>)

gauge: Sequence[Gauge] | None

gridded_forcing: Sequence[GriddedForcing] | None

model_config = {'arbitrary_types_allowed': True, 'extra': 'forbid', 'populate_by_name': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

observation_data: Sequence[ObservationData] | None

station_forcing: Sequence[StationForcing] | None

ravenpy.config.rvs.is_symbolic(params: dict | Any) → bool

Return True if parameters include a symbolic variable.

Emulators

ravenpy.config.emulators.get_model(name)

Return the corresponding Raven emulator configuration class.

Parameters:

name (str) – Model class name or model identifier.

Return type:

Raven model configuration class

Extractors

class ravenpy.extractors.routing_product.BasinMakerExtractor(df, hru_aspect_convention='GRASS', routing_product_version='2.1')

Encapsulate the logic of converting the Routing Product into the required data structures to generate the RVH file format.

Parameters:

• df (GeoDataFrame) – Sub-basin information.

• hru_aspect_convention ({"GRASS", "ArcGIS"}) – How sub-basin aspect is defined.

• routing_product_version ({"2.1", "1.0"}) – Version of the BasinMaker data.

HRU_ASPECT_CONVENTION = 'GRASS'

MANNING_DEFAULT = 0.035

MAX_RIVER_SLOPE = 1e-05

ROUTING_PRODUCT_VERSION = '2.1'

USE_LAKE_AS_GAUGE = False

USE_LAND_AS_GAUGE = False

USE_MANNING_COEFF = False

WEIR_COEFFICIENT = 0.6

extract(hru_from_sb: bool = False) → dict

Extract data from the Routing Product shapefile and return dictionaries that can be parsed into Raven Commands.

Parameters:

hru_from_sb (bool) – If True, draw HRU information from subbasin information. This is likely to yield crude results.

Returns:

“sub_basins”

Sequence of dictionaries with SubBasin attributes.

”sub_basin_group”

Sequence of dictionaries with SubBasinGroup attributes.

”reservoirs”

Sequence of dictionaries with Reservoir attributes.

”channel_profile”

Sequence of dictionaries with ChannelProfile attributes.

”hrus”

Sequence of dictionaries with HRU attributes.

Return type:

dict

class ravenpy.extractors.routing_product.GridWeightExtractor(input_file_path, routing_file_path, dim_names=('lon_dim', 'lat_dim'), var_names=('longitude', 'latitude'), routing_id_field='SubId', netcdf_input_field='NetCDF_col', gauge_ids=None, sub_ids=None, area_error_threshold=0.05)

Class to extract grid weights.

Notes

To use this on HRU GeoJONS created from the Canadian River and Lake Hydrofabric database, set routing_id_field to __INDEX__.

The original version of this algorithm can be found at: https://github.com/julemai/GridWeightsGenerator

AREA_ERROR_THRESHOLD = 0.05

CRS_CAEA = 3573

CRS_LLDEG = 4326

DIM_NAMES = ('lon_dim', 'lat_dim')

NETCDF_INPUT_FIELD = 'NetCDF_col'

ROUTING_ID_FIELD = 'SubId'

VAR_NAMES = ('longitude', 'latitude')

extract() → dict

Return dictionary to create a GridWeights command.

ravenpy.extractors.routing_product.open_shapefile(path: str | PathLike)

Return GeoDataFrame from shapefile path.

ravenpy.extractors.routing_product.upstream_from_coords(lon: float, lat: float, df: DataFrame | geopandas.GeoDataFrame) → DataFrame | geopandas.GeoDataFrame

Return the sub-basins located upstream from outlet.

Parameters:

• lon (float) – Longitude of outlet.

• lat (float) – Latitude of outlet.

• df (pandas.DataFrame or geopandas.GeoDataFrame) – Routing product.

Returns:

Sub-basins located upstream from outlet.

Return type:

pandas.DataFrame or geopandas.GeoDataFrame

ravenpy.extractors.routing_product.upstream_from_id(fid: str | int | float, df: DataFrame | geopandas.GeoDataFrame) → DataFrame | geopandas.GeoDataFrame

Return upstream sub-basins by evaluating the downstream networks.

Parameters:

• fid (str or int or float) – feature ID of the downstream feature of interest.

• df (pandas.DataFrame or geopandas.GeoDataFrame) – A GeoDataframe comprising the watershed attributes.

Returns:

Basins ids including fid and its upstream contributors.

Return type:

pandas.DataFrame or geopandas.GeoDataFrame

ravenpy.extractors.forecasts.get_CASPAR_dataset(climate_model: str, date: datetime, thredds: str = 'https://pavics.ouranos.ca/twitcher/ows/proxy/thredds/', directory: str = 'dodsC/birdhouse/disk2/caspar/daily/', engine: str = 'netcdf4') → tuple[Dataset, list[DatetimeIndex | Series | Timestamp | Any]]

Return CASPAR dataset.

Parameters:

• climate_model (str) – Type of climate model, for now only “GEPS” is supported.

• date (dt.datetime) – The date of the forecast.

• thredds (str) – The thredds server url. Default: “https://pavics.ouranos.ca/twitcher/ows/proxy/thredds/”.

• directory (str) – The directory on the thredds server where the data is stored. Default: “dodsC/birdhouse/disk2/caspar/daily/”.

• engine (str) – The xarray engine to use to open the dataset. Default: “netcdf4”.

Returns:

The forecast dataset.

Return type:

xr.Dataset

ravenpy.extractors.forecasts.get_ECCC_dataset(climate_model: str, thredds: str = 'https://pavics.ouranos.ca/twitcher/ows/proxy/thredds/', directory: str = 'dodsC/datasets/forecasts/eccc_geps/', engine: str = 'netcdf4') → tuple[Dataset, list[DatetimeIndex | Series | Timestamp | Any]]

Return latest GEPS forecast dataset.

Parameters:

• climate_model (str) – Type of climate model, for now only “GEPS” is supported.

• thredds (str) – The thredds server url. Default: “https://pavics.ouranos.ca/twitcher/ows/proxy/thredds/”.

• directory (str) – The directory on the thredds server where the data is stored. Default: “dodsC/datasets/forecasts/eccc_geps/”.

• engine (str) – The xarray engine to use to open the dataset. Default: “netcdf4”.

Returns:

The forecast dataset.

Return type:

xr.Dataset

ravenpy.extractors.forecasts.get_hindcast_day(region_coll: fiona.Collection, date, climate_model='GEPS')

Generate a forecast dataset that can be used to run raven.

Data comes from the CASPAR archive and must be aggregated such that each file contains forecast data for a single day, but for all forecast timesteps and all members.

The code takes the region shapefile, the forecast date required, and the climate_model to use, here GEPS by default, but eventually could be GEPS, GDPS, REPS or RDPS.

ravenpy.extractors.forecasts.get_recent_ECCC_forecast(region_coll: fiona.Collection, climate_model: str = 'GEPS') → Dataset

Generate a forecast dataset that can be used to run raven.

Data comes from the ECCC datamart and collected daily. It is aggregated such that each file contains forecast data for a single day, but for all forecast timesteps and all members.

The code takes the region shapefile and the climate_model to use, here GEPS by default, but eventually could be GEPS, GDPS, REPS or RDPS.

Parameters:

• region_coll (fiona.Collection) – The region vectors.

• climate_model (str) – Type of climate model, for now only “GEPS” is supported.

Returns:

The forecast dataset.

Return type:

xr.Dataset

ravenpy.extractors.forecasts.get_subsetted_forecast(region_coll: fiona.Collection, ds: Dataset, times: datetime | DataArray, is_caspar: bool) → Dataset

Get Subsetted Forecast.

This function takes a dataset, a region and the time sampling array and returns the subsetted values for the given region and times.

Parameters:

• region_coll (fiona.Collection) – The region vectors.

• ds (xr.Dataset) – The dataset containing the raw, worldwide forecast data.

• times (dt.datetime or xr.DataArray) – The array of times required to do the forecast.

• is_caspar (bool) – True if the data comes from Caspar, false otherwise. Used to define lat/lon on rotated grid.

Returns:

The forecast dataset.

Return type:

xr.Dataset

Utilities

Geospatial

Tools for reading and writing geospatial data formats.

ravenpy.utilities.io.address_append(address: str | Path) → str

Format a URL/URI to be more easily read with libraries such as “rasterstats”.

Parameters:

address (Union[str, Path]) – URL/URI to a potential zip or tar file.

Returns:

URL/URI prefixed for archive type.

Return type:

str

ravenpy.utilities.io.archive_sniffer(archives: str | Path | list[str | Path], working_dir: str | Path | None = None, extensions: Sequence[str] | None = None) → list[str | Path]

Return a list of locally unarchived files that match the desired extensions.

Parameters:

• archives (str or Path or list of str or Path) – Archive location or list of archive locations.

• working_dir (str or Path, optional) – String or Path to a working location.

• extensions (Sequence of str, optional) – List of accepted extensions.

Returns:

A list of files with matching accepted extensions.

Return type:

list of str or Path

ravenpy.utilities.io.crs_sniffer(*args: str | Path | Sequence[str | Path]) → list[str | int] | str | int

Return the list of CRS found in files.

Parameters:

*args (str or Path or sequence of str or Path) – Path(s) to the file(s) to examine.

Returns:

Returns either a list of CRSes or a single CRS definition, depending on the number of instances found.

Return type:

str or list of str

ravenpy.utilities.io.generic_extract_archive(resources: str | Path | list[bytes | str | Path], output_dir: str | Path | None = None) → list[str]

Extract archives (tar/zip) to a working directory.

Parameters:

• resources (str or Path or list of bytes or str or Path) – List of archive files (if netCDF files are in list, they are passed and returned as well in the return).

• output_dir (str or Path, optional) – String or Path to a working location (default: temporary folder).

Returns:

A list of original or of extracted files.

Return type:

list of str

ravenpy.utilities.io.get_bbox(vector: str | Path, all_features: bool = True) → tuple[float, float, float, float]

Return bounding box of all features or the first feature in file.

Parameters:

• vector (str or Path) – A path to file storing vector features.

• all_features (bool) – Return the bounding box for all features. Default: True.

Returns:

Geographic coordinates of the bounding box (lon0, lat0, lon1, lat1).

Return type:

float, float, float, float

ravenpy.utilities.io.is_within_directory(directory: str | PathLike, target: str | PathLike) → bool

ravenpy.utilities.io.raster_datatype_sniffer(file: str | Path) → str

Return the type of the raster stored in the file.

Parameters:

file (str or Path) – Path to file.

Returns:

A rasterio datatype of array values.

Return type:

str

ravenpy.utilities.io.safe_extract(tar: TarFile, path: str = '.', members=None, *, numeric_owner=False) → None

Tools for performing geospatial translations and transformations.

ravenpy.utilities.geo.determine_upstream_ids(fid: str | int | float, df: DataFrame | geopandas.GeoDataFrame, *, basin_field: str, downstream_field: str, basin_family: str | None = None) → DataFrame | geopandas.GeoDataFrame

Return a list of upstream features by evaluating the downstream networks.

Parameters:

• fid (str or int or float) – The feature ID of the downstream feature of interest.

• df (pd.DataFrame) – A Dataframe comprising the watershed attributes.

• basin_field (str) – The field used to determine the id of the basin according to hydro project.

• downstream_field (str) – The field identifying the downstream sub-basin for the hydro project.

• basin_family (str, optional) – Regional watershed code (For HydroBASINS dataset).

Returns:

Basins ids including fid and its upstream contributors.

Return type:

pd.DataFrame

ravenpy.utilities.geo.find_geometry_from_coord(lon: float, lat: float, df: geopandas.GeoDataFrame) → geopandas.GeoDataFrame

Return the geometry containing the given coordinates.

lonfloat

Longitude.

latfloat

Latitude.

dfGeoDataFrame

Data.

Returns:

Record whose geometry contains the point.

Return type:

GeoDataFrame

ravenpy.utilities.geo.generic_raster_clip(raster: str | Path, output: str | Path, geometry: shapely.geometry.Polygon | shapely.geometry.MultiPolygon | list[shapely.geometry.Polygon | shapely.geometry.MultiPolygon], touches: bool = False, fill_with_nodata: bool = True, padded: bool = True, raster_compression: str = 'lzw') → None

Crop a raster file to a given geometry.

Parameters:

• raster (Union[str, Path]) – Path to input raster.

• output (Union[str, Path]) – Path to output raster.

• geometry (Union[Polygon, MultiPolygon, List[Union[Polygon, MultiPolygon]]) – Geometry defining the region to crop.

• touches (bool) – Whether to include cells that intersect the geometry or not. Default: True.

• fill_with_nodata (bool) – Whether to keep pixel values for regions outside of shape or set as nodata or not. Default: True.

• padded (bool) – Whether to add a half-pixel buffer to shape before masking or not. Default: True.

• raster_compression (str) – Level of data compression. Default: ‘lzw’.

Return type:

None

ravenpy.utilities.geo.generic_raster_warp(raster: str | Path, output: str | Path, target_crs: str | dict | pyproj.CRS, raster_compression: str = 'lzw') → None

Reproject a raster file.

Parameters:

• raster (Union[str, Path]) – Path to input raster.

• output (Union[str, Path]) – Path to output raster.

• target_crs (str or dict) – Target projection identifier.

• raster_compression (str) – Level of data compression. Default: ‘lzw’.

Return type:

None

ravenpy.utilities.geo.generic_vector_reproject(vector: str | Path, projected: str | Path, source_crs: str | pyproj.CRS = 4326, target_crs: str | pyproj.CRS = None) → None

Reproject all features and layers within a vector file and return a GeoJSON.

Parameters:

• vector (Union[str, Path]) – Path to a file containing a valid vector layer.

• projected (Union[str, Path]) – Path to a file to be written.

• source_crs (Union[str, pyproj.crs.CRS]) – CRS for the source geometry. Default: 4326.

• target_crs (Union[str, pyproj.crs.CRS]) – CRS for the target geometry.

Return type:

None

ravenpy.utilities.geo.geom_transform(geom: shapely.geometry.GeometryCollection | shapely.geometry.shape, source_crs: str | int | pyproj.CRS = 4326, target_crs: str | int | pyproj.CRS = None) → shapely.geometry.GeometryCollection

Change the projection of a geometry.

Assuming a geometry’s coordinates are in a source_crs, compute the new coordinates under the target_crs.

Parameters:

• geom (Union[GeometryCollection, shape]) – Source geometry.

• source_crs (Union[str, int, CRS]) – Projection identifier (proj4) for the source geometry, e.g. ‘+proj=longlat +datum=WGS84 +no_defs’.

• target_crs (Union[str, int, CRS]) – Projection identifier (proj4) for the target geometry.

Returns:

Reprojected geometry.

Return type:

GeometryCollection

GeoServer interaction operations.

Working assumptions for this module: * Point coordinates are passed as shapely.geometry.Point instances. * BBox coordinates are passed as (lon1, lat1, lon2, lat2). * Shapes (polygons) are passed as shapely.geometry.shape parsable objects. * All functions that require a CRS have a CRS argument with a default set to WGS84. * GEOSERVER_URL points to the GeoServer instance hosting all files. * For legacy reasons, we also accept the GEO_URL environment variable.

TODO: Refactor to remove functions that are just 2-lines of code. For example, many function’s logic essentially consists in creating the layer name. We could have a function that returns the layer name, and then other functions expect the layer name.

ravenpy.utilities.geoserver.filter_hydro_routing_attributes_wfs(attribute: str, value: str | float | int, level: int = 12, lakes: str = '1km', geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → str

Return a URL that formats and returns a remote GetFeatures request from hydro routing dataset.

For geographic rasters, subsetting is based on WGS84 (Long, Lat) boundaries. If not geographic, subsetting based on projected coordinate system (Easting, Northing) boundaries.

Parameters:

• attribute (list) – Attributes/fields to be queried.

• value (str or int or float) – The requested value for the attribute.

• level (int) – Level of granularity requested for the lakes vector (range(7,13)). Default: 12.

• lakes ({"1km", "all"}) – Query the version of dataset with lakes under 1km in width removed (“1km”) or return all lakes (“all”).

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

URL to the GeoJSON-encoded WFS response.

Return type:

str

ravenpy.utilities.geoserver.filter_hydrobasins_attributes_wfs(attribute: str, value: str | float | int, domain: str, geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → str

Return a URL that formats and returns a remote GetFeatures request from the USGS HydroBASINS dataset.

For geographic raster grids, subsetting is based on WGS84 (Long, Lat) boundaries. If not geographic, subsetting based on projected coordinate system (Easting, Northing) boundaries.

Parameters:

• attribute (str) – Attribute/field to be queried.

• value (str or float or int) – Value for attribute queried.

• domain ({"na", "ar"}) – The domain of the HydroBASINS data.

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

URL to the GeoJSON-encoded WFS response.

Return type:

str

ravenpy.utilities.geoserver.get_hydro_routing_attributes_wfs(attribute: Sequence[str], level: int = 12, lakes: str = '1km', geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → str

Return a URL that formats and returns a remote GetFeatures request from hydro routing dataset.

For geographic rasters, subsetting is based on WGS84 (Long, Lat) boundaries. If not geographic, subsetting based on projected coordinate system (Easting, Northing) boundaries.

Parameters:

• attribute (list) – Attributes/fields to be queried.

• level (int) – Level of granularity requested for the lakes vector (range(7,13)). Default: 12.

• lakes ({"1km", "all"}) – Query the version of dataset with lakes under 1km in width removed (“1km”) or return all lakes (“all”).

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

URL to the GeoJSON-encoded WFS response.

Return type:

str

ravenpy.utilities.geoserver.get_hydro_routing_location_wfs(coordinates: tuple[int | float | str, str | float | int], lakes: str, level: int = 12, geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → dict

Return features from the hydro routing data set using bounding box coordinates.

For geographic rasters, subsetting is based on WGS84 (Long, Lat) boundaries. If not geographic, subsetting based on projected coordinate system (Easting, Northing) boundaries.

Parameters:

• coordinates (Tuple[str or float or int, str or float or int]) – Geographic coordinates of the bounding box (left, down, right, up).

• lakes ({"1km", "all"}) – Query the version of dataset with lakes under 1km in width removed (“1km”) or return all lakes (“all”).

• level (int) – Level of granularity requested for the lakes vector (range(7,13)). Default: 12.

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

A GeoJSON-derived dictionary of vector features (FeatureCollection).

Return type:

dict

ravenpy.utilities.geoserver.get_hydrobasins_location_wfs(coordinates: tuple[str | float | int, str | float | int], domain: str, geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → dict[str, str | int | float]

Return features from the USGS HydroBASINS data set using bounding box coordinates.

For geographic raster grids, subsetting is based on WGS84 (Long, Lat) boundaries. If not geographic, subsetting based on projected coordinate system (Easting, Northing) boundaries.

Parameters:

• coordinates (Tuple[str or float or int, str or float or int]) – Geographic coordinates of the bounding box (left, down, right, up).

• domain ({"na", "ar"}) – The domain of the HydroBASINS data.

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

A GeoJSON-encoded vector feature.

Return type:

dict

ravenpy.utilities.geoserver.get_raster_wcs(coordinates: Iterable | Sequence[float | str], geographic: bool = True, *, layer: str, geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → bytes

Return a subset of a raster image from the local GeoServer via WCS 2.0.1 protocol.

For geographic raster grids, subsetting is based on WGS84 (Long, Lat) boundaries. If not geographic, subsetting based on projected coordinate system (Easting, Northing) boundaries.

Parameters:

• coordinates (Sequence of int or float or str) – Geographic coordinates of the bounding box (left, down, right, up).

• geographic (bool) – If True, uses “Long” and “Lat” in WCS call. Otherwise, uses “E” and “N”.

• layer (str) – Layer name of raster exposed on GeoServer instance, e.g. ‘public:CEC_NALCMS_LandUse_2010’.

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

A GeoTIFF array.

Return type:

bytes

ravenpy.utilities.geoserver.hydro_routing_upstream(fid: str | float | int, level: int = 12, lakes: str = '1km', geoserver: str = 'https://pavics.ouranos.ca//geoserver/') → geopandas.GeoDataFrame

Return a list of hydro routing features located upstream.

Parameters:

• fid (str or float or int) – Basin feature ID code of the downstream feature.

• level (int) – Level of granularity requested for the lakes vector (range(7,13)). Default: 12.

• lakes ({"1km", "all"}) – Query the version of dataset with lakes under 1km in width removed (“1km”) or return all lakes (“all”).

• geoserver (str) – The address of the geoserver housing the layer to be queried. Default: https://pavics.ouranos.ca/geoserver/.

Returns:

Basins ids including fid and its upstream contributors.

Return type:

gpd.GeoDataFrame

ravenpy.utilities.geoserver.hydrobasins_aggregate(gdf: DataFrame) → DataFrame

Aggregate multiple HydroBASINS watersheds into a single geometry.

Parameters:

gdf (pd.DataFrame) – Watershed attributes indexed by HYBAS_ID.

Return type:

pd.DataFrame

ravenpy.utilities.geoserver.hydrobasins_upstream(feature: dict, domain: str) → DataFrame

Return a list of HydroBASINS features located upstream.

Parameters:

• feature (dict) – Basin feature attributes, including the fields [“HYBAS_ID”, “NEXT_DOWN”, “MAIN_BAS”].

• domain ({"na", "ar"}) – Domain of the feature, North America or Arctic.

Returns:

Basins ids including fid and its upstream contributors.

Return type:

pd.Series

ravenpy.utilities.geoserver.select_hybas_domain(bbox: tuple[int | float, int | float, int | float, int | float] | None = None, point: tuple[int | float, int | float] | None = None) → str

Provided a given coordinate or boundary box, return the domain name of the geographic region the coordinate is located within.

Parameters:

• bbox (Optional[Tuple[Union[float, int], Union[float, int], Union[float, int], Union[float, int]]]) – Geographic coordinates of the bounding box (left, down, right, up).

• point (Optional[Tuple[Union[float, int], Union[float, int]]]) – Geographic coordinates of an intersecting point (lon, lat).

Returns:

The domain that the coordinate falls within. Possible results: “na”, “ar”.

Return type:

str

Graphics

Library to perform graphs for the streamflow time series analysis.

The following graphs can be plotted:

• hydrograph

• mean_annual_hydrograph

• spaghetti_annual_hydrograph

ravenpy.utilities.graphs.forecast(file: str | Path, fcst_var: str = 'q_sim') → Figure

Create a graphic of the hydrograph for each forecast member.

Parameters:

• file (str or Path) – Raven output file containing simulated streamflows.

• fcst_var (str) – Name of the streamflow variable.

Return type:

matplotlib.pyplot.Figure

ravenpy.utilities.graphs.hindcast(file: str | Path, fcst_var: str, qobs: str | Path, qobs_var: str) → Figure

Create a graphic of the hydrograph for each hindcast member.

Parameters:

• file (str or Path) – Raven output file containing simulated streamflows.

• fcst_var (str) – Name of the streamflow variable.

• qobs (str or Path) – Streamflow observation file, with times matching the hindcast.

• qobs_var (str) – Name of the streamflow observation variable.

Return type:

matplotlib.pyplot.Figure

ravenpy.utilities.graphs.hydrograph(file_list: Sequence[str | Path])

Create a graphic of the hydrograph for each model simulation.

Parameters:

file_list (Sequence of str or Path) – Raven output files containing simulated streamflows.

ravenpy.utilities.graphs.mean_annual_hydrograph(file_list: Sequence[str | Path])

Create a graphic of the mean hydrological cycle for each model simulation.

Parameters:

file_list (Sequence of str or Path) – Raven output files containing simulated streamflows.

ravenpy.utilities.graphs.spaghetti_annual_hydrograph(file: str | Path)

Create a spaghetti plot of the mean hydrological cycle for one model simulations.

The mean simulation is also displayed.

Parameters:

file (str or Path) – Raven output files containing simulated streamflows of one model.

ravenpy.utilities.graphs.ts_fit_graph(ts: DataArray, params: DataArray) → Figure

Create graphic showing a histogram of the data and the distribution fitted to it.

The graphic contains one panel per watershed.

Parameters:

• ts (xr.DataArray) – Stream flow time series with dimensions (time, nbasins).

• params (xr.DataArray) – Fitted distribution parameters returned by xclim.land.fit indicator.

Returns:

Figure showing a histogram and the parameterized pdf.

Return type:

matplotlib.pyplot.Figure

ravenpy.utilities.graphs.ts_graphs(file, trend: bool = True, alpha: float = 0.05)

Create a figure with the statistics so one can see a trend in the data.

Graphs for time series statistics.

Parameters:

• file (str or Path) – xarray-compatible file containing streamflow statistics for one run.

• trend (bool) – If True, the slope will be plotted.

• alpha (float) – Significance level for the Mann-Kendall test.

Regionalization

Tools for hydrological regionalization.

ravenpy.utilities.regionalization.IDW(qsims: DataArray, dist: Series) → DataArray

Inverse distance weighting.

Parameters:

• qsims (xr.DataArray) – Ensemble of hydrogram stacked along the members dimension.

• dist (pd.Series) – Distance from catchment which generated each hydrogram to target catchment.

Returns:

Inverse distance weighted average of ensemble.

Return type:

xr.DataArray

ravenpy.utilities.regionalization.distance(gauged: DataFrame, ungauged: Series) → Series

Return geographic distance [km] between ungauged and database of gauged catchments.

Parameters:

• gauged (pd.DataFrame) – Table containing columns for longitude and latitude of catchment’s centroid.

• ungauged (pd.Series) – Coordinates of the ungauged catchment.

Return type:

pd.Series

ravenpy.utilities.regionalization.multiple_linear_regression(source: DataFrame, params: DataFrame, target: DataFrame) → tuple[list[Any], list[Callable[[], Any]]]

Multiple Linear Regression for model parameters over catchment properties.

Uses known catchment properties and model parameters to estimate model parameter over an ungauged catchment using its properties.

Parameters:

• source (pd.DataFrame) – Properties of gauged catchments.

• params (pd.DataFrame) – Model parameters of gauged catchments.

• target (pd.DataFrame) – Properties of the ungauged catchment.

Returns:

A named tuple of the estimated model parameters and the R2 of the linear regression.

Return type:

list of Any, list of Callable or Any

ravenpy.utilities.regionalization.read_gauged_params(model) → tuple[Series, DataFrame]

Return table of NASH-Sutcliffe Efficiency values and model parameters for North American catchments.

Returns:

• pd.Series – Nash-Sutcliffe Efficiency keyed by catchment ID.

• pd.DataFrame – Model parameters keyed by catchment ID.

ravenpy.utilities.regionalization.read_gauged_properties(properties) → DataFrame

Return table of gauged catchments properties over North America.

Returns:

Catchment properties keyed by catchment ID.

Return type:

pd.DataFrame

ravenpy.utilities.regionalization.regionalization_params(method: str, gauged_params: DataFrame, gauged_properties: DataFrame, ungauged_properties: DataFrame, filtered_params: DataFrame, filtered_prop: DataFrame) → list[list[float]] | list[ndarray]

Return the model parameters to use for the regionalization.

Parameters:

• method ({'MLR', 'SP', 'PS', 'SP_IDW', 'PS_IDW', 'SP_IDW_RA', 'PS_IDW_RA'}) – Name of the regionalization method to use.

• gauged_params (pd.DataFrame) – A DataFrame of parameters for donor catchments (size = number of donors).

• gauged_properties (pd.DataFrame) – A DataFrame of properties of the donor catchments (size = number of donors).

• ungauged_properties (pd.DataFrame) – A DataFrame of properties of the ungauged catchment (size = 1).

• filtered_params (pd.DataFrame) – A DataFrame of parameters of all filtered catchments (size = all catchments with NSE > min_NSE).

• filtered_prop (pd.DataFrame) – A DataFrame of properties of all filtered catchments (size = all catchments with NSE > min_NSE).

Returns:

A list of model parameters to be used for the regionalization.

Return type:

list

ravenpy.utilities.regionalization.regionalize(config: Config, method: str, nash: Series, params: DataFrame | None = None, props: DataFrame | None = None, target_props: Series | dict | None = None, size: int = 5, min_NSE: float = 0.6, workdir: str | Path | None = None, overwrite: bool = False) → tuple[DataArray, Dataset]

Perform regionalization for catchment whose outlet is defined by coordinates.

Parameters:

• config (ravenpy.config.rvs.Config) – Symbolic emulator configuration. Only GR4JCN, HMETS and Mohyse are supported.

• method ({'MLR', 'SP', 'PS', 'SP_IDW', 'PS_IDW', 'SP_IDW_RA', 'PS_IDW_RA'}) – Name of the regionalization method to use.

• nash (pd.Series) – NSE values for the parameters of gauged catchments.

• params (pd.DataFrame) – Model parameters of gauged catchments. Needed for all but MRL method.

• props (pd.DataFrame) – Properties of gauged catchments to be analyzed for the regionalization. Needed for MLR and RA methods.

• target_props (pd.Series or dict) – Properties of ungauged catchment. Needed for MLR and RA methods.

• size (int) – Number of catchments to use in the regionalization.

• min_NSE (float) – Minimum calibration NSE value required to be considered as a donor.

• workdir (Union[str, Path]) – Work directory. If None, a temporary directory will be created.

• overwrite (bool) – If True, existing files will be overwritten.

Returns:

• qsim (DataArray (time, )) – Multi-donor averaged predicted streamflow.

• ensemble (Dataset) – A Dataset containing the ensemble of simulations and parameters used:

  • q_sim : DataArray (realization, time) Ensemble of members based on number of donors.

  • parameter : DataArray (realization, param) Parameters used to run the model.

ravenpy.utilities.regionalization.similarity(gauged: DataFrame, ungauged: DataFrame, kind: str = 'ptp') → Series

Return similarity measure between gauged and ungauged catchments.

Parameters:

• gauged (pd.DataFrame) – Gauged catchment properties.

• ungauged (pd.DataFrame) – Ungauged catchment properties.

• kind ({'ptp', 'std', 'iqr'}) – Normalization method: peak to peak (maximum - minimum), standard deviation, inter-quartile range.

Return type:

pd.Series