From f0f05d513f5e73d84c946fc11833b18cfbea4149 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Thu, 6 Aug 2026 15:02:13 -0700 Subject: [PATCH 01/12] Port real-time resource state to main --- .../backends/hamilton/STAR_backend.py | 28 +++++++++++-------- .../legacy/liquid_handling/liquid_handler.py | 18 +++++++++++- .../liquid_handling/liquid_handler_tests.py | 23 +++++++++++++++ pylabrobot/legacy/liquid_handling/standard.py | 6 ++++ 4 files changed, 63 insertions(+), 12 deletions(-) diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py index f57ec4b3582..d7fe21aeb45 100644 --- a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_backend.py @@ -5049,11 +5049,16 @@ async def drop_resource( # This means that the center vector has to be rotated from the child local space by the # new child absolute rotation. The moved resource's rotation will be the original child # rotation plus the rotation applied by the movement. - # The resource is moved by drop.rotation - # The new resource absolute location is - # drop.resource.get_absolute_rotation().z + drop.rotation + # The resource is detached immediately after pickup. Use its captured + # parent-derived rotation rather than asking the detached resource tree. + resource_absolute_rotation_at_pickup = ( + drop.resource_absolute_rotation_at_pickup or drop.resource.get_absolute_rotation() + ) + resource_absolute_rotation_after_move = resource_absolute_rotation_at_pickup + Rotation( + z=drop.rotation + ) center_in_absolute_space = drop.resource.center().rotated( - Rotation(z=drop.resource.get_absolute_rotation().z + drop.rotation) + resource_absolute_rotation_after_move ) x, y, z = drop.destination + center_in_absolute_space + drop.offset @@ -5063,15 +5068,16 @@ async def drop_resource( ) z_position_at_the_command_end = z_position_at_the_command_end or self._iswap_traversal_height assert ( - drop.resource.get_absolute_rotation().x == 0 - and drop.resource.get_absolute_rotation().y == 0 + resource_absolute_rotation_at_pickup.x == 0 + and resource_absolute_rotation_at_pickup.y == 0 ) - assert drop.resource.get_absolute_rotation().z % 90 == 0 + assert resource_absolute_rotation_at_pickup.z % 90 == 0 - # Use the pickup direction to determine how wide the plate is gripped. - # Note that the plate is still in the original orientation at this point, - # so get_absolute_size_{x,y}() will return the size of the plate in the original orientation. - if ( + # Keep the release width captured before detaching the resource. Recomputing + # its absolute size after pickup can select the long-side plate width. + if drop.resource_width_at_pickup is not None: + plate_width = drop.resource_width_at_pickup + elif ( drop.pickup_direction == GripDirection.FRONT or drop.pickup_direction == GripDirection.BACK ): plate_width = drop.resource.get_absolute_size_x() diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler.py b/pylabrobot/legacy/liquid_handling/liquid_handler.py index a10242c5c95..d1ac95cc487 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler.py @@ -2069,11 +2069,20 @@ async def pick_up_resource( if self._resource_pickup is not None: raise RuntimeError(f"Resource {self._resource_pickup.resource.name} already picked up") + if backend_kwargs.get("plate_width") is not None: + resource_width_at_pickup = backend_kwargs["plate_width"] + elif direction in (GripDirection.FRONT, GripDirection.BACK): + resource_width_at_pickup = resource.get_absolute_size_x() + else: + resource_width_at_pickup = resource.get_absolute_size_y() + self._resource_pickup = ResourcePickup( resource=resource, offset=offset, pickup_distance_from_top=pickup_distance_from_top, direction=direction, + resource_absolute_rotation_at_pickup=resource.get_absolute_rotation(), + resource_width_at_pickup=resource_width_at_pickup, ) extras = self._check_args( @@ -2091,6 +2100,7 @@ async def pick_up_resource( self._resource_pickup = None raise e + resource.unassign() self._state_updated() async def move_picked_up_resource( @@ -2182,8 +2192,12 @@ async def drop_resource( # should be and subtract the rotation of the new parent. # moving from a resource from a rotated parent to a non-rotated parent means child inherits/'houses' the rotation after move + resource_absolute_rotation_at_pickup = ( + self._resource_pickup.resource_absolute_rotation_at_pickup + or resource.get_absolute_rotation() + ) resource_absolute_rotation_after_move = ( - resource.get_absolute_rotation().z + rotation_applied_by_move + resource_absolute_rotation_at_pickup.z + rotation_applied_by_move ) destination_rotation = ( destination.get_absolute_rotation().z if not isinstance(destination, Coordinate) else 0 @@ -2257,6 +2271,8 @@ async def drop_resource( pickup_direction=self._resource_pickup.direction, direction=direction, rotation=rotation_applied_by_move, + resource_absolute_rotation_at_pickup=resource_absolute_rotation_at_pickup, + resource_width_at_pickup=self._resource_pickup.resource_width_at_pickup, ) result = await self.backend.drop_resource(drop=drop, **backend_kwargs) diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py index 1973933b84f..7bce30a164e 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py @@ -28,6 +28,7 @@ Plate, Resource, ResourceNotFoundError, + ResourceHolder, ResourceStack, TipRack, cor_96_wellplate_360uL_Fb, @@ -41,6 +42,7 @@ HasTipError, NoTipError, ) +from pylabrobot.resources.rotation import Rotation from pylabrobot.resources.hamilton import ( STARLetDeck, hamilton_96_tiprack_300uL_filter, @@ -1231,6 +1233,27 @@ async def test_serialize_state_after_setup(self): # 1 arm, no resource picked up self.assertEqual(state["arm_state"], {0: None}) + async def test_resource_drop_uses_pose_and_width_captured_at_pickup(self): + source = ResourceHolder("rotated_source", size_x=130, size_y=90, size_z=0) + source.rotation = Rotation(z=90) + destination = ResourceHolder("destination", size_x=130, size_y=90, size_z=0) + self.plate.unassign() + self.deck.assign_child_resource(source, location=Coordinate(100, 100, 0)) + self.deck.assign_child_resource(destination, location=Coordinate(300, 100, 0)) + source.assign_child_resource(self.plate, location=Coordinate.zero()) + + width_at_pickup = self.plate.get_absolute_size_x() + rotation_at_pickup = self.plate.get_absolute_rotation() + self.assertNotEqual(width_at_pickup, self.plate.get_size_x()) + + await self.lh.pick_up_resource(self.plate, direction=GripDirection.FRONT) + self.assertIsNone(self.plate.parent) + await self.lh.drop_resource(destination, direction=GripDirection.FRONT) + + drop = self.backend.drop_resource.call_args.kwargs["drop"] + self.assertEqual(drop.resource_absolute_rotation_at_pickup, rotation_at_pickup) + self.assertAlmostEqual(drop.resource_width_at_pickup, width_at_pickup) + async def test_serialize_state_no_head96(self): backend = _create_mock_backend(num_channels=8) type(backend).head96_installed = PropertyMock(return_value=False) diff --git a/pylabrobot/legacy/liquid_handling/standard.py b/pylabrobot/legacy/liquid_handling/standard.py index fdc057b7c94..e9f6d2a7202 100644 --- a/pylabrobot/legacy/liquid_handling/standard.py +++ b/pylabrobot/legacy/liquid_handling/standard.py @@ -151,6 +151,10 @@ class ResourcePickup: offset: Coordinate pickup_distance_from_top: float direction: GripDirection + # The resource is detached once the grip succeeds, so capture parent-derived + # geometry before it is no longer available from the resource tree. + resource_absolute_rotation_at_pickup: Optional[Rotation] = None + resource_width_at_pickup: Optional[float] = None @dataclass(frozen=True) @@ -175,6 +179,8 @@ class ResourceDrop: pickup_direction: GripDirection direction: GripDirection rotation: float + resource_absolute_rotation_at_pickup: Optional[Rotation] = None + resource_width_at_pickup: Optional[float] = None PipettingOp = Union[Pickup, Drop, SingleChannelAspiration, SingleChannelDispense] From 6bd4e649a5e0364e0aab33fd0dd06803443fbf2e Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 17:46:09 -0700 Subject: [PATCH 02/12] Add opt-in contextual execution events --- pylabrobot/events/__init__.py | 29 +++ pylabrobot/events/bus.py | 211 ++++++++++++++++++ pylabrobot/events/bus_tests.py | 85 +++++++ pylabrobot/hamilton/transport/usb/usb.py | 39 +++- pylabrobot/io/ftdi.py | 25 ++- pylabrobot/io/serial.py | 22 ++ pylabrobot/io/usb.py | 19 +- .../legacy/liquid_handling/liquid_handler.py | 34 +++ pylabrobot/legacy/machines/machine.py | 10 + pylabrobot/legacy/storage/incubator.py | 32 +++ pylabrobot/resources/resource.py | 18 ++ 11 files changed, 512 insertions(+), 12 deletions(-) create mode 100644 pylabrobot/events/__init__.py create mode 100644 pylabrobot/events/bus.py create mode 100644 pylabrobot/events/bus_tests.py diff --git a/pylabrobot/events/__init__.py b/pylabrobot/events/__init__.py new file mode 100644 index 00000000000..4d223c421bf --- /dev/null +++ b/pylabrobot/events/__init__.py @@ -0,0 +1,29 @@ +"""Structured, opt-in execution events for PyLabRobot.""" + +from .bus import ( + EventBus, + PLREvent, + coordinate_reference, + emit_event, + event_context, + evented_operation, + get_event_bus, + is_event_bus_active, + resource_reference, + set_default_event_bus, + use_event_bus, +) + +__all__ = [ + "EventBus", + "PLREvent", + "coordinate_reference", + "emit_event", + "event_context", + "evented_operation", + "get_event_bus", + "is_event_bus_active", + "resource_reference", + "set_default_event_bus", + "use_event_bus", +] diff --git a/pylabrobot/events/bus.py b/pylabrobot/events/bus.py new file mode 100644 index 00000000000..4c837dbbe36 --- /dev/null +++ b/pylabrobot/events/bus.py @@ -0,0 +1,211 @@ +"""Opt-in structured events for PLR execution and state transitions. + +The bus is deliberately synchronous and in-process. Subscribers should enqueue or persist +quickly; they must not block or alter hardware control flow. +""" + +from __future__ import annotations + +import contextvars +import datetime as dt +import logging +import threading +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from functools import wraps +from typing import Any, Awaitable, Callable, Dict, Iterator, List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class PLREvent: + """One structured PLR event emitted at a state or command boundary.""" + + sequence: int + name: str + timestamp: dt.datetime + context: Dict[str, Any] + data: Dict[str, Any] + + def as_dict(self) -> Dict[str, Any]: + """Return a JSONL-friendly representation of the event.""" + return { + "sequence": self.sequence, + "name": self.name, + "timestamp": self.timestamp.isoformat(), + "context": self.context.copy(), + "data": self.data.copy(), + } + + +EventListener = Callable[[PLREvent], None] +OperationContextFactory = Callable[..., Dict[str, Any]] + + +class EventBus: + """In-process event fan-out that never lets observer failures affect PLR control flow.""" + + def __init__(self): + self._listeners: List[EventListener] = [] + self._lock = threading.Lock() + self._sequence = 0 + + def subscribe(self, listener: EventListener) -> Callable[[], None]: + """Register a listener and return an unsubscribe callback.""" + with self._lock: + self._listeners.append(listener) + + def unsubscribe() -> None: + with self._lock: + if listener in self._listeners: + self._listeners.remove(listener) + + return unsubscribe + + @property + def has_listeners(self) -> bool: + with self._lock: + return bool(self._listeners) + + def emit(self, name: str, *, context: Optional[Dict[str, Any]] = None, **data: Any) -> PLREvent: + """Publish an event to a stable listener snapshot.""" + with self._lock: + self._sequence += 1 + event = PLREvent( + sequence=self._sequence, + name=name, + timestamp=dt.datetime.now(dt.timezone.utc), + context=(context or {}).copy(), + data=data.copy(), + ) + listeners = self._listeners.copy() + + for listener in listeners: + try: + listener(event) + except Exception: + logger.exception("PLR event listener failed while handling %s", name) + return event + + +_default_event_bus: Optional[EventBus] = None +_active_event_bus: contextvars.ContextVar[Optional[EventBus]] = contextvars.ContextVar( + "pylabrobot_active_event_bus", default=None +) +_event_context: contextvars.ContextVar[Dict[str, Any]] = contextvars.ContextVar( + "pylabrobot_event_context", default={} +) + + +def set_default_event_bus(event_bus: Optional[EventBus]) -> None: + """Install or clear the process-wide event bus for code without an explicit scope.""" + global _default_event_bus + _default_event_bus = event_bus + + +def get_event_bus() -> Optional[EventBus]: + """Return the scoped event bus, falling back to the process-wide bus.""" + return _active_event_bus.get() or _default_event_bus + + +def is_event_bus_active() -> bool: + """Whether the current execution context has an interested event subscriber.""" + event_bus = get_event_bus() + return event_bus is not None and event_bus.has_listeners + + +@contextmanager +def use_event_bus(event_bus: EventBus) -> Iterator[EventBus]: + """Temporarily install an event bus for the current async/task context.""" + token = _active_event_bus.set(event_bus) + try: + yield event_bus + finally: + _active_event_bus.reset(token) + + +@contextmanager +def event_context(**values: Any) -> Iterator[None]: + """Attach context to all nested events in the current execution context.""" + context = _event_context.get().copy() + context.update({key: value for key, value in values.items() if value is not None}) + token = _event_context.set(context) + try: + yield + finally: + _event_context.reset(token) + + +def emit_event(name: str, **data: Any) -> Optional[PLREvent]: + """Emit an event if an event bus with listeners is installed.""" + event_bus = get_event_bus() + if event_bus is None or not event_bus.has_listeners: + return None + return event_bus.emit(name, context=_event_context.get(), **data) + + +def resource_reference(resource: Any) -> Optional[Dict[str, Any]]: + """Return a compact resource identity without serializing its full subtree.""" + if resource is None: + return None + result: Dict[str, Any] = { + "name": getattr(resource, "name", None), + "type": type(resource).__name__, + } + model = getattr(resource, "model", None) + if model is not None: + result["model"] = str(model) + rotation = getattr(resource, "rotation", None) + if rotation is not None: + result["rotation"] = { + "x": rotation.x, + "y": rotation.y, + "z": rotation.z, + } + return result + + +def coordinate_reference(coordinate: Any) -> Optional[Dict[str, float]]: + """Return a JSON-friendly coordinate, or ``None`` for an unlocated resource.""" + if coordinate is None: + return None + return {"x": coordinate.x, "y": coordinate.y, "z": coordinate.z} + + +def evented_operation( + name: str, context_factory: OperationContextFactory +) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]: + """Decorate an async frontend call with correlated lifecycle events. + + The wrapper is a no-op when no listener is installed, preserving normal PLR performance and + behaviour. Nested resource and transport events inherit the generated operation context. + """ + + def decorator(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]: + @wraps(func) + async def wrapper(*args: Any, **kwargs: Any) -> Any: + if not is_event_bus_active(): + return await func(*args, **kwargs) + + operation_data = context_factory(*args, **kwargs) + operation_id = uuid.uuid4().hex + with event_context(operation=name, operation_id=operation_id, **operation_data): + emit_event(f"{name}.started", **operation_data) + try: + result = await func(*args, **kwargs) + except BaseException as error: + emit_event( + f"{name}.failed", + **operation_data, + error_type=type(error).__name__, + error_message=str(error), + ) + raise + emit_event(f"{name}.completed", **operation_data) + return result + + return wrapper + + return decorator diff --git a/pylabrobot/events/bus_tests.py b/pylabrobot/events/bus_tests.py new file mode 100644 index 00000000000..4bec0cf8143 --- /dev/null +++ b/pylabrobot/events/bus_tests.py @@ -0,0 +1,85 @@ +import json +import unittest + +from pylabrobot.events import EventBus, emit_event, event_context, evented_operation, use_event_bus +from pylabrobot.resources import Coordinate, Resource + + +class TestEventBus(unittest.TestCase): + def test_context_is_attached_to_events(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus), event_context(run_id="run-1", batch_id="batch-1"): + emit_event("example.completed", value=42) + + self.assertEqual(events[0].name, "example.completed") + self.assertEqual(events[0].context, {"run_id": "run-1", "batch_id": "batch-1"}) + self.assertEqual(events[0].data, {"value": 42}) + self.assertEqual(json.loads(json.dumps(events[0].as_dict()))["name"], "example.completed") + + def test_resource_assignment_and_unassignment_emit_contextual_events(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + parent = Resource("parent", size_x=10, size_y=10, size_z=1) + child = Resource("child", size_x=5, size_y=5, size_z=1) + + with use_event_bus(event_bus), event_context(run_id="run-1"): + parent.assign_child_resource(child, location=Coordinate(1, 2, 3)) + child.unassign() + + assigned, unassigned = events + self.assertEqual(assigned.name, "resource.assigned") + self.assertEqual(assigned.context["run_id"], "run-1") + self.assertEqual(assigned.data["resource"]["name"], "child") + self.assertEqual(assigned.data["parent"]["name"], "parent") + self.assertEqual(assigned.data["location"], {"x": 1, "y": 2, "z": 3}) + self.assertEqual(unassigned.name, "resource.unassigned") + self.assertEqual(unassigned.data["previous_parent"]["name"], "parent") + self.assertEqual(assigned.data["resource"]["rotation"], {"x": 0, "y": 0, "z": 0}) + + +class TestEventedOperation(unittest.IsolatedAsyncioTestCase): + async def test_operation_context_is_inherited_by_nested_events(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + @evented_operation("device.action", lambda value: {"device": "device-1", "value": value}) + async def action(value): + emit_event("firmware.command.started", command="AB") + return value * 2 + + with use_event_bus(event_bus): + self.assertEqual(await action(3), 6) + + self.assertEqual([event.name for event in events], [ + "device.action.started", + "firmware.command.started", + "device.action.completed", + ]) + operation_id = events[0].context["operation_id"] + self.assertEqual(events[1].context["operation_id"], operation_id) + self.assertEqual(events[2].context["operation_id"], operation_id) + + async def test_failed_operation_emits_a_correlated_failure(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + @evented_operation("device.action", lambda: {"device": "device-1"}) + async def action(): + raise RuntimeError("expected failure") + + with use_event_bus(event_bus): + with self.assertRaisesRegex(RuntimeError, "expected failure"): + await action() + + self.assertEqual([event.name for event in events], [ + "device.action.started", + "device.action.failed", + ]) + self.assertEqual(events[1].data["error_type"], "RuntimeError") + self.assertEqual(events[1].context["operation_id"], events[0].context["operation_id"]) diff --git a/pylabrobot/hamilton/transport/usb/usb.py b/pylabrobot/hamilton/transport/usb/usb.py index d0e37ac5cd3..fd19457e7c8 100644 --- a/pylabrobot/hamilton/transport/usb/usb.py +++ b/pylabrobot/hamilton/transport/usb/usb.py @@ -13,6 +13,7 @@ TypeVar, ) +from pylabrobot.events import emit_event from pylabrobot.io.usb import USB T = TypeVar("T") @@ -241,16 +242,34 @@ async def send_command( auto_id=auto_id, **kwargs, ) - resp = await self._write_and_read_command( - id_=id_, - cmd=cmd, - write_timeout=write_timeout, - read_timeout=read_timeout, - wait=wait, - ) - if resp is not None and fmt is not None: - return self._parse_response(resp, fmt) - return resp + event_data = { + "transport": "hamilton_usb", + "driver": type(self).__name__, + "module": module, + "command": command, + "command_id": id_, + "raw_command": cmd, + } + emit_event("firmware.command.started", **event_data) + try: + resp = await self._write_and_read_command( + id_=id_, + cmd=cmd, + write_timeout=write_timeout, + read_timeout=read_timeout, + wait=wait, + ) + result = self._parse_response(resp, fmt) if resp is not None and fmt is not None else resp + except BaseException as error: + emit_event( + "firmware.command.failed", + **event_data, + error_type=type(error).__name__, + error_message=str(error), + ) + raise + emit_event("firmware.command.completed", **event_data, response=resp) + return result async def _write_and_read_command( self, diff --git a/pylabrobot/io/ftdi.py b/pylabrobot/io/ftdi.py index 17df02e7603..c8f966baa02 100644 --- a/pylabrobot/io/ftdi.py +++ b/pylabrobot/io/ftdi.py @@ -5,6 +5,7 @@ from io import IOBase from typing import Optional, cast +from pylabrobot.events import emit_event try: import pylibftdi.driver from pylibftdi import Device, FtdiError @@ -307,7 +308,15 @@ async def write(self, data: bytes) -> int: """Write data to the device. Returns the number of bytes written.""" logger.log(LOG_LEVEL_IO, "[%s] write %s", self._device_id, data) capturer.record(FTDICommand(device_id=self.device_id, action="write", data=data.hex())) - return cast(int, self.dev.write(data)) + bytes_written = cast(int, self.dev.write(data)) + emit_event( + "io.write", + transport="ftdi", + device=self.human_readable_device_name, + device_id=self.device_id, + data=data.hex(), + ) + return bytes_written async def read(self, num_bytes: int = 1) -> bytes: data = self.dev.read(num_bytes) @@ -320,6 +329,13 @@ async def read(self, num_bytes: int = 1) -> bytes: data=data if isinstance(data, str) else data.hex(), ) ) + emit_event( + "io.read", + transport="ftdi", + device=self.human_readable_device_name, + device_id=self.device_id, + data=data if isinstance(data, str) else data.hex(), + ) return cast(bytes, data) async def readline(self) -> bytes: # type: ignore # very dumb it's reading from pyserial @@ -327,6 +343,13 @@ async def readline(self) -> bytes: # type: ignore # very dumb it's reading from if len(data) != 0: logger.log(LOG_LEVEL_IO, "[%s] readline %s", self._device_id, data) capturer.record(FTDICommand(device_id=self.device_id, action="readline", data=data.hex())) + emit_event( + "io.read", + transport="ftdi", + device=self.human_readable_device_name, + device_id=self.device_id, + data=data if isinstance(data, str) else data.hex(), + ) return cast(bytes, data) def serialize(self): diff --git a/pylabrobot/io/serial.py b/pylabrobot/io/serial.py index 26ddf801d96..0030a5bb926 100644 --- a/pylabrobot/io/serial.py +++ b/pylabrobot/io/serial.py @@ -6,6 +6,7 @@ from io import IOBase from typing import Iterator, Optional, cast +from pylabrobot.events import emit_event from pylabrobot.io.errors import ValidationError try: @@ -240,6 +241,13 @@ async def write(self, data: bytes): capturer.record( SerialCommand(device_id=self.port, action="write", data=data.decode("unicode_escape")) ) + emit_event( + "io.write", + transport="serial", + device=self.human_readable_device_name, + device_id=self.port, + data=data.decode("utf-8", errors="backslashreplace"), + ) async def read(self, num_bytes: int = 1) -> bytes: """Read data from the serial device.""" @@ -255,6 +263,13 @@ async def read(self, num_bytes: int = 1) -> bytes: capturer.record( SerialCommand(device_id=self.port, action="read", data=data.decode("unicode_escape")) ) + emit_event( + "io.read", + transport="serial", + device=self.human_readable_device_name, + device_id=self.port, + data=data.decode("utf-8", errors="backslashreplace"), + ) return cast(bytes, data) @@ -272,6 +287,13 @@ async def readline(self) -> bytes: # type: ignore # very dumb it's reading from capturer.record( SerialCommand(device_id=self.port, action="readline", data=data.decode("unicode_escape")) ) + emit_event( + "io.read", + transport="serial", + device=self.human_readable_device_name, + device_id=self.port, + data=data.decode("utf-8", errors="backslashreplace"), + ) return cast(bytes, data) diff --git a/pylabrobot/io/usb.py b/pylabrobot/io/usb.py index 8e4469f1577..adc9a7a7067 100644 --- a/pylabrobot/io/usb.py +++ b/pylabrobot/io/usb.py @@ -5,6 +5,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Callable, List, Optional +from pylabrobot.events import emit_event from pylabrobot.io.capture import Command, capturer, get_capture_or_validation_active from pylabrobot.io.errors import ValidationError from pylabrobot.io.io import IOBase @@ -159,6 +160,13 @@ async def write(self, data: bytes, timeout: Optional[float] = None): data=data.decode("unicode_escape", errors="backslashreplace"), ) ) + emit_event( + "io.write", + transport="usb", + device=self.human_readable_device_name, + device_id=self._unique_id, + data=data.decode("utf-8", errors="backslashreplace"), + ) def _read_packet( self, @@ -277,7 +285,16 @@ def read_or_timeout(): loop = asyncio.get_running_loop() if self._read_executor is None or self.dev is None: raise RuntimeError(f"Call setup() first for USB device '{self.human_readable_device_name}'.") - return await loop.run_in_executor(self._read_executor, read_or_timeout) + response = await loop.run_in_executor(self._read_executor, read_or_timeout) + # Emit on the calling task, not the executor thread, so the operation context survives. + emit_event( + "io.read", + transport="usb", + device=self.human_readable_device_name, + device_id=self._unique_id, + data=response.decode("utf-8", errors="backslashreplace"), + ) + return response def get_available_devices(self) -> List["usb.core.Device"]: """Get a list of available devices that match the specified vendor and product IDs, and serial diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler.py b/pylabrobot/legacy/liquid_handling/liquid_handler.py index d1ac95cc487..bdfdcabedfc 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler.py @@ -24,6 +24,7 @@ cast, ) +from pylabrobot.events import evented_operation, resource_reference from pylabrobot.legacy.liquid_handling.channel_positioning import ( compute_channel_offsets, ) @@ -88,6 +89,36 @@ ] +def _resource_pickup_event_context( + liquid_handler: "LiquidHandler", resource: Resource, **_: Any +) -> Dict[str, Any]: + return { + "device": resource_reference(liquid_handler), + "resources": [resource_reference(resource)], + } + + +def _picked_resource_event_context(liquid_handler: "LiquidHandler", **_: Any) -> Dict[str, Any]: + pickup = liquid_handler._resource_pickup + return { + "device": resource_reference(liquid_handler), + "resources": [] if pickup is None else [resource_reference(pickup.resource)], + } + + +def _resource_drop_event_context( + liquid_handler: "LiquidHandler", + destination: Union[ResourceStack, ResourceHolder, Resource, Coordinate], + **_: Any, +) -> Dict[str, Any]: + context = _picked_resource_event_context(liquid_handler) + if isinstance(destination, Resource): + context["destination"] = resource_reference(destination) + else: + context["destination"] = repr(destination) + return context + + class BlowOutVolumeError(Exception): pass @@ -2035,6 +2066,7 @@ async def stamp( await self.aspirate96(resource=source, volume=volume, flow_rate=aspiration_flow_rate) await self.dispense96(resource=source, volume=volume, flow_rate=dispense_flow_rate) + @evented_operation("liquid_handler.resource_pickup", _resource_pickup_event_context) async def pick_up_resource( self, resource: Resource, @@ -2103,6 +2135,7 @@ async def pick_up_resource( resource.unassign() self._state_updated() + @evented_operation("liquid_handler.resource_move", _picked_resource_event_context) async def move_picked_up_resource( self, to: Coordinate, @@ -2139,6 +2172,7 @@ async def move_picked_up_resource( **backend_kwargs, ) + @evented_operation("liquid_handler.resource_drop", _resource_drop_event_context) async def drop_resource( self, destination: Union[ResourceStack, ResourceHolder, Resource, Coordinate], diff --git a/pylabrobot/legacy/machines/machine.py b/pylabrobot/legacy/machines/machine.py index 16731d30b74..93660f7e92a 100644 --- a/pylabrobot/legacy/machines/machine.py +++ b/pylabrobot/legacy/machines/machine.py @@ -5,6 +5,7 @@ from abc import ABC from typing import Any, Awaitable, Callable, TypeVar +from pylabrobot.events import evented_operation, resource_reference from pylabrobot.legacy.machines.backend import MachineBackend from pylabrobot.serializer import SerializableMixin @@ -17,6 +18,13 @@ _R = TypeVar("_R", bound=Awaitable[Any]) +def _machine_event_context(machine: "Machine", **_: Any) -> dict: + return { + "device": resource_reference(machine), + "backend": type(machine.backend).__name__, + } + + def need_setup_finished(func: Callable[_P, _R]) -> Callable[_P, _R]: """Decorator for methods that require the machine to be set up. @@ -60,11 +68,13 @@ def deserialize(cls, data: dict): data_copy["backend"] = backend return cls(**data_copy) + @evented_operation("machine.setup", _machine_event_context) async def setup(self, **backend_kwargs): await self.backend.setup(**backend_kwargs) self._setup_finished = True @need_setup_finished + @evented_operation("machine.stop", _machine_event_context) async def stop(self): await self.backend.stop() self._setup_finished = False diff --git a/pylabrobot/legacy/storage/incubator.py b/pylabrobot/legacy/storage/incubator.py index 416d319223e..4c48a649ed3 100644 --- a/pylabrobot/legacy/storage/incubator.py +++ b/pylabrobot/legacy/storage/incubator.py @@ -1,6 +1,7 @@ import random from typing import List, Literal, Optional, Union, cast +from pylabrobot.events import evented_operation, resource_reference from pylabrobot.legacy.machines import Machine from pylabrobot.resources import ( Coordinate, @@ -20,6 +21,35 @@ class NoFreeSiteError(Exception): pass +def _fetch_plate_event_context( + incubator: "Incubator", plate_name: str, **_: object +) -> dict: + try: + site = incubator.get_site_by_plate_name(plate_name) + plate = site.resource + except ResourceNotFoundError: + site = None + plate = None + return { + "device": resource_reference(incubator), + "resources": [] if plate is None else [resource_reference(plate)], + "source": resource_reference(site), + "destination": resource_reference(incubator.loading_tray), + } + + +def _take_in_plate_event_context( + incubator: "Incubator", site: Union[PlateHolder, Literal["random", "smallest"]], **_: object +) -> dict: + plate = incubator.loading_tray.resource + return { + "device": resource_reference(incubator), + "resources": [] if plate is None else [resource_reference(plate)], + "source": resource_reference(incubator.loading_tray), + "destination": resource_reference(site) if isinstance(site, PlateHolder) else site, + } + + class Incubator(Machine, Resource): def __init__( self, @@ -73,6 +103,7 @@ def get_site_by_plate_name(self, plate_name: str) -> PlateHolder: return site raise ResourceNotFoundError(f"Plate {plate_name} not found in incubator '{self.name}'") + @evented_operation("incubator.fetch_plate", _fetch_plate_event_context) async def fetch_plate_to_loading_tray(self, plate_name: str, **backend_kwargs) -> Plate: """Fetch a plate from the incubator and put it on the loading tray.""" @@ -112,6 +143,7 @@ def find_smallest_site_for_plate(self, plate: Plate) -> PlateHolder: def find_random_site(self, plate: Plate) -> PlateHolder: return random.choice(self._find_available_sites_sorted(plate)) + @evented_operation("incubator.take_in_plate", _take_in_plate_event_context) async def take_in_plate( self, site: Union[PlateHolder, Literal["random", "smallest"]], **backend_kwargs ): diff --git a/pylabrobot/resources/resource.py b/pylabrobot/resources/resource.py index e93394502f2..f9e9730519c 100644 --- a/pylabrobot/resources/resource.py +++ b/pylabrobot/resources/resource.py @@ -6,6 +6,7 @@ import sys from typing import Any, Callable, Dict, List, Optional, Union, cast +from pylabrobot.events import coordinate_reference, emit_event, resource_reference from pylabrobot.serializer import SerializableMixin, deserialize, serialize from pylabrobot.utils.linalg import matrix_vector_multiply_3x3 from pylabrobot.utils.object_parsing import find_subclass @@ -389,6 +390,13 @@ def assign_child_resource( for callback in self._did_assign_resource_callbacks: callback(resource) + emit_event( + "resource.assigned", + resource=resource_reference(resource), + parent=resource_reference(self), + location=coordinate_reference(resource.location), + ) + def assign_child_by_anchor( self, resource: Resource, @@ -545,6 +553,9 @@ def unassign_child_resource(self, resource: Resource): for callback in self._will_unassign_resource_callbacks: callback(resource) + # Preserve the pose for the event before unassignment clears it. + previous_location = coordinate_reference(resource.location) + # Update the tree structure resource.parent = None resource.location = None @@ -560,6 +571,13 @@ def unassign_child_resource(self, resource: Resource): for callback in self._did_unassign_resource_callbacks: callback(resource) + emit_event( + "resource.unassigned", + resource=resource_reference(resource), + previous_parent=resource_reference(self), + previous_location=previous_location, + ) + def unassign(self): """Unassign this resource from its parent.""" if self.parent is not None: From 48f5a547a3f59e99da72327ce9827b09811e066b Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 18:17:06 -0700 Subject: [PATCH 03/12] Expose default event bus for subscribers --- pylabrobot/events/__init__.py | 2 ++ pylabrobot/events/bus.py | 11 +++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/pylabrobot/events/__init__.py b/pylabrobot/events/__init__.py index 4d223c421bf..1d561596d6c 100644 --- a/pylabrobot/events/__init__.py +++ b/pylabrobot/events/__init__.py @@ -7,6 +7,7 @@ emit_event, event_context, evented_operation, + get_default_event_bus, get_event_bus, is_event_bus_active, resource_reference, @@ -21,6 +22,7 @@ "emit_event", "event_context", "evented_operation", + "get_default_event_bus", "get_event_bus", "is_event_bus_active", "resource_reference", diff --git a/pylabrobot/events/bus.py b/pylabrobot/events/bus.py index 4c837dbbe36..f0194b0a0a8 100644 --- a/pylabrobot/events/bus.py +++ b/pylabrobot/events/bus.py @@ -99,10 +99,17 @@ def emit(self, name: str, *, context: Optional[Dict[str, Any]] = None, **data: A ) -def set_default_event_bus(event_bus: Optional[EventBus]) -> None: - """Install or clear the process-wide event bus for code without an explicit scope.""" +def get_default_event_bus() -> Optional[EventBus]: + """Return the process-wide fallback event bus, if one is installed.""" + return _default_event_bus + + +def set_default_event_bus(event_bus: Optional[EventBus]) -> Optional[EventBus]: + """Install or clear the process-wide event bus and return the previous one.""" global _default_event_bus + previous_event_bus = _default_event_bus _default_event_bus = event_bus + return previous_event_bus def get_event_bus() -> Optional[EventBus]: From 05752ffc41fefc17dca7ff2705094d6928d1ebf4 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 19:11:00 -0700 Subject: [PATCH 04/12] Emit PreciseFlex execution events --- .../brooks/precise_flex/precise_flex.py | 267 +++++++++++++++++- .../precise_flex/tests/precise_flex_tests.py | 45 +++ 2 files changed, 309 insertions(+), 3 deletions(-) diff --git a/pylabrobot/brooks/precise_flex/precise_flex.py b/pylabrobot/brooks/precise_flex/precise_flex.py index e70b8141582..716b03ce264 100644 --- a/pylabrobot/brooks/precise_flex/precise_flex.py +++ b/pylabrobot/brooks/precise_flex/precise_flex.py @@ -10,6 +10,7 @@ from pylabrobot.brooks.precise_flex import kinematics from pylabrobot.brooks.precise_flex.config import Axis, PreciseFlexConfiguration from pylabrobot.brooks.precise_flex.kinematics import JointPose +from pylabrobot.events import coordinate_reference, emit_event, evented_operation from pylabrobot.io.socket import Socket from pylabrobot.resources.coordinate import Coordinate from pylabrobot.resources.rotation import Rotation @@ -33,6 +34,42 @@ BLEND_IN_RANGE = -1 +def _controller_reference(controller: "PreciseFlex") -> dict[str, object]: + """Return the stable controller identity used by structured execution events.""" + return { + "name": "precise_flex", + "type": type(controller).__name__, + "host": controller.io._host, + "port": controller.io._port, + } + + +def _joint_pose_reference(position: JointPose) -> dict[str, float]: + """Convert an axis-keyed joint pose into a JSON-friendly target description.""" + return { + (axis.name.lower() if isinstance(axis, Axis) else str(axis)): float(value) + for axis, value in position.items() + } + + +def _cartesian_target_reference( + location: Coordinate, + direction: float, + *, + orientation: Optional["ElbowOrientation"] = None, + wrist: Optional["Wrist"] = None, + rail_position: Optional[float] = None, +) -> dict[str, object]: + """Describe a Cartesian controller target without serializing a full pose object.""" + return { + "location": coordinate_reference(location), + "direction": float(direction), + "orientation": orientation, + "wrist": wrist, + "rail_position": rail_position, + } + + class MotionProfile(NamedTuple): """A controller motion profile, as reported by ``Profile `` (field order matches the wire).""" @@ -189,9 +226,26 @@ def __init__( # -- communication --------------------------------------------------------- async def send_command(self, command: str) -> str: - await self.io.write(command.encode("utf-8") + b"\n") - reply = await self.io.readline() - return self._parse_reply_ensure_successful(reply) + """Send one firmware command while retaining its enclosing operation context.""" + event_data = { + "device": _controller_reference(self), + "command": command, + } + emit_event("precise_flex.firmware_command.started", **event_data) + try: + await self.io.write(command.encode("utf-8") + b"\n") + reply = await self.io.readline() + result = self._parse_reply_ensure_successful(reply) + except BaseException as error: + emit_event( + "precise_flex.firmware_command.failed", + **event_data, + error_type=type(error).__name__, + error_message=str(error), + ) + raise + emit_event("precise_flex.firmware_command.completed", **event_data, response=result) + return result def _parse_reply_ensure_successful(self, reply: bytes) -> str: """Parse reply from Precise Flex. @@ -215,6 +269,13 @@ def _parse_reply_ensure_successful(self, reply: bytes) -> str: # -- lifecycle ------------------------------------------------------------- + @evented_operation( + "precise_flex.setup", + lambda self, skip_home=False: { + "device": _controller_reference(self), + "skip_home": skip_home, + }, + ) async def setup(self, skip_home: bool = False): """Initialize the PreciseFlex driver. @@ -251,6 +312,10 @@ async def setup(self, skip_home: bool = False): self._assess_configuration(self._configuration) await self._handle_out_of_range_axes() + @evented_operation( + "precise_flex.stop", + lambda self: {"device": _controller_reference(self)}, + ) async def stop(self): """Stop the PreciseFlex driver.""" await self.detach() @@ -307,6 +372,10 @@ async def request_system_state(self) -> int: """ return int(await self.send_command("sysState")) + @evented_operation( + "precise_flex.power_on", + lambda self: {"device": _controller_reference(self)}, + ) async def power_on_robot(self): """Power on the robot.""" error: Optional[PreciseFlexError] = None @@ -323,6 +392,10 @@ async def power_on_robot(self): raise error raise RuntimeError("Failed to power on robot after 3 attempts for unknown reasons.") + @evented_operation( + "precise_flex.recover_from_fault", + lambda self: {"device": _controller_reference(self)}, + ) async def recover_from_fault(self) -> None: """Recover after a collision / fault that stopped the arm and dropped power, leaving it usable. @@ -348,6 +421,10 @@ async def recover_from_fault(self) -> None: await self.attach(1) await self.home() + @evented_operation( + "precise_flex.power_off", + lambda self: {"device": _controller_reference(self)}, + ) async def power_off_robot(self): """Power off the robot.""" await self.set_power(False) @@ -402,6 +479,10 @@ async def detach(self): """Detach the robot.""" await self.attach(0) + @evented_operation( + "precise_flex.home", + lambda self: {"device": _controller_reference(self)}, + ) async def home(self) -> None: """Home the robot associated with this thread. @@ -1134,6 +1215,13 @@ async def zero_torque(self, enable: bool, axis_mask: int = 1) -> None: else: await self.send_command("zeroTorque 0") + @evented_operation( + "precise_flex.start_freedrive", + lambda self, free_axes=None: { + "device": _controller_reference(self), + "free_axes": [int(axis) for axis in free_axes] if free_axes is not None else None, + }, + ) async def start_freedrive_mode(self, free_axes: Optional[List[int]] = None) -> None: """Enter freedrive mode, allowing manual movement of the specified joints. @@ -1154,10 +1242,18 @@ async def start_freedrive_mode(self, free_axes: Optional[List[int]] = None) -> N for axis in free_axes: await self.send_command(f"freemode {axis}") + @evented_operation( + "precise_flex.stop_freedrive", + lambda self: {"device": _controller_reference(self)}, + ) async def stop_freedrive_mode(self) -> None: """Exit freedrive mode for all axes.""" await self.send_command("freemode -1") + @evented_operation( + "precise_flex.halt", + lambda self: {"device": _controller_reference(self)}, + ) async def halt(self): """Stops the current robot immediately but leaves power on.""" await self.send_command("halt") @@ -1941,6 +2037,14 @@ async def request_joint_position(self) -> JointPose: raise PreciseFlexError(-1, "Unexpected response format from wherej command.") return self._parse_angles_response(parts) + @evented_operation( + "precise_flex.move_to_joint_position", + lambda self, position, speed_pct=None: { + "device": _controller_reference(self), + "target_joint_position": _joint_pose_reference(position), + "speed_pct": speed_pct, + }, + ) async def move_to_joint_position( self, position: JointPose, @@ -1965,6 +2069,26 @@ async def request_gripper_pose(self) -> PreciseFlexCartesianPose: # -- cartesian motion --------------------------------------------------------------------- + @evented_operation( + "precise_flex.move_to_location", + lambda self, + location, + direction, + speed_pct=None, + orientation=None, + wrist=None, + rail_position=None: { + "device": _controller_reference(self), + "target": _cartesian_target_reference( + location, + direction, + orientation=orientation, + wrist=wrist, + rail_position=rail_position, + ), + "speed_pct": speed_pct, + }, + ) async def move_to_location( self, location: Coordinate, @@ -2045,6 +2169,37 @@ async def _plan_cartesian_pose_route( prev_pose = cart return targets + @evented_operation( + "precise_flex.move_through_cartesian_poses", + lambda self, poses, speed_pct=None, blend=True: { + "device": _controller_reference(self), + "waypoint_count": len(poses), + "start_target": ( + _cartesian_target_reference( + poses[0].location, + poses[0].rotation.z, + orientation=poses[0].orientation, + wrist=poses[0].wrist, + rail_position=poses[0].rail_position, + ) + if poses + else None + ), + "end_target": ( + _cartesian_target_reference( + poses[-1].location, + poses[-1].rotation.z, + orientation=poses[-1].orientation, + wrist=poses[-1].wrist, + rail_position=poses[-1].rail_position, + ) + if poses + else None + ), + "speed_pct": speed_pct, + "blend": blend, + }, + ) async def move_through_cartesian_poses( self, poses: Sequence[PreciseFlexCartesianPose], @@ -2175,6 +2330,14 @@ async def here_c(self, location_index: int) -> None: _gripper_soft_min: Optional[float] = None _gripper_soft_max: Optional[float] = None + @evented_operation( + "precise_flex.move_gripper", + lambda self, width, force_sensing=False: { + "device": _controller_reference(self), + "width_mm": float(width), + "force_sensing": force_sensing, + }, + ) async def move_gripper( self, width: float, @@ -2215,6 +2378,31 @@ async def move_gripper( await self._set_grip_open_pos(units) await self.send_command("gripper 1") + @evented_operation( + "precise_flex.move_gripper_joint_position", + lambda self, position, force_sensing=False: { + "device": _controller_reference(self), + "gripper_joint_position": float(position), + "force_sensing": force_sensing, + }, + ) + async def move_gripper_joint_position( + self, + position: float, + force_sensing: bool = False, + ) -> None: + """Move the gripper to a controller-native joint position. + + This is the counterpart to :meth:`move_gripper` for integrations with + taught joint-space routes. The caller owns the joint calibration. + """ + if force_sensing: + await self._set_grip_close_pos(position) + await self.send_command("gripper 2") + else: + await self._set_grip_open_pos(position) + await self.send_command("gripper 1") + async def is_gripper_closed(self) -> bool: """(Single Gripper Only) Tests if the gripper is fully closed by checking the end-of-travel sensor. @@ -2238,6 +2426,13 @@ async def are_grippers_closed(self) -> tuple[bool, bool]: # -- rail --------------------------------------------------------------------------------- + @evented_operation( + "precise_flex.move_rail", + lambda self, rail_position: { + "device": _controller_reference(self), + "rail_position": float(rail_position), + }, + ) async def move_rail(self, rail_position: float) -> None: """Move the rail to the specified position. @@ -2254,6 +2449,16 @@ async def move_rail(self, rail_position: float) -> None: # -- pick & place ------------------------------------------------------------------------- + @evented_operation( + "precise_flex.pick_up_at_joint_position", + lambda self, position, resource_width, finger_speed_pct=50.0, grasp_force=10.0: { + "device": _controller_reference(self), + "target_joint_position": _joint_pose_reference(position), + "resource_width_mm": float(resource_width), + "finger_speed_pct": float(finger_speed_pct), + "grasp_force": float(grasp_force), + }, + ) async def pick_up_at_joint_position( self, position: JointPose, @@ -2282,6 +2487,14 @@ async def pick_up_at_joint_position( ) await self._pick_plate_j(position) + @evented_operation( + "precise_flex.drop_at_joint_position", + lambda self, position, resource_width: { + "device": _controller_reference(self), + "target_joint_position": _joint_pose_reference(position), + "resource_width_mm": float(resource_width), + }, + ) async def drop_at_joint_position( self, position: JointPose, @@ -2301,6 +2514,30 @@ async def drop_at_joint_position( ) await self._place_plate_j(position) + @evented_operation( + "precise_flex.pick_up_at_location", + lambda self, + location, + direction, + resource_width, + finger_speed_pct=50.0, + grasp_force=10.0, + orientation=None, + wrist=None, + rail_position=None: { + "device": _controller_reference(self), + "target": _cartesian_target_reference( + location, + direction, + orientation=orientation, + wrist=wrist, + rail_position=rail_position, + ), + "resource_width_mm": float(resource_width), + "finger_speed_pct": float(finger_speed_pct), + "grasp_force": float(grasp_force), + }, + ) async def pick_up_at_location( self, location: Coordinate, @@ -2353,6 +2590,26 @@ async def pick_up_at_location( ) await self._pick_plate_c(cartesian_position=coords) + @evented_operation( + "precise_flex.drop_at_location", + lambda self, + location, + direction, + resource_width, + orientation=None, + wrist=None, + rail_position=None: { + "device": _controller_reference(self), + "target": _cartesian_target_reference( + location, + direction, + orientation=orientation, + wrist=wrist, + rail_position=rail_position, + ), + "resource_width_mm": float(resource_width), + }, + ) async def drop_at_location( self, location: Coordinate, @@ -2442,6 +2699,10 @@ def parking_position(self, position: Optional[JointPose]) -> None: self._validate_parking_position(position) self._parking_position: Optional[JointPose] = dict(position) if position is not None else None + @evented_operation( + "precise_flex.park", + lambda self: {"device": _controller_reference(self)}, + ) async def park(self) -> None: """Move to ``self.parking_position``; defaults at setup, reassignable at runtime. diff --git a/pylabrobot/brooks/precise_flex/tests/precise_flex_tests.py b/pylabrobot/brooks/precise_flex/tests/precise_flex_tests.py index 72516530619..52eda85592c 100644 --- a/pylabrobot/brooks/precise_flex/tests/precise_flex_tests.py +++ b/pylabrobot/brooks/precise_flex/tests/precise_flex_tests.py @@ -8,6 +8,7 @@ PreciseFlex, PreciseFlexCartesianPose, ) +from pylabrobot.events import EventBus, event_context, use_event_bus from pylabrobot.resources import Coordinate, Rotation @@ -87,6 +88,50 @@ def test_mm_to_firmware_units_helper(self): self.assertEqual(self.arm._mm_to_firmware_units(100.0), 540.0) +class TestPreciseFlexEvents(unittest.IsolatedAsyncioTestCase): + async def test_gripper_event_and_nested_firmware_commands_inherit_resource_context(self): + arm = PreciseFlex( + host="localhost", + gripper_length=162.0, + gripper_z_offset=0.0, + closed_gripper_position=500.0, + ) + arm.io.write = AsyncMock() # type: ignore[method-assign] + arm.io.readline = AsyncMock(return_value=b"0\n") # type: ignore[method-assign] + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus), event_context( + resources=[{"name": "sample_plate"}], + source={"name": "source_nest"}, + destination={"name": "destination_nest"}, + ): + await arm.move_gripper_joint_position(520.0) + + self.assertEqual(events[0].name, "precise_flex.move_gripper_joint_position.started") + self.assertEqual(events[-1].name, "precise_flex.move_gripper_joint_position.completed") + self.assertEqual(events[0].data["gripper_joint_position"], 520.0) + self.assertEqual(events[0].data["device"]["name"], "precise_flex") + self.assertEqual(events[0].context["resources"], [{"name": "sample_plate"}]) + self.assertEqual(events[0].context["source"], {"name": "source_nest"}) + self.assertEqual(events[0].context["destination"], {"name": "destination_nest"}) + + firmware_events = [ + event for event in events if event.name == "precise_flex.firmware_command.started" + ] + self.assertEqual( + [event.data["command"] for event in firmware_events], + [ + "GripOpenPos 520.0", + "gripper 1", + ], + ) + self.assertTrue( + all(event.context["resources"] == [{"name": "sample_plate"}] for event in firmware_events) + ) + + class TestPreciseFlex400OutOfRangeRecovery(unittest.IsolatedAsyncioTestCase): def setUp(self): self.arm = _make_arm() From a7d83eb5d01b2f6596315e95aa2c89ee961d1691 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 19:43:37 -0700 Subject: [PATCH 05/12] Emit liquid handling execution events --- .../legacy/liquid_handling/liquid_handler.py | 69 +++++++++++++++++++ .../liquid_handling/liquid_handler_tests.py | 30 ++++++++ 2 files changed, 99 insertions(+) diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler.py b/pylabrobot/legacy/liquid_handling/liquid_handler.py index bdfdcabedfc..e8836733ddd 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler.py @@ -119,6 +119,73 @@ def _resource_drop_event_context( return context +def _liquid_operation_plate(resource: Container) -> Resource: + """Return the plate that owns a well, or the container itself when it is not plate-backed.""" + + current: Optional[Resource] = resource + while current is not None: + if isinstance(current, Plate): + return current + current = current.parent + return resource + + +def _safe_event_volume(value: Any) -> Any: + """Normalize a volume for event consumers without changing liquid-handler validation.""" + + try: + return float(value) + except (TypeError, ValueError): + return repr(value) + + +def _liquid_operation_event_context( + liquid_handler: "LiquidHandler", + resources: Sequence[Container], + vols: Sequence[Any], + use_channels: Optional[List[int]] = None, + **_: Any, +) -> Dict[str, Any]: + """Describe requested liquid operations using plate lanes and per-channel well detail.""" + + resource_list = list(resources) + channels = use_channels or liquid_handler._default_use_channels or list(range(len(resource_list))) + operation_resources = resource_list + if len(operation_resources) == 1 and len(channels) > 1: + operation_resources = operation_resources * len(channels) + + plate_resources = [_liquid_operation_plate(resource) for resource in resource_list] + unique_plate_resources: List[Dict[str, Any]] = [] + seen_plate_names: Set[str] = set() + for plate_resource in plate_resources: + reference = resource_reference(plate_resource) + if reference is None: + continue + name = reference.get("name") + if not isinstance(name, str) or name in seen_plate_names: + continue + seen_plate_names.add(name) + unique_plate_resources.append(reference) + + liquid_operations = [] + for channel, resource, volume in zip(channels, operation_resources, vols): + liquid_operations.append( + { + "channel": channel, + "resource": resource_reference(resource), + "plate": resource_reference(_liquid_operation_plate(resource)), + "volume_ul": _safe_event_volume(volume), + } + ) + + return { + "device": resource_reference(liquid_handler), + # Gantt lanes are plate-level; exact well information remains available per channel below. + "resources": unique_plate_resources, + "liquid_operations": liquid_operations, + } + + class BlowOutVolumeError(Exception): pass @@ -905,6 +972,7 @@ def _check_containers(self, resources: Sequence[Resource]): if len(not_containers) > 0: raise TypeError(f"Resources must be `Container`s, got {not_containers}") + @evented_operation("liquid_handler.aspirate", _liquid_operation_event_context) @need_setup_finished async def aspirate( self, @@ -1097,6 +1165,7 @@ async def aspirate( if error is not None: raise error + @evented_operation("liquid_handler.dispense", _liquid_operation_event_context) @need_setup_finished async def dispense( self, diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py index 7bce30a164e..f639ef99cda 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py @@ -7,6 +7,7 @@ import pytest +from pylabrobot.events import EventBus, use_event_bus from pylabrobot.legacy.liquid_handling.backends.backend import LiquidHandlerBackend from pylabrobot.legacy.liquid_handling.backends.chatterbox import LiquidHandlerChatterboxBackend from pylabrobot.legacy.liquid_handling.channel_positioning import ( @@ -622,6 +623,35 @@ async def test_offsets_asp_disp(self): ops=[_make_disp(well, vol=10, offset=Coordinate(x=1, y=1, z=1), tip=t)], ) + async def test_aspirate_and_dispense_emit_plate_aware_events(self): + well = self.plate.get_item("A1") + well.tracker.set_volume(20) + self.lh.update_head_state({0: self.tip_rack.get_item("A1").get_tip()}) + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus): + await self.lh.aspirate([well], vols=[10]) + await self.lh.dispense([well], vols=[10]) + + self.assertEqual( + [event.name for event in events], + [ + "liquid_handler.aspirate.started", + "liquid_handler.aspirate.completed", + "liquid_handler.dispense.started", + "liquid_handler.dispense.completed", + ], + ) + context = events[0].context + self.assertEqual([resource["name"] for resource in context["resources"]], ["plate"]) + operation = context["liquid_operations"][0] + self.assertEqual(operation["channel"], 0) + self.assertEqual(operation["resource"]["name"], well.name) + self.assertEqual(operation["plate"]["name"], "plate") + self.assertEqual(operation["volume_ul"], 10.0) + async def test_return_tips(self): tip_spot = self.tip_rack.get_item("A1") tip = tip_spot.get_tip() From ad369c6b97994518988097b4d6ee7f4a31beddc8 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 22:33:01 -0700 Subject: [PATCH 06/12] fix: preserve direct liquid operation resources in events --- pylabrobot/events/bus.py | 21 +++++++++++++++++- .../legacy/liquid_handling/liquid_handler.py | 22 +++++++++---------- .../liquid_handling/liquid_handler_tests.py | 11 ++++++++-- 3 files changed, 39 insertions(+), 15 deletions(-) diff --git a/pylabrobot/events/bus.py b/pylabrobot/events/bus.py index f0194b0a0a8..7c179bbcaf3 100644 --- a/pylabrobot/events/bus.py +++ b/pylabrobot/events/bus.py @@ -154,7 +154,12 @@ def emit_event(name: str, **data: Any) -> Optional[PLREvent]: def resource_reference(resource: Any) -> Optional[Dict[str, Any]]: - """Return a compact resource identity without serializing its full subtree.""" + """Return a compact resource identity and its assigned-resource ancestry. + + The reference always identifies the resource directly involved in an operation. Its + ancestry is structural context only: consumers can, for example, present a well on + the lane for its owning plate without replacing the well in the event payload. + """ if resource is None: return None result: Dict[str, Any] = { @@ -171,6 +176,20 @@ def resource_reference(resource: Any) -> Optional[Dict[str, Any]]: "y": rotation.y, "z": rotation.z, } + ancestors = [] + current = getattr(resource, "parent", None) + while current is not None: + ancestor: Dict[str, Any] = { + "name": getattr(current, "name", None), + "type": type(current).__name__, + } + ancestor_model = getattr(current, "model", None) + if ancestor_model is not None: + ancestor["model"] = str(ancestor_model) + ancestors.append(ancestor) + current = getattr(current, "parent", None) + if ancestors: + result["ancestors"] = ancestors return result diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler.py b/pylabrobot/legacy/liquid_handling/liquid_handler.py index e8836733ddd..69d5b0012b8 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler.py @@ -146,7 +146,7 @@ def _liquid_operation_event_context( use_channels: Optional[List[int]] = None, **_: Any, ) -> Dict[str, Any]: - """Describe requested liquid operations using plate lanes and per-channel well detail.""" + """Describe requested liquid operations using their directly operated containers.""" resource_list = list(resources) channels = use_channels or liquid_handler._default_use_channels or list(range(len(resource_list))) @@ -154,18 +154,17 @@ def _liquid_operation_event_context( if len(operation_resources) == 1 and len(channels) > 1: operation_resources = operation_resources * len(channels) - plate_resources = [_liquid_operation_plate(resource) for resource in resource_list] - unique_plate_resources: List[Dict[str, Any]] = [] - seen_plate_names: Set[str] = set() - for plate_resource in plate_resources: - reference = resource_reference(plate_resource) + unique_operation_resources: List[Dict[str, Any]] = [] + seen_resource_keys: Set[Tuple[Any, Any]] = set() + for operation_resource in operation_resources: + reference = resource_reference(operation_resource) if reference is None: continue - name = reference.get("name") - if not isinstance(name, str) or name in seen_plate_names: + key = (reference.get("type"), reference.get("name")) + if key in seen_resource_keys: continue - seen_plate_names.add(name) - unique_plate_resources.append(reference) + seen_resource_keys.add(key) + unique_operation_resources.append(reference) liquid_operations = [] for channel, resource, volume in zip(channels, operation_resources, vols): @@ -180,8 +179,7 @@ def _liquid_operation_event_context( return { "device": resource_reference(liquid_handler), - # Gantt lanes are plate-level; exact well information remains available per channel below. - "resources": unique_plate_resources, + "resources": unique_operation_resources, "liquid_operations": liquid_operations, } diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py index f639ef99cda..c4b401aef3f 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py @@ -623,7 +623,7 @@ async def test_offsets_asp_disp(self): ops=[_make_disp(well, vol=10, offset=Coordinate(x=1, y=1, z=1), tip=t)], ) - async def test_aspirate_and_dispense_emit_plate_aware_events(self): + async def test_aspirate_and_dispense_emit_well_accurate_events(self): well = self.plate.get_item("A1") well.tracker.set_volume(20) self.lh.update_head_state({0: self.tip_rack.get_item("A1").get_tip()}) @@ -645,7 +645,14 @@ async def test_aspirate_and_dispense_emit_plate_aware_events(self): ], ) context = events[0].context - self.assertEqual([resource["name"] for resource in context["resources"]], ["plate"]) + self.assertEqual([resource["name"] for resource in context["resources"]], [well.name]) + self.assertEqual(context["resources"][0]["type"], "Well") + self.assertTrue( + any( + ancestor["name"] == "plate" and ancestor["type"] == "Plate" + for ancestor in context["resources"][0]["ancestors"] + ) + ) operation = context["liquid_operations"][0] self.assertEqual(operation["channel"], 0) self.assertEqual(operation["resource"]["name"], well.name) From 4d385f7a432d130a257d1a5cb556732d7cdfab8b Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 22:56:01 -0700 Subject: [PATCH 07/12] feat: add semantic event operation scopes --- pylabrobot/events/__init__.py | 2 ++ pylabrobot/events/bus.py | 55 ++++++++++++++++++++++++---------- pylabrobot/events/bus_tests.py | 43 +++++++++++++++++++++++++- 3 files changed, 84 insertions(+), 16 deletions(-) diff --git a/pylabrobot/events/__init__.py b/pylabrobot/events/__init__.py index 1d561596d6c..270d4beacc3 100644 --- a/pylabrobot/events/__init__.py +++ b/pylabrobot/events/__init__.py @@ -6,6 +6,7 @@ coordinate_reference, emit_event, event_context, + event_operation, evented_operation, get_default_event_bus, get_event_bus, @@ -21,6 +22,7 @@ "coordinate_reference", "emit_event", "event_context", + "event_operation", "evented_operation", "get_default_event_bus", "get_event_bus", diff --git a/pylabrobot/events/bus.py b/pylabrobot/events/bus.py index 7c179bbcaf3..e172128b15b 100644 --- a/pylabrobot/events/bus.py +++ b/pylabrobot/events/bus.py @@ -42,6 +42,7 @@ def as_dict(self) -> Dict[str, Any]: EventListener = Callable[[PLREvent], None] OperationContextFactory = Callable[..., Dict[str, Any]] +CompletionDataFactory = Callable[[], Dict[str, Any]] class EventBus: @@ -200,6 +201,43 @@ def coordinate_reference(coordinate: Any) -> Optional[Dict[str, float]]: return {"x": coordinate.x, "y": coordinate.y, "z": coordinate.z} +@contextmanager +def event_operation( + name: str, + *, + completed_data_factory: CompletionDataFactory | None = None, + **operation_data: Any, +) -> Iterator[None]: + """Emit correlated lifecycle events around a semantic operation block. + + This is the block-oriented counterpart to :func:`evented_operation`, for + integrations whose operation spans several frontend calls rather than one + decorated method. ``completed_data_factory`` can capture the final resource + state after a successful operation; the started event always describes the + invocation state. + """ + + if not is_event_bus_active(): + yield + return + + operation_id = uuid.uuid4().hex + with event_context(operation=name, operation_id=operation_id, **operation_data): + emit_event(f"{name}.started", **operation_data) + try: + yield + except BaseException as error: + emit_event( + f"{name}.failed", + **operation_data, + error_type=type(error).__name__, + error_message=str(error), + ) + raise + completed_data = completed_data_factory() if completed_data_factory is not None else operation_data + emit_event(f"{name}.completed", **completed_data) + + def evented_operation( name: str, context_factory: OperationContextFactory ) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]: @@ -216,21 +254,8 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any: return await func(*args, **kwargs) operation_data = context_factory(*args, **kwargs) - operation_id = uuid.uuid4().hex - with event_context(operation=name, operation_id=operation_id, **operation_data): - emit_event(f"{name}.started", **operation_data) - try: - result = await func(*args, **kwargs) - except BaseException as error: - emit_event( - f"{name}.failed", - **operation_data, - error_type=type(error).__name__, - error_message=str(error), - ) - raise - emit_event(f"{name}.completed", **operation_data) - return result + with event_operation(name, **operation_data): + return await func(*args, **kwargs) return wrapper diff --git a/pylabrobot/events/bus_tests.py b/pylabrobot/events/bus_tests.py index 4bec0cf8143..3ff98d3f95b 100644 --- a/pylabrobot/events/bus_tests.py +++ b/pylabrobot/events/bus_tests.py @@ -1,7 +1,14 @@ import json import unittest -from pylabrobot.events import EventBus, emit_event, event_context, evented_operation, use_event_bus +from pylabrobot.events import ( + EventBus, + emit_event, + event_context, + event_operation, + evented_operation, + use_event_bus, +) from pylabrobot.resources import Coordinate, Resource @@ -42,6 +49,40 @@ def test_resource_assignment_and_unassignment_emit_contextual_events(self): class TestEventedOperation(unittest.IsolatedAsyncioTestCase): + def test_block_operation_emits_a_correlated_lifecycle(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus), event_operation("device.action", device="device-1"): + emit_event("firmware.command.started", command="AB") + + self.assertEqual( + [event.name for event in events], + ["device.action.started", "firmware.command.started", "device.action.completed"], + ) + operation_id = events[0].context["operation_id"] + self.assertEqual(events[1].context["operation_id"], operation_id) + self.assertEqual(events[2].context["operation_id"], operation_id) + + def test_block_operation_can_capture_final_completion_data(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + location = {"name": "source"} + + with use_event_bus(event_bus), event_operation( + "resource.transfer", + resources=[{"name": "plate", "location": location["name"]}], + completed_data_factory=lambda: { + "resources": [{"name": "plate", "location": location["name"]}], + }, + ): + location["name"] = "destination" + + self.assertEqual(events[0].data["resources"][0]["location"], "source") + self.assertEqual(events[1].data["resources"][0]["location"], "destination") + async def test_operation_context_is_inherited_by_nested_events(self): events = [] event_bus = EventBus() From d09dfb02985208a00841109d1d04d34689e84d67 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 23:02:35 -0700 Subject: [PATCH 08/12] feat: emit semantic liquid handler tip events --- .../legacy/liquid_handling/liquid_handler.py | 51 +++++++++++++++++ .../liquid_handling/liquid_handler_tests.py | 56 +++++++++++++++++++ 2 files changed, 107 insertions(+) diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler.py b/pylabrobot/legacy/liquid_handling/liquid_handler.py index 69d5b0012b8..6e671e7a9e3 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler.py @@ -184,6 +184,53 @@ def _liquid_operation_event_context( } +def _tip_operation_event_context( + liquid_handler: "LiquidHandler", + tip_spots: Sequence[Union[TipSpot, Trash]], + use_channels: Optional[List[int]] = None, + **_: Any, +) -> Dict[str, Any]: + """Describe channel tip operations using their direct pickup/drop resources.""" + + resources: List[Dict[str, Any]] = [] + seen_resource_keys: Set[Tuple[Any, Any]] = set() + for tip_spot in tip_spots: + reference = resource_reference(tip_spot) + if reference is None: + continue + key = (reference.get("type"), reference.get("name")) + if key in seen_resource_keys: + continue + seen_resource_keys.add(key) + resources.append(reference) + + channels = use_channels or liquid_handler._default_use_channels or list(range(len(tip_spots))) + tip_operations = [ + {"channel": channel, "resource": resource_reference(tip_spot)} + for channel, tip_spot in zip(channels, tip_spots) + ] + return { + "device": resource_reference(liquid_handler), + "resources": resources, + "tip_operations": tip_operations, + } + + +def _tip_rack_operation_event_context( + liquid_handler: "LiquidHandler", + tip_rack: Optional[TipRack] = None, + resource: Optional[Union[TipRack, Trash]] = None, + **_: Any, +) -> Dict[str, Any]: + """Describe a 96-head operation by its directly operated rack or trash resource.""" + + operation_resource = tip_rack if tip_rack is not None else resource + return { + "device": resource_reference(liquid_handler), + "resources": [] if operation_resource is None else [resource_reference(operation_resource)], + } + + class BlowOutVolumeError(Exception): pass @@ -530,6 +577,7 @@ def get_picked_up_resource(self) -> Optional[Resource]: return None return self._resource_pickup.resource + @evented_operation("liquid_handler.tip_pickup", _tip_operation_event_context) @need_setup_finished async def pick_up_tips( self, @@ -679,6 +727,7 @@ def get_mounted_tips(self) -> List[Optional[Tip]]: """ return [tracker.get_tip() if tracker.has_tip else None for tracker in self.head.values()] + @evented_operation("liquid_handler.tip_drop", _tip_operation_event_context) @need_setup_finished async def drop_tips( self, @@ -1543,6 +1592,7 @@ async def use_tips( else: await self.return_tips(use_channels=channels) + @evented_operation("liquid_handler.tip_pickup_96", _tip_rack_operation_event_context) async def pick_up_tips96( self, tip_rack: TipRack, @@ -1612,6 +1662,7 @@ async def pick_up_tips96( tip_spot.tracker.commit() self.head96[i].commit() + @evented_operation("liquid_handler.tip_drop_96", _tip_rack_operation_event_context) async def drop_tips96( self, resource: Union[TipRack, Trash], diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py index c4b401aef3f..c30392e7d56 100644 --- a/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py +++ b/pylabrobot/legacy/liquid_handling/liquid_handler_tests.py @@ -659,6 +659,62 @@ async def test_aspirate_and_dispense_emit_well_accurate_events(self): self.assertEqual(operation["plate"]["name"], "plate") self.assertEqual(operation["volume_ul"], 10.0) + async def test_tip_pickup_and_discard_emit_direct_tip_resources(self): + tip_spot = self.tip_rack.get_item("A1") + trash = self.deck.get_trash_area() + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus): + await self.lh.pick_up_tips([tip_spot]) + await self.lh.discard_tips() + + self.assertEqual( + [event.name for event in events], + [ + "liquid_handler.tip_pickup.started", + "liquid_handler.tip_pickup.completed", + "liquid_handler.tip_drop.started", + "liquid_handler.tip_drop.completed", + ], + ) + pickup = events[0].context + self.assertEqual(pickup["resources"][0]["name"], tip_spot.name) + self.assertEqual(pickup["resources"][0]["type"], "TipSpot") + self.assertTrue( + any( + ancestor["name"] == self.tip_rack.name and ancestor["type"] == "TipRack" + for ancestor in pickup["resources"][0]["ancestors"] + ) + ) + self.assertEqual(pickup["tip_operations"], [{"channel": 0, "resource": pickup["resources"][0]}]) + discard = events[2].context + self.assertEqual(discard["resources"][0]["name"], trash.name) + self.assertEqual(discard["resources"][0]["type"], "Trash") + self.assertEqual(discard["tip_operations"][0]["resource"], discard["resources"][0]) + + async def test_96_head_tip_operations_emit_direct_rack_resources(self): + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus): + await self.lh.pick_up_tips96(self.tip_rack) + await self.lh.drop_tips96(self.tip_rack) + + self.assertEqual( + [event.name for event in events], + [ + "liquid_handler.tip_pickup_96.started", + "liquid_handler.tip_pickup_96.completed", + "liquid_handler.tip_drop_96.started", + "liquid_handler.tip_drop_96.completed", + ], + ) + self.assertEqual(events[0].context["resources"][0]["name"], self.tip_rack.name) + self.assertEqual(events[0].context["resources"][0]["type"], "TipRack") + async def test_return_tips(self): tip_spot = self.tip_rack.get_item("A1") tip = tip_spot.get_tip() From 618ce238ab026bf40105d9e7e3a766b13c1a0099 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Fri, 7 Aug 2026 23:11:24 -0700 Subject: [PATCH 09/12] feat: emit shaker and temperature controller events --- pylabrobot/legacy/shaking/shaker.py | 21 +++++++++++- pylabrobot/legacy/shaking/shaker_tests.py | 33 ++++++++++++++++++ .../temperature_controller.py | 29 +++++++++++++++- .../temperature_controller_tests.py | 34 +++++++++++++++++++ 4 files changed, 115 insertions(+), 2 deletions(-) diff --git a/pylabrobot/legacy/shaking/shaker.py b/pylabrobot/legacy/shaking/shaker.py index 78ee93fca82..0c8476e6353 100644 --- a/pylabrobot/legacy/shaking/shaker.py +++ b/pylabrobot/legacy/shaking/shaker.py @@ -1,12 +1,29 @@ import asyncio -from typing import Optional +from typing import Any, Optional +from pylabrobot.events import evented_operation, resource_reference from pylabrobot.legacy.machines.machine import Machine from pylabrobot.resources import Coordinate, ResourceHolder from .backend import ShakerBackend +def _shaker_event_context( + shaker: "Shaker", + speed: Optional[float] = None, + duration: Optional[float] = None, + **_: Any, +) -> dict[str, Any]: + """Describe a shaker operation without inferring a plate association.""" + + context: dict[str, Any] = {"device": resource_reference(shaker)} + if speed is not None: + context["speed_rpm"] = float(speed) + if duration is not None: + context["duration_seconds"] = float(duration) + return context + + class Shaker(ResourceHolder, Machine): """A shaker machine""" @@ -34,6 +51,7 @@ def __init__( Machine.__init__(self, backend=backend) self.backend: ShakerBackend = backend # fix type + @evented_operation("shaker.shake", _shaker_event_context) async def shake(self, speed: float, duration: Optional[float] = None, **backend_kwargs): """Shake the shaker at the given speed @@ -53,6 +71,7 @@ async def shake(self, speed: float, duration: Optional[float] = None, **backend_ if self.backend.supports_locking: await self.backend.unlock_plate() + @evented_operation("shaker.stop_shaking", _shaker_event_context) async def stop_shaking(self, **backend_kwargs): await self.backend.stop_shaking(**backend_kwargs) diff --git a/pylabrobot/legacy/shaking/shaker_tests.py b/pylabrobot/legacy/shaking/shaker_tests.py index 4a1f391a4ca..9296237e6fc 100644 --- a/pylabrobot/legacy/shaking/shaker_tests.py +++ b/pylabrobot/legacy/shaking/shaker_tests.py @@ -1,5 +1,6 @@ import unittest +from pylabrobot.events import EventBus, use_event_bus from pylabrobot.legacy.shaking import Shaker, ShakerChatterboxBackend from pylabrobot.resources.coordinate import Coordinate @@ -18,3 +19,35 @@ def test_serialization(self): serialized = s.serialize() deserialized = Shaker.deserialize(serialized) self.assertEqual(s, deserialized) + + +class ShakerEventTests(unittest.IsolatedAsyncioTestCase): + async def test_shake_and_stop_emit_device_scoped_events(self): + shaker = Shaker( + name="test_shaker", + size_x=10, + size_y=10, + size_z=10, + backend=ShakerChatterboxBackend(), + child_location=Coordinate.zero(), + ) + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus): + await shaker.shake(speed=900, duration=0) + await shaker.stop_shaking() + + self.assertEqual( + [event.name for event in events], + [ + "shaker.shake.started", + "shaker.shake.completed", + "shaker.stop_shaking.started", + "shaker.stop_shaking.completed", + ], + ) + self.assertEqual(events[0].context["device"]["name"], "test_shaker") + self.assertEqual(events[0].context["speed_rpm"], 900.0) + self.assertEqual(events[0].context["duration_seconds"], 0.0) diff --git a/pylabrobot/legacy/temperature_controlling/temperature_controller.py b/pylabrobot/legacy/temperature_controlling/temperature_controller.py index 85a8c52e2fa..61570acf897 100644 --- a/pylabrobot/legacy/temperature_controlling/temperature_controller.py +++ b/pylabrobot/legacy/temperature_controlling/temperature_controller.py @@ -1,13 +1,37 @@ import asyncio import time -from typing import Optional +from typing import Any, Optional +from pylabrobot.events import evented_operation, resource_reference from pylabrobot.legacy.machines.machine import Machine from pylabrobot.resources import Coordinate, ResourceHolder from .backend import TemperatureControllerBackend +def _temperature_event_context( + temperature_controller: "TemperatureController", + temperature: Optional[float] = None, + passive: Optional[bool] = None, + timeout: Optional[float] = None, + tolerance: Optional[float] = None, + **_: Any, +) -> dict[str, Any]: + """Describe a thermal-device command without inferring a plate association.""" + + context: dict[str, Any] = {"device": resource_reference(temperature_controller)} + target_temperature = temperature_controller.target_temperature if temperature is None else temperature + if target_temperature is not None: + context["target_temperature_c"] = float(target_temperature) + if passive is not None: + context["passive"] = passive + if timeout is not None: + context["timeout_seconds"] = float(timeout) + if tolerance is not None: + context["tolerance_c"] = float(tolerance) + return context + + class TemperatureController(ResourceHolder, Machine): """Temperature controller, for heating or for cooling.""" @@ -36,6 +60,7 @@ def __init__( self.backend: TemperatureControllerBackend = backend # fix type self.target_temperature: Optional[float] = None + @evented_operation("temperature_controller.set_temperature", _temperature_event_context) async def set_temperature(self, temperature: float, passive: bool = False): """Set the temperature of the temperature controller. @@ -68,6 +93,7 @@ async def get_temperature(self) -> float: """Get the current temperature of the temperature controller in Celsius.""" return await self.backend.get_current_temperature() + @evented_operation("temperature_controller.wait_for_temperature", _temperature_event_context) async def wait_for_temperature(self, timeout: float = 300.0, tolerance: float = 0.5) -> None: """Wait for the temperature to reach the target temperature. The target temperature must be set by `set_temperature()`. @@ -86,6 +112,7 @@ async def wait_for_temperature(self, timeout: float = 300.0, tolerance: float = await asyncio.sleep(1.0) raise TimeoutError(f"Temperature did not reach target temperature within {timeout} seconds.") + @evented_operation("temperature_controller.deactivate", _temperature_event_context) async def deactivate(self): """Deactivate the temperature controller. This will stop the heating or cooling, and return the temperature to ambient temperature. The target temperature will be reset to `None`. diff --git a/pylabrobot/legacy/temperature_controlling/temperature_controller_tests.py b/pylabrobot/legacy/temperature_controlling/temperature_controller_tests.py index 14f8b9a1787..49fe2f3fda8 100644 --- a/pylabrobot/legacy/temperature_controlling/temperature_controller_tests.py +++ b/pylabrobot/legacy/temperature_controlling/temperature_controller_tests.py @@ -1,5 +1,6 @@ import unittest +from pylabrobot.events import EventBus, use_event_bus from pylabrobot.legacy.temperature_controlling import ( TemperatureController, TemperatureControllerChatterboxBackend, @@ -55,6 +56,39 @@ async def test_passive_cooling_without_support(self): self.assertEqual(await backend.get_current_temperature(), 20.0) +class TemperatureControllerEventTests(unittest.IsolatedAsyncioTestCase): + async def test_set_and_wait_for_temperature_emit_device_scoped_events(self): + temperature_controller = TemperatureController( + name="test_temperature_module", + size_x=1, + size_y=1, + size_z=1, + backend=TemperatureControllerChatterboxBackend(dummy_temperature=20.0), + child_location=Coordinate.zero(), + ) + events = [] + event_bus = EventBus() + event_bus.subscribe(events.append) + + with use_event_bus(event_bus): + await temperature_controller.set_temperature(37.0) + await temperature_controller.wait_for_temperature(timeout=1.0, tolerance=0.5) + + self.assertEqual( + [event.name for event in events], + [ + "temperature_controller.set_temperature.started", + "temperature_controller.set_temperature.completed", + "temperature_controller.wait_for_temperature.started", + "temperature_controller.wait_for_temperature.completed", + ], + ) + self.assertEqual(events[0].context["device"]["name"], "test_temperature_module") + self.assertEqual(events[0].context["target_temperature_c"], 37.0) + self.assertEqual(events[2].context["timeout_seconds"], 1.0) + self.assertEqual(events[2].context["tolerance_c"], 0.5) + + class _FakeBackend(TemperatureControllerBackend): def __init__(self, temperature: float = 25.0): super().__init__() From 75f42319a731a2ec12e10733025c75e3630b3b67 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Mon, 10 Aug 2026 18:30:02 -0700 Subject: [PATCH 10/12] docs: define EventBus operation semantics --- docs/contributor_guide/event-bus.md | 233 ++++++++++++++++++++++++++++ docs/contributor_guide/index.md | 1 + 2 files changed, 234 insertions(+) create mode 100644 docs/contributor_guide/event-bus.md diff --git a/docs/contributor_guide/event-bus.md b/docs/contributor_guide/event-bus.md new file mode 100644 index 00000000000..984648334f1 --- /dev/null +++ b/docs/contributor_guide/event-bus.md @@ -0,0 +1,233 @@ +# EventBus Operation Semantics + +The PLR EventBus provides optional, structured execution events for applications that need +observability without parsing device logs. This guide defines the contract for contributors +instrumenting a frontend or driver. + +The EventBus is intentionally in-process and synchronous. It is an observation mechanism, +not a control mechanism: listener failures must never affect hardware control flow. + +## When to emit events + +Instrument **semantic public operations**: an operation that a protocol author would recognize +as one action, such as fetching a plate, aspirating, shaking, or moving an arm to a location. +Do not instrument every serial, USB, transport, or firmware command as a semantic event. +Those lower-level details remain available through normal device logging for diagnostics. + +An instrumented method remains a no-op with respect to events unless an EventBus with at least +one subscriber is active. Use one of these helpers: + +- `@evented_operation(...)` for one public async method. +- `with event_operation(...):` for one logical operation implemented by several calls. +- `emit_event(...)` only for a meaningful state transition that is not an operation lifecycle. + +## Universal event contract + +A `PLREvent` always contains: + +```python +{ + "sequence": 42, + "name": "liquid_handler.aspirate.completed", + "timestamp": "2026-08-10T12:34:56.789012+00:00", + "context": {...}, + "data": {...}, +} +``` + +Use names in this form: + +```text +.. +``` + +Examples from the current implementation: + +```text +incubator.fetch_plate.started +liquid_handler.resource_pickup.completed +liquid_handler.tip_pickup.failed +shaker.shake.completed +temperature_controller.wait_for_temperature.completed +precise_flex.move_through_cartesian_poses.completed +``` + +Every operation scope emits exactly one correlated lifecycle sequence: + +```text +.started -> .completed +.started -> .failed +``` + +The EventBus adds these values to `context` for every lifecycle event: + +- `operation`: `.` +- `operation_id`: a UUID shared by the lifecycle sequence + +Callers may add higher-level execution context with `event_context(...)`; for example, a batch +identifier or protocol run identifier. Device code must not invent protocol-specific context. + +## Required fields for semantic device operations + +Every instrumented hardware operation should provide: + +- `device`: `resource_reference()` for the device or controller issuing the operation. +- `resources`: direct PLR resources acted on by the operation, represented with + `resource_reference()`. + +`resources` describes the literal object operated on. Do not replace a `Well` with its owning +`Plate`, or a `TipSpot` with its parent `TipRack`, merely for a downstream display. The +reference includes `ancestors` so a consumer can make that presentation choice without changing +the event's meaning. + +Use `source` and `destination` only when the operation genuinely transfers a resource between +physical locations. Represent a resource endpoint with `resource_reference()`. Use +`coordinate_reference()` for a geometric endpoint when no PLR resource exists. + +Avoid adding fields solely to simplify one consumer. An event should describe what the PLR API +actually did; dashboards, logs, and integrations can derive their own views from the structured +references. + +## Operation templates + +### Machine lifecycle + +Use for `setup()` and `stop()` on a public PLR machine frontend. + +```python +@evented_operation( + "machine.setup", + lambda self, **_: {"device": resource_reference(self), "resources": []}, +) +async def setup(self, **backend_kwargs): + ... +``` + +Current example: `legacy.machines.Machine.setup` and `.stop`. + +### Resource transfer + +Use for a plate, lid, carrier, or other resource transfer. The direct moved resource is listed in +`resources`; locations are named separately. + +```python +with event_operation( + "incubator.fetch_plate", + device=resource_reference(self), + resources=[resource_reference(plate)], + source=resource_reference(site), + destination=resource_reference(self.loading_tray), +): + await self.backend.fetch_plate_to_loading_tray(plate) +``` + +Current examples: + +- `legacy.storage.Incubator.fetch_plate_to_loading_tray` +- `legacy.storage.Incubator.take_in_plate` +- `legacy.liquid_handling.LiquidHandler.resource_pickup`, `.resource_move`, and `.resource_drop` + +For pickup and drop, record the resource's invocation state in `.started`. A +`completed_data_factory` may capture its final assignment or pose for `.completed`. + +### Liquid handling + +Use the direct operated containers in `resources`, plus one `liquid_operations` record per +channel. Each item should include the channel, direct resource reference, owning plate reference +when applicable, and `volume_ul`. + +```python +{ + "device": resource_reference(liquid_handler), + "resources": [resource_reference(well)], + "liquid_operations": [{ + "channel": 0, + "resource": resource_reference(well), + "plate": resource_reference(well.parent), + "volume_ul": 50.0, + }], +} +``` + +Current examples: `legacy.liquid_handling.LiquidHandler.aspirate` and `.dispense`. + +### Tip handling + +Report each direct `TipSpot`, `TipRack`, or `Trash` resource. For channelized tip actions, +include `tip_operations` with the channel and direct resource. + +```python +{ + "device": resource_reference(liquid_handler), + "resources": [resource_reference(tip_spot)], + "tip_operations": [{"channel": 0, "resource": resource_reference(tip_spot)}], +} +``` + +Current examples: `LiquidHandler.pick_up_tips`, `.drop_tips`, `.pick_up_tips96`, and +`.drop_tips96`. + +### Thermal and shaking operations + +Represent the issuing controller as `device`. Include an operated resource only when the PLR API +has one. Use explicit unit-suffixed fields for operation parameters: + +```python +{ + "device": resource_reference(controller), + "resources": [], + "temperature_c": 37.0, + "duration_s": 300.0, + "speed_rpm": 800.0, +} +``` + +Current examples: + +- `legacy.shaking.Shaker.shake` and `.stop_shaking` +- `legacy.temperature_controlling.TemperatureController.set_temperature`, + `.wait_for_temperature`, and `.deactivate` + +### Arm/controller motion + +Public controller operations should identify the controller in `device` and use explicit, +unit-bearing motion arguments where relevant. Low-level controller operations can be useful to a +diagnostic listener, but higher-level resource-aware wrappers should emit the resource-transfer +events when an arm is actually approaching, picking up, moving, or dropping a PLR resource. + +Current examples: `brooks.precise_flex.PreciseFlex` lifecycle, joint/cartesian/rail/gripper +motion, pick/drop, and park operations. + +## Failure events + +A failed operation retains the original invocation context and adds: + +```python +{ + "error_type": type(error).__name__, + "error_message": str(error), +} +``` + +Do not swallow or transform the original exception merely to emit an event. The EventBus emits +the `.failed` event and re-raises the same failure. Add structured device-specific error details +only when they are stable and useful independently of the raw message. + +## Contributor checklist + +When adding EventBus support to a frontend or driver: + +1. Choose public semantic operation boundaries; do not decorate transport primitives by default. +2. Use a stable `.` name and one event scope per logical action. +3. Include `device` and direct `resources` in the context factory. +4. Preserve PLR resource semantics; use ancestry for context rather than substituting resources. +5. Use explicit units in quantitative field names, such as `volume_ul`, `duration_s`, and + `temperature_c`. +6. Include `source` and `destination` only for actual resource transfers. +7. Add tests for `.started`, `.completed`, and `.failed`, including operation-ID correlation. +8. Verify no EventBus listener is required for normal device operation and that listener failures + cannot alter hardware control flow. + +The first implementation instruments a limited set of shared and legacy frontends. New or +existing drivers should adopt these conventions incrementally at their public semantic API +boundaries. diff --git a/docs/contributor_guide/index.md b/docs/contributor_guide/index.md index 4f3f8feb1fb..0da06bc00a7 100644 --- a/docs/contributor_guide/index.md +++ b/docs/contributor_guide/index.md @@ -15,6 +15,7 @@ contributing-to-docs :caption: Adding Backends/Drivers device-driver-guide +event-bus ```
From 274b46b65c89f2f0c5ebbc2b04f3ce7ad370092d Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Mon, 10 Aug 2026 18:40:10 -0700 Subject: [PATCH 11/12] docs: distinguish semantic and diagnostic events --- docs/contributor_guide/event-bus.md | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/contributor_guide/event-bus.md b/docs/contributor_guide/event-bus.md index 984648334f1..5781d0271c9 100644 --- a/docs/contributor_guide/event-bus.md +++ b/docs/contributor_guide/event-bus.md @@ -11,8 +11,13 @@ not a control mechanism: listener failures must never affect hardware control fl Instrument **semantic public operations**: an operation that a protocol author would recognize as one action, such as fetching a plate, aspirating, shaking, or moving an arm to a location. -Do not instrument every serial, USB, transport, or firmware command as a semantic event. -Those lower-level details remain available through normal device logging for diagnostics. +Do not represent every serial, USB, transport, or firmware command as a *semantic* operation. + +The current EventBus also supports a separate diagnostic layer. Instrumented transports emit +`io.read` and `io.write`, and the Hamilton USB transport emits `firmware.command.started`, +`.completed`, and `.failed`. These records are useful for correlation and diagnostics, but are +not a replacement for semantic frontend events. Consumers should normally filter them from +operator timelines and high-level notifications. An instrumented method remains a no-op with respect to events unless an EventBus with at least one subscriber is active. Use one of these helpers: @@ -21,6 +26,10 @@ one subscriber is active. Use one of these helpers: - `with event_operation(...):` for one logical operation implemented by several calls. - `emit_event(...)` only for a meaningful state transition that is not an operation lifecycle. +Use a low-level diagnostic event only when the transport boundary itself is useful to observe. +Diagnostic events may inherit the enclosing semantic operation context, but do not define a new +protocol-level action or resource-transfer meaning. + ## Universal event contract A `PLREvent` always contains: From 0841d8b5e750e998bee598c6f568bc4dfbaf8248 Mon Sep 17 00:00:00 2001 From: Jon Chen Date: Mon, 10 Aug 2026 18:50:13 -0700 Subject: [PATCH 12/12] docs: keep EventBus consumer guidance neutral --- docs/contributor_guide/event-bus.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/contributor_guide/event-bus.md b/docs/contributor_guide/event-bus.md index 5781d0271c9..ba277832e97 100644 --- a/docs/contributor_guide/event-bus.md +++ b/docs/contributor_guide/event-bus.md @@ -15,9 +15,9 @@ Do not represent every serial, USB, transport, or firmware command as a *semanti The current EventBus also supports a separate diagnostic layer. Instrumented transports emit `io.read` and `io.write`, and the Hamilton USB transport emits `firmware.command.started`, -`.completed`, and `.failed`. These records are useful for correlation and diagnostics, but are -not a replacement for semantic frontend events. Consumers should normally filter them from -operator timelines and high-level notifications. +`.completed`, and `.failed`. Semantic and diagnostic events are complementary: semantic events +describe PLR-level operations, while diagnostic events describe the transport and controller +activity performed to execute them. An instrumented method remains a no-op with respect to events unless an EventBus with at least one subscriber is active. Use one of these helpers: