tools.helpers

class tools.helpers.MatildaBulkProcessor(scenarios, matilda_settings, matilda_parameters)[source]

A class to run multiple MATILDA simulations for different input scenarios and models in single or multi-processing mode and store the results in a dictionary. .. attribute:: scenarios

A dictionary with scenario names as keys and a dictionary of climate models as values.

type:

dict

matilda_settings

A dictionary of MATILDA settings.

Type:: dict

matilda_parameters

A dictionary of MATILDA parameter values.

Type:: dict

run_single_process():: Runs the MATILDA simulations for the scenarios and models in single-processing mode and returns a dictionary of results.

run_multi_process():: Runs the MATILDA simulations for the scenarios and models in multi-processing mode and returns a dictionary of results.

matilda_headless(df, matilda_settings, matilda_parameters):: A helper function to run a single MATILDA simulation given a dataframe, MATILDA settings and parameter values.

static matilda_headless(df, matilda_settings, matilda_parameters)[source]

A helper function to run a single MATILDA simulation given a dataframe, MATILDA settings and parameter values. :param df: The input dataframe for the MATILDA simulation. :type df: pandas.DataFrame :param matilda_settings: A dictionary of MATILDA settings. :type matilda_settings: dict :param matilda_parameters: A dictionary of MATILDA parameter values. :type matilda_parameters: dict

Returns:: A dictionary containing the MATILDA model output and glacier rescaling factor.
Return type:: dict

run_multi_process(num_cores=2)[source]: Runs the MATILDA simulations for the scenarios and models in multi-processing mode and returns a dictionary of results. :returns: A dictionary of MATILDA simulation results. :rtype: dict

run_single_process()[source]: Runs the MATILDA simulations for the scenarios and models in single-processing mode and returns a dictionary of results. :returns: A dictionary of MATILDA simulation results. :rtype: dict

tools.helpers.adjust_bias(predictand, predictor, method='normal_mapping')[source]: Applies bias correction to discrete periods individually.

tools.helpers.confidence_interval(df)[source]

Calculate the mean and 95% confidence interval for each row in a dataframe.

Parameters:: df (pandas.DataFrame) – The input dataframe.
Returns:: A dataframe with the mean and confidence intervals for each row.
Return type:: pandas.DataFrame

tools.helpers.create_scenario_dict(tas: dict, pr: dict, scenario_nums: list) → dict[source]

Create a nested dictionary of scenarios and models from two dictionaries of pandas DataFrames.

Parameters:

tas (dict) – A dictionary of pandas DataFrames where the keys are scenario names and each DataFrame has columns representing different climate model mean daily temperature (K) time series.
pr (dict) – A dictionary of pandas DataFrames where the keys are scenario names and each DataFrame has columns representing different climate model mean daily precipitation (mm/day) time series.
scenario_nums (list) – A list of integers representing the scenario numbers to include in the resulting dictionary.

Returns:

A nested dictionary where the top-level keys are scenario names (e.g. ‘SSP2’, ‘SSP5’) and the values are dictionaries containing climate models as keys and the corresponding pandas DataFrames as values. The DataFrames have three columns: ‘TIMESTAMP’, ‘T2’, and ‘RRR’, where ‘TIMESTAMP’ represents the time step, ‘T2’ the mean daily temperature (K), and ‘RRR’ the mean daily precipitation (mm/day).

Return type:

dict

tools.helpers.crop2wy(df, begin=10)[source]

Crops a DataFrame to include only the rows that fall within a complete water year. :param df: Input DataFrame with a DatetimeIndex and a ‘water_year’ column. :type df: pandas.DataFrame :param begin: The month (1-12) that marks the beginning of the water year. Default is 10. :type begin: int, optional

Returns:: A new DataFrame containing only the rows that fall within a complete water year.
Return type:: pandas.DataFrame or None

tools.helpers.dict_filter(dictionary, filter_string)[source]: Returns a dict with all elements of the input dict that contain a filter string in their keys.

tools.helpers.dict_to_parquet(dictionary: dict, directory_path: str, pbar: bool = True) → None[source]: Recursively stores the dataframes in the input dictionary as parquet files in the specified directory. Nested dictionaries are supported. If the specified directory does not exist, it will be created. :param dictionary: A nested dictionary containing pandas dataframes. :type dictionary: dict :param directory_path: The directory path to store the parquet files. :type directory_path: str :param pbar: A flag indicating whether to display a progress bar. Default is True. :type pbar: bool, optional

tools.helpers.dict_to_pickle(dic, target_path)[source]

Saves a dictionary to a pickle file at the specified target path. Creates target directory if not existing. :param dic: The dictionary to save to a pickle file. :type dic: dict :param target_path: The path of the file where the dictionary shall be stored. :type target_path: str

Return type:: None

tools.helpers.drop_keys(dic, keys_to_drop)[source]

Removes specified keys from a dictionary. :param dict: The dictionary to remove keys from. :type dict: dict :param keys_to_drop: A list of keys to remove from the dictionary. :type keys_to_drop: list

