evaluation.py

from __future__ import print_function

import collections
import logging
from itertools import chain, product
import math
import random

from .representations import Construction

_logger = logging.getLogger(__name__)

EvaluationConfig = collections.namedtuple('EvaluationConfig',
                                          ['num_samples', 'sample_size'])

FORMAT_STRINGS = {
    'default': """Filename   : {name}
Num samples: {samplesize_count}
Sample size: {samplesize_avg}
F-score    : {fscore_avg:.3}
Precision  : {precision_avg:.3}
Recall     : {recall_avg:.3}""",
    'table': "{name:10} {precision_avg:6.3} {recall_avg:6.3} {fscore_avg:6.3}",
    'latex': "{name} & {precision_avg:.3} &"
             " {recall_avg:.3} & {fscore_avg:.3} \\\\"}


def _sample(compound_list, size, seed):
    """Create a specific size sample from the compound list using a specific
    seed"""
    return random.Random(seed).sample(compound_list, size)


class MorfessorEvaluationResult(object):
    """A MorfessorEvaluationResult is returned by a MorfessorEvaluation
    object. It's purpose is to store the evaluation data and provide nice
    formatting options.

    Each MorfessorEvaluationResult contains the data of 1 evaluation
    (which can have multiple samples).

    """

    print_functions = {'avg': lambda x: sum(x) / len(x),
                       'min': min,
                       'max': max,
                       'values': list,
                       'count': len}
    #TODO add maybe std as a print function?

    def __init__(self, meta_data=None):
        self.meta_data = meta_data

        self.precision = []
        self.recall = []
        self.fscore = []
        self.samplesize = []

        self._cache = None

    def __getitem__(self, item):
        """Provide dict style interface for all values (standard values and
        metadata)"""
        if self._cache is None:
            self._fill_cache()

        return self._cache[item]

    def add_data_point(self, precision, recall, f_score, sample_size):
        """Method used by MorfessorEvaluation to add the results of a single
        sample to the object"""
        self.precision.append(precision)
        self.recall.append(recall)
        self.fscore.append(f_score)
        self.samplesize.append(sample_size)

        #clear cache
        self._cache = None

    def __str__(self):
        """Method for default visualization"""
        return self.format(FORMAT_STRINGS['default'])

    def _fill_cache(self):
        """ Pre calculate all variable / function combinations and put them in
        cache"""
        self._cache = {'{}_{}'.format(val, func_name): func(getattr(self, val))
                       for val in ('precision', 'recall', 'fscore',
                                   'samplesize')
                       for func_name, func in self.print_functions.items()}
        self._cache.update(self.meta_data)

    def _get_cache(self):
        """ Fill the cache (if necessary) and return it"""
        if self._cache is None:
            self._fill_cache()
        return self._cache

    def format(self, format_string):
        """ Format this object. The format string can contain all variables,
        e.g. fscore_avg, precision_values or any item from metadata"""
        return format_string.format(**self._get_cache())


class MorfessorEvaluation(object):
    """ Do the evaluation of one model, on one testset. The basic procedure is
    to create, in a stable manner, a number of samples and evaluate them
    independently. The stable selection of samples makes it possible to use
    the resulting values for Pair-wise statistical significance testing.

    reference_annotations is a standard annotation dictionary:
    {compound => ([annoation1],.. ) }
    """
    def __init__(self, reference_annotations):
        self.reference = {}

        for compound, analyses in reference_annotations.items():
            self.reference[compound] = [compound.segmentation_to_splitlist(a) for a in analyses]

        self._samples = {}

    def _create_samples(self, configuration=EvaluationConfig(10, 1000)):
        """Create, in a stable manner, n testsets of size x as defined in
        test_configuration
        """

        #TODO: What is a reasonable limit to warn about a too small testset?
        if len(self.reference) < (configuration.num_samples *
                                  configuration.sample_size):
            _logger.warning("The test set is too small for this sample size")

        compound_list = sorted(self.reference.keys())
        self._samples[configuration] = [
            _sample(compound_list, configuration.sample_size, i) for i in
            range(configuration.num_samples)]

    def get_samples(self, configuration=EvaluationConfig(10, 1000)):
        """Get a list of samples. A sample is a list of compounds.

        This method is stable, so each time it is called with a specific
        test_set and configuration it will return the same samples. Also this
        method caches the samples in the _samples variable.

        """
        if configuration not in self._samples:
            self._create_samples(configuration)
        return self._samples[configuration]

    def _evaluate(self, prediction, locations_only=True):
        """Helper method to get the precision and recall of 1 sample
        NOTE: precision and recall are calculated as AVERAGES on a by-item basis,
        which yields differences from the global calculation when the gold
        standard for an item and/or the system prediction for that item
        have no splits.
        When the gold standard of an item has no splits, the system scores
        1 for recall (0/0).
        When the system prediction has no splits, the system scores 1 for
        precision (0/0).
        On a global basis, cases where the system prediction has no splits
        should not contribute to precision, and cases where the gold standard
        has no splits do not contribute to recall.
        (on top of this, averaging by-item gives every word the same weight;
        on a global basis, words with more splits in the gold and/or system
        have more weight. but a by-item approach is more interpretable.)
        """
        def calc_prop_distance(ref, pred, locations_only=True):
            """Calculate recall based on split locations or properties."""
            if locations_only:
                ref = ref.splitlocs
                pred = pred.splitlocs
            if len(ref) == 0:
                return 1.0
            diff = len(set(ref) - set(pred))
            return (len(ref) - diff) / float(len(ref))

        wordlist = sorted(set(prediction.keys()) & set(self.reference.keys()))

        recall_sum = 0.0
        precis_sum = 0.0

        for word in wordlist:
            if len(word) < 2:
                continue

            recall_sum += max(calc_prop_distance(r, p, locations_only=locations_only)
                              for p, r in product(prediction[word],
                                                  self.reference[word]))

            precis_sum += max(calc_prop_distance(p, r, locations_only=locations_only)
                              for p, r in product(prediction[word],
                                                  self.reference[word]))

        precision = precis_sum / len(wordlist)
        recall = recall_sum / len(wordlist)
        f_score = 2.0 / (1.0 / precision + 1.0 / recall)

        return precision, recall, f_score, len(wordlist)

    def evaluate_model(self, model, configuration=EvaluationConfig(10, 1000),
                       meta_data=None, locations_only=True):
        """Get the prediction of the test samples from the model and do the
        evaluation.
        If locations_only is True, only check that Splits are in the right
        location; do not check that they are the right kind
        (e.g. Split vs. RedSplit).

        The meta_data object has preferably at least the key 'name'.

        """
        if meta_data is None:
            meta_data = {'name': 'UNKNOWN'}

        mer = MorfessorEvaluationResult(meta_data)

        for i, sample in enumerate(self.get_samples(configuration)):
            _logger.debug("Evaluating sample %s", i)
            prediction = {}
            for compound in sample:
                prediction[compound] = [model.viterbi_splitlist(compound)[0]]

            mer.add_data_point(*self._evaluate(prediction, locations_only=locations_only))

        return mer

    def evaluate_segmentation(self, segmentation,
                              configuration=EvaluationConfig(10, 1000),
                              meta_data=None, locations_only=True):
        """Method for evaluating an existing segmentation.
        If locations_only is True, only check that Splits are in the right
        location; do not check that they are the right kind
        (e.g. Split vs. RedSplit)."""

        segmentation = {compound: [compound.segmentation_to_splitlist(analysis)] for compound, analysis in segmentation}

        if meta_data is None:
            meta_data = {'name': 'UNKNOWN'}

        mer = MorfessorEvaluationResult(meta_data)

        for i, sample in enumerate(self.get_samples(configuration)):
            _logger.debug("Evaluating sample %s", i)

            prediction = {k: v for k, v in segmentation.items() if k in sample}
            mer.add_data_point(*self._evaluate(prediction, locations_only=locations_only))

        return mer


