System Building
Before running workflows, you build a System that connects your devices, locations, and workflows.
Overview
The system building process:
- Create
ResourceRegistry- Register all devices and transporters - Create
SystemMap- Define locations from transporter teachpoints - Assign resources to locations - Map devices to teachpoints
- Create
SdkToSystemBuilder- Connect everything - Call
get_system()- Get the runnable system
ResourceRegistry
The registry tracks all devices and transporters:
from orca.sdk.system import ResourceRegistry
registry = ResourceRegistry()
registry.add_resources([shaker, sealer, centrifuge, transporter])
Methods
| Method | Description |
|---|---|
add_resource(resource) | Add a single device or transporter |
add_resources(resources) | Add multiple resources or resource pools |
add_resource_pool(pool) | Add a resource pool |
get_resource(name) | Get resource by name |
get_device(name) | Get device by name |
get_transporter(name) | Get transporter by name |
set_simulating(bool) | Set simulation mode for all resources |
initialize_all() | Initialize all resources (async) |
Properties
resources- All registered resourcesdevices- All devicestransporters- All transportersresource_pools- All resource pools
SystemMap
The system map creates locations from transporter teachpoints and tracks device assignments:
from orca.sdk.system import SystemMap
system_map = SystemMap(registry)
When created, the SystemMap:
- Reads teachpoints from all registered transporters
- Creates a
Locationfor each teachpoint - Connects locations based on transporter reach
Assigning Devices to Locations
Map devices to teachpoint locations:
system_map.assign_resources({
"shaker_teachpoint": shaker,
"sealer_teachpoint": sealer,
"input_stack": input_storage,
})
The keys are teachpoint names from your transporter configuration.
Getting Locations
Retrieve locations for use in threads:
input_loc = system_map.get_location("input_stack")
output_loc = system_map.get_location("output_stack")
Methods
| Method | Description |
|---|---|
get_location(name) | Get location by teachpoint name |
assign_resources(mapping) | Assign multiple devices to locations |
assign_resource_to_location(name, resource) | Assign single device |
add_location(location) | Add a manual location |
location_exists(name) | Check if location exists |
SdkToSystemBuilder
The builder connects all components into a runnable system:
from orca.sdk.system import SdkToSystemBuilder
from orca.sdk.events import EventBus
event_bus = EventBus()
builder = SdkToSystemBuilder(
name="My Lab System",
description="Sample processing system",
labwares=[sample_plate, tip_rack],
resources_registry=registry,
system_map=system_map,
methods=[shake_method, seal_method],
workflows=[main_workflow],
event_bus=event_bus
)
system = builder.get_system()
Parameters
| Parameter | Type | Description |
|---|---|---|
name | str | System name |
description | str | System description |
labwares | Sequence[LabwareTemplate] | All labware templates |
resources_registry | ResourceRegistry | Device registry |
system_map | SystemMap | Location map |
methods | List[MethodTemplate] | All methods |
workflows | List[WorkflowTemplate] | All workflows |
event_bus | IEventBus | Event bus for notifications |
Complete Example
from orca.sdk.devices import Shaker, Sealer, Transporter
from orca.sdk.labware import PlateTemplate
from orca.sdk.workflow import MethodTemplate, ThreadTemplate, WorkflowTemplate
from orca.sdk.actions import Shake, Seal
from orca.sdk.system import ResourceRegistry, SystemMap, SdkToSystemBuilder
from orca.sdk.events import EventBus
# 1. Create devices
shaker = Shaker("shaker", shaker_driver)
sealer = Sealer("sealer", sealer_driver)
arm = Transporter("arm", arm_driver, "teachpoints/arm.json")
# 2. Create registry
registry = ResourceRegistry()
registry.add_resources([shaker, sealer, arm])
# 3. Create system map (loads teachpoints automatically)
system_map = SystemMap(registry)
# 4. Assign devices to locations
system_map.assign_resources({
"shaker": shaker,
"sealer": sealer,
})
# 5. Define labware
plate = PlateTemplate("sample", plate_factory, None)
# 6. Define methods
shake_method = MethodTemplate("shake", [
Shake(resource=shaker, duration=60, speed=500, inputs=[plate], outputs=[plate])
])
seal_method = MethodTemplate("seal", [
Seal(resource=sealer, temperature=165, duration=5, inputs=[plate], outputs=[plate])
])
# 7. Define thread
thread = ThreadTemplate(
labware_template=plate,
start=system_map.get_location("input_stack"),
end=system_map.get_location("output_stack"),
methods=[shake_method, seal_method]
)
# 8. Define workflow
workflow = WorkflowTemplate("main")
workflow.add_thread(thread, is_start=True)
# 9. Build system
event_bus = EventBus()
builder = SdkToSystemBuilder(
name="Sample System",
description="Shake and seal plates",
labwares=[plate],
resources_registry=registry,
system_map=system_map,
methods=[shake_method, seal_method],
workflows=[workflow],
event_bus=event_bus
)
system = builder.get_system()
Next Steps
- Running Workflows - Execute the system
- Events - Subscribe to system events