wolfhece.hydrology.Optimisation

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

wolfhece.hydrology.Optimisation.DLL_FILE = 'WolfDll.dll'[source]
wolfhece.hydrology.Optimisation.DLL_FILE_DEBUG = 'WolfDll_debug.dll'[source]
class wolfhece.hydrology.Optimisation.CaseOpti(*args, **kw)[source]

Bases: wolfhece.PyGui.GenMapManager

Inheritance diagram of wolfhece.hydrology.Optimisation.CaseOpti

Default class for a Wolf Map Manager.

Will be overriden by the specific classes MapManager, GPU2DModel, HydrologyModel, Wolf2DModel.

It is not suitable for direct use.

launcherDir: str[source]
launcherParam: wolfhece.hydrology.Comparison.Wolf_Param[source]
refCatchment: wolfhece.hydrology.Comparison.Catchment[source]
idToolItem: int[source]
mydro: wolfhece.PyGui.HydrologyModel[source]
idMenuItem: int[source]
wx_exists: bool[source]
read_param(dir, copyDefault=False, callback=None, workingDir='')[source]
show_launcherParam(event)[source]
show_mydro(event)[source]
class wolfhece.hydrology.Optimisation.Optimisation(parent=None, title='', w=500, h=500, init_wx=True, debugDLL=False)[source]

Bases: wolfhece.wolf_array.wx.Frame

Inheritance diagram of wolfhece.hydrology.Optimisation.Optimisation
workingDir: str[source]
myParams: dict[source]
myParamsPy: dict[source]
curParams_vec_F: wolfhece.wolf_array.np.ndarray[source]
nbParams: int[source]
optiFactor_F: wolfhece.hydrology.Comparison.ct.c_double[source]
bestFactor: float[source]
comparHowParam: wolfhece.hydrology.Comparison.Wolf_Param[source]
saParam: wolfhece.hydrology.Comparison.Wolf_Param[source]
optiParam: wolfhece.hydrology.Comparison.Wolf_Param[source]
dllFortran: wolfhece.hydrology.Comparison.ct.CDLL[source]
pathDll: str[source]
callBack_proc: dict[source]
callBack_ptr: dict[source]
myCases: list[CaseOpti][source]
myStations: list[str][source]
compareFilesDict: dict[str, str][source]
compareSubBasins: dict[str, wolfhece.hydrology.Comparison.SubBasin][source]
all_intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]][source]
wx_exists: bool[source]
initGUI()[source]
quitGUI(event)[source]
new(event)[source]
load(event, workingDir: str = '', fileName: str = '')[source]
apply_optim(event, idLauncher: int = 0, replace_only_if_better: bool = False, optim_params: wolfhece.wolf_array.np.ndarray = None)[source]

Apply optimal parameters based on the results file of the optimisation : “.rpt”.

Parameters:
  • event – The event from the GUI.

  • (optional (replace_only_if_better) – int(0)): The ID of the launcher.

  • (optional – bool(False) by default): A boolean indicating whether to replace the current parameters if the new ones are better.

Returns:

If replace_only_if_better is False, returns the best parameters found. If replace_only_if_better is True and the new parameters are better, returns the best parameters found. Otherwise, returns None.

init_lumped_hydro(event)[source]
init_with_default_lumped(replace: bool = False)[source]
init_lumped_model()[source]
init_myParams(idLauncher=0)[source]
collect_optim(fileName='')[source]
init_with_reference(idLauncher=0)[source]
get_reference(refFileName='', idLauncher=0)[source]
init_dir_in_params()[source]
update_dir_in_params()[source]
checkIntervals()[source]
update_parameters_launcher(idLauncher=0)[source]
update_parameters_SA()[source]
plot_optim_sub(event, idLauncher=0)[source]
plot_optim_jct(event, idLauncher=0)[source]
load_dll(path, fileName)[source]
default_files(event)[source]
compute_optimizer(idOpti=1)[source]
evaluate_model_optimizer(parameters: wolfhece.wolf_array.np.array, idOpti: int = 1)[source]
write_mesh_results_optimizer(idOpti: int = 1)[source]
init_optimizer(idForced=-1)[source]
init_optimizer_again(event, idForced=1)[source]
compute_distributed_hydro_model(idLauncher=0)[source]
compute0_distributed_hydro_model(event)[source]
associate_ptr(event, which='all', idOpti=1, idLauncher=0)[source]
associate_callback_fct()[source]
associate_callback_fct_update(idOpti=1, idLauncher=0)[source]
associate_callback_fct_getcvg(idOpti=1, idLauncher=0)[source]
associate_ptr_q_all(idOpti=1, idLauncher=0)[source]
associate_ptr_time_delays(idOpti=1, idLauncher=0)[source]
associate_ptr_params(idOpti=1, idLauncher=0)[source]
associate_ptr_opti_factor(idOpti=1, idLauncher=0)[source]
init_distributed_hydro_model(event)[source]
launch_lumped_optimisation(event, idOpti=1)[source]
test_update_hydro_py(event)[source]
launch_semiDistributed_optimisation(event, idOpti=1, idLauncher=0)[source]

Procedure launching the semi-distributed optimisation process.

Parameters:
  • event – The event triggering the optimisation.

  • idOpti (int) – The ID of the optimizer in Fortran.

  • idLauncher (int) – The ID of the launcher.

Returns:

None

