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, submitting 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.
TaskStatus: The following statuses are available:

The following statuses are available for ANNOTATE tasks:
- NEW: Tasks that are unassigned.
- ASSIGNED: Tasks that are assigned to a user.
- RELEASED: Tasks that were released from an assigned user.
- REOPENED: Tasks that were reopened.
- SKIPPED: Tasks that were skipped by annotators.
- COMPLETED: Tasks that are in the Complete stage. Used for Annotation sub-tasks in Consensus Annotation tasks.
ℹ️
Note
Consensus Annotation tasks consist of 1 or more Annotation sub-tasks.

The following statuses are available for REVIEW and CONSENSUS REVIEW tasks:
- NEW: Tasks that are unassigned.
- ASSIGNED: Tasks that are assigned to a user.
- RELEASED: Tasks that were released from an assigned user.
- REOPENED: Tasks that were reopened.

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 Stages

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 Stages

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 Stages

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}")

Filter tasks by status

Filter tasks by status.

TaskStatus:

The following statuses are available for ANNOTATE tasks:
- NEW: Tasks that are unassigned.
- ASSIGNED: Tasks that are assigned to a user.
- RELEASED: Tasks that were released from an assigned user.
- REOPENED: Tasks that were reopened.
- SKIPPED: Tasks that were skipped by annotators.
- COMPLETED: Tasks that are in the Complete stage. Used for Annotation sub-tasks in Consensus Annotation tasks.
ℹ️
Note
Consensus Annotation tasks consist of 1 or more Annotation sub-tasks.

The following statuses are available for REVIEW and CONSENSUS REVIEW tasks:
- NEW: Tasks that are unassigned.
- ASSIGNED: Tasks that are assigned to a user.
- RELEASED: Tasks that were released from an assigned user.
- REOPENED: Tasks that were reopened.

# Import dependencies
from encord import EncordUserClient, Project

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

SSH_PATH = "<ssh-private-key>"
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="<stage name>", type_=<stage-type>)
# 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, <stage-type>)

for task in stage.get_tasks(status: "<type-of-status>"):
    
    print(f"Task: {task}")

# 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 = "f441ca60-c3d3-477d-be94-898aab6131ef"

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 like follows
assert isinstance(stage, AnnotationStage)

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

Filter tasks by data hash

You can filter tasks in any stage using data hashes.

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage

# Constants
SSH_PATH = "<ssh-private-key>"
PROJECT_HASH = "<project-unique-id>"

# Create user client
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)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# Uncomment the following line for different stage
# stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)

# Check instance type. Change the type for each stage
assert isinstance(stage, ReviewStage)

# List of data hashes.
data_hashes = ["<data-unique-id-01>", "<data-unique-id-02>", "<data-unique-id-03>"]

for task in stage.get_tasks(data_hash=data_hashes):
    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).

❗️
CRITICAL INFORMATION
Tasks are automatically assigned to the person executing the task.submit().


# 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.submit()
    
    print(f"Task: {task}")

Reopen Review 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)

# Get tasks from the REVIEW stage
review_stage = project.workflow.get_stage(name="<stage name>", type_=ReviewStage)
task = next(review_stage.get_tasks(data_title="<file name>")) 

# Now that we have the tasks we can show the information about label reviews for each task
for r in task.get_label_reviews():
    print(f"Label Type: {r.label_type}, LabelID: {r.label_id}, status: {r.status}")

If those tasks have reviews we can approve them:

for r in task.get_label_reviews():
    r.approve()

We can then check in the UI that the reviews are approved.

We can the reopen reviews using:

for r in task.get_label_reviews():
    r.reopen()

After that check in the UI to verify that the reviews are back to normal.

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 Stages

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()

End-to-End Example moving tasks through Projects

This example uses a basic non-Consensus Project Workflow moving a number of tasks from ANNOTATE to COMPLETE.

End-to-end Example Workflow

End-to-end Example tasks

