wolfhece.hydrology.hyetograms

Module Contents

class wolfhece.hydrology.hyetograms.ClimateData_IRM(workingDir: str | pathlib.Path = Path('P:\\Donnees\\Pluies\\IRM\\climateGrid'))

Class to handle IRM climate data.

Data are available for academic purposes only on https://opendata.meteo.be/. Please read carefully the terms of use before using this data. Without valid data, the class will not work.

workingDir

_data: pandas.DataFrame = None

_grid: wolfhece.PyVertexvectors.Zones = None

_kdtree = None

_start_time

_end_time

property variables: list[str]

Return the list of variables available in the IRM data.

Available variables should be : - day - temp_max - temp_min - temp_avg - precip_quantity - humidity_relative - pressure - sun_duration - short_wave_from_sky - evapotrans_ref

Values are from +0800 of the day, to +0800 of the next day.

property data: pandas.DataFrame

Return the data as a pandas DataFrame. If the data is not read yet, return None.

read_data(all: bool = True, start_yearmonth: tuple[int, int] = None, end_yearmonth: tuple[int, int] = None)

Read the IRM data from the working directory.

_create_zones()

Create zones for the pixels and their squares.

plot(figax=None, title: str = None, with_ids: bool = False)

Plot the GRID for IRM data.

as_zones() → wolfhece.PyVertexvectors.Zones

Return the grid as a Zones object. If the zones are not created yet, create them.

get_attribute4date(date: datetime.datetime, variable: str) → pandas.DataFrame

Return the attribute for a specific date.

get_attribute4daterange(date_start: datetime.datetime, date_end: datetime.datetime, variable: str) → pandas.DataFrame

Return the attribute for a specific date range.

plot_spatial_attribute(date: datetime.datetime, variable: str, figax=None, title: str = None, cmap: str = 'viridis')

Plot the spatial distribution of the attribute for a specific date.

animation_spatial_attribute(variable: str, figax=None, date_start: datetime.datetime = 0, date_end: datetime.datetime = -1, cmap: str = 'viridis')

Create an animation of the spatial distribution of a specific attribute over time. The animation will show the attribute data for each time step.

Parameters:

• figax – A tuple (fig, ax) to use for the animation. If None, a new figure and axes will be created.

• date_start – The starting date for the animation.

• date_end – The ending date for the animation. If -1, it will use the last date.

• cmap – The colormap to use for the attribute data.

• interval_days – The interval between frames in days.

Returns:

The animation object.

find_pixelid_from_X_Y(x: float, y: float) → int

Find the pixel id from the X and Y coordinates. :param x: The X coordinate. :param y: The Y coordinate. :return: The pixel id or None if not found.

plot_hyetogram(position: list[float, float] | tuple[float, float], date_start: datetime.datetime = 0, date_end: datetime.datetime = -1, figax=None)

Plot the hyetogram for a specific position over a date range.

Parameters:

• position – The position (x, y) for which to plot the hyetogram.

• date_start – The starting date for the hyetogram.

• date_end – The ending date for the hyetogram. If -1, it will use the last date.

• figax – A tuple (fig, ax) to use for the plot. If None, a new figure and axes will be created.

class wolfhece.hydrology.hyetograms.Rainfall_Gridded(workingDir: str | pathlib.Path, type_of_rain: int = source_IRM | source_dist)

Class to handle gridded rainfall data. It can read gridded rainfall data.

Attention

• The grid is provided as a shapefile in the “Grid” subdirectory of the working directory.

• The associated data are provided in the “data” or “IRM” subdirectory of the working directory.

• The link between each polygon and the data is done thorugh the index of the polygon in the shapefile. BUT, as we come from Fortran-world, the index is supposed 1-based, not 0-based. The increment of the index is done in the “PyVertexvectors” module and checked in the “_read_grid” routine.

workingDir

type_of_rain

is_binary = False

_grid: wolfhece.PyVertexvectors.Zones = None

_data: dict[str | int, numpy.ndarray]

_times: list[datetime.datetime] = []

gridDir

dataDir

as_zones() → wolfhece.PyVertexvectors.Zones

Return the grid as a Zones object.

property time_steps: list[datetime.timedelta]

