wolfhece.hydrology.Optimisation =============================== .. py:module:: wolfhece.hydrology.Optimisation .. autoapi-nested-parse:: 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 --------------- .. py:data:: DLL_FILE :value: 'WolfDll.dll' .. py:data:: DLL_FILE_DEBUG :value: 'WolfDll_debug.dll' .. py:data:: DLL_FILE_TEST :value: 'WolfDll_test.dll' .. py:class:: CaseOpti(*args, **kw) Bases: :py:obj:`wolfhece.PyGui.GenMapManager` .. autoapi-inheritance-diagram:: wolfhece.hydrology.Optimisation.CaseOpti :parts: 1 :private-bases: 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. .. py:attribute:: launcherDir :type: str .. py:attribute:: launcherParam :type: wolfhece.hydrology.Comparison.Wolf_Param .. py:attribute:: refCatchment :type: wolfhece.hydrology.Comparison.Catchment .. py:attribute:: idToolItem :type: int .. py:attribute:: mydro :type: wolfhece.PyGui.HydrologyModel .. py:attribute:: idMenuItem :type: int .. py:attribute:: wx_exists :type: bool .. py:method:: read_param(dir, copyDefault=False, callback=None, workingDir='') .. py:method:: show_launcherParam(event) .. py:method:: show_mydro(event) .. py:class:: Optimisation(parent=None, title='', w=500, h=500, init_wx=True, debugDLL=False, for_test: bool = False) Bases: :py:obj:`wolfhece.wolf_array.wx.Frame` .. autoapi-inheritance-diagram:: wolfhece.hydrology.Optimisation.Optimisation :parts: 1 :private-bases: .. py:attribute:: workingDir :type: str .. py:attribute:: myParams :type: dict .. py:attribute:: myParamsPy :type: dict .. py:attribute:: curParams_vec_F :type: wolfhece.wolf_array.np.ndarray .. py:attribute:: nbParams :type: int .. py:attribute:: optiFactor_F :type: wolfhece.hydrology.Comparison.ct.c_double .. py:attribute:: bestFactor :type: float .. py:attribute:: comparHowParam :type: wolfhece.hydrology.Comparison.Wolf_Param .. py:attribute:: saParam :type: wolfhece.hydrology.Comparison.Wolf_Param .. py:attribute:: optiParam :type: wolfhece.hydrology.Comparison.Wolf_Param .. py:attribute:: dllFortran :type: wolfhece.hydrology.Comparison.ct.CDLL .. py:attribute:: pathDll :type: str .. py:attribute:: callBack_proc :type: dict .. py:attribute:: callBack_ptr :type: dict .. py:attribute:: myCases :type: list[CaseOpti] .. py:attribute:: myStations :type: list[str] .. py:attribute:: compareFilesDict :type: dict[str, str] .. py:attribute:: compareSubBasins :type: dict[str, wolfhece.hydrology.Comparison.SubBasin] .. py:attribute:: all_intervals :type: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] .. py:attribute:: wx_exists :type: bool .. py:attribute:: debugDLL :value: False .. py:method:: initGUI() .. py:method:: quitGUI(event) .. py:method:: new(event) .. py:method:: load(event, workingDir: str = '', fileName: str = '') .. py:method:: 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". :param event: The event from the GUI. :param idLauncher (optional: int(0)): The ID of the launcher. :param replace_only_if_better (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. .. py:method:: init_lumped_hydro(event) .. py:method:: init_with_default_lumped(replace: bool = False) .. py:method:: init_lumped_model() .. py:method:: init_myParams(idLauncher=0) .. py:method:: collect_optim(fileName='') .. py:method:: init_with_reference(idLauncher=0) .. py:method:: get_reference(refFileName='', idLauncher=0) .. py:method:: init_dir_in_params() .. py:method:: update_dir_in_params() .. py:method:: checkIntervals() .. py:method:: update_parameters_launcher(idLauncher=0) .. py:method:: update_parameters_SA() .. py:method:: plot_optim_sub(event, idLauncher=0) .. py:method:: plot_optim_jct(event, idLauncher=0) .. py:method:: load_dll(path, fileName) .. py:method:: default_files(event) .. py:method:: compute_optimizer(idOpti=1) .. py:method:: evaluate_model_optimizer(parameters: wolfhece.wolf_array.np.array, idOpti: int = 1) .. py:method:: write_mesh_results_optimizer(idOpti: int = 1) .. py:method:: init_optimizer(idForced=-1) .. py:method:: init_optimizer_again(event, idForced=1) .. py:method:: compute_distributed_hydro_model(idLauncher=0) .. py:method:: compute0_distributed_hydro_model(event) .. py:method:: associate_ptr(event, which='all', idOpti=1, idLauncher=0) .. py:method:: associate_callback_fct() .. py:method:: associate_callback_fct_update(idOpti=1, idLauncher=0) .. py:method:: associate_callback_fct_getcvg(idOpti=1, idLauncher=0) .. py:method:: associate_ptr_q_all(idOpti=1, idLauncher=0) .. py:method:: associate_ptr_time_delays(idOpti=1, idLauncher=0) .. py:method:: associate_ptr_params(idOpti=1, idLauncher=0) .. py:method:: associate_ptr_opti_factor(idOpti=1, idLauncher=0) .. py:method:: init_distributed_hydro_model(event) .. py:method:: launch_lumped_optimisation(event, idOpti=1) .. py:method:: test_update_hydro_py(event) .. py:method:: launch_semiDistributed_optimisation(event, idOpti=1, idLauncher=0) Procedure launching the semi-distributed optimisation process. :param event: The event triggering the optimisation. :param idOpti: The ID of the optimizer in Fortran. :type idOpti: int :param idLauncher: The ID of the launcher. :type idLauncher: int :returns: None .. py:method:: update_hydro(idCompar) .. py:method:: reload_hydro(idCompar, firstLevel: int = 1, lastLevel: int = -1, fromStation: str = '', updateAll: bool = False) .. py:method:: get_cvg(pointerData) .. py:method:: update_timeDelay(index) .. py:method:: prepare_calibration_timeDelay(stationOut, idLauncher=0) .. py:method:: reset(event) .. py:method:: disable_all_MenuBar(exceptions=[]) .. py:method:: enable_MenuBar(menuBar: str) .. py:method:: enable_Menu(menuItem: str, menuBar: str, isEnable: bool) .. py:method:: add_Case() .. py:method:: launch_optimisation(idOpti=1) .. py:method:: show_optiParam(event) .. py:method:: show_saParam(event) .. py:method:: show_comparHowParam(event) .. py:method:: update_nothing(whatever, value=0.0) .. py:method:: apply_timeDelay_dist(idOpti: int = 1, idLauncher: int = 0, junctionKey: str = '') .. py:method:: update_time_delays(idOpti: int = 1, idLauncher: int = 0) .. py:method:: update_myParams(idLauncher=0) .. py:method:: set_compare_stations(idLauncher) .. py:method:: destroyOpti(event) .. py:method:: get_all_outlets(event, idLauncher: int = 0) .. py:method:: write_all_inlets(event, idLauncher: int = 0) .. py:method:: plot_all_landuses(event, idLauncher: int = 0) .. py:method:: plot_all_landuses_hydro(event, idLauncher: int = 0) .. py:method:: generate_semiDist_optim_simul(event, idOpti=1, idLauncher: int = 0) .. py:method:: generate_semiDist_debug_simul(event, idOpti=1, idLauncher: int = 0) .. py:method:: read_all_attempts_SA(format='rpt', all_attempts=False, filter_repetitions=True, stationOut: str = '') .. py:method:: apply_optim_2_params(params: wolfhece.wolf_array.np.array, idLauncher=0) .. py:method:: apply_all_tests(idLauncher=0) .. py:method:: remove_py_params(idLauncher: int = 0) Removes the Python parameters from the optimization configuration. :param idLauncher: The ID of the launcher. Defaults to 0. :type idLauncher: int, optional .. py:method:: _read_opti_intervals(idLauncher: int = 0) -> list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] .. todo:: - Add the measure of the comparison file in properties of the object opti - Check according to the current Observation, which comparision intervals are posssible -> and sort them - Save the comparison intervals somewhere - Save the useful comparison intervals somewhere - Return the useful intervals. .. py:method:: _save_opti_intervals(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = None) -> int .. py:method:: 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] .. todo:: - Add the measure of the comparison file in properties of the object opti - Check according to the current Observation, which comparision intervals are posssible -> and sort them - Save the comparison intervals somewhere - Save the useful comparison intervals somewhere - Return the useful intervals. .. py:method:: _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. :param intervals: A list of intervals represented as tuples of start and end datetime objects. :type intervals: list[tuple[datetime.datetime, datetime.datetime]] :param measures: A dictionary of measures where the keys are station names and the values are SubBasin objects. :type measures: dict[str, SubBasin] :param idLauncher: The id of the launcher. Defaults to 0. :type idLauncher: int, optional :param stationOut: The station name. Defaults to "". :type stationOut: str, optional :returns: A list of intervals with NaN measures. :rtype: list[tuple[datetime.datetime, datetime.datetime]] :raises None: .. py:method:: 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 .. py:method:: prepare_init_params_from_best(best_params: wolfhece.wolf_array.np.array, idLauncher: int = 0) .. py:method:: get_initial_parameters() -> wolfhece.wolf_array.np.array .. py:method:: reset_init_params(init_params: wolfhece.wolf_array.np.array) .. py:method:: extract_internal_variables(event, idLauncher: int = 0, to_plot: bool = True) .. py:method:: _check_presence_of_iv(idLauncher: int = 0) .. py:method:: plot_Nash_vs_Qexcess(event, idLauncher: int = 0) .. py:method:: get_all_Nash() .. py:method:: get_all_params(idLauncher: int = 0) .. py:method:: save_all_params(all_params: dict = {}, idLauncher: int = 0) .. py:method:: save_current_compare_file(stationOut: str) Save the current compare file for a given station to prepare optimisation with Fortran. :param stationOut: The station identifier. :type stationOut: str :returns: None :raises None: .. py:method:: prepare_simulation(opti_intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], idLauncher: int = 0) .. py:method:: reset_simulation_intervals(default_interval: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]], idLauncher: int = 0) .. py:method:: 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. :param idLauncher: The id of the launcher. Defaults to 0. :type idLauncher: int, optional :returns: None :raises None: .. py:method:: get_best_params(stationOut: str, criterion: str = 'Nash', quantile: float = 0.99, std: float = 0.05, eps: float = 0.2, rmv_near_max=0.0001, nb_rand_close: int = 10, objective_fct: bool = True, apply_clustering: bool = False, objective_weight: float = 1.0) .. py:method:: _get_exceedance(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) -> float .. py:method:: _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. :param idOpti: The id of the optimisation. Defaults to 1. :type idOpti: int, optional :param stationOut: The station identifier. Defaults to "". :type stationOut: str, optional :param fct_list: A list of functions. Defaults to []. :type fct_list: list[str], optional :returns: None :raises None: .. py:method:: _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]] .. py:method:: _get_flow_fractions(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) -> dict[list[str], list[float]] .. py:method:: _get_punctual_reservoir_fractions(eval_date: wolfhece.hydrology.Comparison.datetime.datetime, idLauncher: int = 0, stationOut: str = '') -> dict[list[str], list[float]] .. py:method:: _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]] .. py:method:: _get_ratio_max_sim_obs(idLauncher: int = 0, stationOut: str = '', intervals: list[tuple[wolfhece.hydrology.Comparison.datetime.datetime, wolfhece.hydrology.Comparison.datetime.datetime]] = []) -> float .. py:method:: get_param_names(idLauncher: int = 0, stationOut: str = '') .. py:method:: plot_equifinality(event, idLauncher: int = 0) .. py:method:: plot_model_analysis(event, idLauncher: int = 0) .. py:method:: 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] .. py:method:: _prepare_opti_hydro_files(stationOut: str, idLauncher: int = 0, onlyOwnSub: bool = False, doneList: list[str] = [], previousLevel: int = 1) .. py:method:: 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) .. py:method:: launch_models_propertie_with_Nash(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. :param idLauncher: The id of the launcher. Defaults to 0. :type idLauncher: int, optional :returns: None :raises None: .. py:method:: _build_type_to_key_index() -> dict[int, int] .. py:method:: _get_key_from_type_all_parameters(list_type_param: list[int]) -> dict[int | None] .. py:method:: _get_key_from_type_parameter(type_param: int) -> int .. py:method:: make_nd_array(c_pointer, shape, dtype=np.float64, order='C', own_data=True, readonly=False) .. py:method:: _reload_model_analysis(stationOut: str, all_params: wolfhece.wolf_array.np.ndarray) Reload the model analysis for a given station. :param stationOut: The name of the station. :type stationOut: str :param all_params: The parameters to be tested. :type all_params: np.ndarray :returns: None :raises None: