Source code for safe.postprocessors.abstract_postprocessor
# -*- coding: utf-8 -*-
"""**Abstract postprocessor class, do not instantiate directly**
"""
__author__ = 'Marco Bernasocchi <marco@opengis.ch>'
__revision__ = '$Format:%H$'
__date__ = '10/10/2012'
__license__ = "GPL"
__copyright__ = 'Copyright 2012, Australia Indonesia Facility for '
__copyright__ += 'Disaster Reduction'
import logging
from safe.common.exceptions import PostProcessorError
from safe.common.utilities import get_defaults
from safe.impact_functions.core import format_int
from third_party.odict import OrderedDict
LOGGER = logging.getLogger('InaSAFE')
[docs]class AbstractPostprocessor():
"""
Abstract postprocessor class, do not instantiate directly.
but instantiate the PostprocessorFactory class which will take care of
setting up many prostprocessors. Alternatively you can as well instantiate
directly a sub class of AbstractPostprocessor.
Each subclass has to overload the process method and call its parent
like this: AbstractPostprocessor.process(self)
if a postprocessor needs parmeters, then it should override the setup and
clear methods as well and call respectively
AbstractPostprocessor.setup(self) and AbstractPostprocessor.clear(self).
for implementation examples see AgePostprocessor which uses mandatory and
optional parameters
"""
NO_DATA_TEXT = get_defaults('NO_DATA')
def __init__(self):
"""
Constructor for abstract postprocessor class, do not instantiate
directly. It takes care of defining self._results
Needs to be called from the concrete implementation with
AbstractPostprocessor.__init__(self)
"""
self._results = None
[docs] def setup(self, params):
"""Abstract method to be called from the concrete implementation
with AbstractPostprocessor.setup(self, None) it takes care of results
being initialized
Args:
params: dict of parameters to pass to the post processor
Returns:
None
Raises:
None
"""
del params
if self._results is not None:
self._raise_error('clear needs to be called before setup')
self._results = OrderedDict()
[docs] def process(self):
"""Abstract method to be called from the concrete implementation
with AbstractPostprocessor.process(self) it takes care of results
being initialized
Args:
None
Returns:
None
Raises:
None
"""
if self._results is None:
self._raise_error('setup needs to be called before process')
[docs] def clear(self):
"""Abstract method to be called from the concrete implementation
with AbstractPostprocessor.process(self) it takes care of results
being cleared
Args:
None
Returns:
None
Raises:
None
"""
self._results = None
[docs] def results(self):
"""Returns the postprocessors results
Args:
None
Returns:
Odict of results
Raises:
None
"""
return self._results
def _raise_error(self, message=None):
"""internal method to be used by the postprocessors to raise an error
Args:
None
Returns:
None
Raises:
PostProcessorError
"""
if message is None:
message = 'Postprocessor error'
raise PostProcessorError(message)
def _log_message(self, message):
"""internal method to be used by the postprocessors to log a message
Args:
None
Returns:
None
Raises:
None
"""
LOGGER.debug(message)
def _append_result(self, name, result, metadata=None):
"""add an indicator results to the postprocessors result.
internal method to be used by the postprocessors to add an indicator
results to the postprocessors result
Args:
* name: str the name of the indicator
* result the value calculated by the indicator
* metadata Dict of metadata
Returns:
None
Raises:
None
"""
if metadata is None:
metadata = dict()
LOGGER.debug('name : ' + str(name) + '\nresult : ' + str(result))
if result is not None and result != self.NO_DATA_TEXT:
try:
result = format_int(result)
except ValueError as e:
LOGGER.debug(e)
result = result
self._results[name] = {'value': result,
'metadata': metadata}