wolfhece.hydrology.Catchment

Author: HECE - University of Liege, Pierre Archambeau, Christophe Dessers Date: 2024

Copyright (c) 2024 University of Liege. All rights reserved.

This script and its content are protected by copyright law. Unauthorized copying or distribution of this file, via any medium, is strictly prohibited.

Module Contents

class wolfhece.hydrology.Catchment.Catchment(_name, _workingDir, _plotAllSub, _plotNothing, _initWithResults=True, _catchmentFileName='', _rbFileName='', _tz=0, version=cst.VERSION_WOLFHYDRO)

This object contains all the information about the Catchment.

The Catchment is composed of element that can be:

• Subbasin

• RetentionBasin

In the Catchment the following rules are applied:

• there is one and only one outlet.

• one element

property simulation_intervals: list[tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime, wolfhece.hydrology.RetentionBasin.datetime.datetime]]

property temporal_parameters: tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime, wolfhece.hydrology.RetentionBasin.datetime.datetime]

time_delays_F: wolfhece.PyParams.np.ndarray

_version: float

charact_watrshd: wolfhece.hydrology.PyWatershed.Watershed

subBasinCloud: wolfhece.PyVertex.cloud_vertices

retentionBasinCloud: wolfhece.PyVertex.cloud_vertices

iP_Cloud: wolfhece.PyVertex.cloud_vertices

catchmentDict: dict[wolfhece.PyParams.Union[str, int], wolfhece.PyParams.Union[wolfhece.hydrology.RetentionBasin.SubBasin, wolfhece.hydrology.RetentionBasin.RetentionBasin]]

subBasinDict: dict[int, wolfhece.hydrology.RetentionBasin.SubBasin]

get_subBasin(id_sorted_or_name: int | str) → wolfhece.hydrology.RetentionBasin.SubBasin

This method returns the subbasin object associated with the sorted id or name given in argument.

The sorted id is the one given by the Fortran code.

get_time()

This method saves the time characteristics read in the .param file and build a time array. The convention used for the time written in the .param is as follow: YYYYMMDD-HHMMSS It is important to notice that the dateBegin and dateEnd are converted in GMT+0

Internal variables modified : self.deltaT, self.dateBegin, self.dateEnd, self.time.

NBIf any change in the convention is mentionned in the comment following the dates the code will return an error

Otherwise, no procedure will warn the user that a converntion is modified.

get_rain(workingDir)

get_hyeto(fileName)

init_dictIdConversion(workingDir)

Procedure that converts the Id of the intersection points in the input file into the sorted sub-basin ids file read: simul_sorted_interior_points.txt Internal variables modified: self.dictIdConversion

The conversion dictionnary has the given form: self.dictIdConversion[Internal Point id] = sorted subbasin from the lowest to the highest

create_ObjectsInCatchment()

Procedure which creates the objects in the dictionnaries of the subbasins and the RB and each object are pointed in the global catchment dictionnary. This procedre also create the 1st level of the Topo dictionnary. Internal variables modified: subBasinDict, retentionBasinDict, self.catchmentDict,

_fill_cloud_retentionbasin()

This procedure fills the cloud of the retention basin with the vertices of the retention basin and its inlets.

get_retentionbasin_zones() → wolfhece.wolf_array.Zones

This method returns a Zones instance of the retention basins.

link_objects()

This procedure link all the subbasins and the retention basin altogether to form a network. If a subbasin without inlet whatsoever is detected, one adds it to the first level of the Topology tree.

Internal variables modified: subBasinDict, retentionBasinDict, topologyDict

complete_topoDict()

Procedure that finish to complete the topo tree. Before calling this procedure, only the first level were completed in self.link_objects().

Modified internal variables: self.topologyDict

Strategy:

• We save the element on a certain level if all his inlets have inferior levels.

TO DO: Add a test that counts the number of subbasins and check if the number is right or not

-> take into account the fact that a subbasin can be the last element in this computation.

construct_hydro(firstLevel=1, lastLevel=-1, fromStation: str = '', updateAll: bool = False)

This procedure will use the topo tree to build the hydrographs of all elements

Internal variable changed: self.catchmentDict

read_Jsonfile(fileName)

Function which reads a json file as input

save_ExcelFile()

Procedure that saves the data in an Excel file.

save_ExcelFile_noLagTime()

Procedure that saves the data in an Excel file.

save_ExcelFile_inlets_noLagTime()