Project ID: 01bd7084-b04a-4cd1-87f3-73e8e78925c4
Data units: apples_01.jpg, apples_02.jpg, apples_03.jpg, apples_04.jpg, apples_05.jpg

Step 1: View all tasks in the Project

A few tasks have been annotated, but they have not been submitted yet. The following code lists all tasks in the ANNOTATE 1 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 = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

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 like follows
assert isinstance(stage, AnnotationStage)

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

Task: uuid=UUID('387dbbe6-4abb-47f8-9efe-199d1d4652ce') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 51, 51, 139841) data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg' label_branch_name='main' assignee=None
Task: uuid=UUID('80e09443-f649-4624-9cff-222c04115d16') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 53, 58, 858513) data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg' label_branch_name='main' assignee='[email protected]'
Task: uuid=UUID('d5fa67aa-9490-4248-ad0d-287b5dc23189') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 52, 15, 119658) data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg' label_branch_name='main' assignee='[email protected]'
Task: uuid=UUID('e53132b4-fd13-4673-bf65-41fe8e690125') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 55, 9, 87875) data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg' label_branch_name='main' assignee='[email protected]'
Task: uuid=UUID('e793424c-1d2f-4f32-aa18-920ada4b2a7a') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 52, 48, 962443) data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg' label_branch_name='main' assignee='[email protected]'

From the returned results we can see that four of the data units have been assigned (apples_01 to apples_04) while one task has not (apples_05). We do not know the extent that each of the assigned data units has been annotated, but we want the tasks to move through the Workflow.

Step 2: Submit all tasks for review

The following code submits ALL tasks to the REVIEW 1 stage. Tasks are automatically assigned to the user submitting the task.

# 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 = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

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.submit()
    
    print(f"Task: {task}")

Step 3: Verify that the tasks are in REVIEW 1

Each time we perform an action, we should verify the action is successful. After submitting the tasks, we can verify that the tasks are now in REVIEW 1.

# 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 = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

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="Review 1", type_=ReviewStage)
# 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, ReviewStage)

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

Task: uuid=UUID('643dc436-3c90-478b-a7f6-45cd70d0fc19') created_at=datetime.datetime(2024, 6, 30, 22, 46, 57, 229922) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 28, 115530) assignee=None data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg'
Task: uuid=UUID('02ace181-87dd-4628-8b17-7f9f16015c43') created_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 636022) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 709843) assignee=None data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg'
Task: uuid=UUID('6aae1e0b-8297-4efa-9274-e1355633010d') created_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 488760) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 565142) assignee=None data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg'
Task: uuid=UUID('e9623d55-b166-4a7e-bf8e-4e3b635263db') created_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 32894) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 149193) assignee=None data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg'
Task: uuid=UUID('4ff5801d-ad1a-4304-8270-55be654d681a') created_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 422) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 86008) assignee=None data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg'

All the tasks are now in stage REVIEW 1. None of the tasks are assigned to any users.

Step 4: Reject a task

We know that apples-05.jpg does not have any labels, because it was not assigned to any users before we submitted the task. If we REJECT the task, according to the Workflow in the Project, the task returns to the ANNOTATE 1 stage.

The following code filters the task in the REVIEW 1 stage based on the task's data hash, and then rejects the task.

The user rejecting the task is automatically assigned to the task, when the task is resubmitted for review.

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage
from uuid import UUID  # Import the UUID function

# Constants
SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

# Create user client
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)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# Uncomment the following line for different stage

# Check instance type
assert isinstance(stage, ReviewStage)

data_hashes = ["34571f50-90d1-4681-a58b-1fd08e9f6665"]

for task in stage.get_tasks(data_hash=data_hashes):

    task.reject()

    print(f"Task: {task}")

Task: uuid=UUID('02ace181-87dd-4628-8b17-7f9f16015c43') created_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 636022) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 709843) assignee=None data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg'

From what was returned, we can see that only data unit apples_05.jpg was acted upon.

