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'

wolfhece.hydrology.Optimisation.DLL_FILE_DEBUG = 'WolfDll_debug.dll'

wolfhece.hydrology.Optimisation.DLL_FILE_TEST = 'WolfDll_test.dll'

class wolfhece.hydrology.Optimisation.CaseOpti(*args, **kw)

Bases: wolfhece.PyGui.GenMapManager

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

launcherParam: wolfhece.hydrology.Comparison.Wolf_Param

refCatchment: wolfhece.hydrology.Comparison.Catchment

idToolItem: int

mydro: wolfhece.PyGui.HydrologyModel

idMenuItem: int

wx_exists: bool

read_param(dir, copyDefault=False, callback=None, workingDir: wolfhece.wolf_array.Path = '')

show_launcherParam(event)

show_mydro(event)

class wolfhece.hydrology.Optimisation.Optimisation(parent=None, title='', w=500, h=500, init_wx=True, debugDLL=False, for_test: bool = False)

Bases: wolfhece.wolf_array.wx.Frame

workingDir: str

myParams: dict

myParamsPy: dict

curParams_vec_F: wolfhece.wolf_array.np.ndarray

nbParams: int

optiFactor_F: wolfhece.hydrology.Comparison.ct.c_double

bestFactor: float

comparHowParam: wolfhece.hydrology.Comparison.Wolf_Param

saParam: wolfhece.hydrology.Comparison.Wolf_Param

optiParam: wolfhece.hydrology.Comparison.Wolf_Param

dllFortran: wolfhece.hydrology.Comparison.ct.CDLL

pathDll: str

callBack_proc: dict

callBack_ptr: dict

myCases: list[CaseOpti]

myStations: list[str]

compareFilesDict: dict[str, str]

compareSubBasins: dict[str, wolfhece.hydrology.Comparison.SubBasin]

all_intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]]

wx_exists: bool

debugDLL = False

initGUI()

quitGUI(event)

new(event)

Create a new optimisation directory and files.

load(event, workingDir: str = '', fileName: str = '')

apply_optim(event, idLauncher: int = 0, replace_only_if_better: bool = False, optim_params: wolfhece.wolf_array.np.ndarray = None)

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)

init_with_default_lumped(replace: bool = False)

init_lumped_model()

init_myParams(idLauncher=0)

collect_optim(fileName='')

init_with_reference(idLauncher=0)

get_reference(refFileName='', idLauncher=0)

init_dir_in_params()

update_dir_in_params()

checkIntervals()

update_parameters_launcher(idLauncher=0)

update_parameters_SA()

plot_optim_sub(event, idLauncher=0)

plot_optim_jct(event, idLauncher=0)

load_dll(path: str, fileName: str = 'WolfDLL.dll')

Load the Fortran DLL for optimization.

Parameters:

• path – The directory where the DLL is located.

• fileName – The name of the DLL file to load.

default_files()

Create the default optimizer files in the working directory.

compute_optimizer(idOpti=1)

evaluate_model_optimizer(parameters: wolfhece.wolf_array.np.array, idOpti: int = 1)

write_mesh_results_optimizer(idOpti: int = 1)

init_optimizer(idForced=-1)

init_optimizer_again(event, idForced=1)

compute_distributed_hydro_model(idLauncher=0)

compute0_distributed_hydro_model(event)

associate_ptr(event, which='all', idOpti=1, idLauncher=0)

associate_callback_fct()

associate_callback_fct_update(idOpti=1, idLauncher=0)

associate_callback_fct_getcvg(idOpti=1, idLauncher=0)

associate_ptr_q_all(idOpti=1, idLauncher=0)

associate_ptr_iv_saved(idOpti=1, idLauncher=0)

associate_ptr_time_delays(idOpti=1, idLauncher=0)

associate_ptr_params(idOpti=1, idLauncher=0)

associate_ptr_opti_factor(idOpti=1, idLauncher=0)

get_all_activated_iv(idOpti: int = 1, idLauncher: int = 0, iv_variables: tuple[wolfhece.wolf_array.np.ndarray, wolfhece.wolf_array.np.ndarray] = None) → tuple[wolfhece.wolf_array.np.ndarray, wolfhece.wolf_array.np.ndarray]

init_distributed_hydro_model(event)

launch_lumped_optimisation(event, idOpti=1)

test_update_hydro_py(event)