Return the time step between each data point.

has_uniform_time_step() → bool

Check if the time step is uniform.

property time_step: datetime.timedelta

Return the time step between the two first data.

read()

Read grid and data from the working directory.

_read_grid()

Read the grid data from the grid directory. The grid data should be in a shapefile format.

_read_associated_data()

Read the date associeted to the grid.

_read_rain_ascii(filename: str | pathlib.Path) → tuple[str | int, numpy.ndarray, list[datetime.datetime]]

Read data from an ASCII file. The filename should end with .rain.

Structure of the ASCII file: - 4 Header lines with :

• The first line is a name of the series.

• The second line is the number of data columns (n).

• The third line is the total number of columns (n + 6).

• The fourth line is the number of rows.

• Each line represents a time step.

  • The first six columns are the day, month, year, hour, minute, and second.

  • The last column is the rain value.

Parameters:

filename – The name of the file to read.

_read_rain_binary(filename: str | pathlib.Path) → tuple[int, numpy.ndarray, list[datetime.datetime]]

Read data from a binary file. The filename should end with .dat.

Structure of the binary file: - 4 bytes for a “name” as integer - 4 bytes for the number of data columns (n) - 4 bytes for the total number of columns (n + 6) - 4 bytes for the number of rows - For each row:

• 1 byte for the day

• 1 byte for the month

• 2 bytes for the year

• 1 byte for the hour

• 1 byte for the minute

• 1 byte for the second

• n*8 bytes for the rain value as float

Parameters:

filename – The name of the file to read.

Returns:

A numpy array with the rain data.

get_rain4index(index: int) → dict[str | int, float]

Get the rain data for a given index. The index should be an integer representing the time step position.

Parameters:

index – The index time for which to get the rain data (0-based).

Returns:

A dictionary with the zone name as key and the rain value as value.

plot_spatial_rain4index(index: int, figax=None, title: str = None, cmap: str = 'viridis')

Plot the spatial distribution of rain for a given index.

animation_spatial_rain_index(figax=None, idx_start: int = 0, idx_end: int = -1, cmap: str = 'viridis', interval: int = 100)

Create an animation of the spatial distribution of rain over time. The animation will show the rain data for each time step.

Parameters:

• figax – A tuple (fig, ax) to use for the animation. If None, a new figure and axes will be created.

• idx_start – The starting index for the animation (0-based).

• idx_end – The ending index for the animation (0-based). If -1, it will use the last index.

• cmap – The colormap to use for the rain data.

• interval – The interval between frames in milliseconds.

Returns:

The animation object.

animation_spatial_rain_date(figax=None, date_start: datetime.datetime = 0, date_end: int = -1, cmap: str = 'viridis', interval: int = 100)

Create an animation of the spatial distribution of rain over time. The animation will show the rain data for each time step.

Parameters:

• figax – A tuple (fig, ax) to use for the animation. If None, a new figure and axes will be created.

• idx_start – The starting index for the animation (0-based).

• idx_end – The ending index for the animation (0-based). If -1, it will use the last index.

• cmap – The colormap to use for the rain data.

• interval – The interval between frames in milliseconds.

Returns:

The animation object.

class wolfhece.hydrology.hyetograms.Rainfall_Polygons(workingDir: str | pathlib.Path, type_of_rain: int)

For source_point_measurements:

« ind_unique.txt » : contient une matrice avec autant de lignes qu’il y a de configurations et autant de colonne qu’il y a de stations. Les lignes représentent le numéro de la configuration et les colonnes représentent les stations. La variable stockée dans cette matrice est l’indice dans la liste de pluies mesurées à une station auquel commencer pour cette configuration.

« nb_ind_unique.txt » : vecteur contenant le nombre de pas de temps à considérer pour chaque configuration.

« unique.txt » : représente le code de la configuration. La valeur stockée est la conversion en entier sur 8 bytes du code binaire dont le premier élément est la première station et sa valeur est égale à 1 si celle-ci est utilisée dans la configuration présente. Cette définition a pour effet de limiter le nombre de stations exploitable à 64 éléments par bassin versant étudié. Dans l’attribution des pluies aux mailles, toutes les configurations sont parcourues dans l’ordre des lignes de la matrice contenue dans le fichier « ind_unique.txt » pour construire progressivement les pluies de bassin.

