Source code for nilmtk.preprocessing.clip

from __future__ import print_function, division
from warnings import warn
from ..node import Node
from ..utils import index_of_column_name

[docs]class Clip(Node):

    """Ensures that no value is below a lower limit or above an upper limit.
    If self.lower and self.upper are None then will use clip settings from
    'device': {'measurements': {'upper_limit' and 'lower_limit'}}.
    """

    # Not very well specified.  Really want to specify that 
    # we need 'lower_limit' and 'upper_limit' to be specified in
    # each measurement...
    requirements = {'device': {'measurements': 'ANY VALUE'}}
    postconditions =  {'preprocessing_applied': {'clip': {}}}

[docs]    def reset(self):
        self.lower = None
        self.upper = None

[docs]    def process(self):
        self.check_requirements()
        metadata = self.upstream.get_metadata()
        measurements = metadata['device']['measurements']
        for chunk in self.upstream.process():
            for measurement in chunk:
                lower, upper = _find_limits(measurement, measurements)
                lower = lower if self.lower is None else self.lower
                upper = upper if self.upper is None else self.upper
                if lower is not None and upper is not None:
                    # We use `chunk.iloc[:,icol]` instead of iterating
                    # through each column so we can to the clipping in place
                    icol = index_of_column_name(chunk, measurement)
                    chunk.iloc[:,icol] = chunk.iloc[:,icol].clip(lower, upper)

            yield chunk

def _find_limits(measurement, measurements):
    """
    Returns
    -------
    lower, upper : numbers
    """
    for m in measurements:
        if ((m.get('physical_quantity'), m.get('type')) == measurement):
            return m.get('lower_limit'), m.get('upper_limit')

    warn('No measurement limits for {}.'.format(measurement), RuntimeWarning)
    return None, None