launch_semiDistributed_optimisation(event, idOpti=1, idLauncher=0)

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)

reload_hydro(idCompar, firstLevel: int = 1, lastLevel: int = -1, fromStation: str = '', updateAll: bool = False)

get_cvg(pointerData)

update_timeDelay(index)

prepare_calibration_timeDelay(stationOut, idLauncher=0)

reset(event)

disable_all_MenuBar(exceptions=[])

enable_MenuBar(menuBar: str)

enable_Menu(menuItem: str, menuBar: str, isEnable: bool)

add_Case()

launch_optimisation(idOpti=1)

show_optiParam(event)

show_saParam(event)

show_comparHowParam(event)

update_nothing(whatever, value=0.0)

apply_timeDelay_dist(idOpti: int = 1, idLauncher: int = 0, junctionKey: str = '')

update_time_delays(idOpti: int = 1, idLauncher: int = 0)

update_myParams(idLauncher=0)

set_compare_stations(idLauncher)

destroyOpti(event)

get_all_outlets(event, idLauncher: int = 0)

write_all_inlets(event, idLauncher: int = 0)

plot_all_landuses(event, idLauncher: int = 0)

plot_all_landuses_hydro(event, idLauncher: int = 0)

generate_semiDist_optim_simul(event, idOpti=1, idLauncher: int = 0)

generate_semiDist_debug_simul(event, idOpti=1, idLauncher: int = 0)

read_all_attempts_SA(format='rpt', all_attempts=False, filter_repetitions=True, stationOut: str = '')

apply_optim_2_params(params: wolfhece.wolf_array.np.array, idLauncher=0)

apply_all_tests(idLauncher=0)

remove_py_params(idLauncher: int = 0)

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]]

_save_opti_intervals(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = None) → int

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]

_define_intervals_with_ts(intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], time: wolfhece.wolf_array.np.ndarray, ts: wolfhece.wolf_array.np.ndarray, idLauncher: int = 0)

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.

• ts (dict[str, SubBasin]) – A dictionary of time series where the keys are station names and the values are vectors of numpy array.

• 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 –

_define_intervals_with_nan_inlets(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 = '')

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 –

_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 = '')

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]])

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)

get_initial_parameters() → wolfhece.wolf_array.np.array

reset_init_params(init_params: wolfhece.wolf_array.np.array)

extract_internal_variables(event, idLauncher: int = 0, to_plot: bool = True)

_check_presence_of_iv(idLauncher: int = 0)

plot_Nash_vs_Qexcess(event, idLauncher: int = 0)

get_all_Nash()

get_all_params(idLauncher: int = 0)

save_all_params(all_params: dict = {}, idLauncher: int = 0)

save_current_compare_file(stationOut: str)

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)

reset_simulation_intervals(default_interval: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], idLauncher: int = 0)

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)

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 = None, std: float = None, eps: float = 0.2, rmv_near_max=None, nb_rand_close: int = 10, objective_fct: bool = True, apply_clustering: bool = False, objective_weight: float = 1.0)

_get_exceedance(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) → float

_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]]

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]]

_get_flow_fractions(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = [], from_full_matrix: tuple[wolfhece.wolf_array.np.ndarray, wolfhece.wolf_array.np.ndarray] = None) → dict[list[str], list[float]]

This function retrieves the flow fractions for a given sub-basin. It can also take a “full matrix” of internal variables to compute the fractions. This “full matrix” is a tuple containing the ids of the internal variables and the matrix itself.

Parameters:

• idLauncher (int, optional) – ID of the launcher, defaults to 0

• stationOut (str, optional) – Name of the outlet station, defaults to “”

• intervals (list[tuple[datetime.datetime, datetime.datetime]], optional) – List of start date and end date for the intervals to evaluate the flow fractions, defaults to []

• from_full_matrix (tuple[np.ndarray, np.ndarray], optional) – Containing the matrix (nb_t, nb_iv) of I.V. coming from the Fortran code directly (not read in files), defaults to None

Returns:

Dictionary containing the flow fractions for each type of flow with their names as keys and the fractions as values.

Return type:

dict[list[str], list[float]]

_get_max_flow_fractions(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = [], from_full_matrix: tuple[wolfhece.wolf_array.np.ndarray, wolfhece.wolf_array.np.ndarray] = None) → dict[list[str], list[float]]