For source_IRM:

Une seule configuration spatiale est présente. Pas de fichiers « ind_unique.txt » et « nb_ind_unique.txt ».

workingDir

type_of_rain

hyetoDict

configs

_codes: dict[int, int]

_nbsteps4code: dict[int, int]

_steps4eachcode

_geometries: dict[int, dict[str:Zones]]

_hyetograms: dict[str, dict[str, numpy.ndarray]]

_checks()

Perform checks on the data.

get_computed_steps4code(code: int, base: int = 0) → list[int]

Get all computed time steps for a given code.

ATTENTION : it will return index positions. By default, the base is 0 (Python base).

Parameters:

• code – The code for which to get the steps.

• base – The base to use for the steps (default is 0 == Python, 1 == Fortran).

get_config4date(date: datetime.datetime) → int

Get the configuration for a given date. The date should be in the format ‘datetime.datetime’.

Parameters:

date – The date for which to get the configuration.

get_config4timestamp(timestamp: int) → int

Get the configuration for a given timestamp. The timestamp should be an integer representing the seconds since epoch.

Parameters:

timestamp – The timestamp for which to get the configuration.

get_config4index(index: int) → int

Get the configuration for a given index time. The index should be an integer representing the position in time (0-based).

Parameters:

index – The index for which to get the configuration (0-based).

Returns:

The configuration key for the given index (not the code itself).

get_code4index(index: int) → int

Get the code for a given index time. The index should be an integer representing the position in time (0-based).

Parameters:

index – The index for which to get the code (0-based).

Returns:

The configuration code for the given index.

get_code4date(date: datetime.datetime) → int

Get the code for a given date. The date should be in the format ‘datetime.datetime’.

Parameters:

date – The date for which to get the code.

Returns:

The configuration code for the given date.

get_code4timestamp(timestamp: int) → int

Get the code for a given timestamp. The timestamp should be an integer representing the seconds since epoch.

Parameters:

timestamp – The timestamp for which to get the code.

Returns:

The configuration code for the given timestamp.

get_geometry4index(index: int, all_polygons: bool = True) → wolfhece.PyVertexvectors.Zones

Get the geometry for a given index. The index should be an integer representing the time step position.

Parameters:

• index – The index time for which to get the geometry (0-based).

• all_polygons – If True, return all polygons, otherwise return only the used polygons.

Returns:

The geometry for the given index.

get_geometry4code(code: int, all_polygons: bool = True) → wolfhece.PyVertexvectors.Zones

Get the geometry for a given configuration code. The code should be an integer representing the configuration.

Parameters:

• code – The configuration code for which to get the geometry.

• all_polygons – If True, return all polygons, otherwise return only the used polygons.

get_geometry4codeindex(code_index: int, all_polygons: bool = True) → wolfhece.PyVertexvectors.Zones

Get the geometry for a given configuration code. The code should be an integer representing the configuration.

Parameters:

• code_index – The index of the code in the list of codes (1-based index).

• all_polygons – If True, return all polygons, otherwise return only the used polygons.

property nb_steps4code: list[tuple[int, int]]

Get the number of steps for each code. Returns a list of tuples (code, number of steps).

property nb_steps4code_asdict: dict[int, int]

Get the number of steps for each code as a dictionary. Returns a dictionary with the code as key and the number of steps as value.

get_geometries(n_more_frequent: int = 5, all_poygons: bool = False) → dict[int, wolfhece.PyVertexvectors.Zones]

Get the geometries for the most frequent configurations. The n_more_frequent parameter defines how many configurations to return.

Parameters:

• n_more_frequent – The number of most frequent configurations to return.

• all_poygons – If True, return all polygons, otherwise return only the used polygons.

Returns:

A dictionary with the configuration code as key and the Zones object as value.

get_most_frequent_code(n_more_frequent: int = 5) → list[int]

Get the most frequent configurations codes. The n_more_frequent parameter defines how many configurations to return.

Parameters:

n_more_frequent – The number of most frequent configurations to return.

Returns:

A list of the most frequent configuration codes.

get_sorted_codes() → list[int]