Procedure that saves the inlets in an Excel file.

save_ExcelFile_V2()

Procedure that saves the data in an Excel file.

save_ExcelFile_Vesdre_simul2D(fileName='PostProcess/Data_simul2D.xlsx', directory='')

Procedure that saves the data in an Excel file.

save_ExcelFile_Vesdre_all(fileName='PostProcess/Data_simul2D.xlsx', directory='')

Procedure that saves the data in an Excel file.

save_characteristics()

Procedure that saves the data in an Excel file.

read_dbfFile(fileName)

init_hyetoDict()

Procedure that saves the all the hyeto data of the Catchment in a dictionnary

Internal variables modified: self.hyetoDict

Structure du dictionnaire self.hyetoDict:

• Ordered To Nb: self.hyetoDict[‘Ordered To Nb’][nb] = iD

  • iD = hyeto nb of the file to read in the folder “Whole_basin”. E.g.: the file “[iD]evap.hyeto”

  • nb = hyeto number sorted from 1 to nbHyeto. E.g.: if iD=[‘2’, ‘5’, ‘7’] => self.hyetoDict[‘Ordered To Nb’][2] = ‘5’

• Hyetosself.hyetoDict[‘Hyetos’][nb]

  • time: time array read in the .hyeto file

  • rain: rain array read in the .hyeto file

TO DO: Consider a more general way to detect .dbf files. E.G. when the data are read in NetCDF files.

add_rainToAllObjects()

Add rain and hyetographs to all subbasins

plot_intersection(rangeData=[], plot_raw=True, tzPlot=0, axis: str = 'Datetime', show: bool = True)

This procedure will plot all the subbasins or RB with level>1 in the topo tree.

plot_allSub(withEvap=False, withCt=False, selection_by_iD=[], graph_title='', show=True, writeDir='', figure=None, Measures=[])

This procedure plots the hydrographs and hyetographs of all subbasins

plot_allJct(withEvap=False, selection_by_key=[], graph_title='', show=True, writeDir='', figure=None, Measures=[], Measure_unit='m3/s', addTable=False, rangeData=[])

This procedure plots the hydrographs and hyetographs of all junctions

draw_flowChart(flowchart)

This procedure save and plot a flowchart representing the topo tree input: - flowchart: graphviz.Digraph Object -> modified at the end

make_stat_distributionOfslope()

This procedure plot the stat distribution of slopes.

make_stat_distributionOfTime()

This procedure plot the stat distribution of slopes.

check_massConservation()

This procedure check whether the mass conservation is verified ot not.

copy()

get_time_btw_ssbasins(stationNames=[], unit='h', ref=[])

Returns the transfer time between susBasin modules

operation_on_param_file(fileName, which, operation=None)

_correct_Umax_from_old_model(adapt_with_rain: bool = True)

plot_all_diff_cumulRain_with_lagtime(interval, lagTime, selection_by_iD=[], graph_title='', show=True, writeDir='', lawNetRain=0, netRainParams={})

get_all_cumulRain(selection_by_iD=[]) → tuple[wolfhece.PyParams.np.array, list[wolfhece.PyParams.np.array]]

read_measuring_stations_SPW(fileNameIn='')

Function that read the stations, their characteristics and locations and store all information in a dictionnary

get_eff_subBasin()

get_one_surfaces_proportions(whichName)

update_hydro(idCompar, fromLevel: bool = False, level_min: int = 1)

get_cvg(pointerData)

define_station_out(stationOut)

sort_level_given_junctions(givenJct: list, changeNames=False)

construct_surfaceDrainedHydro_RB()

activate_usefulSubs(mask=[], blockJunction=[], onlyItself: bool = False)

get_key_catchmentDict(name)

get_keys_catchmentDict(names: list)

write_eff_subBasin()

read_hydro_eff_subBasin()

update_timeDelay(stationName, value=0.0, reset=False)

update_timeDelays_from_F(stationName, value=0.0, reset=False)

reset_timeDelay(stationOut='')

get_inletsName(stationOut)

get_all_timeDelay(ref='')

get_timeDelays_inlets(ref: str = '') → dict[str, float]

save_timeDelays(junctionList: list)

get_lastJunctionKey()

set_timeDelays(method='wolf_array', junctionKey='', readWolf: bool = True, updateAll: bool = False)

find_all_timeDelayObj()

plot_landuses(selection_by_iD: list = [], graph_title: str = '', onlySub: bool = True, show: bool = True, writeDir: str = '', figure=None)

