EEG-Workflow-System

Blocks Documentation

About blocks and workflow

A block is a method that takes input and produces output when executed. Each block has editable parameters that have some default value.

A workflow is a group of blocks connected to each other where the output of one block is the input for the other. Workflows do not necessarily mean a single flow of execution, it could have multiple parallel flows.

Consider a simple workflow using the Arithmetic module:

• As you can see the output of the constant block is connected to the input of the addition block.
• Each constant block has an editable parameter deciding the value of the constant block, as seen in the below image.

On executing the above workflow, the result produced by the workflow is shown here.

To define a block we need to specify the inputs, outputs, properties, and the execution function.

The execution function may use the inputs and the properties to perform some operation and produce a single visualization and multiple outputs variables. The outputs can then be forwarded as input to another block.

Create New Custom Blocks

Before we learn how to create a new custom block, lets first see predefined classes:

class Block(object):

    family = "Default Family"
    name = "Default Name"
    description = ""

    FILE_BASE_PATH = '{}/.EEGWorkflow/Shared'.format(os.path.expanduser('~'))

    def __init__(self):
        pass

    def input_params(self,data):
        pass

    def execute(self):
        pass

class BlockInput:

    def __init__(self,name:str, min_cardinality:int, max_cardinality:int, attribute_type:str):
        if not (isinstance(name,str) and isinstance(min_cardinality,int) and isinstance(max_cardinality,int) \
            and isinstance(attribute_type,str)):
            raise TypeError("Incorrect parameter type")
        self.name = name
        self.value = None
        self.min_cardinality = min_cardinality
        self.max_cardinality = max_cardinality
        self.attribute_type = attribute_type

    def set_value(self,value):
        self.value = value

class BlockOutput:

    def __init__(self, name:str, attribute_type:str, min_cardinality:int, max_cardinality:int):
        if not (isinstance(name,str) and isinstance(min_cardinality,int) and isinstance(max_cardinality,int) \
            and isinstance(attribute_type,str)):
            raise TypeError("Incorrect parameter type")
        self.name = name
        self.value = None
        self.max_cardinality = max_cardinality
        self.min_cardinality = min_cardinality
        self.attribute_type = attribute_type

    def set_value(self,value):
        self.value = value

class BlockParameter:

    def __init__(self, name:str, attribute_type:str, defaultvalue=None, description=''):
        if not (isinstance(name,str) and isinstance(attribute_type,str)):
            raise TypeError("Incorrect parameter type")
        self.name = name
        self.value = defaultvalue
        self.attribute_type = attribute_type
        self.description = str(description)

    def set_value(self,value):
        self.value = value

class ParameterType:

    NUMBER = "NUMBER"
    NUMBER_ARRAY = "NUMBER[]"
    STRING = "STRING"
    STRING_ARRAY = "STRING[]"
    FILE = "FILE"
    FILE_LOCATION = "FILE_LOCATION"
    FILE_ARRAY = "FILE[]"
    BOOLEAN = "BOOLEAN"
    OBJECT = "OBJECT"

    EEGDATA = "EEGData"
    EPOCHS = "EPOCHS"
    FEATUREVECTOR = "FeatureVector"
    MODEL = "Model"

To define a block in python, as shown in the last image, we need to follow the design pattern

• Every block must be a python class which should inherit the Block class (shown above)
• The new block class that you create will override input_params and execute method of Block class
• Input, Output and editable Parameter variables of the custom block should be an instance of BlockInput, BlockOutput and BlockParameter class respectively
• Create the input, output and editable parameter instance variables in the __init__ method of our custom class
• Assign the value to the above variables (except output variables) in input_params method
• The input_params method takes data as the only parameter. data is a dictionary of string-value pair, where the key is the variable name (string) and value is the value to be assigned to that variable
• Override the execute method in which assign the value to the output variables
• Lastly, override the Block class variables family and name to family the block belongs to and the name of the block

In order to understand it better, let’s take the example of the Arithmetic Module.
The Arithmetic module contains 5 blocks:

• Constant Block
• Addition Block
• Subtraction Block
• Multiplication Block
• Division Block

Let’s see the Addition block class:

import time

from blocks.Block import Block 
from blocks.BlockInput import BlockInput
from blocks.BlockParameter import BlockParameter
from blocks.BlockOutput import BlockOutput
from blocks.ParameterType import ParameterType

class Addition(Block):

    family = 'AddSub'
    name = 'Addition'

    def __init__(self):
        self.num1 = BlockInput(
            name='num1',
            min_cardinality=1,
            max_cardinality=1,
            attribute_type=ParameterType.NUMBER
        )
        self.num2 = BlockInput(
            name='num2',
            min_cardinality=1,
            max_cardinality=1,
            attribute_type=ParameterType.NUMBER
        )
        self.output = BlockOutput(
            name='output',
            min_cardinality=1,
            max_cardinality=1,
            attribute_type=ParameterType.NUMBER
        )
        

    def input_params(self,data):
        self.num1.set_value(data['num1'])
        self.num2.set_value(data['num2'])

    def execute(self):
        value = self.num1.value + self.num2.value
        self.output.set_value(value)
        return (value,'STRING')

As you can see we have defined all the variables in the __init__ method and assigned all the variables (except output variables) in input_params and output variables in execute method.