To verify that the task is now in ANNOTATE 1, run the following:

# 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 = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

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 like follows
assert isinstance(stage, AnnotationStage)

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

Task: uuid=UUID('387dbbe6-4abb-47f8-9efe-199d1d4652ce') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 23, 48, 3, 805733) data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg' label_branch_name='main' assignee='[email protected]'

Step 5: Approve tasks

There are still four tasks in REVIEW 1. The following code approves the tasks. This moves the tasks to the COMPLETE stage. You could approve all the tasks at once the REVIEW 1 stage, but instead we will filter by data hash and then approve.

The user approving the task is automatically assigned to the task.

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage
from uuid import UUID  # Import the UUID function

# Constants
SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

# Create user client
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)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# Uncomment the following line for different stage

# Check instance type
assert isinstance(stage, ReviewStage)

data_hashes = [
    "36b0afd3-6ff8-4201-bf65-35a37f6fe5b5",
    "4dc183ba-4233-4cb1-a940-cb6b2bc38f35",
    "22f8dc9a-deff-49a2-a2d2-3863cf20d4f4",
    "809344fd-8935-4dde-81eb-36efb4e6d558"
    ]

for task in stage.get_tasks(data_hash=data_hashes):

    task.reject()

    print(f"Task: {task}")

Task: uuid=UUID('643dc436-3c90-478b-a7f6-45cd70d0fc19') created_at=datetime.datetime(2024, 6, 30, 22, 46, 57, 229922) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 28, 115530) assignee=None data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg'
Task: uuid=UUID('6aae1e0b-8297-4efa-9274-e1355633010d') created_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 488760) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 565142) assignee=None data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg'
Task: uuid=UUID('e9623d55-b166-4a7e-bf8e-4e3b635263db') created_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 32894) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 149193) assignee=None data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg'
Task: uuid=UUID('4ff5801d-ad1a-4304-8270-55be654d681a') created_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 422) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 86008) assignee=None data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg'

Step 6: Verify tasks are COMPLETE

Use the following code to verify that the tasks apples01 to apples_04 are in the COMPLETE 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 = "09765dc5-08b0-4b32-b37a-e1b932b8b684"

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="Complete", type_=FinalStage)
# 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, FinalStage)

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

Task: uuid=UUID('e27acb09-0300-4bc9-8143-d0b837fd9207') created_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 448867) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 495702) data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg'
Task: uuid=UUID('23351c59-32a6-42b9-8e06-fe20c41513bc') created_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 67363) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 110112) data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg'
Task: uuid=UUID('9e8fce04-af0e-41ca-926a-56bf354a2cba') created_at=datetime.datetime(2024, 7, 1, 0, 18, 21, 168387) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 21, 223495) data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg'
Task: uuid=UUID('41d98c6d-2f4c-4ad1-8f91-576b74e56600') created_at=datetime.datetime(2024, 7, 1, 0, 18, 20, 516722) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 20, 561088) data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg'

And from what the code returns, the tasks are now in the COMPLETE stage.

Workflow Basics

ℹ️Note

View all stages in a Workflow

ℹ️Note

Non-Consensus Workflow Stages

Display Tasks in a Stage

Assign a User to Tasks in a Stage

Get all Tasks

Filter tasks by status

ℹ️Note

Filter tasks by data hash

Move all tasks in a stage to the next stage

❗️CRITICAL INFORMATION

Reopen Review Tasks

Use model predictions as an annotator

Consensus Workflow Stages

Get all Subtasks

Select your labels from a Consensus Project

End-to-End Example moving tasks through Projects

Step 1: View all tasks in the Project

Step 2: Submit all tasks for review

Step 3: Verify that the tasks are in REVIEW 1

Step 4: Reject a task

Step 5: Approve tasks

Step 6: Verify tasks are COMPLETE

ℹ️
Note

ℹ️
Note

ℹ️
Note

❗️
CRITICAL INFORMATION