Workflows and Stages

Workflow Basics

Encord SDK supports programmatic use of Workflows with the following: WorkflowStage, WorkflowAction, and Task. Typical use would be querying a WorkflowStage to get Tasks matching some criteria, and then executing a WorkflowAction against Tasks (for example: assigning, submiting annotation tasks, or accepting review tasks). The type of actions available depends on the stage queried.

Here is a short description of major parts of the Workflows SDK:

  • WorkflowStage: Move your tasks through a Workflow Project using the WorkflowStage class. However, tasks CANNOT "teleport" through your Workflow. Tasks must move through the Workflow in logical order.
    These are the stages currently supported:

    • AnnotationStage
    • ReviewStage
    • ConsensusAnnotationStage
    • ConsensusReviewStage
    • FinalStage (Complete and Archive)
  • WorkflowAction: Applies an action to a task in a workflow stage

    The following actions are supported:

    • assign: Assigns a user to one or more tasks.
    • submit: Moves a task to the next stage.
    • release: Releases a task from the current user.
    • accept: Accepts a task.
    • reject: Rejects a task.
  • Task: Exposes a simple version of the tasks available in a Project.

View all stages in a Workflow

The following code provides you with the method to view all stages in a Workflow in a Project. The code is the same for Consensus and Non-Consensus Projects.

ℹ️

Note

Routers are not displayed in the list of stages when using the SDK.


# Import dependencies
from encord import EncordUserClient, Project


SSH_PATH = "<file-path-to-private-key-file>"
PROJECT_HASH = "<unique-id-for-project>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Print all Workflow stages
for idx, stage in enumerate(project.workflow.stages):
    print(f"Stage #{idx}: {stage}")


# Import dependencies
from encord import EncordUserClient, Project


SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "2c2a74d0-d33d-43b2-b5dd-1c6c3e6d7b6d"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Print all Workflow stages
for idx, stage in enumerate(project.workflow.stages):
    print(f"Stage #{idx}: {stage}")

Non-Consensus Workflow Project

This is an example of a non-Consensus Workflow graph and the result returned after running the code to view all stages in a Workflow.


Stage #0: AnnotationStage(stage_type=WorkflowStageType.ANNOTATION, uuid='2ed1d9f4-5b42-4959-bcbe-93eaf866320c', title='Annotate 1')
Stage #1: ReviewStage(stage_type=WorkflowStageType.REVIEW, uuid='ddb7973d-bae9-4ced-8dcb-ec178e66dba0', title='Review 1')
Stage #2: AnnotationStage(stage_type=WorkflowStageType.ANNOTATION, uuid='58ec51ba-2669-4036-b25c-17aecf805a59', title='Annotate 2')
Stage #3: ReviewStage(stage_type=WorkflowStageType.REVIEW, uuid='8047e0d5-fc41-4ac5-a7ed-335b3dbd887f', title='Review 2')
Stage #4: FinalStage(stage_type=WorkflowStageType.DONE, uuid='90fc8b51-6a2f-4de0-88d0-059f76865385', title='Complete')
Stage #5: FinalStage(stage_type=WorkflowStageType.DONE, uuid='1432dbf0-5c19-41fe-a7b0-811d36b27456', title='Archive')

Consensus Workflow Project

This is an example of a Consensus Workflow graph and the result returned after running the code to view all stages in a Workflow.


Stage #0: ConsensusAnnotationStage(stage_type=WorkflowStageType.CONSENSUS_ANNOTATION, uuid='2adfe595-4a56-41f2-bfcc-eeab3dd58b65', title='Consensus 1')
Stage #1: ConsensusReviewStage(stage_type=WorkflowStageType.CONSENSUS_REVIEW, uuid='1f37fb4c-a74a-4437-b33d-6fc4e3ccc2f5', title='Consensus 1 Review')
Stage #2: ReviewStage(stage_type=WorkflowStageType.REVIEW, uuid='9fc321da-0e62-4cc5-b5e7-5ce8f73c3c20', title='Review 1')
Stage #3: ConsensusAnnotationStage(stage_type=WorkflowStageType.CONSENSUS_ANNOTATION, uuid='518cd79f-a9a9-4ab9-bb92-3dcf0fae29a2', title='Consensus 2')
Stage #4: ConsensusReviewStage(stage_type=WorkflowStageType.CONSENSUS_REVIEW, uuid='81e0cb65-69ae-4d3b-aac8-8917f1480930', title='Consensus 2 Review')
Stage #5: FinalStage(stage_type=WorkflowStageType.DONE, uuid='fd5c3513-c7a5-42c3-a07c-933838aecf62', title='Complete')
Stage #6: FinalStage(stage_type=WorkflowStageType.DONE, uuid='c67d08c5-b93b-49be-87b2-39809ea7b27a', title='Archive')

