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: wolfhece.wolf_array.Path = '') .. 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) Create a new optimisation directory and files. .. 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: str, fileName: str = 'WolfDLL.dll') Load the Fortran DLL for optimization. :param path: The directory where the DLL is located. :param fileName: The name of the DLL file to load. .. py:method:: default_files() Create the default optimizer files in the working directory. .. 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_iv_saved(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:: 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] .. 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_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. :param intervals: A list of intervals represented as tuples of start and end datetime objects. :type intervals: list[tuple[datetime.datetime, datetime.datetime]] :param ts: A dictionary of time series where the keys are station names and the values are vectors of numpy array. :type ts: 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:: _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. :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:: _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 = 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) .. 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]] = [], 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. :param idLauncher: ID of the launcher, defaults to 0 :type idLauncher: int, optional :param stationOut: Name of the outlet station, defaults to "" :type stationOut: str, optional :param intervals: List of start date and end date for the intervals to evaluate the flow fractions, defaults to [] :type intervals: list[tuple[datetime.datetime, datetime.datetime]], optional :param from_full_matrix: Containing the matrix (nb_t, nb_iv) of I.V. coming from the Fortran code directly (not read in files), defaults to None :type from_full_matrix: tuple[np.ndarray, np.ndarray], optional :return: Dictionary containing the flow fractions for each type of flow with their names as keys and the fractions as values. :rtype: dict[list[str], list[float]] .. py:method:: _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]] .. py:method:: _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]] .. 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_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. :param idLauncher: The id of the launcher. Defaults to 0. :type idLauncher: int, optional :returns: None :raises None: .. py:method:: 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. :param idLauncher: The id of the launcher. Defaults to 0. :type idLauncher: int, optional :returns: None :raises None: .. py:method:: 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. :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:: _intersect_intervals(intervals: list[tuple[wolfhece.hydrology.read.date, wolfhece.hydrology.read.date]], flood_intervals: tuple[wolfhece.hydrology.read.date, wolfhece.hydrology.read.date]) .. py:method:: reload_optimal_subbasin(idLauncher: int = 0, idOpti: int = 1, stationOut: str = None) Reload the optimal subbasin for a given idLauncher and idOpti. :param idLauncher: The id of the launcher. :type idLauncher: int :param idOpti: The id of the optimizer. :type idOpti: int :param stationOut: The name of the station. :type stationOut: str :returns: None :raises None: .. py:method:: init_optimize_model_f(idOpti: int = 1) Initialize the optimization model for a given idOpti. :param idOpti: The id of the optimizer. :type idOpti: int :returns: None :raises None: .. py:method:: 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) .. py:method:: 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. :param idLauncher: The id of the launcher. :type idLauncher: int :param idOpti: The id of the optimizer. :type idOpti: int :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: .. 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) -> tuple[list, 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: