Calibration

Calibration

Bases: Problem

Warning

This class is currently under development.

A Problem subclass from the pymoo Python package for performing model calibration against observed data using multi-objective optimization and evolutionary algorithms.

Parameters:

Name	Type	Description	Default
parameters	BoundType	List of dictionaries defining parameter configurations for calibration simulations. Each dictionary contain the following keys: name (str): Required. Name of the parameter in the cal_parms.cal file. change_type (str): Required. Type of change to apply. Must be one of 'absval', 'abschg', or 'pctchg'. lower_bound (float): Required. Lower bound for the parameter. upper_bound (float): Required. Upper bound for the parameter. units (Iterable[int]): Optional. List of unit IDs to which the parameter change should be constrained. conditions (dict[str, list[str]]): Optional. Conditions to apply when changing the parameter. Supported keys include 'hsg', 'texture', 'plant', and 'landuse', each mapped to a list of allowed values. parameters = [ { 'name': 'cn2', 'change_type': 'pctchg', 'lower_bound': 25, 'upper_bound': 75, }, { 'name': 'perco', 'change_type': 'absval', 'lower_bound': 0, 'upper_bound': 1, 'conditions': {'hsg': ['A']} }, { 'name': 'bf_max', 'change_type': 'absval', 'lower_bound': 0.1, 'upper_bound': 2.0, 'units': range(1, 194) } ]	required
calsim_dir	str | Path	Path to the directory where simulations for each individual in each generation will be performed. Raises an error if the folder is not empty. This precaution helps prevent data deletion, overwriting directories, and issues with reading required data files not generated by the simulation.	required
txtinout_dir	str | Path	Path to the TxtInOut directory containing the required files for SWAT+ simulation.	required
extract_data	dict[str, dict[str, Any]]	A nested dictionary specifying how to extract data from SWAT+ simulation output files. The top-level keys are filenames of the output files, without paths (e.g., channel_sd_day.txt). Each key must map to a non-empty dictionary containing the following sub-keys, which correspond to the input variables within the method simulated_timeseries_df: has_units (bool): Required. If True, the third line of the simulated file contains units for the columns. begin_date (str): Optional. Start date in DD-Mon-YYYY format (e.g., 01-Jan-2010). Defaults to the earliest date in the simulated file. end_date (str): Optional. End date in DD-Mon-YYYY format (e.g., 31-Dec-2013). Defaults to the latest date in the simulated file. ref_day (int): Optional. Reference day for monthly and yearly time series. If None (default), the last day of the month or year is used, obtained from simulation. Not applicable to daily time series files (ending with _day). ref_month (int): Optional. Reference month for yearly time series. If None (default), the last month of the year is used, obtained from simulation. Not applicable to monthly time series files (ending with _mon). apply_filter (dict[str, list[typing.Any]]): Optional. Each key is a column name and the corresponding value is a list of allowed values for filtering rows in the DataFrame. By default, no filtering is applied. An error is raised if filtering produces an empty DataFrame. Note The sub-key usecols should not be included here. Although no error will be raised, it will be ignored during class initialization because the sim_col sub-key from the objective_config input is automatically used as usecols. Including it manually has no effect. extract_data = { 'channel_sd_day.txt': { 'has_units': True, 'begin_date': '01-Jun-2014', 'end_date': '01-Oct-2016', 'apply_filter': {'gis_id': [561]} }, 'channel_sd_mon.txt': { 'has_units': True, 'apply_filter': {'name': ['cha561']} } }	required
observe_data	dict[str, dict[str, str]]	A nested dictionary specifying observed data configuration. The top-level keys are same as keys of extract_data (e.g., channel_sd_day.txt). Each key must map to a non-empty dictionary containing the following sub-keys: obs_file (str): Required. Path to the CSV file containing observed data. The file must include a date column with comma as the separator to read the DataFrame by the pandas.read_csv method. The date col will be used to merge simulated and observed data. date_format (str): Required. Date format of the date column in obs_file, used to parse datetime.date objects from date strings. observe_data = { 'channel_sd_day.txt': { 'obs_file': "C:\Users\Username\observed_data\discharge_daily.csv", 'date_format': '%Y-%m-%d' }, 'channel_sd_mon.txt': { 'obs_file': "C:\Users\Username\observed_data\discharge_monthly.csv", 'date_format': '%Y-%m-%d' } }	required
objective_config	dict[str, dict[str, str]]	A nested dictionary specifying objectives configuration. The top-level keys are same as keys of extract_data (e.g., channel_sd_day.txt). Each key must map to a non-empty dictionary containing the following sub-keys: sim_col (str): Required. Name of the column containing simulated values. obs_col (str): Required. Name of the column containing observed values. indicator (str): Required. Name of the performance indicator used for optimization. Available options with their optimization direction are listed below: NSE: Nash–Sutcliffe Efficiency (maximize). KGE: Kling–Gupta Efficiency (maximize). MSE: Mean Squared Error (minimize). RMSE: Root Mean Squared Error (minimize). MARE: Mean Absolute Relative Error (minimize). Tip Avoid using MARE if obs_col contains zero values, as it will cause a division-by-zero error. objective_config = { 'channel_sd_day.txt': { 'sim_col': 'flo_out', 'obs_col': 'discharge', 'indicator': 'NSE' }, 'channel_sd_mon.txt': { 'sim_col': 'flo_out', 'obs_col': 'discharge', 'indicator': 'MSE' } }	required
algorithm	str	Name of the alogrithm. Available options: GA: Genetic Algorithm (single-objective). DE: Differential Evolution Algorithm (single-objective). NSGA2: Non-dominated sorted Genetic Algorithm - II (multi-objective). Note Multi-objective algorithms can be used for single-objective optimization, but not vice versa.	required
n_gen	int	Number of generation that alogirithm will run.	required
pop_size	int	Number of individual in each generation.	required
max_workers	int	Number of logical CPUs to use for parallel processing. If None (default), all available logical CPUs are used. Tip Simulation efficiency can be improved by choosing pop_size strategically. The total number of simulations during optimization equals n_gen × pop_size. Simulations within a generation are executed in parallel batches depending on CPU availability. For example, with 16 CPUs and a pop_size of 20, two batches will be required (16 + 4); therefore, selecting a pop_size of 16 or 32 may improve efficiency.	None