Get the sorted configuration codes based on the number of steps. Returns a list of configuration codes sorted by the number of steps in descending order.

Returns:

A list of configuration codes sorted by the number of steps.

get_hyetograms4index(index: int) → list

Get the hyetograms for a given index time. The index should be an integer representing the time step position.

Parameters:

index – The index time for which to get the hyetograms (0-based).

Returns:

A list of hyetograms for the given index.

get_rains4index(index: int) → numpy.ndarray

Get the rain data for a given index time. The index should be an integer representing the time step position.

Parameters:

index – The index time for which to get the rain data (0-based).

Returns:

A numpy array with the rain data for the given index.

get_rains4timestamp(timestamp: int) → numpy.ndarray

Get the rain data for a given timestamp. The timestamp should be an integer representing the seconds since epoch.

Parameters:

timestamp – The timestamp for which to get the rain data.

Returns:

A numpy array with the rain data for the given timestamp.

get_rains4date(date: datetime.datetime) → numpy.ndarray

Get the rain data for a given date. The date should be in the format ‘datetime.datetime’.

Parameters:

date – The date for which to get the rain data.

Returns:

A numpy array with the rain data for the given date.

get_rains4code(code: int) → numpy.ndarray

Get the rain data for given configuration.

Parameters:

code – The configuration code for which to get the rain data.

Returns:

A numpy array with the rain data for the given code.

get_footprint_and_rain4index(index: int) → tuple

Get the footprint and rain data for a given index. The index should be an integer representing the time step position.

Parameters:

index – The index time for which to get the footprint and rain data (0-based).

Returns:

A tuple containing the footprints (as vector objects) and the rain data (as numpy array).

get_footprint_and_rain4code(code: int) → tuple

Get the footprint and rain data for a given configuration code. The code should be an integer representing the configuration.

Parameters:

code – The configuration code for which to get the footprint and rain data.

Returns:

A tuple containing the footprints (as vector objects), the rain data (as numpy array) and the times.

get_most_rainy_code(n_most: int = 5) → int

Get the configuration code with the most frequently rain.

We identify the number of consecutive steps with rain values greater than zero.

get_most_relative_rainy_code(n_most: int = 5) → int

Get the configuration code with the most frequently rain.

We identify the number of consecutive steps with rain values greater than zero.

property rain_maximum: float

Returns the maximum rain value across all configurations.

property rain_maxima: dict[int, float]

Returns a dictionary with the maximum rain value for each configuration code. The key is the configuration code and the value is the maximum rain value.

property timestamps

Returns the timestamps list

get_timestamp_from_index(index: int) → int

Get the timestamp for a given index. The index should be an integer representing the time step position.

Parameters:

index – The index time for which to get the timestamp (0-based).

property times

Returns the time list for the first configuration.

property time_begin

Returns the beginning time of the first configuration.

property time_end

Returns the end time of the last configuration.

property number_of_configurations

Returns the number of unique configurations.

property nb_records

Returns the number of records (scenarios).

_decode_config(value: int) → list

Decode the unique configuration from an integer value. The value is a binary representation where each bit represents whether a station is included in the configuration.

_code_config(config: list) → int

Encode a unique configuration from a list of 0s and 1s into an integer. The list represents whether each station is included in the configuration.

_read_geometries()

Read the geometries from the directory. The geometries are stored in vector files named as: Rain_basin_geom_<code>.vec and Rain_basin_geom_<code>_all_zones.vec where <code> is the configuration code.

_read_configurations()

Read the unique configurations from a text file.

The configurations are stored in the following files: - unique.txt: contains the unique configurations as integer codes. - nb_ind_unique.txt: contains the number of steps for each configuration. - ind_unique.txt: contains the steps for each configuration. - input_data_gap.txt: contains the gap between the first and the second configuration. - scenarios.txt: contains the scenarios (time, configuration). The files should be located in the “Whole_basin” directory. The working directory should be set to the directory containing the “Whole_basin” directory.

_read_hyetograms()

Read the hyetograms from the directory.

The hyetograms are stored in files named as: rain<code>.hyeto where <code> is the configuration code. The files should be located in the “Whole_basin” directory. The working directory should be set to the directory containing the “Whole_basin” directory.

