Workflow#

class ikomia.dataprocess.workflow.Workflow(name='Untitled', registry=IkomiaRegistry())#

Workflow management of Computer Vision tasks. Implement features to create, modify and run graph-based pipeline of CWorkflowTask objects or derived. Workflows can be created from scratch by using IkomiaRegistry to instanciate and connect task objects. Workflows can also be loaded from JSON file created with the interactive designer of Ikomia Studio. Derived from CWorkflow.

Import

from ikomia.dataprocess.workflow import Workflow

Methods

__init__([name, registry])

Construct Workflow object with the given name and an IkomiaRegistry object.

add_task([task, name, params, auto_connect, ...])

Add task identified by its unique name in the workflow.

connect_tasks(src, target[, edges])

Connect two tasks of the workflow.

find_task(name[, index])

Get identifiers and instance of tasks with the given name in the workflow.

get_task_id(task)

Get task unique identifier from the task instance/

get_task_output([task_obj, task_name, ...])

Get specific output(s) defined by their types (IODataType) for the given task.

get_tasks()

Get all tasks composing the workflow.

get_time_metrics()

Get metrics around workflow execution time.

load(self, path)

Load a workflow previously saved as JSON file.

remove_task([task, name, index])

Remove task from workflow specified by task instance or name (with corresponding index).

root()

Get workflow root node.

run()

Start workflow execution on global input.

run_on([array, path, url, folder])

Convenient function to run the workflow on common inputs.

set_directory_input([folder, index])

Set folder as global input of the workflow.

set_image_input([array, path, url, index, ...])

Set image as global input of the workflow.

set_video_input([path, url, index, datatype])

Set video as global input of the workflow.

set_parameters(params[, task_obj, ...])

Set task parameters as a simple key-value dict.

Inherited methods

add_input(self, input)

Add global input to the workflow.

clear(self)

Remove all tasks and connections from the workflow.

clear_inputs(self)

Remove all inputs of the workflow.

clear_output_data(self)

Clear output data for all the tasks of the workflow.

delete_edge(self, id)

Remove the given connection/edge from the workflow.

delete_task(self, id)

Remove the given task from the workflow.

export_graphviz(self, path)

Export the workflow structure as Graphviz .dot file.

get_children(self, id)

Get child task identifiers of the task specified by the given identifier.

get_edge_info(self, id)

Get connection information (edge) from its unique identifier.

get_edge_source(self, id)

Get source/parent task identifier of a given edge.

get_edge_target(self, id)

Get target/child task identifier of a given edge.

get_elapsed_time_to(self, task_id)

Get the workflow running time in milliseconds from the start to the given task.

get_final_tasks(self)

Get all final or leaf tasks of the workflow.

get_in_edges(self, id)

Get input connections (in-edges) of the task specified by the given identifier.

get_out_edges(self, id)

Get output connections (out-edges) of the task specified by the given identifier.

get_parents(self, id)

Get parent task identifiers of the task specified by the given identifier.

get_required_tasks(self, path)

Get task names required to load and execute the given workflow file.

get_root_id(self)

Get unique identifier of the root node.

get_root_target_types(self)

Get all input data types of the tasks connected to the root node.

get_task(self, id)

Get the task object from the given ID.

get_task_count(self)

Get the number of tasks in the workflow.

get_task_ids(self)

Get the list of all task identifiers.

get_total_elapsed_time(self)

Get the total workflow running time in milliseconds

load(self, path)

Load a workflow previously saved as JSON file.

remove_input(self, index)

Remove global workflow input at the given index.

set_auto_save(self, enable)

Activate/deactivate auto-save mode.

set_input(self, input, index, new_sequence)

Set workflow input at position index.

set_output_folder(self, path)

Set workflow output folder.

save(self, path)

Save the workflow as a JSON file.

stop(self)

Stop workflow execution.Each CWorkflowTask object or derived must reimplement the stop() function that will be called by the workflow.

update_start_time(self)

Reset the starting point for the computation of the running time.

Details

__init__(name='Untitled', registry=IkomiaRegistry())#