Source code in pySWATPlus/calibration.py

class Calibration(pymoo.core.problem.Problem):  # type: ignore[misc]
    '''
    Warning:
        This class is currently under development.

    A `Problem` subclass from the [`pymoo`](https://github.com/anyoptimization/pymoo) Python package
    for performing model calibration against observed data using multi-objective optimization
    and evolutionary algorithms.

    Args:
        parameters (newtype.BoundType): List of dictionaries defining parameter configurations for calibration simulations.
            Each dictionary contain the following keys:

            - `name` (str): **Required.** Name of the parameter in the `cal_parms.cal` file.
            - `change_type` (str): **Required.** Type of change to apply. Must be one of 'absval', 'abschg', or 'pctchg'.
            - `lower_bound` (float): **Required.** Lower bound for the parameter.
            - `upper_bound` (float): **Required.** Upper bound for the parameter.
            - `units` (Iterable[int]): Optional. List of unit IDs to which the parameter change should be constrained.
            - `conditions` (dict[str, list[str]]): Optional.  Conditions to apply when changing the parameter.
              Supported keys include `'hsg'`, `'texture'`, `'plant'`, and `'landuse'`, each mapped to a list of allowed values.

            ```python
            parameters = [
                {
                    'name': 'cn2',
                    'change_type': 'pctchg',
                    'lower_bound': 25,
                    'upper_bound': 75,
                },
                {
                    'name': 'perco',
                    'change_type': 'absval',
                    'lower_bound': 0,
                    'upper_bound': 1,
                    'conditions': {'hsg': ['A']}
                },
                {
                    'name': 'bf_max',
                    'change_type': 'absval',
                    'lower_bound': 0.1,
                    'upper_bound': 2.0,
                    'units': range(1, 194)
                }
            ]
            ```

        calsim_dir (str | pathlib.Path): Path to the directory where simulations for each individual in each generation will be performed.
            Raises an error if the folder is not empty. This precaution helps prevent data deletion, overwriting directories,
            and issues with reading required data files not generated by the simulation.

        txtinout_dir (str | pathlib.Path): Path to the `TxtInOut` directory containing the required files for SWAT+ simulation.

        extract_data (dict[str, dict[str, typing.Any]]): A nested dictionary specifying how to extract data from SWAT+ simulation output files.
            The top-level keys are filenames of the output files, without paths (e.g., `channel_sd_day.txt`). Each key must map to a non-empty dictionary
            containing the following sub-keys, which correspond to the input variables within the method
            [`simulated_timeseries_df`](https://swat-model.github.io/pySWATPlus/api/data_manager/#pySWATPlus.DataManager.simulated_timeseries_df):

            - `has_units` (bool): **Required.** If `True`, the third line of the simulated file contains units for the columns.
            - `begin_date` (str): Optional. Start date in `DD-Mon-YYYY` format (e.g., 01-Jan-2010). Defaults to the earliest date in the simulated file.
            - `end_date` (str): Optional. End date in `DD-Mon-YYYY` format (e.g., 31-Dec-2013). Defaults to the latest date in the simulated file.
            - `ref_day` (int): Optional. Reference day for monthly and yearly time series.
               If `None` (default), the last day of the month or year is used, obtained from simulation. Not applicable to daily time series files (ending with `_day`).
            - `ref_month` (int): Optional. Reference month for yearly time series. If `None` (default), the last month of the year is used, obtained from simulation.
              Not applicable to monthly time series files (ending with `_mon`).
            - `apply_filter` (dict[str, list[typing.Any]]): Optional. Each key is a column name and the corresponding value
              is a list of allowed values for filtering rows in the DataFrame. By default, no filtering is applied.
              An error is raised if filtering produces an empty DataFrame.

            !!! note
                The sub-key `usecols` should **not** be included here. Although no error will be raised, it will be ignored during class initialization
                because the `sim_col` sub-key from the `objective_config` input is automatically used as `usecols`. Including it manually has no effect.

            ```python
            extract_data = {
                'channel_sd_day.txt': {
                    'has_units': True,
                    'begin_date': '01-Jun-2014',
                    'end_date': '01-Oct-2016',
                    'apply_filter': {'gis_id': [561]}
                },
                'channel_sd_mon.txt': {
                    'has_units': True,
                    'apply_filter': {'name': ['cha561']}
                }
            }
            ```

        observe_data (dict[str, dict[str, str]]): A nested dictionary specifying observed data configuration. The top-level keys
            are same as keys of `extract_data` (e.g., `channel_sd_day.txt`). Each key must map to a non-empty dictionary containing the following sub-keys:

            - `obs_file` (str): **Required.** Path to the CSV file containing observed data. The file must include a `date` column with comma as the
              separator to read the `DataFrame` by the [`pandas.read_csv`](https://pandas.pydata.org/docs/reference/api/pandas.read_csv.html#pandas-read-csv)
              method. The `date` col will be used to merge simulated and observed data.
            - `date_format` (str): **Required.** Date format of the `date` column in `obs_file`, used to parse `datetime.date` objects from date strings.

            ```python
            observe_data = {
                'channel_sd_day.txt': {
                    'obs_file': "C:\\Users\\Username\\observed_data\\discharge_daily.csv",
                    'date_format': '%Y-%m-%d'
                },
                'channel_sd_mon.txt': {
                    'obs_file': "C:\\Users\\Username\\observed_data\\discharge_monthly.csv",
                    'date_format': '%Y-%m-%d'
                }
            }
            ```

        objective_config (dict[str, dict[str, str]]): A nested dictionary specifying objectives configuration. The top-level keys
            are same as keys of `extract_data` (e.g., `channel_sd_day.txt`). Each key must map to a non-empty dictionary containing the following sub-keys:

            - `sim_col` (str): **Required.** Name of the column containing simulated values.
            - `obs_col` (str): **Required.** Name of the column containing observed values.
            - `indicator` (str): **Required.** Name of the performance indicator used for optimization.
              Available options with their optimization direction are listed below:

                - `NSE`: Nash–Sutcliffe Efficiency (**maximize**).
                - `KGE`: Kling–Gupta Efficiency (**maximize**).
                - `MSE`: Mean Squared Error (**minimize**).
                - `RMSE`: Root Mean Squared Error (**minimize**).
                - `MARE`: Mean Absolute Relative Error (**minimize**).

            !!! tip
                Avoid using `MARE` if `obs_col` contains zero values, as it will cause a division-by-zero error.

            ```python
            objective_config = {
                'channel_sd_day.txt': {
                    'sim_col': 'flo_out',
                    'obs_col': 'discharge',
                    'indicator': 'NSE'
                },
                'channel_sd_mon.txt': {
                    'sim_col': 'flo_out',
                    'obs_col': 'discharge',
                    'indicator': 'MSE'
                }
            }
            ```

        algorithm (str): Name of the alogrithm. Available options:

            - `GA`: Genetic Algorithm (**single-objective**).
            - `DE`: [Differential Evolution Algorithm](https://doi.org/10.1007/3-540-31306-0) (**single-objective**).
            - `NSGA2`: [Non-dominated sorted Genetic Algorithm - II](https://doi.org/10.1109/4235.996017) (**multi-objective**).

            !!! Note
                Multi-objective algorithms can be used for single-objective optimization, but not vice versa.

        n_gen (int): Number of generation that alogirithm will run.

        pop_size (int): Number of individual in each generation.

        max_workers (int): Number of logical CPUs to use for parallel processing. If `None` (default), all available logical CPUs are used.

            !!! tip
                Simulation efficiency can be improved by choosing `pop_size` strategically.
                The total number of simulations during optimization equals `n_gen × pop_size`.
                Simulations within a generation are executed in parallel batches depending on CPU availability.
                For example, with 16 CPUs and a `pop_size` of 20, two batches will be required (16 + 4);
                therefore, selecting a `pop_size` of 16 or 32 may improve efficiency.
    '''

    def __init__(
        self,
        parameters: newtype.BoundType,
        calsim_dir: str | pathlib.Path,
        txtinout_dir: str | pathlib.Path,
        extract_data: dict[str, dict[str, typing.Any]],
        observe_data: dict[str, dict[str, str]],
        objective_config: dict[str, dict[str, str]],
        algorithm: str,
        n_gen: int,
        pop_size: int,
        max_workers: typing.Optional[int] = None
    ) -> None:

        # Check input variables type
        validators._variable_origin_static_type(
            vars_types=typing.get_type_hints(
                obj=self.__class__.__init__
            ),
            vars_values=locals()
        )

        # Absolute directory path
        calsim_dir = pathlib.Path(calsim_dir).resolve()
        txtinout_dir = pathlib.Path(txtinout_dir).resolve()

        # Validate same top-level keys in dictionaries
        validators._dict_key_equal(
            extract_data=extract_data,
            observe_data=observe_data,
            objective_config=objective_config
        )

        # Dictionary of metric key name
        df_key = {
            obj: obj.split('.')[0] + '_df' for obj in objective_config
        }

        # Validate initialization of TxtinoutReader class
        tmp_reader = TxtinoutReader(
            tio_dir=txtinout_dir
        )

        # Disable CSV print to save time
        tmp_reader.disable_csv_print()

        # List of BoundDict objects
        params_bounds = utils._parameters_bound_dict_list(
            parameters=parameters
        )

        # Validate configuration of simulation parameters
        validators._simulation_preliminary_setup(
            sim_dir=calsim_dir,
            tio_dir=txtinout_dir,
            parameters=params_bounds
        )

        # Validate objectives configuration
        validators._metric_config(
            input_dict=objective_config,
            var_name='objective_config'
        )
        for obj in objective_config:
            if objective_config[obj]['indicator'] == 'PBIAS':
                raise ValueError(
                    'Indicator "PBIAS" is invalid in objective_config; it lacks a defined optimization direction'
                )

        # Validate observe_data configuration
        validators._observe_data_config(
            observe_data=observe_data
        )

        # Dictionary of observed DataFrames
        observe_dict = utils._observe_data_dict(
            observe_data=observe_data,
            metric_config=objective_config,
            df_key=df_key
        )

        # Validate extract_data configuration
        for key in extract_data:
            extract_data[key]['usecols'] = [objective_config[key]['sim_col']]
        validators._extract_data_config(
            extract_data=extract_data
        )

        # Change variable name with counter (for multiple occurence only) in BoundType
        var_names = utils._parameters_name_with_counter(
            parameters=params_bounds
        )

        # List of variable lower and upper bounds
        var_lb = []
        var_ub = []
        for param in params_bounds:
            var_lb.append(param.lower_bound)
            var_ub.append(param.upper_bound)

        # Initalize parameters
        self.params_bounds = params_bounds
        self.extract_data = extract_data
        self.objective_config = objective_config
        self.var_names = var_names
        self.df_key = df_key
        self.observe_dict = observe_dict
        self.calsim_dir = calsim_dir
        self.txtinout_dir = txtinout_dir
        self.algorithm = algorithm
        self.n_gen = n_gen
        self.pop_size = pop_size
        self.total_sim = n_gen * pop_size
        self.max_workers = max_workers
        self.track_gen = 0

        # Access properties and methods from Problem class
        super().__init__(
            n_var=len(params_bounds),
            n_obj=len(objective_config),
            xl=numpy.array(var_lb),
            xu=numpy.array(var_ub)
        )

    def _evaluate(
        self,
        x: numpy.typing.NDArray[numpy.float64],
        out: dict[str, numpy.typing.NDArray[numpy.float64]]
    ) -> None:

        # Track starting simulation number for each generation
        start_sim = self.track_gen * self.pop_size + 1

        # Display start of current generation number
        self.track_gen = self.track_gen + 1
        print(
            f'\nStarted generation: {self.track_gen}/{self.n_gen}\n'
        )

        # Simulation in separate CPU
        cpu_sim = functools.partial(
            cpu._simulation_output,
            num_sim=self.total_sim,
            var_names=self.var_names,
            sim_dir=self.calsim_dir,
            tio_dir=self.txtinout_dir,
            params_bounds=self.params_bounds,
            extract_data=self.extract_data,
            clean_setup=True
        )

        # Assign model simulation in individual computer CPU and collect results
        cpu_dict = {}
        with concurrent.futures.ProcessPoolExecutor(max_workers=self.max_workers) as executor:
            # Multicore simulation
            futures = [
                executor.submit(cpu_sim, idx, arr) for idx, arr in enumerate(x, start=start_sim)
            ]
            for future in concurrent.futures.as_completed(futures):
                # Display end of current simulation for tracking
                print(f'Completed simulation: {start_sim + futures.index(future)}/{self.total_sim}', flush=True)
                # Collect simulation results
                future_result = future.result()
                cpu_dict[tuple(future_result['array'])] = {
                    k: v for k, v in future_result.items() if k != 'array'
                }

        # Mapping betten objectives and their directions
        objs_dirs = self._objectives_directions()

        # An empty list to store population objectives in current generation
        gen_objs = []

        # Iterate population array in the current generation
        for pop in x:
            # Empty list to collect individual population objectives
            pop_objs = []
            # Simulation output for the population
            pop_sim = cpu_dict[tuple(pop)]
            # Iterate objectives
            for obj in self.objective_config:
                # Simulated DataFrame
                sim_df = pop_sim[self.df_key[obj]]
                sim_df.columns = ['date', 'sim']
                # Observed DataFrame
                obs_df = self.observe_dict[self.df_key[obj]]
                # Merge simulated and observed DataFrames by 'date' column
                merge_df = sim_df.merge(
                    right=obs_df,
                    how='inner',
                    on='date'
                )
                # Normalized DataFrame
                norm_df = utils._df_normalize(
                    df=merge_df[['sim', 'obs']],
                    norm_col='obs'
                )
                # Indicator method from abbreviation
                obj_ind = self.objective_config[obj]['indicator']
                indicator_method = getattr(
                    PerformanceMetrics(),
                    f'compute_{obj_ind.lower()}'
                )
                # Indicator value computed from method
                ind_val = indicator_method(
                    df=norm_df,
                    sim_col='sim',
                    obs_col='obs'
                )
                # Objective value based on maximize or minimize direction
                obj_val = - ind_val if objs_dirs[obj_ind] == 'max' else ind_val
                # Store objective value for sample
                pop_objs.append(obj_val)
            # Store sample objectives in population objective list
            gen_objs.append(pop_objs)

        # Array of objective values for the current generation
        out["F"] = numpy.array(gen_objs)

        # Print end of current generation number
        print(
            f'\nCompleted generation: {self.track_gen}/{self.n_gen}\n'
        )

    def _objectives_directions(
        self
    ) -> dict[str, str]:
        '''
        Provide a dictionary mapping optimization objectives to their respective directions.
        '''

        objs_dirs = {
            'NSE': 'max',
            'KGE': 'max',
            'MSE': 'min',
            'RMSE': 'min',
            'MARE': 'min'
        }

        return objs_dirs

    def _algorithm_class(
        self,
        algorithm: str
    ) -> type:
        '''
        Retrieve the optimization algorithm class from the `pymoo` package.
        '''

        single_obj = ['GA', 'DE']
        multi_obj = ['NSGA2']

        # Dictionary mapping between algorithm name and module
        api_module = {
            'GA': importlib.import_module('pymoo.algorithms.soo.nonconvex.ga'),
            'DE': importlib.import_module('pymoo.algorithms.soo.nonconvex.de'),
            'NSGA2': importlib.import_module('pymoo.algorithms.moo.nsga2')
        }

        # Check algorithm name is valid
        if algorithm not in api_module:
            raise NameError(
                f'Invalid algorithm "{algorithm}"; valid names are {list(api_module.keys())}'
            )

        # Check single objective algorithm cannot be used for multiple objectives
        if len(self.objective_config) >= 2 and algorithm in single_obj:
            raise ValueError(
                f'Algorithm "{algorithm}" cannot handle multiple objectives; use one of {multi_obj}'
            )

        # Algorithm class
        alg_class = typing.cast(
            type,
            getattr(api_module[algorithm], algorithm)
        )

        return alg_class

    def parameter_optimization(
        self
    ) -> dict[str, typing.Any]:
        '''
        Run the optimization using the configured algorithm, population size, and number of generations.

        This method executes the optimization process and returns a dictionary containing the optimized
        parameters, corresponding objective values, and total execution time.

        The following JSON files are saved in `calsim_dir`:

        - `optimization_history.json`: A dictionary containing the optimization history. Each key in this
          file is an integer starting from 1, representing the generation number. The corresponding value
          is a sub-dictionary with two keys: `pop` for the population data (decision variables) and `obj`
          for the objective function values. This file is useful for analyzing optimization progress,
          convergence trends, performance indicators, and visualization.

        - `optimization_result.json`: A dictionary containing the final output dictionary.

        Returns:
            Dictionary with the following keys:

                - `algorithm`: Name of the algorithm.
                - `generation`: Number of generation to run.
                - `population`: Number of individual in each generation.
                - `total_simulation`: Number of total simulation.
                - `time_sec`: Total execution time in seconds.
                - `variables`: Array of optimized decision variables.
                - `objectives`: Array of objective values corresponding to the optimized decision variables.

        Note:
            - For multi-objective optimization, the number of `variables` and `objectives` may exceed one,
              representing non-dominated solutions where a solution cannot be improved in one objective
              without worsening another.

            - The computation progress can be tracked through the following `console` messages.
              The simulation index ranges from 1 to the total number of simulations, which equals the product
              of the population size and the number of generations:

                - `Started generation: <current_started_generation>/<total_generations>`
                - `Started simulation: <current_started_index>/<total_simulations>`
                - `Completed simulation: <current_compeleted_index>/<total_simulations>`
                - `Completed generation: <current_completed_generation>/<total_generations>`

            - The disk space on the computer for `calsim_dir` must be sufficient to run
              parallel simulations (at least `pop_size` times the size of the `TxtInOut` folder).
              Otherwise, no error will be raised by the system, but simulation outputs may not be generated.
        '''

        # Algorithm object
        alg_class = self._algorithm_class(
            algorithm=self.algorithm
        )
        alg_object = alg_class(
            pop_size=self.pop_size
        )

        # Optimization result
        result = pymoo.optimize.minimize(
            problem=self,
            algorithm=alg_object,
            termination=('n_gen', self.n_gen),
            save_history=True
        )

        # Sign of objective directions
        objs_dirs = self._objectives_directions()
        dir_list = [
            objs_dirs[v['indicator']] for k, v in self.objective_config.items()
        ]
        dir_sign = numpy.where(numpy.array(dir_list) == 'max', -1, 1)

        # Save optimization history for each generation
        opt_hist = {}
        for i, gen in enumerate(result.history, start=1):
            opt_hist[i] = {
                'pop': gen.pop.get('X').tolist(),
                'obj': (gen.pop.get('F') * dir_sign).tolist()
            }
        with open(self.calsim_dir / 'optimization_history.json', 'w') as output_write:
            json.dump(opt_hist, output_write, indent=4)

        # Optimized output of parameters, objectives, and execution times
        opt_output = {
            'algorithm': self.algorithm,
            'generation': self.n_gen,
            'population': self.pop_size,
            'total_simulation': self.pop_size * self.n_gen,
            'time_sec': round(result.exec_time),
            'variables': result.X,
            'objectives': result.F * dir_sign
        }

        # Save optimized outputs
        save_output = copy.deepcopy(opt_output)
        save_output = {
            k: v.tolist() if k.startswith(('var', 'obj')) else v for k, v in save_output.items()
        }
        with open(self.calsim_dir / 'optimization_result.json', 'w') as output_write:
            json.dump(save_output, output_write, indent=4)

        return opt_output