class WilcoxonSignedRank(object):
    """Class for doing statistical signficance testing with the Wilcoxon
    Signed-Rank test

    It implements the Pratt method for handling zero-differences and
    applies a 0.5 continuity correction for the z-statistic.

    """

    @staticmethod
    def _wilcoxon(d, method='pratt', correction=True):
        if method not in ('wilcox', 'pratt'):
            raise ValueError
        if method == 'wilcox':
            d = list(filter(lambda a: a != 0, d))

        count = len(d)

        ranks = WilcoxonSignedRank._rankdata([abs(v) for v in d])
        rank_sum_pos = sum(r for r, v in zip(ranks, d) if v > 0)
        rank_sum_neg = sum(r for r, v in zip(ranks, d) if v < 0)

        test = min(rank_sum_neg, rank_sum_pos)

        mean = count * (count + 1) * 0.25
        stdev = (count*(count + 1) * (2 * count + 1))
        # compensate for duplicate ranks
        no_zero_ranks = [r for i, r in enumerate(ranks) if d[i] != 0]
        stdev -= 0.5 * sum(x * (x*x-1) for x in
                           collections.Counter(no_zero_ranks).values())

        stdev = math.sqrt(stdev / 24.0)

        if correction:
            correction = +0.5 if test > mean else -0.5
        else:
            correction = 0
        z = (test - mean - correction) / stdev

        return 2 * WilcoxonSignedRank._norm_cum_pdf(abs(z))

    @staticmethod
    def _rankdata(d):
        od = collections.Counter()
        for v in d:
            od[v] += 1

        rank_dict = {}
        cur_rank = 1
        for val, count in sorted(od.items(), key=lambda x: x[0]):
            rank_dict[val] = (cur_rank + (cur_rank + count - 1)) / 2
            cur_rank += count

        return [rank_dict[v] for v in d]

    @staticmethod
    def _norm_cum_pdf(z):
        """Pure python implementation of the normal cumulative pdf function"""
        return 0.5 - 0.5 * math.erf(z / math.sqrt(2))

    def significance_test(self, evaluations, val_property='fscore_values',
                          name_property='name'):
        """Takes a set of evaluations (which should have the same
        test-configuration) and calculates the p-value for the Wilcoxon signed
        rank test

        Returns a dictionary with (name1,name2) keys and p-values as values.
        """
        results = {r[name_property]: r[val_property] for r in evaluations}
        if any(len(x) < 10 for x in results.values()):
            _logger.error("Too small number of samples for the Wilcoxon test")
            return {}
        p = {}
        for r1, r2 in product(results.keys(), results.keys()):
            p[(r1, r2)] = self._wilcoxon([v1-v2
                                          for v1, v2 in zip(results[r1],
                                                            results[r2])])

        return p

    @staticmethod
    def print_table(results):
        """Nicely format a results table as returned by significance_test"""
        names = sorted(set(r[0] for r in results.keys()))

        col_width = max(max(len(n) for n in names), 5)

        for h in chain([""], names):
            print('{:{width}}'.format(h, width=col_width), end='|')
        print()

        for name in names:
            print('{:{width}}'.format(name, width=col_width), end='|')

            for name2 in names:
                print('{:{width}.5}'.format(results[(name, name2)],
                                            width=col_width), end='|')
            print()