Returns:: A modified dictionary with the specified keys removed.
Return type:: dict

tools.helpers.drop_model(col_names, dict_or_df)[source]

Drop columns with given names from either a dictionary of dataframes or a single dataframe.

Parameters:

col_names (list of str) – The list of model names to drop.
dict_or_df (dict of pandas.DataFrame or pandas.DataFrame) – If a dict of dataframes, all dataframes in the dict will be edited. If a single dataframe, only that dataframe will be edited.

Returns:

The updated dictionary of dataframes or dataframe with dropped columns.

Return type:

dict of pandas.DataFrame or pandas.DataFrame

tools.helpers.get_si(fast_results: str, to_csv: bool = False) → pandas.DataFrame[source]

Computes the sensitivity indices of a given FAST simulation results file.

Parameters:

fast_results (str) – The path of the FAST simulation results file.
to_csv (bool, optional) – If True, the sensitivity indices are saved to a CSV file with the same name as fast_results, but with ‘_sensitivity_indices.csv’ appended to the end (default is False).

Returns:

A pandas DataFrame containing the sensitivity indices and parameter names.

Return type:

pd.DataFrame

tools.helpers.handle_dash_availability()[source]

Check whether the notebook is running locally.

Returns:: True if Dash dashboards should be displayed. False if Dash should be skipped.
Return type:: bool

tools.helpers.hydrologicalize(df, begin_of_water_year=10)[source]

Adds a ‘water_year’ column to a DataFrame and crops it to include only complete water years. :param df: Input DataFrame with a DatetimeIndex. :type df: pandas.DataFrame :param begin_of_water_year: The month (1-12) that marks the beginning of the water year. Default is 10. :type begin_of_water_year: int, optional

Returns:: A new DataFrame with a ‘water_year’ column and only rows that fall within complete water years.
Return type:: pandas.DataFrame or None

tools.helpers.mean_elevation_from_raster(raster_path, geometry_gdf)[source]

Calculate mean elevation from a raster within a polygon geometry.

Parameters:

raster_path (str or Path) – Path to the DEM raster.
geometry_gdf (geopandas.GeoDataFrame) – Polygon geometry used to clip the raster.

Returns:

Mean elevation of valid raster cells within the polygon.

Return type:

float

tools.helpers.parquet_to_dict(directory_path: str, pbar: bool = True) → dict[source]

Recursively loads dataframes from the parquet files in the specified directory and returns a dictionary. Nested directories are supported. :param directory_path: The directory path containing the parquet files. :type directory_path: str :param pbar: A flag indicating whether to display a progress bar. Default is True. :type pbar: bool, optional

Returns:: A dictionary containing the loaded pandas dataframes.
Return type:: dict

tools.helpers.pickle_to_dict(file_path)[source]

Loads a dictionary from a pickle file at a specified file path. :param file_path: The path of the pickle file to load. :type file_path: str

Returns:: The dictionary loaded from the pickle file.
Return type:: dict

tools.helpers.read_era5l(file)[source]: Reads ERA5-Land data, drops redundant columns, and adds DatetimeIndex. Resamples the dataframe to reduce the DatetimeIndex to daily resolution.

tools.helpers.read_yaml(file_path)[source]

Read a YAML file and return the contents as a dictionary. :param file_path: The path of the YAML file to read. :type file_path: str

Returns:: The contents of the YAML file as a dictionary.
Return type:: dict

tools.helpers.replace_values(target_df, source_df, source_column)[source]

Replaces values in the overlapping period in the target dataframe with values from the source dataframe using the specified source column.

Parameters:

target_df (pd.DataFrame) – Target dataframe where values will be replaced.
source_df (pd.DataFrame) – Source dataframe from which values will be taken.
source_column (str) – Column name in the source dataframe to use for replacement.

Returns:

The target dataframe with updated values.

Return type:

pd.DataFrame

tools.helpers.restore_output_archive(zip_file='output_download.zip', target_dir='output')[source]: Extract a ZIP archive into target_dir. If target_dir already contains files, ask whether to replace its contents.

tools.helpers.update_yaml(file_path, new_items)[source]

Update a YAML file with the contents of a dictionary. :param file_path: The path of the YAML file to update. :type file_path: str :param new_items: The dictionary of new key-value pairs to add to the existing YAML file. :type new_items: dict

Return type:: None

tools.helpers.water_year(df, begin=10)[source]

Calculates the water year for each date in the index of the input DataFrame. :param df: Input DataFrame with a DatetimeIndex. :type df: pandas.DataFrame :param begin: The month (1-12) that marks the beginning of the water year. Default is 10. :type begin: int, optional

Returns:: An array of integers representing the water year for each date in the input DataFrame index.
Return type:: numpy.ndarray

tools.helpers.write_yaml(data, file_path)[source]

Write a dictionary to a YAML file. Ensures all values are in standard Python types before writing.

Parameters:

data (dict) – The dictionary to write to a YAML file.
file_path (str) – The path of the file where the YAML data shall be stored.

Return type:

None