_write_rain_binary(name_serie: int, filename: str | pathlib.Path, data: numpy.ndarray, times: list[datetime.datetime] = None)

Write data to a binary file. The filename should end with .dat.

Structure of the binary file: - 4 bytes for a “name” as integer - 4 bytes for the number of data columns (n) - 4 bytes for the total number of columns (n + 6) - 4 bytes for the number of rows - For each row:

• 1 byte for the day

• 1 byte for the month

• 2 bytes for the year

• 1 byte for the hour

• 1 byte for the minute

• 1 byte for the second

• n*8 bytes for the rain value as float

Parameters:

• filename – The name of the file to write.

• data – The data to write, should be a 1D numpy array.

_read_rain_binary(filename: str | pathlib.Path) → tuple[int, numpy.ndarray, list[datetime.datetime]]

Read data from a binary file. The filename should end with .dat.

Structure of the binary file: - 4 bytes for a “name” as integer - 4 bytes for the number of data columns (n) - 4 bytes for the total number of columns (n + 6) - 4 bytes for the number of rows - For each row:

• 1 byte for the day

• 1 byte for the month

• 2 bytes for the year

• 1 byte for the hour

• 1 byte for the minute

• 1 byte for the second

• n*8 bytes for the rain value as float

Parameters:

filename – The name of the file to read.

Returns:

A numpy array with the rain data.

_write_rain_ascii(name_serie: str | int, filename: str | pathlib.Path, data: numpy.ndarray, times: list[datetime.datetime] = None)

Write data to an ASCII file. The filename should end with .rain.

Structure of the ASCII file: - 4 Header lines with :

• The first line is a name of the series.

• The second line is the number of data columns (n).

• The third line is the total number of columns (n + 6).

• The fourth line is the number of rows.

• Each line represents a time step.

  • The first six columns are the day, month, year, hour, minute, and second.

  • The last column is the rain value.

Parameters:

• name_serie – The name of the series, can be an integer or a string.

• filename – The name of the file to write.

• data – The data to write, should be a 1D numpy array.

_read_rain_ascii(filename: str | pathlib.Path) → tuple[str | int, numpy.ndarray, list[datetime.datetime]]

Read data from an ASCII file. The filename should end with .rain.

Structure of the ASCII file: - 4 Header lines with :

• The first line is a name of the series.

• The second line is the number of data columns (n).

• The third line is the total number of columns (n + 6).

• The fourth line is the number of rows.

• Each line represents a time step.

  • The first six columns are the day, month, year, hour, minute, and second.

  • The last column is the rain value.

Parameters:

filename – The name of the file to read.

convert2grid(grid: pathlib.Path | str | wolfhece.PyVertexvectors.Zones, output: pathlib.Path | str = None, overwrite: bool = True, parallel: bool = True)

Convert the data to a grid.

The grid can be a Path or a Zones object. If a Path is provided, it should point to a grid file. If a Zones object is provided, it should contain the grid polygons. The output will be written to the specified output directory. If the output directory already exists, it will be overwritten if overwrite is set to True. If output is None, the output will be written to the parent directory of the grid file, ‘data’ subdirectory. If parallel is set to True, the computation will be done in parallel using threads.

Parameters:

• grid – The grid to convert the data to.

• output – The output directory where the data will be written.

• overwrite – If True, the output directory will be overwritten if it already exists.

• parallel – If True, the computation will be done in parallel using threads.

plot_spatial_rain4index(index: int, figax=None, title: str = None, cmap: str = 'viridis')

Plot the spatial distribution of rain for a given index. The index should be an integer representing the time step position.

animation_spatial_rain_index(code: int = 1, idx_start: int = 0, idx_end: int = -1, figax=None, cmap: str = 'viridis', interval: int = 100)

Create an animation of the spatial distribution of rain for all indices. The animation will be displayed using matplotlib’s FuncAnimation.

animation_spatial_rain_date(code: int = 1, date_start: datetime.datetime = 0, date_end: datetime.datetime = -1, figax=None, cmap: str = 'viridis', interval: int = 100)

Create an animation of the spatial distribution of rain for all indices within a date range. The animation will be displayed using matplotlib’s FuncAnimation.