Construct Workflow object with the given name and an IkomiaRegistry object. The latter is used to instanciate algorithm from their unique name when added to the workflow. Thus, you are able to use any Ikomia algorithms (built-in and Ikomia HUB) in your workflow.

add_input((CWorkflow)self, (CWorkflowTaskIO)input) None :#

Add global input to the workflow.

Parameters:

input (CWorkflowTaskIO based object) – input object

add_task(task=None, name: str = '', params: dict = {}, auto_connect: bool = False, public_hub: bool = True, private_hub=False)#

Add task identified by its unique name in the workflow. If the given task is not yet in the registry, it will be firstly downloaded and installed from Ikomia HUB. Task unique identifier can then be retrieved with get_task_id().

Parameters:
  • name (str) – algorithm unique name

  • param (CWorkflowTaskParam based object) – algorithm parameters

Returns:

task instance

Return type:

CWorkflowTask based object

clear((CWorkflow)self) None :#

Remove all tasks and connections from the workflow. The workflow is thus empty after that.

clear_inputs((CWorkflow)self) None :#

Remove all inputs of the workflow.

clear_output_data((CWorkflow)self) None :#

Clear output data for all the tasks of the workflow.

connect_tasks(src, target, edges=None)#

Connect two tasks of the workflow. Depending of the inputs/outputs configuration, multiple connections between the two tasks can be set. A connection is a pair (ie tuple) composed by the output index of the source task and the input index of the target task.

Parameters:
  • src (CWorkflowTask based object) – source task or None (connect to root)

  • target (CWorkflowTask based object) – target task

  • edges (list of pair) – connections. If None is passed, auto-connection is enabled so that the system will try to find the best connections automatically with respect to inputs and outputs data types.

delete_edge((CWorkflow)self, (int)id) None :#

Remove the given connection/edge from the workflow. The identifier becomes invalid after this operation.

Parameters:

id (int) – edge identifier

delete_task((CWorkflow)self, (int)id) None :#

Remove the given task from the workflow. The identifier becomes invalid after this operation.

Parameters:

id (int) – task identifier

export_graphviz((CWorkflow)self, (str)path) None :#

Export the workflow structure as Graphviz .dot file. You can then visualize it with the dot command or with Graphviz Python package.

Parameters:

path (str) – path where the .dot file is saved

find_task(name: str, index=-1)#

Get identifiers and instance of tasks with the given name in the workflow.

Parameters:
  • name (str) – algorithm name. Multiple candidates may exist, so use index parameter to specify one.

  • index (int) – zero-based index of the wanted task. If -1, the function returns all candidates.

Returns:

if index == -1 CWorkflowTask based object : if index != -1

Return type:

list of CWorkflowTask based objects

get_children((CWorkflow)self, (int)id) object :#

Get child task identifiers of the task specified by the given identifier. Task connected to the outputs of a given task is designated as child or target.

Parameters:

id (int) – task identifier on which to get childs

Returns:

child identifiers

Return type:

int list

get_edge_info((CWorkflow)self, (int)id) tuple :#

Get connection information (edge) from its unique identifier.

Parameters:

id (int) – edge identifier

Returns:

pair composed by the output index of the source/parent task and the input index of the target/child task

Return type:

tuple (int)

get_edge_source((CWorkflow)self, (int)id) int :#

Get source/parent task identifier of a given edge.

Parameters:

id (int) – edge identifier

Returns:

task identifier

Return type:

id (int)

get_edge_target((CWorkflow)self, (int)id) int :#

Get target/child task identifier of a given edge.

Parameters:

id (int) – edge identifier

Returns:

task identifier

Return type:

id (int)

get_elapsed_time_to((CWorkflow)self, (int)task_id) float :#

Get the workflow running time in milliseconds from the start to the given task.

Parameters:

id (int) – task identifier

Returns:

elapsed time

Return type:

float

get_final_tasks((CWorkflow)self) object :#

Get all final or leaf tasks of the workflow.

Returns:

leaf task identifiers

Return type:

int list

get_in_edges((CWorkflow)self, (int)id) object :#

Get input connections (in-edges) of the task specified by the given identifier. Edge information can then be retrieved by the function get_edge_info().

Parameters:

