Manual operator actions in a Jupyter notebook#

This recipe models a common hybrid workflow: a plate is fetched from an incubator, a person moves it into a plate reader, and the protocol then reads it.

ManualOperator keeps the protocol independent of the acknowledgement interface. This notebook uses a small local provider built on input() because a notebook prompt is often the right level of complexity for a simple manual handoff. The same ManualOperator calls can later use a dashboard, LIMS, or message-broker provider without changing the protocol logic.

Prerequisites#

  • PyLabRobot with the manual_operator and EventBus features available.

  • This recipe uses legacy chatterbox backends, so it does not connect to hardware.

  • Run the notebook interactively. The manual-transfer cell pauses until the operator presses Enter. Set INTERACTIVE = False to run the chatterbox demonstration without a prompt.

1. Define a notebook-local provider#

An OperatorActionProvider turns a transport-independent OperatorActionRequest into an acknowledgement interaction. This minimal provider prints the request and treats Enter as a successful acknowledgement.

It intentionally pauses this notebook’s event loop while waiting. That is appropriate when the protocol should wait for the manual handoff before proceeding. Applications that need a richer UI or concurrent orchestration can supply their own provider instead.

from pylabrobot.manual_operator import (
    ManualOperator,
    OperatorActionRequest,
    OperatorActionResult,
)


INTERACTIVE = True


class NotebookOperatorActionProvider:
    """Minimal Jupyter-friendly provider for interactive protocol pauses."""

    async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
        print(f"\n{action.title}\n\n{action.instructions}\n")
        if INTERACTIVE:
            input(f"{action.confirmation_text}: ")
        else:
            print(f"[auto-confirmed] {action.confirmation_text}")
        return OperatorActionResult.completed(confirmed_by="notebook operator")


operator = ManualOperator(NotebookOperatorActionProvider(), name="notebook_operator")

2. Model the incubator, plate reader, and sample plate#

The sample plate begins in a modeled incubator storage site. The incubator and reader use chatterbox backends, which print their actions and return deterministic dummy data.

from pylabrobot.events import EventBus, PLREvent, use_event_bus
from pylabrobot.legacy.plate_reading import PlateReader, PlateReaderChatterboxBackend
from pylabrobot.legacy.storage import Incubator, IncubatorChatterboxBackend
from pylabrobot.resources import Coordinate, PlateCarrier, PlateHolder
from pylabrobot.resources.corning import cor_96_wellplate_360uL_Fb


incubator_slot = PlateHolder(
    name="incubator_slot_1",
    size_x=127.76,
    size_y=85.48,
    size_z=20,
    pedestal_size_z=0,
).at(Coordinate.zero())
incubator_rack = PlateCarrier(
    name="incubator_rack",
    size_x=140,
    size_y=100,
    size_z=100,
    sites={0: incubator_slot},
)

incubator = Incubator(
    name="incubator",
    size_x=200,
    size_y=200,
    size_z=300,
    backend=IncubatorChatterboxBackend(),
    racks=[incubator_rack],
    loading_tray_location=Coordinate.zero(),
)
plate_reader = PlateReader(
    name="plate_reader",
    size_x=160,
    size_y=160,
    size_z=100,
    backend=PlateReaderChatterboxBackend(),
)

sample_plate = cor_96_wellplate_360uL_Fb(name="sample_plate")
incubator_slot.assign_child_resource(sample_plate)

print(f"{sample_plate.name} starts in {incubator_slot.name}.")

3. Optional: observe semantic EventBus events#

ManualOperator does not require EventBus. It works with no subscriber installed. This optional section demonstrates that the incubator fetch and manual resource transfer emit semantic lifecycle events when a subscriber is active. Set ENABLE_EVENT_BUS_DEMO = False to run the exact same manual workflow without EventBus.

from contextlib import nullcontext


ENABLE_EVENT_BUS_DEMO = True

event_bus = EventBus()


def print_operation_outcome(event: PLREvent) -> None:
    operation = event.context.get("operation")
    if not isinstance(operation, str):
        return
    outcome = event.name.removeprefix(f"{operation}.")
    if outcome in {"completed", "failed"}:
        print(f"[event] {event.name}")


if ENABLE_EVENT_BUS_DEMO:
    event_bus.subscribe(print_operation_outcome)

4. Fetch, manually transfer, and read the plate#

The model remains unchanged while the request is pending: the sample plate stays on the incubator tray until the operator reports completion. move_resource() then validates the model and assigns the plate to the reader. On real hardware, ensure the reader is open before moving the plate and close it before reading. The ENABLE_EVENT_BUS_DEMO setting changes only observation; it does not change the manual action or resource-model behavior.

async def run_manual_transfer_and_read() -> list[dict]:
    event_scope = use_event_bus(event_bus) if ENABLE_EVENT_BUS_DEMO else nullcontext()
    with event_scope:
        await incubator.setup()
        await plate_reader.setup()
        try:
            await incubator.fetch_plate_to_loading_tray(sample_plate.name)
            assert incubator.loading_tray.resource is sample_plate

            await plate_reader.open()
            await operator.move_resource(
                resource=sample_plate,
                source=incubator.loading_tray,
                destination=plate_reader,
                title="Move plate to reader",
                instructions=(
                    "Move sample_plate from the incubator loading tray into the open plate reader, "
                    "and confirm after it is seated correctly."
                ),
                confirmation_text="Press Enter after the plate is seated in the reader",
                details={"reason": "manual incubator-to-reader handoff"},
            )
            assert plate_reader.get_plate() is sample_plate

            # Close the reader only after the model is updated to show the plate inside it.
            await plate_reader.close()
            return await plate_reader.read_absorbance(
                wavelength=450,
                use_new_return_type=True,
            )
        finally:
            await plate_reader.stop()
            await incubator.stop()


readings = await run_manual_transfer_and_read()
print(readings[0]["data"][0][:3])

What the protocol guarantees#

  • The incubator fetch updates the model from its storage site to the loading tray.

  • A cancelled or failed manual action leaves the plate on the tray in the PLR model.

  • A successful move_resource() acknowledgement updates the model only if the source and destination are still consistent.

  • If an EventBus subscriber is active, it sees the incubator fetch and the manual_operator.resource.move lifecycle. No subscriber is required for the manual action or resource-model update to work.

For manual actions that do not move a modeled resource, call await operator.perform(...) with a stable action name and structured details instead.