Workflows

Workflows coordinate multiple threads working together.

WorkflowTemplate

Import from orca.sdk.workflow:

from orca.sdk.workflow import WorkflowTemplate

workflow = WorkflowTemplate(name="my_assay")
workflow.add_thread(plate_thread, is_start=True)

Methods

Method	Description
add_thread(thread, is_start=False)	Add a thread. Set is_start=True for threads that begin when the workflow starts.
set_spawn_point(spawn_thread, from_thread, at, join=False)	Configure thread spawning and synchronization
add_event_handler(event_name, handler)	Subscribe to workflow events

Properties

• name - Workflow name
• thread_templates - All threads in the workflow
• entry_thread_templates - Threads that start when the workflow starts
• spawns - Configured spawn points
• event_hooks - Registered event handlers

Adding Threads

Add threads using add_thread(). Mark threads that start immediately with is_start=True:

from orca.sdk.workflow import WorkflowTemplate

workflow = WorkflowTemplate(name="my_assay")
workflow.add_thread(plate_thread, is_start=True)  # Starts when workflow starts
workflow.add_thread(dest_thread)  # Will be spawned later

Spawn Points

Spawn points control when threads begin. A spawned thread starts when its parent reaches a specific method.

workflow.set_spawn_point(
    spawn_thread=dest_thread,
    from_thread=source_thread,
    at=transfer_method,
    join=False
)

Parameters

Parameter	Description
spawn_thread	Thread to spawn
from_thread	Parent thread that triggers the spawn
at	Method where spawn occurs
join	Whether threads must synchronize (default: False)

Without Join (Parallel Execution)

When join=False, the spawned thread runs independently:

source_thread: ─[method1]─[transfer_method]─[method2]─▶
                              │ spawn
dest_thread:                  └──[method_a]─[method_b]─▶

With Join (Synchronized Execution)

When join=True, both threads must arrive before the method executes. The spawned thread uses SharedMethodTemplate() as a placeholder:

from orca.sdk.workflow import SharedMethodTemplate

# Source has the actual method
source_thread = ThreadTemplate(
    labware_template=source_plate,
    start=start_loc,
    end=end_loc,
    methods=[transfer_method]
)

# Destination uses SharedMethodTemplate as placeholder
dest_thread = ThreadTemplate(
    labware_template=dest_plate,
    start=start_loc,
    end=end_loc,
    methods=[SharedMethodTemplate()]
)

workflow.set_spawn_point(
    spawn_thread=dest_thread,
    from_thread=source_thread,
    at=transfer_method,
    join=True  # Both must arrive
)

source_thread: ─[prep_source]─────────────┐
                                          ├─[transfer_method]─▶
dest_thread:   ─[prep_dest]───────────────┘
                              (both arrive before transfer)

When join=True, the SharedMethodTemplate placeholder in dest_thread is replaced with the actual transfer_method, allowing both threads to participate in the same operation with their own labware.

Multi-Plate Operations

For operations requiring multiple plates (liquid transfers, cherrypicking):

from orca.sdk.workflow import MethodTemplate, SharedMethodTemplate, ThreadTemplate
from orca.sdk.actions import RunProtocol

# Transfer needs both source and destination
transfer_method = MethodTemplate(
    name="transfer",
    actions=[
        RunProtocol(
            resource=liquid_handler,
            protocol_filepath="protocols/transfer.hsl",
            parameters={"volume": 100},
            inputs=[source_plate, dest_plate],
            outputs=[source_plate, dest_plate]
        )
    ]
)

# Source thread owns the method
source_thread = ThreadTemplate(
    labware_template=source_plate,
    start=source_input,
    end=source_output,
    methods=[transfer_method]
)

# Destination thread participates via SharedMethodTemplate
dest_thread = ThreadTemplate(
    labware_template=dest_plate,
    start=dest_input,
    end=dest_output,
    methods=[SharedMethodTemplate()]
)

# Configure the workflow
workflow = WorkflowTemplate("transfer_workflow")
workflow.add_thread(source_thread, is_start=True)
workflow.add_thread(dest_thread)
workflow.set_spawn_point(
    spawn_thread=dest_thread,
    from_thread=source_thread,
    at=transfer_method,
    join=True
)

Event Handlers

Subscribe to workflow events:

def on_method_complete(event, context):
    print(f"Method completed: {context.method_name}")

workflow.add_event_handler("METHOD.COMPLETED", on_method_complete)

See Events for available events and handler signatures.

Next Steps

• System Building - Connect workflows to devices
• Running Workflows - Execute workflows
• Events - React to workflow events