Source code for node_find_maxima

# This file is part of Sympathy for Data.
# Copyright (c) 2018, Combine Control Systems AB
#
# SYMPATHY FOR DATA COMMERCIAL LICENSE
# You should have received a link to the License with Sympathy for Data.

import numpy as np
import scipy.signal as signal

from sympathy.api import node
from sympathy.api.exceptions import SyConfigurationError
from sympathy.api.nodeconfig import Port, Ports, Tag, Tags, adjust




class FindLocalMaxima(node.Node):
    name = 'Find local maxima/minima'
    author = 'Mathias Broxvall'
    icon = 'maxima.svg'
    description = (
        'Finds local maxima, minima. A point is a maximum(minimum) at scale S '
        'if it is larger(smaller) than all other points within radius S.')
    nodeid = 'com.sympathyfordata.timeseriesanalysis.find_maxima'
    tags = Tags(Tag.Analysis.Features)

    parameters = node.parameters()
    parameters.set_string(
        'method',
        label='Type',
        value='maxima',
        description='Selects type of extreme points to detect',
        editor=node.editors.combo_editor(
            options=['maxima', 'minima']))
    parameters.set_integer(
        'scale',
        label='Scale level',
        value=1,
        description='Scale level at which to search for maxima/minima')
    parameters.set_list(
        'column',
        label='Select column',
        description='Select column used for finding local extreme points',
        value=[],
        editor=node.editors.combo_editor())

    inputs = Ports([
        Port.Table('input', name='input')
    ])
    outputs = Ports([
        Port.Table('indices', name='indices')
    ])

    def adjust_parameters(self, node_context):
        adjust(node_context.parameters['column'],
               node_context.input['input'])

    def execute(self, node_context):
        in_tbl = node_context.input['input']
        out_tbl = node_context.output['indices']
        scale = node_context.parameters['scale'].value
        method = node_context.parameters['method'].value
        col_name = node_context.parameters['column'].selected

        if not col_name:
            raise SyConfigurationError(
                'You need to configure which column the maxima/minima '
                'should be identified for')

        if method == 'maxima':
            cmp = np.greater
        elif method == 'minima':
            cmp = np.less
        else:
            raise SyConfigurationError('Invalid method {} selected'
                                       .format(method))

        col = in_tbl.col(col_name)
        name = col.name
        indices = signal.argrelextrema(col.data, cmp, order=scale)[0]
        out_tbl.set_column_from_array(name, indices)





class FindLocalPlateau(node.Node):
    name = 'Find Plateaus'
    author = 'Mathias Broxvall'
    icon = 'plateaux.svg'
    description = (
        'Finds plateaus in input data and returns starting and stopping '
        'indices for them')
    nodeid = 'com.sympathyfordata.timeseriesanalysis.find_plateaus'
    tags = Tags(Tag.Analysis.Features)

    parameters = node.parameters()
    parameters.set_integer(
        'scale',
        label='Minimum size',
        value=1,
        description='Minimum size of plateau in samples')
    parameters.set_float(
        'epsilon',
        label='Epsilon',
        value=1e-3,
        description='Maximum absolute acceptable derivation within a plateau')
    parameters.set_list(
        'column',
        label='Select column',
        description='Select column used for finding plateaus.',
        value=[],
        editor=node.editors.combo_editor())

    inputs = Ports([
        Port.Table('input', name='input')
    ])
    outputs = Ports([
        Port.Table('indices', name='indices')
    ])

    def adjust_parameters(self, node_context):
        adjust(node_context.parameters['column'],
               node_context.input['input'])

    def execute(self, node_context):
        in_tbl = node_context.input['input']
        out_tbl = node_context.output['indices']
        scale = node_context.parameters['scale'].value
        col_name = node_context.parameters['column'].selected
        epsilon = node_context.parameters['epsilon'].value

        if not col_name:
            raise SyConfigurationError(
                'You need to configure which column the plateaux should be '
                'identified for')

        col = in_tbl.col(col_name)

        diffs = col.data[1:] - col.data[:-1]
        is_flat = np.r_[(np.abs(diffs) < epsilon), False]
        prev = 0
        start = []
        stop = []
        for idx in range(len(col.data)):
            if not is_flat[idx]:
                if idx - prev >= scale:
                    start.append(prev)
                    stop.append(idx)
                    prev = idx+1
                else:
                    prev = idx+1

        out_tbl.set_column_from_array('start', np.array(start))
        out_tbl.set_column_from_array('stop', np.array(stop))