TaskRunner

This is used to run a Task. It can be used in unit test to test a Task.

Simply provide the task type, config params and inputs. then you can call the run method to test your task and check outputs

:raises MissingInputResourcesException: [description] :raises Exception: [description] :return: [description] :rtype: [type]

Functions

__init__

Initialize self. See help(type(self)) for accurate signature.

task_type : type

params : dict[str, typing.Any] | None

inputs : dict[str, gws_core.resource.resource.Resource] | None

config_model_id : str | None

input_specs : gws_core.io.io_specs.InputSpecs | None

output_specs : gws_core.io.io_specs.OutputSpecs | None

config_specs : gws_core.config.config_specs.ConfigSpecs | None

task_id : str | None

scenario_id : str | None

message_dispatcher : gws_core.core.classes.observer.message_dispatcher.MessageDispatcher | None

add_log_observer

Method to create an observer and attached it to the task. The log will be available in the retuned BasicMessageObserver. This can be useful for testings

Return type : BasicMessageObserver

add_observer

Method to create an observer and attached it to the task. The log will be available in the retuned BasicMessageObserver. This can be useful for testings

observer : MessageObserver

check_before_run

This method check the config and inputs and then execute the check before run of the task

Return type : CheckBeforeTaskResult

force_dispatch_waiting_messages

get_output

output_name : str

Return type : Resource

get_outputs

Return type : dict

get_task

Return type : gws_core.task.task.Task | None

run

This method, checks the config, inputs and then run the task

Return type : dict

run_after_task

set_input

input_name : str

resource : Resource

set_progress_bar

progress_bar : ProgressBar

Have you developed a brick?