Step

A step is a single action on one device. It completes or fails as a whole, so a step never leaves a partial result at the scientific level.

For example, a shaking step can lock a shaker, set its RPM, run it for a duration, and unlock it. The step succeeds only when the complete operation finishes.

Example

from unitelabs.sdk.automate import step, phase

@step()
async def shake(shaker: ShakerDevice, rpm: int, duration: int):
    await shaker.elm_controller.lock(retry=3)
    await shaker.shake_controller.set_rpm(rpm)
    await shaker.shake_controller.shake(duration)
    await shaker.elm_controller.unlock()

Steps are called from inside a phase:

@phase()
async def agitation(target: Plate, shaker: ShakerDevice):
    await mix(target)
    await shake(shaker=shaker, rpm=300, duration=60)
    await shake(shaker=shaker, rpm=150, duration=30)

Key properties

• No partial results: raise and handle errors inside the step so it returns a clear success or failure.
• One device per step: a step drives a single device. Coordinating two or more devices belongs in a phase.
• Automatically retryable: the workflow engine retries technical errors (device timeouts, transient communication failures) automatically before propagating the error upward
• Sequenced within a phase: steps within a phase always run sequentially, in the order they are called

Retries

The workflow engine retries steps on technical failures automatically. You can configure retry behavior per step:

@step(retries=5, retry_delay_seconds=2)
async def aspirate(liquid_handler: LiquidHandler, volume: float):
    await liquid_handler.pipettes.aspirate(volume)

A technical error is a transient device-level failure: a timeout, a lost connection, a device that did not acknowledge a command. These are retried automatically.

A scientific error: a sample gone missing, a volume that cannot be aspirated — is propagated to the phase for user-defined recovery. Steps do not handle scientific errors; phases do (see Error Handling).

Device typing

Steps declare the device type they require as a function parameter. The workflow engine resolves the actual hardware instance at run time based on availability and capability — the step code never hardcodes a specific instrument:

@step()
async def centrifuge_samples(centrifuge: CentrifugeDevice, rpm: int, duration: int):
    await centrifuge.centrifugation_controller.spin(rpm=rpm, duration=duration)

This means the same step can run on any compatible centrifuge in the lab.

Related concepts

• Phase: the group of steps that ends in a stable state a run can resume from
• Error Handling: retry behavior and error propagation
• Workflow: the top-level process, defined by the scientific result it produces