parameter_optimization() -> dict[str, typing.Any]

Run the optimization using the configured algorithm, population size, and number of generations.

This method executes the optimization process and returns a dictionary containing the optimized parameters, corresponding objective values, and total execution time.

The following JSON files are saved in calsim_dir:

• optimization_history.json: A dictionary containing the optimization history. Each key in this file is an integer starting from 1, representing the generation number. The corresponding value is a sub-dictionary with two keys: pop for the population data (decision variables) and obj for the objective function values. This file is useful for analyzing optimization progress, convergence trends, performance indicators, and visualization.

• optimization_result.json: A dictionary containing the final output dictionary.

Returns:

Type	Description
dict[str, Any]	Dictionary with the following keys: algorithm: Name of the algorithm. generation: Number of generation to run. population: Number of individual in each generation. total_simulation: Number of total simulation. time_sec: Total execution time in seconds. variables: Array of optimized decision variables. objectives: Array of objective values corresponding to the optimized decision variables.

Note

• For multi-objective optimization, the number of variables and objectives may exceed one, representing non-dominated solutions where a solution cannot be improved in one objective without worsening another.

• The computation progress can be tracked through the following console messages. The simulation index ranges from 1 to the total number of simulations, which equals the product of the population size and the number of generations:

  • Started generation: <current_started_generation>/<total_generations>
  • Started simulation: <current_started_index>/<total_simulations>
  • Completed simulation: <current_compeleted_index>/<total_simulations>
  • Completed generation: <current_completed_generation>/<total_generations>

