Source code for node_conditional_error

# This file is part of Sympathy for Data.
# Copyright (c) 2016, Combine Control Systems AB
#
# SYMPATHY FOR DATA COMMERCIAL LICENSE
# You should have received a link to the License with Sympathy for Data.
from sympathy.api import node as synode
from sympathy.api.nodeconfig import Port, Ports, Tag, Tags
from sympathy.api.exceptions import SyNodeError, sywarn
from sylib import util


ERROR_LEVEL = ['Error', 'Warning', 'Output']




class ConditionalError(synode.Node):
    """
    Raise an error if the supplied predicate function returns True.

    See https://docs.python.org/3/tutorial/controlflow.html#lambda-expressions
    for a description of lambda functions. Have a look at the :ref:`Data type
    APIs<datatypeapis>` to see what methods and attributes are available on the
    data type that you are working with.
    """

    name = 'Conditional error/warning'
    author = 'Magnus Sandén'
    icon = ''
    description = "Raise an error if a predicate is True."
    nodeid = 'org.sysess.sympathy.data.table.conditionalerror'
    icon = 'error.svg'
    tags = Tags(Tag.Development.Test)

    inputs = Ports([Port.Custom('<a>', 'Input', name='in')])
    outputs = Ports([Port.Custom('<a>', 'Output', name='out')])

    parameters = synode.parameters()
    parameters.set_string(
        'predicate',
        label='Predicate function:',
        value='lambda arg: True',
        description='Error message is printed if this function returns True.',
        editor=synode.editors.code_editor())
    parameters.set_string(
        'error_msg',
        label='Error message:',
        value='Error!',
        description='Error message to display to the user.')
    parameters.set_list(
        'error_type',
        label='Severity:',
        list=ERROR_LEVEL,
        value=[0],
        description='The level "Error" stops flow execution.',
        editor=synode.editors.combo_editor())

    def execute(self, node_context):
        predicate_str = node_context.parameters['predicate'].value
        error_type = node_context.parameters['error_type'].selected
        error_msg = node_context.parameters['error_msg'].value
        in_table = node_context.input['in']
        out_table = node_context.output['out']

        # Evaluate predicate function
        if util.base_eval(predicate_str)(in_table):
            if error_type == 'Error':
                raise SyNodeError(error_msg)
            elif error_type == 'Warning':
                sywarn(error_msg)
            elif error_type == 'Output':
                print(error_msg)
            else:
                raise ValueError('Unknown error_type')

        # Output is equal to input
        out_table.source(in_table)