read_all_landuses()

build_compare_file(hydroFile: str, unit: str = 'mm/h', otherSurf: float = 0.0, lag: int = 0, junction: str = '', dateBegin: wolfhece.hydrology.RetentionBasin.datetime.datetime = None, dateEnd: wolfhece.hydrology.RetentionBasin.datetime.datetime = None, deltaT: wolfhece.hydrology.RetentionBasin.datetime.timedelta = None)

change_version(newVersion=None)

get_version()

get_sub_Nash(measure: wolfhece.hydrology.RetentionBasin.SubBasin, selection_by_iD, intervals: list[tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime]] = [])

get_sub_peak(selection_by_iD, intervals: list[tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime]] = [])

set_eff_outlet_coord()

update_charact_watrshd()

get_all_x_production(selection_by_iD: list = [], to_save: bool = True, to_plot: bool = False) → dict

Retrieves the x production values for all sub-basins or a specific selection of sub-basins.

Parameters:

selection_by_iD (list, optional) – A list of sub-basin IDs to retrieve x production values for. If empty, retrieves x production values for all sub-basins. Defaults to [].

Returns:

A dictionary containing the sub-basin names or IDs as keys and their corresponding x production values as values.

Return type:

dict

get_all_fractions(plt_dict: dict[str:np.array] = {}, selection_by_iD: list = [], to_save: bool = True, to_plot: bool = False, summary: str = None, summary_interval: list[wolfhece.hydrology.RetentionBasin.datetime.datetime] = None, add_info: dict[dict[str, float]] = {}) → dict

Retrieves the physical flux fractions values for all sub-basins or a specific selection of sub-basins.

Parameters:

selection_by_iD (list, optional) – A list of sub-basin IDs to retrieve fractions values for. If empty, retrieves fractions values for all sub-basins. Defaults to [].

Returns:

A dictionary containing the sub-basin names or IDs as keys and their corresponding fractions values as values.

Return type:

dict

plot_all_fractions(all_fractions: dict[str:np.array] = {}, selection_by_iD: list = [], to_show: bool = False, writeDir: str = '', range_data: list[wolfhece.hydrology.RetentionBasin.datetime.datetime] = [])

get_all_iv_production(selection_by_iD: list = [], to_save: bool = True) → dict

Retrieves the x production values for all sub-basins or a specific selection of sub-basins.

Parameters:

selection_by_iD (list, optional) – A list of sub-basin IDs to retrieve x production values for. If empty, retrieves x production values for all sub-basins. Defaults to [].

Returns:

A dictionary containing the sub-basin names or IDs as keys and their corresponding x production values as values.

Return type:

dict

activate_all_internal_variables(selection_by_iD: list = []) → dict

Activates all internal variables for all sub-basins or a specific selection of sub-basins.

Parameters:

selection_by_iD (list, optional) – A list of sub-basin IDs to activate internal variables for. If empty, activates internal variables for all sub-basins. Defaults to [].

Returns:

A dictionary containing the sub-basin names or IDs as keys and their corresponding internal variables as values.

Return type:

dict

check_presence_of_iv(selection_by_iD: list = []) → dict

Checks the presence of internal variables for all sub-basins or a specific selection of sub-basins.

Parameters:

selection_by_iD (list, optional) – A list of sub-basin IDs to check the presence of internal variables for. If empty, checks the presence of internal variables for all sub-basins. Defaults to [].

Returns:

A dictionary containing the sub-basin names or IDs as keys and their corresponding internal variables as values.

Return type:

dict

get_all_Qtest(selection_by_iD: list = [], nb_atttempts: int = -1)

_get_simulation_intervals()

This procedure is getting the simulation intervals of the current module.

_set_simulation_intervals(simulation_intervals: list[tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime, wolfhece.hydrology.RetentionBasin.datetime.datetime]])

This procedure is setting the simulation intervals of the current module.

_get_temporal_parameters() → tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime, wolfhece.hydrology.RetentionBasin.datetime.datetime]

This procedure is getting the temporal parameters of the current module.

_set_temporal_parameters(simulation_intervals: tuple[wolfhece.hydrology.RetentionBasin.datetime.datetime, wolfhece.hydrology.RetentionBasin.datetime.datetime])

This procedure is setting the temporal parameters of the current module.

make_nd_array(c_pointer, shape, dtype=np.float64, order='C', own_data=True, readonly=False)