id (int) – task identifier

Returns:

edge identifiers

Return type:

int list

get_out_edges((CWorkflow)self, (int)id) object :#

Get output connections (out-edges) of the task specified by the given identifier. Edge information can then be retrieved by the function get_edge_info().

Parameters:

id (int) – task identifier

Returns:

edge identifiers

Return type:

int list

get_parents((CWorkflow)self, (int)id) object :#

Get parent task identifiers of the task specified by the given identifier. Task connected to the inputs of a given task is designated as parent or source.

Parameters:

id (int) – task identifier on which to get parents

Returns:

parent identifiers

Return type:

int list

get_required_tasks((CWorkflow)self, (str)path) object :#

Get task names required to load and execute the given workflow file.

Parameters:

path (str) – path to the workflow file (JSON)

Returns:

task names

Return type:

list of str

get_root_id((CWorkflow)self) int :#

Get unique identifier of the root node.

Returns:

root node ID

Return type:

int

get_root_target_types((CWorkflow)self) object :#

Get all input data types of the tasks connected to the root node.

Returns:

data types

Return type:

list of IODataType

get_task((CWorkflow)self, (int)id) CWorkflowTask :#

Get the task object from the given ID. Unique task identifiers can be retrieved with the functions get_task_ids() or find_task().

Returns:

CWorkflowTask object or derived

get_task_id(task)#

Get task unique identifier from the task instance/

Parameters:

task (CWorkflowTask based object) – task instance

Returns:

task unique identifier

Return type:

int

get_task_count((CWorkflow)self) int :#

Get the number of tasks in the workflow.

Returns:

task count

Return type:

int

get_task_ids((CWorkflow)self) object :#

Get the list of all task identifiers. You can then retrieve task object from ID with the function get_task().

get_task_output(task_obj=None, task_name='', task_index=0, types=[ikomia.core.pycore.IODataType.IMAGE], output_index=-1)#

Get specific output(s) defined by their types (IODataType) for the given task.

