What is a task ?

A task is a fundamental object used in any Brick. It is one of the main object of the workflow that you can find in the playground.

A task is a special function that works with Constellab object. It contains a run method to execute any code. It takes resources as inputs and resources as outputs. It also takes a configuration object.

For example, the object TableTransposer is a task that takes a Table as input, transpose it rand return the transposed table as output.

Tasks can be connected to each other to be executed, this is what we called a protocol.

How to create a task ?

To create a new task in your brick, you need to create a new class that:

• Extend the Task class (import from gws_core)
• Is decorated with the @task_decorator (import from gws_core).
• To see how to define the name, description and style of the task with the decorator, view this page : https://constellab.community/bricks/gws_core/latest/doc/developer-guide/type/decorator
• Define the input_specs, output_specs and config_specs

Here is an example of our RobotMove task. This task takes a Robot resource as input, update its positions based on the configuration and return the Robot.

from gws_core import (ConfigParams, FloatParam, InputSpec, OutputSpec,
                      StrParam, Task, TaskInputs, TaskOutputs, task_decorator)
from .robot_resource import Robot  # local import
@task_decorator("RobotMove", human_name="Move robot",
                short_description="This task emulates a short moving step of the robot", hide=True)
class RobotMove(Task):
    input_specs = {'robot': InputSpec(Robot, human_name="Robot",  short_description="The robot to move")}
    output_specs = {'robot': OutputSpec(Robot)}
    config_specs = {
                   'moving_step': FloatParam(default_value=0.1, short_description="The moving step of the robot"),
                   'direction': StrParam( default_value="north", allowed_values=["north", "south", "east", "west"], short_description="The                   moving direction")
     }
    def run(self, params: ConfigParams, inputs: TaskInputs) -> TaskOutputs:
        print(f"Moving {params.get_value('moving_step')}", flush=True)
        robot: Robot = inputs['robot']
        robot.move(direction=params.get_value('direction'), moving_step=params.get_value('moving_step'))
        return {'robot': robot}

Run method

The run method is the method executed when the system runs the task. You must right your code logic for this task in this method.

The run method has 2 parameters

• params of type ConfigParams : this is a dict object that contains the values of the configuration provided by the user. See Configuration for more information.
• inputs of type TaskInputs : this is a dict object that contains all the resources provided as input of the task. See Input & Outputs for more information.

The run method must return a TaskOutputs which is a dictionary of resources.

The task will be executed in a protocol, so the outputs of one task can be connected to the one or more inputs of the next task. With this you can pass resources that a task generated to another task. 

Inputs and Outputs

When we create a task we need to define what are the task IO (inputs and outputs) and the type of each IO. A task takes only Resource as IO so the type provided in the IO must be Resource. When the task is run the inputs are provided in the run method.

To define the specifications of the task IO we must define the input_specs and output_specs.

Input

The input_specs object is a dictionary of InputSpec. The keys of the dictionary are important because it helps you retrieve the input in the run method. In the above example we retrieve the Robot resource by calling inputs['robot'] robot Is the key define in the input_specs.

The InputSpec object help you to define your inputs. It takes multiple parameters:

• resource_types: a resource type or a List of resource types. If 1 type is provided (recommended), the system only accepts this type and sub-classes for this input. If multiple types are provided, the system will accept any of this classes and sub-classes. Before running the task, the system checks the inputs and if 1 input is not compatible the task is not run.
• human_name: pretty name that will be displayed in the playground.
• short_description: small description to add information about this input (also for the playground).
• is_optional: this input might not be connected to another task output and the task will still be executed. If the input is connected, the system will wait for the input to be provided before running the task. Also tells that None value is allowed as input.  

Outputs

The output_specs object is a dictionary of OutputSpec. The keys of the dictionary are important because it must match the keys of the returned object in the run method. For example in the RobotMove task, we returned : return {'robot': robot} which means we return the resource robot with the key robot. The robot key is the same as defined in the output_specs of the task.

The OutputSpec object help you to define your outputs. It takes multiple parameters (some like InputSpec):

