feat(robot-server): add unique IDs to GET /modules response (#9758)

Closes #9533

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

@@ -13,6 +13,11 @@
     ModuleAtPort,
     ModuleType,
     ModuleModel,
+    TemperatureStatus,
+    MagneticStatus,
+    HeaterShakerStatus,
+    SpeedStatus,
+    LiveData,
 )
 
 __all__ = [
@@ -31,4 +36,10 @@
     "HeaterShaker",
     "ModuleType",
     "ModuleModel",
+    "TemperatureStatus",
+    "MagneticStatus",
+    "HeaterShakerStatus",
+    "SpeedStatus",
+    "ThermocyclerLidStatus",
+    "LiveData",
 ]

api/src/opentrons/hardware_control/modules/types.py

@@ -8,10 +8,10 @@
     Any,
     Tuple,
     Awaitable,
-    Mapping,
     Union,
     TYPE_CHECKING,
 )
+from typing_extensions import TypedDict
 from pathlib import Path
 
 from opentrons.drivers.rpi_drivers.types import USBPort
@@ -28,7 +28,10 @@
 
 UploadFunction = Callable[[str, str, Dict[str, Any]], Awaitable[Tuple[bool, str]]]
 
-LiveData = Mapping[str, Union[str, Mapping[str, Union[float, str, None]]]]
+
+class LiveData(TypedDict):
+    status: str
+    data: Dict[str, Union[float, str, bool, None]]
 
 
 class ModuleType(str, Enum):
@@ -86,10 +89,18 @@ class ModuleInfo(NamedTuple):
 
 # TODO(mc, 2022-01-18): replace with enum
 ModuleModel = Union[
-    MagneticModuleModel, TemperatureModuleModel, ThermocyclerModuleModel
+    MagneticModuleModel,
+    TemperatureModuleModel,
+    ThermocyclerModuleModel,
+    HeaterShakerModuleModel,
 ]
 
 
+class MagneticStatus(str, Enum):
+    ENGAGED = "engaged"
+    DISENGAGED = "disengaged"
+
+
 class TemperatureStatus(str, Enum):
     HOLDING = "holding at target"
     COOLING = "cooling"

robot-server/robot_server/modules/__init__.py

@@ -0,0 +1,4 @@
+"""Endpoints for getting information about the robot's attached modules."""
+from .router import modules_router
+
+__all__ = ["modules_router"]

robot-server/robot_server/modules/module_data_mapper.py

