Source code for safe.postprocessors.aggregation_categorical_postprocessor

# -*- coding: utf-8 -*-
"""**Postprocessors package.**

"""

__author__ = 'Marco Bernasocchi <[email protected]>'
__revision__ = '$Format:%H$'
__date__ = '10/10/2012'
__license__ = "GPL"
__copyright__ = 'Copyright 2012, Australia Indonesia Facility for '
__copyright__ += 'Disaster Reduction'

from safe.postprocessors.abstract_postprocessor import AbstractPostprocessor
from safe.utilities.i18n import tr


[docs]class AggregationCategoricalPostprocessor(AbstractPostprocessor):
    """
    Postprocessor that calculates categorical statistics.
    see the _calculate_* methods to see indicator specific documentation

    see :mod:`safe.defaults` for default values information
    """

    def __init__(self):
        """
        Constructor for postprocessor class,
        It takes care of defining self.impact_classes
        """
        AbstractPostprocessor.__init__(self)
        self.impact_classes = None
        self.impact_attrs = None
        self.target_field = None

[docs]    def description(self):
        """Describe briefly what the post processor does.
        """
        return tr('Calculates generic categorical statistics.')

[docs]    def setup(self, params):
        """Initialise needed parameters.
        """
        AbstractPostprocessor.setup(self, None)
        if (
                self.impact_classes is not None or
                self.impact_attrs is not None or
                self.target_field is not None):
            self._raise_error('clear needs to be called before setup')

        self.impact_classes = params['impact_classes']
        self.impact_attrs = params['impact_attrs']
        self.target_field = params['target_field']
        self._log_message(self.impact_attrs)

[docs]    def process(self):
        """Performs all the indicators calculations.
        """
        AbstractPostprocessor.process(self)
        if (
                self.impact_classes is None or
                self.impact_attrs is None or
                self.target_field is None):
            self._log_message(
                '%s not all params have been correctly '
                'initialized, setup needs to be called before '
                'process. Skipping this postprocessor'
                % self.__class__.__name__)
        else:
            self._calculate_categories()

[docs]    def clear(self):
        """Clear properly parameters.
        """
        AbstractPostprocessor.clear(self)
        self.impact_classes = None
        self.impact_attrs = None
        self.target_field = None

    def _calculate_categories(self):
        """Indicator that shows total population.
        """

        impact_name = tr(self.target_field).lower()

        results = {}
        for impact_class in self.impact_classes:
            results[impact_class] = 0

        for feature in self.impact_attrs:
            myTarget = feature[self.target_field]
            results[myTarget] += 1

        for impact_class in self.impact_classes:
            result = results[impact_class]
            self._append_result('%s %s' % (impact_name, impact_class), result)