update_hydro(idCompar)[source]
reload_hydro(idCompar, firstLevel: int = 1, lastLevel: int = -1, fromStation: str = '', updateAll: bool = False)[source]
get_cvg(pointerData)[source]
update_timeDelay(index)[source]
prepare_calibration_timeDelay(stationOut, idLauncher=0)[source]
reset(event)[source]
disable_all_MenuBar(exceptions=[])[source]
enable_MenuBar(menuBar: str)[source]
enable_Menu(menuItem: str, menuBar: str, isEnable: bool)[source]
add_Case()[source]
launch_optimisation(idOpti=1)[source]
show_optiParam(event)[source]
show_saParam(event)[source]
show_comparHowParam(event)[source]
update_nothing(whatever, value=0.0)[source]
apply_timeDelay_dist(idOpti: int = 1, idLauncher: int = 0, junctionKey: str = '')[source]
update_time_delays(idOpti: int = 1, idLauncher: int = 0)[source]
update_myParams(idLauncher=0)[source]
set_compare_stations(idLauncher)[source]
destroyOpti(event)[source]
get_all_outlets(event, idLauncher: int = 0)[source]
write_all_inlets(event, idLauncher: int = 0)[source]
plot_all_landuses(event, idLauncher: int = 0)[source]
plot_all_landuses_hydro(event, idLauncher: int = 0)[source]
generate_semiDist_optim_simul(event, idOpti=1, idLauncher: int = 0)[source]
generate_semiDist_debug_simul(event, idOpti=1, idLauncher: int = 0)[source]
read_all_attempts_SA(format='rpt', all_attempts=False, filter_repetitions=True)[source]
apply_optim_2_params(params: wolfhece.wolf_array.np.array, idLauncher=0)[source]
apply_all_tests(idLauncher=0)[source]
remove_py_params(idLauncher: int = 0)[source]

Removes the Python parameters from the optimization configuration.

Parameters:

idLauncher (int, optional) – The ID of the launcher. Defaults to 0.

_read_opti_intervals(idLauncher: int = 0) list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]][source]
_save_opti_intervals(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = None) int[source]
select_opti_intervals(all_intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = None, idLauncher: int = 0, stationOut='', filter_nan: bool = True) list[tuple][source]
_define_intervals_with_nan_measures(intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], measures: dict[str, wolfhece.hydrology.Comparison.SubBasin], idLauncher: int = 0, stationOut: str = '')[source]

Defines new intervals excluding all NaN measures based on the given intervals and measures dictionary. For instance, if there is continuous NaN measures within a given interval, the function will split that interval into smaller that do not contain NaN measures.

Parameters:
  • intervals (list[tuple[datetime.datetime, datetime.datetime]]) – A list of intervals represented as tuples of start and end datetime objects.

  • measures (dict[str, SubBasin]) – A dictionary of measures where the keys are station names and the values are SubBasin objects.

  • idLauncher (int, optional) – The id of the launcher. Defaults to 0.

  • stationOut (str, optional) – The station name. Defaults to “”.

Returns:

A list of intervals with NaN measures.

Return type:

list[tuple[datetime.datetime, datetime.datetime]]

Raises:

None

save_opti_dates_to_file(opti_dates: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]])[source]

Here the procedure is saving the intervals of dates for calibration in the compare.how.param

prepare_init_params_from_best(best_params: wolfhece.wolf_array.np.array, idLauncher: int = 0)[source]
get_initial_parameters() wolfhece.wolf_array.np.array[source]
reset_init_params(init_params: wolfhece.wolf_array.np.array)[source]
extract_internal_variables(event, idLauncher: int = 0, to_plot: bool = True)[source]
_check_presence_of_iv(idLauncher: int = 0)[source]
plot_Nash_vs_Qexcess(event, idLauncher: int = 0)[source]
get_all_Nash()[source]
get_all_params(idLauncher: int = 0)[source]
save_all_params(all_params: dict = {}, idLauncher: int = 0)[source]
save_current_compare_file(stationOut: str)[source]

Save the current compare file for a given station to prepare optimisation with Fortran.

Parameters:

stationOut (str) – The station identifier.

Returns:

None

Raises:

None

prepare_simulation(opti_intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], idLauncher: int = 0)[source]
reset_simulation_intervals(default_interval: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], idLauncher: int = 0)[source]
test_equifinality_with_Nash(event, idLauncher: int = 0, idOpti: int = 1, quantile_Nash: float = 0.01, std_Nash: float = 0.3, clustering_Nash: bool = True)[source]

Test the equifinality of the model.

Parameters:

idLauncher (int, optional) – The id of the launcher. Defaults to 0.

Returns:

None

Raises:

None

get_best_params(stationOut: str, criterion: str = 'Nash', quantile: float = 0.99, std: float = 0.05, eps: float = 0.1, objective_fct: bool = True, apply_clustering: bool = False)[source]
_get_exceedance(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) float[source]
_get_cur_fractions(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) dict[list[str], list[float]][source]

Save the evaluations of the model.

Parameters:
  • idOpti (int, optional) – The id of the optimisation. Defaults to 1.

  • stationOut (str, optional) – The station identifier. Defaults to “”.

  • fct_list (list[str], optional) – A list of functions. Defaults to [].

Returns:

None

Raises:

None

_get_volume_fractions(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) dict[list[str], list[float]][source]
_get_max_runoff(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) dict[list[str], list[float]][source]
_get_ratio_max_sim_obs(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) float[source]
get_param_names(idLauncher: int = 0, stationOut: str = '')[source]
plot_equifinality(event, idLauncher: int = 0)[source]
add_Case(idLauncher: int = 0)[source]
make_nd_array(c_pointer, shape, dtype=np.float64, order='C', own_data=True, readonly=False)[source]