• resource_types: a resource type or a List of resource types. If 1 type is provided (recommended), it tells the system that this task returns a specific type(and sub-classes) for this output. If multiple it tells that the task will return one of the provided type (and sub-classes) for this output. 
• human_name: pretty name that will be displayed in the playground.
• short_description: small description to add information about this output (also for the playground).
• is_optional: tell that this output may return None or not being provided.  
• is_constant (for expert): When true, it tells the system that the output resource was not modified from the input resource. The system will not need to create a new resource after the task.
• sub_class (for expert): When true, it tells that the resource_types are compatible with any child class of the provided resource type (useful for task that work with generic resource type).

Configuration

The config_specs attribute allows you to expose configuration to the playground. You can define multiple parameters with different type and this will generate a form available in the playground when the user uses the task.

The config_specs object is a dictionary of Param. The keys are important because it helps you retrieve the parameters value from the params object (first parameter of the run method). In the above example we retrieve the two parameters’ values direction and moving_step by calling params.get_value('direction') moving_step=params.get_value('moving_step'). The params object is a python dict with additional methods.

There are different parameter types (int, float, bool, str..) but they all have common options when we define a new parameter:

• human_name: pretty name that will be displayed in the playground when configuring the task.

• short_description: small description to add information about this parameter (also for the playground).

• default_value: default value to use when the user did not provide a value for this parameter. If a default value is provided, it automatically set the optional to True. If not provided and the optional is set to False, the parameter is mandatory. If you want to set a default value to None, set optional to True.

• optional: define the parameters as optional. If default_value is not provided and the user did not define this parameter, the value will be None
• visibility
• public : main param visible in the first config section in the playground
• protected : considered as advanced param, it will be in the advanced section in the playground. It must have a default value or be optional.
• private (for expert) : it will not be visible in the playground. It must have a default value or be optional)

• allowed_values: list of allowed value for the parameter. It will be rendered as a select in the playground unit: define the measure unit nit of this parameter

Basic Params

Here is the list of all the basic param types with additional options for some of them:

• StrParam: string param. It will show a basic text input in the playground. Use allowed value to restrict possible values min_length: minimal length for the provided text max_length: maximal length for the provided text
• IntParam & FloatParam : numeric param. It will show an input with only number possible.
• min_value: minimum value for the param
• max_value: maximum value for the param
• BoolParam: Boolean param. It will show a checkbox.
• ListParam: List param. Show a text input in the playground where the user can provide multiple values. The value for this param will be a list of strings. It also works with allowed_value, to allow user to select multiple values from a list of choices.
• DictParam (for expert) : only to use for hidden parameters (it has no representation in the playground).
• TagsParams : Show a special input to select multiple tags. The value of the parameter will be a list of tags.

ParamSet

Use to define a group of parameters that can be added multiple times. The ParamSet contains a dictionary of parameters. Use this when there are 2 or more configurations that are related and you want the user to be able to provided multiple combinations of theses parameters.

When not optional, the user will need to provide at least one value (a combination) in the playground.

Use the max_number_of_occurence attribute to limit the number of combinations the user can provide.

The value of this parameter will be a list of dictionaires.

Task Special methods

Log messages

There are multiple method that exist to write logs during the task. The logs are available in the playground and are update in real time. The logs are really usefull when the task takes some time and it is also possible to log progress.

Here are the log methods:

• update_progress_value: log a progress message and update the progress value (value from 0 to 100). Ex: update_progress_value(10, 'Reached 10%')
• log_info_message: log an information message
• log_success_message: log a success message
• log_warning_message: log a warning message
• log_error_message: log an error message
• log_message: log a message with the type in parameter

You can call theses methods in the run methods like this : self.log_info_message('Hello world').

Check before run

The check_before_run method can be overwritten to perform custom check before running task. It must return a CheckBeforeTaskResult that contains 2 objects :