• The disk space on the computer for calsim_dir must be sufficient to run parallel simulations (at least pop_size times the size of the TxtInOut folder). Otherwise, no error will be raised by the system, but simulation outputs may not be generated.

Source code in pySWATPlus/calibration.py

def parameter_optimization(
    self
) -> dict[str, typing.Any]:
    '''
    Run the optimization using the configured algorithm, population size, and number of generations.

    This method executes the optimization process and returns a dictionary containing the optimized
    parameters, corresponding objective values, and total execution time.

    The following JSON files are saved in `calsim_dir`:

    - `optimization_history.json`: A dictionary containing the optimization history. Each key in this
      file is an integer starting from 1, representing the generation number. The corresponding value
      is a sub-dictionary with two keys: `pop` for the population data (decision variables) and `obj`
      for the objective function values. This file is useful for analyzing optimization progress,
      convergence trends, performance indicators, and visualization.

    - `optimization_result.json`: A dictionary containing the final output dictionary.

    Returns:
        Dictionary with the following keys:

            - `algorithm`: Name of the algorithm.
            - `generation`: Number of generation to run.
            - `population`: Number of individual in each generation.
            - `total_simulation`: Number of total simulation.
            - `time_sec`: Total execution time in seconds.
            - `variables`: Array of optimized decision variables.
            - `objectives`: Array of objective values corresponding to the optimized decision variables.

    Note:
        - For multi-objective optimization, the number of `variables` and `objectives` may exceed one,
          representing non-dominated solutions where a solution cannot be improved in one objective
          without worsening another.

        - The computation progress can be tracked through the following `console` messages.
          The simulation index ranges from 1 to the total number of simulations, which equals the product
          of the population size and the number of generations:

            - `Started generation: <current_started_generation>/<total_generations>`
            - `Started simulation: <current_started_index>/<total_simulations>`
            - `Completed simulation: <current_compeleted_index>/<total_simulations>`
            - `Completed generation: <current_completed_generation>/<total_generations>`

        - The disk space on the computer for `calsim_dir` must be sufficient to run
          parallel simulations (at least `pop_size` times the size of the `TxtInOut` folder).
          Otherwise, no error will be raised by the system, but simulation outputs may not be generated.
    '''

    # Algorithm object
    alg_class = self._algorithm_class(
        algorithm=self.algorithm
    )
    alg_object = alg_class(
        pop_size=self.pop_size
    )

    # Optimization result
    result = pymoo.optimize.minimize(
        problem=self,
        algorithm=alg_object,
        termination=('n_gen', self.n_gen),
        save_history=True
    )

    # Sign of objective directions
    objs_dirs = self._objectives_directions()
    dir_list = [
        objs_dirs[v['indicator']] for k, v in self.objective_config.items()
    ]
    dir_sign = numpy.where(numpy.array(dir_list) == 'max', -1, 1)

    # Save optimization history for each generation
    opt_hist = {}
    for i, gen in enumerate(result.history, start=1):
        opt_hist[i] = {
            'pop': gen.pop.get('X').tolist(),
            'obj': (gen.pop.get('F') * dir_sign).tolist()
        }
    with open(self.calsim_dir / 'optimization_history.json', 'w') as output_write:
        json.dump(opt_hist, output_write, indent=4)

    # Optimized output of parameters, objectives, and execution times
    opt_output = {
        'algorithm': self.algorithm,
        'generation': self.n_gen,
        'population': self.pop_size,
        'total_simulation': self.pop_size * self.n_gen,
        'time_sec': round(result.exec_time),
        'variables': result.X,
        'objectives': result.F * dir_sign
    }

    # Save optimized outputs
    save_output = copy.deepcopy(opt_output)
    save_output = {
        k: v.tolist() if k.startswith(('var', 'obj')) else v for k, v in save_output.items()
    }
    with open(self.calsim_dir / 'optimization_result.json', 'w') as output_write:
        json.dump(save_output, output_write, indent=4)

    return opt_output