Context API reference
Your @orca.action, @orca.method and @orca.thread function is handed one argument, ctx. It is your handle on the running system: the device you are at, the labware you are holding, variable values, and events.
Each decorator hands you a different shape, tuned to what makes sense at that level.
| Decorator | Context | What it does |
|---|---|---|
@orca.action | ActionContext | Talks to the action's device, reads labware, resolves variables, emits and waits for events. |
@orca.method | MethodContext | Device and labware lookups by name; can emit events. No wait_for. |
@orca.thread | ThreadContext | Event coordination across threads, variable reads, receiver-slot state. No device(): that lives at action scope. |
A fourth shape, WorkflowContext, is not an execution context at all. It is the builder @orca.workflow hands you once, at registration. It is under Advanced.
In an action
@orca.action(device=shaker, inputs=[sample_plate])
async def shake(ctx: ActionContext):
await ctx.device().shake(duration=30, speed=500)
| Call | Gives you |
|---|---|
ctx.device() | The action's device. Pass a type, ctx.device(IShaker), for autocomplete. |
ctx.labware("sample") | The labware instance the thread is carrying, by template name. |
ctx.plate("sample") | The PyLabRobot object inside it. Also ctx.tip_rack and ctx.trough. |
await ctx.next_tips("tips", 8) | The next 8 positions the rack still holds a tip at. |
await ctx.param("dilution_factor") | A variable value. |
await ctx.emit("PLATE.READY"), await ctx.wait_for("PLATE.READY", timeout=60.0) | Publish an event, or block until one arrives. |
ctx.device() takes no name argument. Each action targets the single device declared on @orca.action(device=...).
Labware lookup is by the labware template's name, as listed in inputs=[...]. A name that is not an input raises ValueError and lists the ones that are.
ActionContext is the richest of the four, and it is created once per action execution. Everything else it carries is under Advanced.
In a method
@orca.method
async def warm_and_shake(ctx: MethodContext):
yield warm
yield shake
| Call | Gives you |
|---|---|
ctx.device("shaker_1") | A device by name. Pass a type second: ctx.device("shaker_1", IShaker). |
ctx.labware("sample") | The labware instance, by template name. |
await ctx.param("dilution_factor") | A variable value. |
await ctx.emit("STEP.DONE", value="warm_and_shake") | Publish an event. |
The device name comes first here, because a method can span devices. Compare ActionContext.device(), which takes no name because an action is bound to one device.
There is no wait_for on a method. Wait inside an @orca.action, or at thread level.
MethodContext is created once per method execution. It is smaller than ActionContext, because most work in a method happens inside the yielded actions.
In a thread
@orca.thread(labware=sample_plate, start="stacker", end="waste")
async def sample_thread(ctx: ThreadContext):
yield shake_step
yield seal_step
| Call | Gives you |
|---|---|
ctx.labware | The labware this thread carries. A property, not a call. |
await ctx.param("dilution_factor") | A variable value. |
await ctx.emit("RESULT.READY", value=plate.barcode) | Publish an event. |
await ctx.wait_for("RESULT.READY", timeout=120) | Block until another thread publishes one. |
ctx.has_more_work() | Whether this thread's receiver slot can still take contributions. |
Thread-level code does not issue device commands. There is no ctx.device() on ThreadContext. If you want one, the work belongs in an @orca.action.
ThreadContext is the execution context for a labware thread. It is created once per thread, and it lives as long as the thread does.
Advanced
The rest of each context, grouped by which one you are holding, plus the workflow builder and the full side-by-side table.
ActionContext in full
Typed lookups
ctx.device() # DeviceHandle
ctx.device(IShaker) # the same handle, typed for autocomplete
plate = ctx.labware("sample") # LabwareInstance
plate = ctx.labware("sample", PlateInstance) # narrowed; raises TypeError on mismatch
PLR-object shortcuts skip the wrapper:
plr_plate = ctx.plate("sample") # IPlate
plr_tip_rack = ctx.tip_rack("tips") # ITipRack
plr_trough = ctx.trough("reservoir") # ITrough
These are the same as ctx.labware("sample", PlateInstance).plate and so on.
Tips the rack actually holds
spots = await ctx.next_tips("tips", 8) # list[ITipSpot]
await lh.pick_up_tips(spots)
next_tips returns the next count positions the rack still holds a tip at, column-major, so an 8-channel head gets a whole column. It is async. Use it instead of naming positions: a rack left on the deck between runs has no full first column by the second run, and a hard-coded A1 picks air. It raises when the rack cannot supply that many, and when nothing has ever described the rack's contents. Recording a pick-up advances what the next call returns.
Well selectors
selector = ctx.get_well_selector("sample") # WellSelector
Returns the selector declared on the action via well_selectors={...}, or all_wells() if none was.
Pooling
quadrant = ctx.pool_index("read_plate") # 0-based index of this contribution
When several threads contribute into one shared receiver, for example four 96-well plates pooled into one 384-well read plate, pool_index(receiver_name) gives the 0-based index of this action's contribution to that receiver's current instance. Use it to route each contributor into its own region. Raises ValueError if this action does not contribute to that receiver. ActionContext only.
Variables
factor = await ctx.param("dilution_factor") # OptionValue
factor = await ctx.param("dilution_factor", float) # narrowed; raises TypeError on mismatch
Async, so await it. The narrowing form takes str, int, float or bool. If the action is running inside a submission, that submission's own overrides are read before the execution-wide ones. See Variables.
Events
await ctx.emit("PLATE.READY", value=plate.barcode, data={"step": "sealed"})
value, data = await ctx.wait_for("PLATE.READY", timeout=60.0)
Both are async. emit publishes to the in-process channel other threads read, and bridges to the runtime event surface as CUSTOM.PLATE_READY.EMITTED, so sinks, archives and the events API see it. Dots in your name become underscores in the bridged name only; the event context keeps the name verbatim.
wait_for blocks until a publish arrives after the counter this context last saw, or raises TimeoutError.
Manual operator step
await ctx.manual_step(
"Refill the reservoir on deck slot A1, then confirm.",
timeout_hours=2,
)
Logs the instruction to the orca.operator logger, emits OPERATOR.INSTRUCTION with a unique step_id, and blocks until a targeted OPERATOR.CONFIRM.{step_id} event fires. Each call gets its own step id, so concurrent manual steps do not interfere. timeout_hours=0 (the default) waits indefinitely.
A manual step is a safe point: nothing is moving and the operator is at the instrument. So if the execution is paused while the step is pending, the confirm is still accepted, and the thread then waits for resume before running the rest of the body. timeout_hours bounds the wait for the operator's confirm, not the hold for a pause they asked for themselves.
MethodContext in full
ctx.device("shaker_1") # DeviceHandle by name
ctx.device("shaker_1", IShaker) # typed handle
plate = ctx.labware("sample") # LabwareInstance
There is no narrowing overload on labware here. Narrow inside an action with ActionContext.labware(name, kind) if you need it.
factor = await ctx.param("dilution_factor") # OptionValue
Async, and no narrowing form. Submission-aware, the same as the other two.
await ctx.emit("STEP.DONE", value="warm_and_shake")
emit only, and it bridges to CUSTOM.STEP_DONE.EMITTED like the others. No wait_for on MethodContext: wait inside an @orca.action, or at thread level.
ThreadContext in full
Labware
plate = ctx.labware # property, not a call
The labware instance this thread is carrying, from the @orca.thread(labware=...) declaration. Raises RuntimeError in a context with no bound labware.
Variables
factor = await ctx.param("dilution_factor")
Async, and submission-aware: if the thread carries a submission id, that submission's own partition is read before the execution-wide one. See Variables.
Events
await ctx.emit("RESULT.READY", value=plate.barcode)
value, data = await ctx.wait_for("RESULT.READY", timeout=120)
Same channel and same bridge to CUSTOM.RESULT_READY.EMITTED as the other two contexts.
Receiver slots
while ctx.has_more_work():
yield orca.join(allows=[transfer_to_read_plate])
ctx.has_more_work() says whether this thread's receiver slot can still take contributions. Four gates, in order:
- Queued items give
True, even on a closed slot. - A closed slot gives
False. - A receiver whose labware is used up gives
False, whatever room the slot has left. A contributor discovers the depletion when it asks the labwarecan_continue(), and records it on the slot, so this can land while the receiver is already parked in its join. - Otherwise, the slot's own room. An open slot already at
max_contributionsgivesFalse, finalizing a full receiver rather than looping for a contribution the capacity policy would overflow to a fresh one.
The slot closes when its contribution window is over: no thread that transitively feeds this receiver up the contributes_to chain is still live, or, for a receiver with no declared feeder, no in-scope worker thread is still live. A thread waiting for an operator to collect its labware counts as finished for both. Nothing waits on the chance that a later submission might feed the slot; only an injection already in flight defers the close.
A thread with no receiver slot always gets True. Terminate that loop some other way, or do not loop.
Partner constraints
ctx.set_partner_constraint("tips_template", {"barcode": "RACK-007"})
yield use_tips
When this thread's next action triggers auto-spawn for labware matching tips_template, the constraints go to the labware finder to pick the right partner. Raises RuntimeError in a context with no partner-constraint setter wired.
WorkflowContext, the workflow builder
@orca.workflow(name="smc_assay")
def smc_assay(wf: WorkflowContext):
wf.start(sample_thread)
wf.thread(tips_thread)
wf.on("THREAD.CREATED", MyHandler())
wf.variable("plate_count", VariableDefinition(type="int", default=1))
Builder, not an execution context. It is created once per workflow registration, and it is called once, at registration. Declare structure here; do not put per-run logic in it.
wf.start(thread)
Register an entry thread that spawns when the workflow starts. More than one is allowed. A thread whose start uses REUSE_EXISTING, or one declared immovable=True, is refused as an entry at build time.
wf.thread(thread, capacity=None, overflow_strategy=None)
Register a thread for auto-spawn, keyed by the labware template it carries. When an owning thread resolves an action whose inputs name that template and nothing is carrying one, the engine spawns this thread. Its generator yields orca.join(...) at the point it joins the shared method.
| Parameter | Effect |
|---|---|
capacity | A CapacityPolicy for the receiver slot: max_contributions plus an OverflowAction of NEW, REJECT or RECOVERABLE_REJECT. |
overflow_strategy | An IOverflowStrategy for handling overflow yourself. |
Two threads registered for the same labware template raise ValueError.
wf.on(event_name, handler)
Subscribe an event handler for this workflow's executions. The handler is an IEventHandler (one handle(event, context) method) or a plain callable taking the same two arguments.
wf.variable(name, definition)
Declare a workflow-scoped variable. See Variables for the VariableDefinition fields.
wf.template returns the WorkflowTemplate being built.
The full surface, side by side
| Surface | ActionContext | MethodContext | ThreadContext | WorkflowContext |
|---|---|---|---|---|
device() | yes (no name) | yes (by name) | no | no |
device(T) typed | yes | yes (name first) | no | no |
labware(name) | yes | yes | ctx.labware (property) | no |
labware(name, kind) typed | yes | no | no | no |
plate / tip_rack / trough shortcuts | yes | no | no | no |
next_tips(name, count) | yes (async) | no | no | no |
get_well_selector(name) | yes | no | no | no |
pool_index(receiver) | yes | no | no | no |
param(name) | yes (async) | yes (async) | yes (async) | no |
param(name, kind) typed | yes (async) | no | no | no |
emit(name, value, data) | yes (async) | yes (async) | yes (async) | no |
wait_for(name, timeout) | yes (async) | no | yes (async) | no |
manual_step(instruction, timeout_hours) | yes (async) | no | no | no |
has_more_work() | no | no | yes | no |
set_partner_constraint(...) | no | no | yes | no |
start(thread) | no | no | no | yes |
thread(thread, ...) | no | no | no | yes |
on(event_name, handler) | no | no | no | yes |
variable(name, definition) | no | no | no | yes |
Pitfalls
ActionContext.device()takes NO name. The device comes from the action's@orca.action(device=...)declaration.MethodContext.device("shaker_1")does take a name, because a method can span devices.- No
wait_foronMethodContext. Events flow at action scope or thread scope. Methods do not wait; their actions do. ctx.wait_foradvances;orca.ondoes not. BothActionContext.wait_forandThreadContext.wait_forkeep a per-context seen-counter: the first call sees the latest publish, and later calls only see newer ones. The counter lives as long as its context, so a thread'swhile True: await ctx.wait_for("X")gets each new publish, while a fresh action starts from the latest publish again. Theorca.on(...)andorca.branch(...)steps read from zero every time, so they always see the latest publish.WorkflowContextis a builder. The body of@orca.workflow def my_assay(wf):runs ONCE at registration, not per execution.- The
@orca.workflowfunction must not beasync. It raisesTypeErrorif it returns a coroutine. ctx.labwareonThreadContextis a property, not a call.ctx.labware("sample")raisesTypeError. Usectx.labware.paramnarrowing raisesTypeErroron mismatch.ctx.param("plate_count", int)against a workflow that definedplate_countas a float raises at the call site. Fix the type at the definition.