feat(robot-server, api): Implement RobotServerPyroResource providing proxies of server assets (#21240)

Introduces RobotServerPyroResource to provide hardware listener callbacks, camera provider, and file provider over pyro.

Author: CaseyBatten

api/src/opentrons/__init__.py

@@ -1,8 +1,13 @@
 import logging
 import os
 import re
+import socket
+import time
 from pathlib import Path
-from typing import Any, List, Tuple
+from typing import Any, List, Tuple, cast
+
+import Pyro5.api as pyro
+import Pyro5.errors as pyro_errors
 
 from ._version import version
 from opentrons.config import (
@@ -15,19 +20,22 @@
     feature_flags as ff,
 )
 from opentrons.drivers.serial_communication import get_ports_by_name
+from opentrons.hardware_control import API as HardwareAPI
 from opentrons.hardware_control import (
-    API as HardwareAPI,
-)
-from opentrons.hardware_control import (
+    HardwareControlAPI,
     ThreadManagedHardware,
     ThreadManager,
 )
 from opentrons.hardware_control import (
     types as hw_types,
 )
+from opentrons.hardware_control.pyro_utils.serpent_type_registry import (
+    register_hardware_types,
+)
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocols.types import ApiDeprecationError
 from opentrons.util import logging_config
+from opentrons.util.pyro.pyro_client_async_adapter import AsyncClientPyroObject
 
 HERE = os.path.abspath(os.path.dirname(__file__))
 __version__ = version
@@ -151,3 +159,41 @@ async def initialize() -> ThreadManagedHardware:
     log.info(f"Robot Name: {name()}")
 
     return await _create_thread_manager()
+
+
+def identify_hardware_process() -> HardwareControlAPI:
+    """
+    Identify the Pyro Proxy for the OT3API and return a wrapped hardware instance.
+    """
+    robot_conf = robot_configs.load()
+    logging_config.log_init(robot_conf.log_level)
+    pyro.config.COMMTIMEOUT = 100
+    try:
+        # Find the OT3API on the nameserver
+        # todo(chb, 03-31-2026): for now this is using the same methodology as the DirectedRunProcess work, consolidate
+        ot3_process_proxy = None
+        start_time = time.monotonic()
+        with pyro.locate_ns() as ns:
+            while time.monotonic() - start_time < 60:
+                if "OT3API" in ns.list():
+                    ot3_process_proxy = pyro.Proxy(ns.list()["OT3API"])  # type: ignore[no-untyped-call]
+                    break
+                time.sleep(0.01)
+
+        if ot3_process_proxy is None:
+            raise pyro_errors.CommunicationError(
+                "Opentrons-robot-server could not find OT3API URI on Pyro5 Nameserver."
+            )
+        else:
+            ot3_process_async_client = AsyncClientPyroObject(ot3_process_proxy)
+            hardware_api = cast(HardwareControlAPI, ot3_process_async_client)
+            # Register hardware types for the robot server process
+            register_hardware_types()
+            log.info("Opentrons Hardware API Subprocess identified and ready for use.")
+
+            return hardware_api
+
+    except (pyro_errors.NamingError, pyro_errors.CommunicationError, socket.timeout):
+        raise pyro_errors.CommunicationError(
+            "Opentrons Pyro5 Nameserver not found within 60 seconds."
+        )

api/src/opentrons/hardware_control/ot3api.py

@@ -18,6 +18,7 @@
     Sequence,
     Set,
     Tuple,
+    Type,
     TypeVar,
     Union,
     cast,
@@ -92,6 +93,7 @@
 from .ot3_calibration import OT3RobotCalibrationProvider, OT3Transforms
 from .pause_manager import PauseManager
 from .protocols import FlexHardwareControlInterface
+from .protocols.types import FlexRobotType
 from .types import (
     AsynchronousModuleErrorNotification,
     Axis,
@@ -144,6 +146,7 @@
 from opentrons.util.pyro.pyro_synchronous_adapter import (
     convert_result_to_proxy,
     convert_result_to_wrapped_dict,
+    convert_type_to_instance,
     pyro_behavior,
 )
 
@@ -256,6 +259,10 @@ def estop_cb(event: HardwareEvent) -> None:
         OT3RobotCalibrationProvider.__init__(self, self._config)
         ExecutionManagerProvider.__init__(self, isinstance(backend, OT3Simulator))
 
+    @pyro_behavior(specialty_func=convert_type_to_instance, apply_local=False)
+    def get_robot_type(self) -> Type[FlexRobotType]:
+        return FlexRobotType
+
     def is_idle_mount(self, mount: Union[top_types.Mount, OT3Mount]) -> bool:
         """Only the gripper mount or the 96-channel pipette mount would be idle
         (disengaged).
@@ -527,12 +534,14 @@ def is_simulator(self) -> bool:
         """`True` if this is a simulator; `False` otherwise."""
         return isinstance(self._backend, OT3Simulator)
 
+    @pyro_behavior(specialty_func=convert_result_to_proxy, apply_local=False)
     def register_callback(self, cb: HardwareEventHandler) -> Callable[[], None]:
         """Allows the caller to register a callback, and returns a closure
         that can be used to unregister the provided callback
         """
         self._callbacks.add(cb)
 
+        # todo(chb: 04-08-2026): Do we need to add a LOCAL @pyro_behavior to this, which will destroy the proxy object when the time comes?
         def unregister() -> None:
             self._callbacks.remove(cb)
 
@@ -2693,7 +2702,8 @@ def get_attached_instrument(
         return self.get_attached_pipette(mount)
 
     @property
-    def attached_instruments(self) -> Any:
+    @pyro_behavior(specialty_func=convert_result_to_wrapped_dict, apply_local=False)
+    def attached_instruments(self) -> Dict[top_types.Mount, PipetteDict]:
         # Warning: don't use this in new code, used `attached_pipettes` instead
         return self.attached_pipettes
 

api/src/opentrons/hardware_control/protocols/__init__.py

@@ -81,9 +81,6 @@ class FlexHardwareControlInterface(
     with some additional functionality and parameterization not supported on the OT-2.
     """
 
-    def get_robot_type(self) -> Type[FlexRobotType]:
-        return FlexRobotType
-
 
 __all__ = [
     "HardwareControlInterface",

api/src/opentrons/hardware_control/pyro_utils/serpent_type_registry.py

@@ -1,9 +1,12 @@
 """Registry for use with a Pyro Daemon client and server to allow serialization of Opentrons Hardware types and classes."""
 
+import datetime
 from typing import Dict
 
 import opentrons.config.types
 import opentrons.hardware_control.dev_types
+import opentrons.hardware_control.instruments.ot3.instrument_calibration
+import opentrons.hardware_control.protocols.types
 import opentrons.hardware_control.types
 import opentrons.types
 from opentrons.util.pyro.pyro_serialization import (
@@ -40,6 +43,135 @@ def _estop_overall_status_class_to_dict(obj) -> Dict:  # type: ignore
     }
 
 
+# GRIPPER CALIBRATION
+# todo(chb, 04-08-2026): This should be consumed into an automated registry process
+def _GripperCalibrationOffset_dict_to_class(  # type: ignore
+    classname, d
+) -> opentrons.hardware_control.instruments.ot3.instrument_calibration.GripperCalibrationOffset:
+    modified = (
+        None
+        if d["last_modified"] is None
+        else datetime.datetime.fromisoformat(d["last_modified"])
+    )
+    markedAt = (
+        None
+        if d["status_markedAt"] is None
+        else datetime.datetime.fromisoformat(d["status_markedAt"])
+    )
+    status_source = (
+        None
+        if d["status_source"] is None
+        else opentrons.hardware_control.instruments.ot3.instrument_calibration.SourceType(
+            d["status_source"]
+        )
+    )
+    return opentrons.hardware_control.instruments.ot3.instrument_calibration.GripperCalibrationOffset(
+        offset=opentrons.types.Point(x=d["offset_x"], y=d["offset_y"], z=d["offset_z"]),
+        source=opentrons.hardware_control.instruments.ot3.instrument_calibration.SourceType(
+            d["source"]
+        ),
+        status=opentrons.hardware_control.instruments.ot3.instrument_calibration.CalibrationStatus(
+            markedBad=(d["status_markedBad"] == "True"),
+            source=status_source,
+            markedAt=markedAt,
+        ),
+        last_modified=modified,
+    )
+
+
+def _GripperCalibrationOffset_class_to_dict(obj) -> Dict:  # type: ignore
+    if isinstance(obj.last_modified, datetime.datetime):
+        modified = obj.last_modified.isoformat()
+    else:
+        modified = None
+    if isinstance(obj.status.markedAt, datetime.datetime):
+        markedAt = obj.status.markedAt.isoformat()
+    else:
+        markedAt = None
+    return {
+        "__class__": "opentrons.hardware_control.instruments.ot3.instrument_calibration.GripperCalibrationOffset",
+        "offset_x": obj.offset.x,
+        "offset_y": obj.offset.y,
+        "offset_z": obj.offset.z,
+        "source": obj.source,
+        "status_markedBad": obj.status.markedBad,
+        "status_source": obj.status.source,
+        "status_markedAt": markedAt,
+        "last_modified": modified,
+    }
+
+
+# PIPETTER CALIBRATION
+# todo(chb, 04-08-2026): This should be consumed into an automated registry process
+def _PipetteOffsetSummary_dict_to_class(  # type: ignore
+    classname, d
+) -> opentrons.hardware_control.instruments.ot3.instrument_calibration.PipetteOffsetSummary:
+    modified = (
+        None
+        if d["last_modified"] is None
+        else datetime.datetime.fromisoformat(d["last_modified"])
+    )
+    markedAt = (
+        None
+        if d["status_markedAt"] is None
+        else datetime.datetime.fromisoformat(d["status_markedAt"])
+    )
+    status_source = (
+        None
+        if d["status_source"] is None
+        else opentrons.hardware_control.instruments.ot3.instrument_calibration.SourceType(
+            d["status_source"]
+        )
+    )
+    return opentrons.hardware_control.instruments.ot3.instrument_calibration.PipetteOffsetSummary(
+        offset=opentrons.types.Point(x=d["offset_x"], y=d["offset_y"], z=d["offset_z"]),
+        source=opentrons.hardware_control.instruments.ot3.instrument_calibration.SourceType(
+            d["source"]
+        ),
+        status=opentrons.hardware_control.instruments.ot3.instrument_calibration.CalibrationStatus(
+            markedBad=(d["status_markedBad"] == "True"),
+            source=status_source,
+            markedAt=markedAt,
+        ),
+        last_modified=modified,
+        reasonability_check_failures=[],  # todo(chb: 04-09-2026): These are skipped for integration simplicity, they should be handled by automatic process
+    )
+
+
+def _PipetteOffsetSummary_class_to_dict(obj) -> Dict:  # type: ignore
+    if isinstance(obj.last_modified, datetime.datetime):
+        modified = obj.last_modified.isoformat()
+    else:
+        modified = None
+    if isinstance(obj.status.markedAt, datetime.datetime):
+        markedAt = obj.status.markedAt.isoformat()
+    else:
+        markedAt = None
+    return {
+        "__class__": "opentrons.hardware_control.instruments.ot3.instrument_calibration.PipetteOffsetSummary",
+        "offset_x": obj.offset.x,
+        "offset_y": obj.offset.y,
+        "offset_z": obj.offset.z,
+        "source": obj.source,
+        "status_markedBad": obj.status.markedBad,
+        "status_source": obj.status.source,
+        "status_markedAt": markedAt,
+        "last_modified": modified,
+        "reasonability_check_failures": None,  # todo(chb: 04-09-2026): These are skipped for integration simplicity, they should be handled by automatic process
+    }
+
+
+# Robot type registry - of note, this is meant to return a "pure" type
+def _robot_type_class_to_dict(obj) -> Dict:  # type: ignore
+    return {"__class__": ".".join((obj.__module__, obj.__class__.__name__))}
+
+
+def _robot_type_dict_to_class(  # type: ignore
+    classname, d
+) -> type[opentrons.hardware_control.protocols.types.FlexRobotType]:
+    return opentrons.hardware_control.protocols.types.FlexRobotType
+
+
 # Handy function to map all registries for the Hardware controller
 def register_hardware_types() -> None:
     """Registers serialize and deserialize behavior for Opentrons Hardware types and classes.
@@ -59,7 +191,12 @@ def register_hardware_types() -> None:
             OpentronsPyroSerializer.register_enum(enum_type)
 
     opentrons_pydantic_types = find_pydantic_classes_in_packages(
-        [opentrons.types, opentrons.config.types, opentrons.hardware_control.types]
+        [
+            opentrons.types,
+            opentrons.config.types,
+            opentrons.hardware_control.types,
+            opentrons.hardware_control.protocols.types,
+        ]
     )
     for pydantic_type in opentrons_pydantic_types:
         OpentronsPyroSerializer.register_pydantic_model(pydantic_type)
@@ -72,9 +209,32 @@ def register_hardware_types() -> None:
 
     OpentronsPyroSerializer.register_unhashable_dicts()
 
+    # Specialized registrations:
+    register_type_to_serpent(
+        class_type=opentrons.hardware_control.protocols.types.FlexRobotType,
+        dict_to_class=_robot_type_dict_to_class,
+        class_to_dict=_robot_type_class_to_dict,
+    )
+
+    # todo(chb, 04-03-2026): This one should probably be removed and classes like it converted to an appropriate, automated format
     # E-Stop Overall registration
     register_type_to_serpent(
         class_type=opentrons.hardware_control.types.EstopOverallStatus,
         dict_to_class=_estop_overall_status_dict_to_class,
         class_to_dict=_estop_overall_status_class_to_dict,
     )
+
+    # todo(chb: 04-09-2026): These are termporary direct serializations to support the initial robot server intergration, replace with automated solution
+    # gripper calibration
+    register_type_to_serpent(
+        class_type=opentrons.hardware_control.instruments.ot3.instrument_calibration.GripperCalibrationOffset,
+        dict_to_class=_GripperCalibrationOffset_dict_to_class,
+        class_to_dict=_GripperCalibrationOffset_class_to_dict,
+    )
+
+    # pipette calibration
+    register_type_to_serpent(
+        class_type=opentrons.hardware_control.instruments.ot3.instrument_calibration.PipetteOffsetSummary,
+        dict_to_class=_PipetteOffsetSummary_dict_to_class,
+        class_to_dict=_PipetteOffsetSummary_class_to_dict,
+    )