Note:
While uploading custom module, path to import predefined Block, BlockInput, BlockOutput, BlockParameter and ParameterType class should be:

from blocks.Block import Block 
from blocks.BlockInput import BlockInput
from blocks.BlockParameter import BlockParameter
from blocks.BlockOutput import BlockOutput
from blocks.ParameterType import ParameterType

Now lets see Contant block class:

from blocks.Block import Block 
from blocks.BlockInput import BlockInput
from blocks.BlockParameter import BlockParameter
from blocks.BlockOutput import BlockOutput
from blocks.ParameterType import ParameterType

class Constant(Block):

    family = 'Constant'
    name = 'Constant'

    def __init__(self):
        self.num = BlockParameter(
            name='constant value',
            attribute_type=ParameterType.NUMBER,
            defaultvalue=10
            # defaultvalue=''
        )
        self.output = BlockOutput(
            name='output',
            min_cardinality=1,
            max_cardinality=1,
            attribute_type=ParameterType.NUMBER
        )

    def input_params(self,data):
        self.num.set_value(int(data['constant value']))

    def execute(self):
        value = self.num.value
        self.output.set_value(value)
        return (value,'STRING')

As you can see it is similar to the Addition block. Note that BlockParameter needs to have a default value.

Note: Value for each BlockParameter instance variable comes as a string from the frontend, hence it always needs to be type-casted. As you can see in the input_params method of Constant block, we type-casted the num variable to int which is a block parameter.

Each variable input, output, or parameter has attribute_type. The arrtibute_type is a string used by the frontend to prevent connections where types do not match. For example, if block1 needs an input of list of integers and block2 outputs just an integer, then connecting the output of block2 to the input of block1 should not be allowed. Hence atrribute_type helps in avoiding type mismatch.

Apart from input, output, and parameter, there is something called stdout.
stdout is used to display some output on the block. In the above image, you can see that each block is displaying some output. Constant block displays the constant value, Multiplication block displays the output after multiplying the two values and Plot block outputs a link containing the image of the plotted graph.
These outputs are called the stdout. This stdout is determined from the return value of the execute method. In order to set the stdout as a STRING or GRAPH you need to return a tuple in execute method. The tuple must be (value, type) format. It should be one of the two:

- ("[string you want to display]", "STRING")
- (None, "GRAPH")
- ("<Keras model Object>","Model")

• Note: Nothing will be displayed if you return anything other than the above two formats.
• For type=”STRING”, check the return format for the Constant block shown above.
• The value for type=”GRAPH” is None because your block should only use matplotlib to plot a graph, and in the execute method, you need to plot the graph (plt.plot(...)). Below you can see the example of the Plot block. Check the return type and logic of the execute method.

from blocks.Block import Block 
from blocks.BlockInput import BlockInput
from blocks.BlockParameter import BlockParameter
from blocks.BlockOutput import BlockOutput
from blocks.ParameterType import ParameterType

import numpy as np
import matplotlib.pyplot as plt

class Plot(Block):

    family = 'Graph'
    name = 'Plot'

    def __init__(self):
        self.function = BlockParameter(
            name='function equation',
            attribute_type=ParameterType.STRING,
            defaultvalue='x'
        )
        self.graph_domain = BlockParameter(
            name='range',
            attribute_type=ParameterType.NUMBER,
            defaultvalue=100
        )

    def input_params(self,data):
        self.function.set_value(data['function equation'])
        self.graph_domain.set_value(int(data['range']))

    def execute(self):
        x = np.array(range(-self.graph_domain.value,self.graph_domain.value))
        y = eval(self.function.value)
        plt.figure()
        plt.plot(x,y)
        plt.xlabel('x-axis')
        plt.ylabel('y-axis')
        plt.title('Graph Equation: {}'.format(self.function.value))
        plt.grid()
        return (None,'GRAPH')
        

Create Module

The Group of all the custom blocks forms a module. Arithmetic module consists of 5 blocks as mentioned above.

Note: For a module, always follow a proper directory structure of the family. Taking Arithmetic module as an example, the structure followed by it is shown below:

├── Arithmetic
   ├── AddSub
   │   ├── __init__.py
   │   ├── Addition.py
   │   └── Subraction.py
   ├── MultiDiv
   │   ├── __init__.py
   │   ├── Multiplication.py
   │   └── Division.py
   ├── Constant
   │   ├── __init__.py
   │   └── Constant.py
   ├── Graph
   │   ├── __init__.py
   │   └── Plot.py
   └── __init__.py

Lastly (VERY IMPORTANT):

• The outermost __init__.py file must contain a dictionary variable called string_classobject_mapping. The string_classobject_mapping is dict with key as CustomBlockclass.name and value as the instance of your custom block class, CustomBlockclass()

For Arithmetic module, the outermost __init__.py file contains:

from .AddSub.Addition import Addition
from .AddSub.Subraction import Subraction
from .Constant.Constant import Constant
from .MultiDiv.Multiplication import Multiplication
from .MultiDiv.Division import Division

string_classobject_mapping = {
    Addition.name: Addition(),
    Subraction.name: Subraction(),
    Multiplication.name: Multiplication(),
    Division.name: Division(),
    Constant.name: Constant()
}

Check Arithmetic Module and EEG_Basil Module for better understanding.

This site is open source. Improve this page.