@@ -0,0 +1,124 @@
+"""Module identification and response data mapping."""
+from typing import Type, cast
+
+from opentrons.hardware_control.modules import (
+    LiveData,
+    ModuleType,
+    MagneticStatus,
+    TemperatureStatus,
+    HeaterShakerStatus,
+    SpeedStatus,
+)
+from opentrons.drivers.types import (
+    ThermocyclerLidStatus,
+    HeaterShakerLabwareLatchStatus,
+)
+from opentrons.drivers.rpi_drivers.types import USBPort as HardwareUSBPort
+
+from opentrons.protocol_engine import ModuleModel
+
+from .module_identifier import ModuleIdentity
+from .module_models import (
+    AttachedModule,
+    AttachedModuleData,
+    MagneticModule,
+    MagneticModuleData,
+    TemperatureModule,
+    TemperatureModuleData,
+    ThermocyclerModule,
+    ThermocyclerModuleData,
+    HeaterShakerModule,
+    HeaterShakerModuleData,
+    UsbPort,
+)
+
+
+class ModuleDataMapper:
+    """Map hardware control modules to module response."""
+
+    def map_data(
+        self,
+        model: str,
+        module_identity: ModuleIdentity,
+        has_available_update: bool,
+        live_data: LiveData,
+        usb_port: HardwareUSBPort,
+    ) -> AttachedModule:
+        """Map hardware control data to an attached module response."""
+        module_model = ModuleModel(model)
+        module_type = module_model.as_type()
+
+        module_cls: Type[AttachedModule]
+        module_data: AttachedModuleData
+
+        # rely on Pydantic to check/coerce data fields from dicts at run time
+        if module_type == ModuleType.MAGNETIC:
+            module_cls = MagneticModule
+            module_data = MagneticModuleData(
+                status=MagneticStatus(live_data["status"]),
+                engaged=cast(bool, live_data["data"].get("engaged")),
+                height=cast(float, live_data["data"].get("height")),
+            )
+
+        elif module_type == ModuleType.TEMPERATURE:
+            module_cls = TemperatureModule
+            module_data = TemperatureModuleData(
+                status=TemperatureStatus(live_data["status"]),
+                targetTemperature=cast(float, live_data["data"].get("targetTemp")),
+                currentTemperature=cast(float, live_data["data"].get("currentTemp")),
+            )
+
+        elif module_type == ModuleType.THERMOCYCLER:
+            module_cls = ThermocyclerModule
+            module_data = ThermocyclerModuleData(
+                status=TemperatureStatus(live_data["status"]),
+                targetTemperature=cast(float, live_data["data"].get("targetTemp")),
+                currentTemperature=cast(float, live_data["data"].get("currentTemp")),
+                lidStatus=cast(ThermocyclerLidStatus, live_data["data"].get("lid")),
+                lidTemperature=cast(float, live_data["data"].get("lidTemp")),
+                lidTargetTemperature=cast(float, live_data["data"].get("lidTarget")),
+                holdTime=cast(float, live_data["data"].get("holdTime")),
+                rampRate=cast(float, live_data["data"].get("rampRate")),
+                currentCycleIndex=cast(int, live_data["data"].get("currentCycleIndex")),
+                totalCycleCount=cast(int, live_data["data"].get("totalCycleCount")),
+                currentStepIndex=cast(int, live_data["data"].get("currentStepIndex")),
+                totalStepCount=cast(int, live_data["data"].get("totalStepCount")),
+            )
+
+        elif module_type == ModuleType.HEATER_SHAKER:
+            module_cls = HeaterShakerModule
+            module_data = HeaterShakerModuleData(
+                status=HeaterShakerStatus(live_data["status"]),
+                labwareLatchStatus=cast(
+                    HeaterShakerLabwareLatchStatus,
+                    live_data["data"].get("labwareLatchStatus"),
+                ),
+                speedStatus=cast(SpeedStatus, live_data["data"].get("speedStatus")),
+                currentSpeed=cast(int, live_data["data"].get("currentSpeed")),
+                targetSpeed=cast(int, live_data["data"].get("targetSpeed")),
+                temperatureStatus=cast(
+                    TemperatureStatus, live_data["data"].get("temperatureStatus")
+                ),
+                currentTemperature=cast(float, live_data["data"].get("currentTemp")),
+                targetTemperature=cast(float, live_data["data"].get("targetTemp")),
+                errorDetails=cast(str, live_data["data"].get("errorDetails")),
+            )
+        else:
+            assert False, f"Invalid module type {module_type}"
+
+        return module_cls(
+            id=module_identity.module_id,
+            serialNumber=module_identity.serial_number,
+            firmwareVersion=module_identity.firmware_version,
+            hardwareRevision=module_identity.hardware_revision,
+            hasAvailableUpdate=has_available_update,
+            usbPort=UsbPort(
+                port=usb_port.port_number,
+                hub=usb_port.hub,
+                path=usb_port.device_path,
+            ),
+            # types of below fields are already checked at runtime
+            moduleType=module_type,  # type: ignore[arg-type]
+            moduleModel=module_model,  # type: ignore[arg-type]
+            data=module_data,  # type: ignore[arg-type]
+        )

robot-server/robot_server/modules/module_identifier.py

@@ -0,0 +1,43 @@
+"""Identify hardware modules in the robot server."""
+import hashlib
+from dataclasses import dataclass
+from typing import Mapping
+
+
+@dataclass(frozen=True)
+class ModuleIdentity:
+    """Unique identifying features of an attached module."""
+
+    module_id: str
+    serial_number: str
+    firmware_version: str
+    hardware_revision: str
+
+
+class ModuleIdentifier:
+    """Hardware control interface to logical identifier mapping."""
+
+    @staticmethod
+    def identify(device_info: Mapping[str, str]) -> ModuleIdentity:
+        """Identify a module based on its hardware control device info."""
+        serial_number = device_info.get("serial")
+        firmware_version = device_info.get("version")
+        hardware_revision = device_info.get("model")
+
+        assert serial_number, f"Expected serial number in {device_info}"
+        assert firmware_version, f"Expected firmware version in {device_info}"
+        assert hardware_revision, f"Expected hardware revision in {device_info}"
+
+        return ModuleIdentity(
+            module_id=ModuleIdentifier._generate_id(serial_number, hardware_revision),
+            serial_number=serial_number,
+            firmware_version=firmware_version,
+            hardware_revision=hardware_revision,
+        )
+
+    @staticmethod
+    def _generate_id(serial_number: str, hardware_revision: str) -> str:
+        hasher = hashlib.blake2s(digest_size=20)
+        hasher.update(serial_number.encode())
+        hasher.update(hardware_revision.encode())
+        return hasher.hexdigest()