Running Workflows

Once you've built a system, use executors to run workflows.

WorkflowExecutor

The primary way to run workflows:

from orca.sdk.system import WorkflowExecutor

async def main():
    # Build system first (see System Building)
    system = builder.get_system()

    # Initialize devices
    await system.initialize_all()

    # Create executor and run
    executor = WorkflowExecutor(workflow, system)
    await executor.start()

import asyncio
asyncio.run(main())

Parameters

Parameter	Type	Description
workflow	WorkflowTemplate	Workflow to execute
system	System	Built system from SdkToSystemBuilder

Methods

Method	Description
start(sim=False)	Start workflow execution. Set sim=True for simulation.

Simulation Mode

Run workflows without hardware using simulation mode.

System-Level Simulation

Enable simulation for the entire system:

# Option 1: Pass sim=True to start()
await executor.start(sim=True)

# Option 2: Set on system before running
system.set_simulating(True)
await executor.start()

When sim=True:

• All devices use their simulation drivers
• Operations complete instantly
• No hardware communication occurs

Device-Level Simulation

Simulate specific devices while others run on hardware:

# Configure individual devices
shaker.set_simulating(True)   # Simulates
sealer.set_simulating(False)  # Uses real hardware

# Or during construction
shaker = Shaker("shaker", driver, sim=True)

This is useful for:

• Testing workflows before all hardware is ready
• Debugging specific device interactions
• Running partial hardware tests

StandaloneMethodExecutor

Run a single method without defining a full workflow. This is essential for testing methods in isolation before integrating them into larger workflows.

from orca.sdk.system import StandalonMethodExecutor

executor = StandalonMethodExecutor(
    template=transfer_method,
    labware_start_mapping={
        source_plate: "input_stack",
        dest_plate: "input_stack"
    },
    labware_end_mapping={
        source_plate: "output_stack",
        dest_plate: "output_stack"
    },
    system=system
)

await executor.start(sim=True)

Why Use StandaloneMethodExecutor?

1. Unit testing methods - Test individual methods before combining them into workflows
2. Incremental development - Build and validate methods one at a time, then assemble them
3. Faster iteration - Debug a single method without running an entire workflow
4. Confidence - Know each piece works before putting them together

This pattern makes Orca workflows easier to test and maintain.

Parameters

Parameter	Type	Description
template	MethodTemplate	Method to execute
labware_start_mapping	Dict[LabwareTemplate, str]	Starting locations (labware → location name)
labware_end_mapping	Dict[LabwareTemplate, str]	Ending locations (labware → location name)
system	System	Built system
name	str (optional)	Name for the execution
event_hooks	List[EventHookInfo] (optional)	Event handlers

Execution Flow

When a workflow runs:

1. Initialization (if not already done)

   • Devices connect to hardware
   • Transporters home axes

2. Thread Execution

   • Entry threads start immediately
   • Spawned threads start when triggered
   • Methods execute sequentially within threads

3. Resource Coordination

   • Scheduler manages device access
   • Transporters coordinate movements
   • Deadlock resolution when needed

4. Completion

   • All threads finish
   • Labware at end locations

Complete Example

import asyncio
from orca.sdk.system import SdkToSystemBuilder, WorkflowExecutor
from orca.sdk.events import EventBus

async def run_assay():
    # Build system
    event_bus = EventBus()
    builder = SdkToSystemBuilder(
        name="Assay System",
        description="Process samples",
        labwares=[sample_plate],
        resources_registry=registry,
        system_map=system_map,
        methods=[prep_method, read_method],
        workflows=[assay_workflow],
        event_bus=event_bus
    )
    system = builder.get_system()

    # Initialize hardware
    await system.initialize_all()

    # Run workflow
    executor = WorkflowExecutor(assay_workflow, system)
    await executor.start()  # Or sim=True for simulation

if __name__ == "__main__":
    asyncio.run(run_assay())

Next Steps

• Events - Monitor workflow execution
• System Building - Configure the system