Actions
An action is one operation on one device, with one or more pieces of labware as input. Actions are the leaves of the workflow tree — they're where you actually call driver methods.
Defining an action
@orca.action(device=shaker, inputs=[plate])
async def shake(ctx: ActionContext):
await ctx.device().shake(duration=30, speed=500)
The decorator declares which device the action runs on and which labware it consumes. The body is an async function that talks to the device through ctx.device().
Action parameters
@orca.action(
device=sealer,
inputs=[plate],
outputs=[plate],
failure_policy=FailurePolicy.PAUSE,
tag="critical_seal",
deck_positions={plate: "carrier-7"},
well_selectors={...},
declares=DeclaredTracking(...),
)
async def seal(ctx: ActionContext):
...
| Parameter | Default | Effect |
|---|---|---|
device | required | Device or ResourcePool to run on. |
inputs | required | Labware templates this action consumes. |
outputs | inputs | Labware templates this action produces (often same as inputs). |
failure_policy | None | Per-action override of method's failure policy. |
tag | None | Optional tag for filtering events / ops history. |
deck_positions | None | For liquid handlers: which deck slot each labware goes in. |
well_selectors | None | Named well selectors used by the action body. |
declares | None | Declared tracking (volumes, contamination, etc.) this action updates. |
The real signatures are deck_positions: dict[LabwareTemplate, str], well_selectors: dict[str, WellSelector], and declares: DeclaredTracking.
TODO: deck_positions, well_selectors, declares each warrant their own walkthrough.
The action context
The ctx parameter gives you everything the action can do:
| Call | Returns / does |
|---|---|
ctx.device() | The device this action runs on (untyped). |
ctx.device(InterfaceClass) | The device, narrowed to a typed driver interface. |
ctx.labware(name) | The labware instance the thread is carrying for this template name. |
ctx.plate(name) | The underlying PLR Plate for the named plate input (typed accessor). |
ctx.tip_rack(name) | The underlying PLR TipRack for the named tip-rack input (typed accessor). |
ctx.trough(name) | The underlying PLR Trough for the named trough input (typed accessor). |
await ctx.param(name) | A workflow variable's resolved value (untyped). Async, must be awaited. |
await ctx.param(name, kind) | A workflow variable's value, narrowed to kind (e.g. float, int, str, bool); raises TypeError if it isn't already that type. Async. |
ctx.pool_index(receiver_name) | 0-based index of this action's contribution to the named receiver's current instance (raises ValueError if the action doesn't contribute to it). Used to route each contributor into a distinct region, e.g. four 96-well plates into the four quadrants of one 384 read plate. |
await ctx.emit(event_name, value=, data=) | Emit an event for other parts of the workflow to react to. Async. |
await ctx.wait_for(event_name) | Wait for a named event to fire (returns (value, data) tuple). Async. |
ctx.get_well_selector(name) | The WellSelector for a named labware input (falls back to all_wells()). |
await ctx.manual_step(instruction, timeout_hours=0) | Pause and display an operator instruction; blocks until confirmed. Async. |
Calling the device
For dispatch on the driver interface, narrow the device handle:
from orca.devices.device_interfaces import ILiquidHandler
@orca.action(device=liquid_handler, inputs=[plate, tips])
async def transfer(ctx: ActionContext):
lh = ctx.device(ILiquidHandler)
rack = ctx.tip_rack("tips")
plate = ctx.plate("sample_plate")
await lh.pick_up_tips([rack.tip_spot("A1")])
await lh.aspirate([plate.well("A1")], [50.0])
await lh.dispense([plate.well("B1")], [50.0])
await lh.drop_tips([rack.tip_spot("A1")])
Events from actions
Emit:
await ctx.emit("qc_result", value="pass", data={"avg_absorbance": 0.5})
Wait:
qc_value, qc_data = await ctx.wait_for("qc_result")
Events are how actions in different threads coordinate. The workflow can also branch on event values via orca.branch(...) — see Workflows.