Non-Consensus Workflow Projects

In the SDK, a class is defined for every stage type. This allows you to see what properties and actions each stage offers. The following stages are supported for non-Consensus Projects:

  • AnnotationStage
  • ReviewStage
  • FinalStage (Complete and Archive)

class AnnotationStage(WorkflowStage):
	uuid: str
	title: str
  
  def get_tasks()
    

Tasks provides programmatic access to your tasks available in the UI.
The following tasks are supported for non-Consensus Projects::

  • assign: Assigns a user to one or more tasks.
  • submit: Moves a task to the next stage.
  • release: Releases a task from the current user.
class AnnotationTask:
    uuid: UUID
    data_hash: UUID
    data_title: str
    label_branch: str  # main by default

Display Tasks in a Stage


# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert as follows
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Assign a User to Tasks in a Stage


# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter. Any type of narrowing technique can be used here.
# For example:
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():

    task.assign("<[email protected]>")
    
    print(f"Task: {task}")

Get all Tasks

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Consensus 1", type_=ConsensusAnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, ConsensusAnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Move all tasks in a stage to the next stage

Before tasks can be moved along in the Workflow, they must be assigned to one user (for example, your admin).


# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter. Any type of narrowing technique can be used here.
# For example:
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():

    task.assign("[email protected]")

    task.submit()
    
    print(f"Task: {task}")

Use model predictions as an annotator


# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Automated Annotation", type_=AnnotationStage)
# type_ is an optional parameter. Any type of narrowing technique can be used here.
# For example:
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():
    # All the filters and ordering in the UI is supported in the SDK.
    # You can filter by filename, dataset, and so on.
    task.assign("[email protected]")
    
    print(f"Task: {task}")
    if match_some_criteria(task)
        for label_row in project.list_label_rows_v2(data_hashes=[task.data_hash]):
            label_row.initialise_labels()
        # Custom logic that populates label rows with labels
        do_preannotate(label_row)
        label_row.save()

    task.submit()

Consensus Workflow Projects

Consensus annotation tasks can be listed, but cannot be assigned or moved into the Consensus review stage.

Consensus Workflow Projects support the following stages:

  • AnnotationStage
  • ReviewStage
  • ConsensusAnnotationStage
  • ConsensusReviewStage
  • FinalStage (Complete and Archive)

The following actions are supported for Consensus review and Review and Refine:

  • assign: Assigns a user to one or more tasks.
  • approve: Moves a task to the next stage.
  • release: Releases a task from the current user.
  • reject: Rejects a task.

Get all Subtasks

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Consensus 1", type_=ConsensusAnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, ConsensusAnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")
    for subtask in task.subtasks:
        print(f"   subtask: {subtask}")

Select your labels from a Consensus Project

The following code includes produce_result, that is undefined. You need to create and define the custom logic that selects the labels from the label options you have available.


# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Obtaining the stage
stage = project.workflow.get_stage(name="Consensus Review 1", type_=ConsensusReviewStage)

# Get all the tasks (or rather iterator over tasks), 
# Do NOT download them all at once.
for task in stage.get_tasks():
    # Each consensus branch has a unique label hash, so use them as filter
    label_hashes = [option.label_hash for option in task.options]
    label_rows_for_consensus = project.list_label_rows_v2(label_hashes=label_hashes)

    # Download labels
    for label_row in label_rows_for_consensus:
        label_row.initialise_labels()

        # Get and initialise the result label row (which is "main")
        target_label_row = project.list_label_rows_v2(data_hashes=[task.data_hash])[0]
        target_label_row.initialise_labels()
	
        # Perform custom user logic and populate the target label row.
        # This means defining produce_result in your custom logic as produce_result
        # is not defined in the Encord SDK.
        if is_consensus := produce_result(label_rows_for_consensus, target_label_row):
            target_label_row.save()
            task.assign("<[email protected]>")
            task.approve()
        else:
            task.reject()