Skip to content

Documentation for ResultsHandler

Provides all functions that operate on calculated results. As IO for the results object the ResultsHandler is able to handle results on its own.

__init__(self, results_object=None, output_settings=None) special

Initialize the object.

Parameters:

Name Type Description Default
results_object MDBHyperpipe

All results are stored here. An initial setting is not necessary, because a later loading via file or MongoDB is possible.

None
output_settings OutputSettings

Setting for creation and storage of the results_object.

None
Source code in photonai/processing/results_handler.py
def __init__(self, results_object: MDBHyperpipe = None, output_settings=None):
    """
    Initialize the object.

    Parameters:
        results_object:
            All results are stored here.
            An initial setting is not necessary,
            because a later loading via file or MongoDB is possible.

        output_settings (OutputSettings):
            Setting for creation and storage of the results_object.

    """
    self.results = results_object
    self.output_settings = output_settings

get_config_evaluations(self)

Return the test performance of every tested configuration in every outer fold.

Returns:

Type Description
dict

Test performance of every configuration.

Source code in photonai/processing/results_handler.py
def get_config_evaluations(self) -> dict:
    """
    Return the test performance of every tested configuration in every outer fold.

    Returns:
        Test performance of every configuration.

    """
    config_performances = list()
    maximum_fold = None
    for outer_fold in self.results.outer_folds:
        if maximum_fold is None or len(outer_fold.tested_config_list) > maximum_fold:
            maximum_fold = len(outer_fold.tested_config_list)

    for outer_fold in self.results.outer_folds:
        performance = dict()
        for metric in self.results.hyperpipe_info.metrics:
            performance[metric] = list()

        for i in range(maximum_fold):
            # for config in outer_fold.tested_config_list:
            for metric in self.results.hyperpipe_info.metrics:
                if i >= len(outer_fold.tested_config_list):
                    performance[metric].append(np.nan)
                    continue
                config = outer_fold.tested_config_list[i]
                if config.config_failed:
                    performance[metric].append(np.nan)
                else:
                    for item in config.metrics_test:
                        if (item.operation == 'mean') and (item.metric_name == metric):
                            performance[metric].append(item.value)
        config_performances.append(performance)

    config_performances_dict = dict()
    for metric in self.results.hyperpipe_info.metrics:
        config_performances_dict[metric] = list()
        for fold in config_performances:
            config_performances_dict[metric].append(fold[metric])

    return config_performances_dict

get_methods() staticmethod

This function returns a list of all methods available for ResultsHandler.

Returns:

Type Description
list

List of all available methods.

Source code in photonai/processing/results_handler.py
@staticmethod
def get_methods() -> list:
    """
    This function returns a list of all methods available for ResultsHandler.

    Returns:
        List of all available methods.

    """
    methods_list = [s for s in dir(ResultsHandler) if '__' not in s]
    return methods_list

get_performance_table(self)

This function returns a summary table of the overall results.

ToDo: add best_config information!

Source code in photonai/processing/results_handler.py
def get_performance_table(self):
    """This function returns a summary table of the overall results.

    ToDo: add best_config information!
    """
    res_tab = pd.DataFrame()
    for i, folds in enumerate(self.results.outer_folds):
        # add best config infos
        res_tab.loc[i, 'best_config'] = str(folds.best_config.human_readable_config)

        # add fold index
        res_tab.loc[i, 'fold'] = folds.fold_nr

        # add sample size infos
        res_tab.loc[i, 'n_train'] = folds.best_config.best_config_score.number_samples_training
        res_tab.loc[i, 'n_validation'] = folds.best_config.best_config_score.number_samples_validation

        # add performance metrics
        d = folds.best_config.best_config_score.validation.metrics
        for key, value in d.items():
            res_tab.loc[i, key] = value

    # add row with overall info
    res_tab.loc[i + 1, 'n_validation'] = np.sum(res_tab['n_validation'])
    for key, value in d.items():
        m = res_tab.loc[:, key]
        res_tab.loc[i+1, key] = np.mean(m)
        res_tab.loc[i + 1, key + '_sem'] = sem(m)   # standard error of the mean
    res_tab.loc[i + 1, 'best_config'] = 'Overall'
    return res_tab

load_from_file(self, results_file)

Read results_file from json into MDBHyperpipe object self.results.

Parameters:

Name Type Description Default
results_file str

Full path to json file.

required
Source code in photonai/processing/results_handler.py
def load_from_file(self, results_file: str):
    """
    Read results_file from json into MDBHyperpipe object self.results.

    Parameters:
        results_file:
            Full path to json file.

    """
    self.results = MDBHyperpipe.from_document(json.load(open(results_file, 'r')))

load_from_mongodb(self, mongodb_connect_url, pipe_name)

Read results_file from MongoDB into MDBHyperpipe object self.results.

Parameters:

Name Type Description Default
mongodb_connect_url str

MongoDB connection string.

required
pipe_name str

Name of the stored hyperpipe.

required
Source code in photonai/processing/results_handler.py
def load_from_mongodb(self, mongodb_connect_url: str, pipe_name: str):
    """
    Read results_file from MongoDB into MDBHyperpipe object self.results.

    Parameters:
        mongodb_connect_url:
            MongoDB connection string.

        pipe_name:
            Name of the stored hyperpipe.

    """
    connect(mongodb_connect_url, alias="photon_core")
    results = list(MDBHyperpipe.objects.raw({'name': pipe_name}))
    if len(results) == 1:
        self.results = results[0]
    elif len(results) > 1:
        self.results = MDBHyperpipe.objects.order_by([("computation_start_time", DESCENDING)]).raw({'name': pipe_name}).first()
        warn_text = 'Found multiple hyperpipes with that name. Returning most recent one.'
        logger.warning(warn_text)
        warnings.warn(warn_text)
    else:
        raise FileNotFoundError('Could not load hyperpipe from MongoDB.')