_get_punctual_reservoir_fractions(eval_date: wolfhece.hydrology.Comparison.datetime.datetime, idLauncher: int = 0, stationOut: str = '', from_full_matrix: tuple[wolfhece.wolf_array.np.ndarray, wolfhece.wolf_array.np.ndarray] = None) → dict[list[str], list[float]]

_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]]

_get_ratio_max_sim_obs(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) → float

get_param_names(idLauncher: int = 0, stationOut: str = '')

plot_equifinality(event, idLauncher: int = 0)

plot_model_analysis(event, idLauncher: int = 0)

launch_semi_dist_parameters(idLauncher: int = 0, idOpti: int = 1, params_to_test: dict[str, wolfhece.wolf_array.np.ndarray] = {}, return_outflows: bool = False) → dict[str, wolfhece.wolf_array.np.ndarray]

_prepare_opti_hydro_files(stationOut: str, idLauncher: int = 0, onlyOwnSub: bool = False, doneList: list[str] = [], previousLevel: int = 1)

write_one_opti_param(filPath: wolfhece.wolf_array.Path, fileName: wolfhece.wolf_array.Path, myGroup: str, myKey: str, value: float, convers_factor: int = 1.0)

launch_models_propertie_with_Nash_old(event, idLauncher: int = 0, idOpti: int = 1, quantile_Nash: float = 0.01, std_Nash: float = 0.03, clustering_Nash: bool = True, save_every: int = 100, restart_from_file: bool = True)

Analyse the properties of the model and compare them with the Nash coefficient.

Parameters:

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

Returns:

None

Raises:

None –

launch_models_propertie_with_Nash_old2(event, idLauncher: int = 0, idOpti: int = 1, quantile_Nash: float = 0.01, std_Nash: float = 0.03, clustering_Nash: bool = True, save_every: int = 1000, restart_from_file: bool = True, intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = [])

Analyse the properties of the model and compare them with the Nash coefficient.

Parameters:

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

Returns:

None

Raises:

None –

launch_models_properties_with_Nash(event, idLauncher: int = 0, idOpti: int = 1, quantile_Nash: float = 0.01, std_Nash: float = 0.03, clustering_Nash: bool = False, save_every: int = 1000, restart_from_file: bool = False, evaluation_interval: tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime] = [])

Analyse the properties of the model and compare them with the Nash coefficient.

Parameters:

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

Returns:

None

Raises:

None –

_build_type_to_key_index() → dict[int, int]

_get_key_from_type_all_parameters(list_type_param: list[int]) → dict[int | None]

_get_key_from_type_parameter(type_param: int) → int

_intersect_intervals(intervals: list[tuple[wolfhece.hydrology.read.date, wolfhece.hydrology.read.date]], flood_intervals: tuple[wolfhece.hydrology.read.date, wolfhece.hydrology.read.date])

reload_optimal_subbasin(idLauncher: int = 0, idOpti: int = 1, stationOut: str = None)

Reload the optimal subbasin for a given idLauncher and idOpti.

Parameters:

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

• idOpti (int) – The id of the optimizer.

• stationOut (str) – The name of the station.

Returns:

None

Raises:

None –

init_optimize_model_f(idOpti: int = 1)

Initialize the optimization model for a given idOpti.

Parameters:

idOpti (int) – The id of the optimizer.

Returns:

None

Raises:

None –

prepare_optimize_model_F_one_station(idLauncher: int = 0, idOpti: int = 1, stationOut: str = None, all_intervals: list[tuple[wolfhece.hydrology.read.date, wolfhece.hydrology.read.date]] = [], onlyOwnSub: bool = True, previousLevel: int = 1, already_done_subbasins: list[str] = [], return_intervals: bool = False)

extract_hydro_from_params(idLauncher: int = 0, idOpti: int = 1, stationOut: str = None, all_params: wolfhece.wolf_array.np.ndarray = None)

Extract the hydro from the parameters.

Parameters:

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

• idOpti (int) – The id of the optimizer.

• stationOut (str) – The name of the station.

• all_params (np.ndarray) – The parameters to be tested.

Returns:

None

Raises:

None –

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

_reload_model_analysis(stationOut: str, all_params: wolfhece.wolf_array.np.ndarray) → tuple[list, wolfhece.wolf_array.np.ndarray]

Reload the model analysis for a given station.

Parameters:

• stationOut (str) – The name of the station.

• all_params (np.ndarray) – The parameters to be tested.

Returns:

None

Raises:

None –