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.
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.
In Consensus Projects, tasks in the COMPLETE and ARCHIVE blocks CANNOT be reopened.
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.
Routers are not displayed in the list of stages when using the SDK.
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:
Use the following scripts as templates for annotation and review tasks.
Display Stage Tasks
# Import dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, AnnotationStage)for task in stage.get_tasks():print(f"Task: {task}")
Annotation Tasks can only be submitted by users with the following roles: Annotator, Annotator + Reviewer, Team Manager, or Admin.
Review Tasks can only be submitted by users with the following roles: Reviewer, Annotator + Reviewer, Team Manager, or Admin.
from encord.user_client import EncordUserClientfrom encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatusfrom encord.workflow.stages.consensus_annotation import ConsensusAnnotationStagefrom encord.workflow.stages.consensus_review import ConsensusReviewStagefrom encord.workflow.stages.review import ReviewStageSSH_PATH ="<file-path-to-private-key>"PROJECT_ID="<project-unique-id>"user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( Path(SSH_PATH).read_text())project = user_client.get_project(PROJECT_ID)workflow = project.workflowannotation_stage = workflow.get_stage(name="<name-of-workflow-stage>", type_=<type-of-workflow-stage>)BUNDLE_SIZE =100# You can adjust this value as needed, but keep it <= 1000with project.create_bundle(bundle_size=BUNDLE_SIZE)as bundle:for annotation_task in annotation_stage.get_tasks():print(annotation_task.data_title)# Retain current assignee and submit annotation_task.submit(retain_assignee=True, bundle=bundle)
# Import dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, ConsensusAnnotationStage)for task in stage.get_tasks():print(f"Task: {task}")
You can filter tasks in any stage using data hashes.
Filter tasks by data hash
# Import dependenciesfrom encord import EncordUserClient, Projectfrom encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage# ConstantsSSH_PATH ="<ssh-private-key>"PROJECT_HASH ="<project-unique-id>"# Create user clientuser_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( ssh_private_key_path=SSH_PATH)# Get projectproject: Project = user_client.get_project(PROJECT_HASH)# Get stagestage = 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 stageassertisinstance(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}")
You can submit tasks as the currently assigned user or as a different user. If you do not specify the user to submit the task, you are assigned as the user submitting the task.
# Import dependenciesfrom encord import EncordUserClient, Projectfrom encord.workflow import AnnotationStageSSH_PATH ="<file-path-to-private-key>"PROJECT_HASH ="<project-unique-id>"# Authenticateuser_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( ssh_private_key_path=SSH_PATH)# Get projectproject: Project = user_client.get_project(PROJECT_HASH)# Get the specific stage (in this case, "Annotate 1")stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)assertisinstance(stage, AnnotationStage)# Define bundle sizeBUNDLE_SIZE =100# You can adjust this value as needed, but keep it <= 1000# Create a bundle and move taskswith project.create_bundle(bundle_size=BUNDLE_SIZE)as bundle:for task in stage.get_tasks():# The task is submitted as the user who is currently assigned to the task task.submit(retain_assignee=True, bundle=bundle)print(f"Task: {task}")print("All tasks have been processed and moved to the next stage.")
# Import dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, ConsensusAnnotationStage)for task in stage.get_tasks():print(f"Task: {task}")for subtask in task.subtasks:print(f" subtask: {subtask}")
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 dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: Project = user_client.get_project(PROJECT_HASH)# Obtaining the stagestage = project.workflow.get_stage(name="Consensus 1 Review", type_=ConsensusReviewStage)BUNDLE_SIZE =100# Define bundle size# 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) target_label_row = project.list_label_rows_v2(data_hashes=[task.data_hash])[0]with project.create_bundle(bundle_size=BUNDLE_SIZE)as label_initialize_bundle:# Bundling all our initialisation operations in one operationfor label_row in label_rows_for_consensus: label_row.initialise_labels(bundle=label_initialize_bundle) target_label_row.initialise_labels(bundle=label_initialize_bundle)# Download labelsfor label_row in label_rows_for_consensus:# 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("<your-admin@example.com>") task.approve()else: task.reject()
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 dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, AnnotationStage)for task in stage.get_tasks():print(f"Task: {task}")
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.
The following code submits ALL tasks to the REVIEW 1 stage. Before the tasks can be submitted they are first assigned to a user in the Project. The user could be any user capable of annotating in the Project (because this is the ANNOTATE stage). For this example, the Admin of the Project (the user with SDK access) is assigned.
Submit all tasks for review
# Import dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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:assertisinstance(stage, AnnotationStage)BUNDLE_SIZE =100with project.create_bundle(bundle_size=BUNDLE_SIZE)as bundle:for task in stage.get_tasks(): task.assign("chris@acme.com") task.submit(bundle=bundle)print(f"Task: {task}")
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 dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, ReviewStage)for task in stage.get_tasks():print(f"Task: {task}")
All of the tasks are now in stage REVIEW 1. None of the tasks are assigned to any users.
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.
# Import dependenciesfrom encord import EncordUserClient, Projectfrom encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStagefrom uuid import UUID # Import the UUID function# ConstantsSSH_PATH ="/Users/chris-encord/sdk-ssh-private-key.txt"PROJECT_HASH ="01bd7084-b04a-4cd1-87f3-73e8e78925c4"# Create user clientuser_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( ssh_private_key_path=SSH_PATH)# Get projectproject: Project = user_client.get_project(PROJECT_HASH)# Get stagestage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)# Uncomment the following line for different stage# Check instance typeassertisinstance(stage, ReviewStage)data_hashes =["34571f50-90d1-4681-a58b-1fd08e9f6665"]BUNDLE_SIZE =100with project.create_bundle(bundle_size=BUNDLE_SIZE)as bundle:for task in stage.get_tasks(data_hash=data_hashes): task.assign("chris@acme.com") task.reject(bundle=bundle)print(f"Task: {task}")
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 dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, AnnotationStage)for task in stage.get_tasks():print(f"Task: {task}")
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.
# Import dependenciesfrom encord import EncordUserClient, Projectfrom encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStagefrom uuid import UUID # Import the UUID function# ConstantsSSH_PATH ="/Users/chris-encord/sdk-ssh-private-key.txt"PROJECT_HASH ="01bd7084-b04a-4cd1-87f3-73e8e78925c4"# Create user clientuser_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( ssh_private_key_path=SSH_PATH)# Get projectproject: Project = user_client.get_project(PROJECT_HASH)# Get stagestage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)# Uncomment the following line for different stage# Check instance typeassertisinstance(stage, ReviewStage)data_hashes =["36b0afd3-6ff8-4201-bf65-35a37f6fe5b5","4dc183ba-4233-4cb1-a940-cb6b2bc38f35","22f8dc9a-deff-49a2-a2d2-3863cf20d4f4","809344fd-8935-4dde-81eb-36efb4e6d558"]BUNDLE_SIZE =100with project.create_bundle(bundle_size=BUNDLE_SIZE)as bundle:for task in stage.get_tasks(data_hash=data_hashes): task.assign("chris@acme.com") task.approve(bundle=bundle)print(f"Task: {task}")
Use the following code to verify that the tasks apples01 to apples_04 are in the COMPLETE stage.
# Import dependenciesfrom encord import EncordUserClient, Projectfrom 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 projectproject: 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 followsassertisinstance(stage, FinalStage)for task in stage.get_tasks():print(f"Task: {task}")
And from what the code returns, the tasks are now in the COMPLETE stage.