• result: boolean
• If True, everything is ok
• If False, the task will not be executed after this check it might be run later if they are some SKippableIn inputs. If all the input values were provided and the check retuns False. the task will endup in error because it won't be run
• message: Optional[str]
• If False, a message can be provided to log the error message if the task will not be called

Run after task

The run_after_task method can be overwritten to perform action after the task run. This method is called after the resource save. This method is useful to delete temporary objects (like files) to clear the server after the task is run.

Create temp directory

The create_tmp_dir method can be called inside the task to create a temporary directory that will be automatically deleted after the task is run (on the run_after_task method). Output files and folders of the task are moved out of this directory before it is deleted. 

Test the task

There are 2 ways of testing the task and we recommend to do both of them:

• Write unit test
• Test in dev environment

Write unit test

Writing a unit test in the best and easiest way to test your task and make sure it still works after some changes (you change it or update a dependency for example).

Create a new test file in the tests folder. The name of the file must start with test_.

Here is a example on how to test the RobotMove task.

from gws_core import TaskRunner
from gws_core.impl.robot.robot_resource import \
    Robot  # local import of the resource to test
from gws_core.impl.robot.robot_tasks import \
    RobotMove  # local import of the task to test
from unittest import TestCase

class TestRobotMove(TestCase):
    def test_robot_move(self):
        # create an empty robot with position [0,0]
        robot_output = Robot.empty()
        
        # create a task runner and configure it, then run it
        runner = TaskRunner(
            task_type=RobotMove,
            params={'moving_step': 3, 'direction': 'south'},
            inputs={'robot': robot_output},
        )
        outputs = await runner.run()
        # retrieve the robot output
        robot_output: Robot = outputs['robot']
        # check the robot position, y should be -3 as he went south
        self.assertEqual(robot_output.position, [0, -3])

Let's break down this example.

Firstly we declared the class TestRobotMove that extend the TestCase. The test class must extend the TestCase to be considered as a test file by VsCode.

We manually create our Robot resource.

The TaskRunner is the main part of this test. This class allows you to run a task manually with custom config and input and retrieve the outputs. Then we create the task runner by passing the task type RobotMove, the configuration we want to test, and the input (the previously created robot).

Then we run the task by calling the runner.run() , this method will return a TaskOutputs. We can then retrieve the robot from the outputs and check that the value of the robot is correct. Here we check that the position of the robot was changed.

To learn how to run the test and debug it, see the following link.

Dev environment > debug-tests

You can also test the check_before_run method with the task runner. It may also be necessary to call the run_after_task at the end of your test if your task generated side data that need to be cleared.

Test in dev environment

Once you unit tests are ok, you can test your task in the dev environment. See the following link:

Dev environment > start-dev-server

Document the task

Once your task is developed and tested, we recommend to document it. Here is a list of things you can do to improve the documentation of your task:

• Define the human_name and short_description in the @task_decorator
• Define the human_name and short_description in the input_specs, output_specs and config_specs
• Write comment in the python code of your classe.

You can write complete documentation using python comment. It supports the markdown syntax: (https://www.markdownguide.org/basic-syntax/). This documentation is really important and it will be available in Constellab community and in the playground. Here is an example on how to do it.

class RobotMove(Task):
    """
    This is a documentation that support **markdown** syntax. 
    And this documentation will be available in 
    - the hub 
    - the playground.
    """

Deploy your task

Once you created your test and write your documentation, you can deploy your task in a new version of your brick. To do so, check the following link:  Deploy a brick version

Use external data

If your task rely on one or more external data (like a database) that needs to be downloaded first you can use the TaskFileDownloader. This class let you download external file resource (using http protocol) and store it in a specific folder for your brick.

To create the TaskFileDownloder from your task simple do the following :

# create the file_downloader from a task.
file_downloader = TaskFileDownloader(MyTaskClass.get_brick_name(), self.message_dispatcher)
# download a file
file_path = file_downloader.download_file_if_missing("my_file_url", "file_name.json")

The external downloaded data can be viewed directly in the lab in the Monitoring > Other > Brick data.

In this page you can delete manually a data so it will be force to be re-downloaded next time the task is executed.