Parameters:
  • task_obj (task_obj (CWorkflowTask based object) – task instance. See also add_task() and find_task().

  • task_name (str) – algorithm name to be found. Multiple candidates may exist, so use task_index parameter to specify one. Method find_task() is used to retrieve corresponding task(s).

  • task_index (int) – zero-based index of the wanted task. If -1, the function returns all candidates outputs.

  • types (list of IODataType) – output data types.

  • output_index (int) – zero-based index of he wanted output.

Returns:

task output of the given data types (can be a list).

Return type:

CWorkflowTaskIO based object

get_tasks()#

Get all tasks composing the workflow.

Returns:

list of CWorkflowTask based object

get_time_metrics()#

Get metrics around workflow execution time. This includes the total execution time of the workflow, and for each task, the execution time and the execution time from the start.

Returns:

metrics

Return type:

dict

get_total_elapsed_time((CWorkflow)self) float :#

Get the total workflow running time in milliseconds

Returns:

elapsed time

Return type:

float

load((CWorkflow)self, (str)path) None :#

Load a workflow previously saved as JSON file. Common usage is to use Ikomia Studio to create your workflow interactively and without code. With this method, it is very easy and fast to build your pipeline, visualize outputs on test data and save the workflow. But you can also create your workflow from scratch with this API and use the dedicated function save().

Parameters:

path (str) – path of the JSON file

remove_input((CWorkflow)self, (int)index) None :#

Remove global workflow input at the given index.

Parameters:

index (int) – zero-based index of the input to remove

remove_task(task=None, name: str = '', index: int = 0)#

Remove task from workflow specified by task instance or name (with corresponding index).

Parameters:
  • task (CWorkflowTask) – task object instance

  • name (str) – algorithm name to be found. Multiple candidates may exist, so use task_index parameter to specify one. Method find_task() is used to retrieve corresponding task(s).

  • index (int) – zero-based index of the wanted task.

root()#

Get workflow root node.

run()#

Start workflow execution on global input. Each CWorkflowTask object or derived must reimplement the run() function that will be called in the right order by the workflow. Please note that global inputs should be set before calling this function (see set_input(), set_image_input(), set_directory_input()).

run_on(array=None, path='', url='', folder='')#

Convenient function to run the workflow on common inputs. For more advanced use, please consult Workflow. See also set_image_input() and set_directory_input()

Parameters:
  • array (Numpy array) – image or generic array as numpy array

  • path (str) – path to image (valid formats are those managed by OpenCV)

  • url (str) – URL to image file (valid formats are those managed by OpenCV)

  • folder (str) – image folder

set_auto_save((CWorkflow)self, (bool)enable) None :#

Activate/deactivate auto-save mode. If activated, outputs of each tasks of the workflow will be saved in the workflow output folder (see set_output_folder() to set your custom folder). By default, outputs are saved in user_folder/Ikomia/Workflows.

Parameters:

enable (bool) – True or False

set_directory_input(folder='', index=-1)#

Set folder as global input of the workflow. For image-based workflows, all images inside the directory (recursively) will be processed.

Parameters:
  • folder (str) – images folder

  • index (int) – zero-based input index, if -1 a new input is added

set_image_input(array=None, path='', url='', index=-1, datatype=ikomia.core.pycore.IODataType.IMAGE)#

Set image as global input of the workflow. Image can be specified by a Numpy array, a path or an URL thanks to keyword arguments, you have to choose one of them.

Parameters:
  • array – image input as Numpy array

  • path (str) – image input as file path (valid formats are those managed by OpenCV)

  • url (str) – valid URL to image file (valid formats are those managed by OpenCV)

  • index (int) – zero-based input index, if -1 a new input is added

  • datatype (IODataType) – image type

set_video_input(path='', url='', index=-1, datatype=ikomia.core.pycore.IODataType.VIDEO)#

Set video as global input of the workflow. Video can be specified by a path or an URL thanks to keyword arguments, you have to choose one of them.

Parameters:
  • path (str) – image input as file path (valid formats are those managed by OpenCV)

  • url (str) – valid URL to image file (valid formats are those managed by OpenCV)

  • index (int) – zero-based input index, if -1 a new input is added

  • datatype (IODataType) – image type

set_input((CWorkflow)self, (CWorkflowTaskIO)input, (int)index, (bool)new_sequence) None :#

Set workflow input at position index. If index is greater than the input count, the function adds the right number of inputs automatically. Derived class that handles common data type already exists: CImageIO, CVideoIO, CNumericIO, CGraphicsInput, CDatasetIO, CPathIO, CArrayIO, CBlobMeasureIO.

Parameters:
  • input (CWorkflowTaskIO object or derived) – global input of the workflow

  • index (int) – zero-based index

  • new_sequence (bool) – True if it is a new input sequence, False if it is just a new frame of a video or camera stream

set_output_folder((CWorkflow)self, (str)path) None :#

Set workflow output folder. If auto-save mode is activated (see set_auto_save()) outputs of each tasks of the workflow will be saved automatically in this folder. Behind the scene, each task implements a save() function that calls sequentially the save() function of all these outputs.

Parameters:

path (str) – path to the desired directory

set_parameters(params: dict, task_obj=None, task_name='', index=-1)#

Set task parameters as a simple key-value dict. You can get parameters keys for each by calling:

print(task_obj.get_param_object())
Parameters:
  • params (dict) – key-value pairs of parameters to modify.

  • task_obj (CWorkflowTask based object) – task instance. See also add_task() and find_task().

  • task_name (str) – algorithm name to be found. Multiple candidates may exist, so use task_index parameter to specify one. Method find_task() is used to retrieve corresponding task(s).

  • index (int) – zero-based index of the wanted task. If -1, the function modifies all candidates parameters.

save((CWorkflow)self, (str)path) None :#

Save the workflow as a JSON file.

Parameters:

path (str) – path where the JSON is saved

stop((CWorkflow)self) None :#

Stop workflow execution.Each CWorkflowTask object or derived must reimplement the stop() function that will be called by the workflow. Depending on the process implementation, stop may not be instantaneous.

update_start_time((CWorkflow)self) None :#

Reset the starting point for the computation of the running time. This feature could be interesting if you want to process a list of images and monitor the time per image.