feat(api): add protocol api and engine support for absorbance plate reader

api-client/src/modules/api-types.ts

@@ -80,6 +80,12 @@ export interface HeaterShakerData {
   errorDetails: string | null
   status: HeaterShakerStatus
 }
+export interface AbsorbanceReaderData {
+  lidStatus: 'open' | 'closed' | 'unknown'
+  platePresence: 'present' | 'absent' | 'unknown'
+  sampleWavelength: number | null
+  status: AbsorbanceReaderStatus
+}
 
 export type TemperatureStatus =
   | 'idle'
@@ -112,3 +118,5 @@ export type LatchStatus =
   | 'idle_closed'
   | 'idle_unknown'
   | 'unknown'
+
+export type AbsorbanceReaderStatus = 'idle' | 'measuring' | 'error'

api-client/src/modules/types.ts

@@ -3,10 +3,12 @@ import type {
   ThermocyclerModuleModel,
   MagneticModuleModel,
   HeaterShakerModuleModel,
+  AbsorbanceReaderModel,
   TEMPERATURE_MODULE_TYPE,
   MAGNETIC_MODULE_TYPE,
   THERMOCYCLER_MODULE_TYPE,
   HEATERSHAKER_MODULE_TYPE,
+  ABSORBANCE_READER_TYPE,
 } from '@opentrons/shared-data'
 
 import type * as ApiTypes from './api-types'
@@ -44,11 +46,18 @@ export interface HeaterShakerModule extends CommonModuleInfo {
   data: ApiTypes.HeaterShakerData
 }
 
+export interface AbsorbanceReaderModule extends CommonModuleInfo {
+  moduleType: typeof ABSORBANCE_READER_TYPE
+  moduleModel: AbsorbanceReaderModel
+  data: ApiTypes.AbsorbanceReaderData
+}
+
 export type AttachedModule =
   | TemperatureModule
   | MagneticModule
   | ThermocyclerModule
   | HeaterShakerModule
+  | AbsorbanceReaderModule
 
 export interface ModulesMeta {
   cursor: number

api/src/opentrons/drivers/absorbance_reader/abstract.py

@@ -1,6 +1,9 @@
 from abc import ABC, abstractmethod
 from typing import Dict, List
-from opentrons.drivers.types import AbsorbanceReaderLidStatus
+from opentrons.drivers.types import (
+    AbsorbanceReaderLidStatus,
+    AbsorbanceReaderDeviceState,
+)
 
 
 class AbstractAbsorbanceReaderDriver(ABC):
@@ -36,7 +39,7 @@ async def initialize_measurement(self, wavelength: int) -> None:
         ...
 
     @abstractmethod
-    async def get_status(self) -> None:
+    async def get_status(self) -> AbsorbanceReaderDeviceState:
         ...
 
     @abstractmethod

api/src/opentrons/drivers/absorbance_reader/async_byonoy.py

@@ -5,13 +5,17 @@
 from concurrent.futures.thread import ThreadPoolExecutor
 from functools import partial
 from typing import Optional, List, Dict
-import usb.core as usb_core  # type: ignore[import-untyped]
 
 
-from .hid_protocol import AbsorbanceHidInterface as AbsProtocol, ErrorCodeNames
+from .hid_protocol import (
+    AbsorbanceHidInterface as AbsProtocol,
+    ErrorCodeNames,
+    DeviceStateNames,
+)
 from opentrons.drivers.types import (
     AbsorbanceReaderLidStatus,
     AbsorbanceReaderPlatePresence,
+    AbsorbanceReaderDeviceState,
 )
 from opentrons.drivers.rpi_drivers.types import USBPort
 
@@ -36,6 +40,8 @@ def serial_number_from_port(name: str) -> str:
         """
         Get the serial number from a port using pyusb.
         """
+        import usb.core as usb_core  # type: ignore[import-untyped]
+
         port_numbers = tuple(int(s) for s in name.split("-")[1].split("."))
         device = usb_core.find(port_numbers=port_numbers)
         if device:
@@ -232,6 +238,7 @@ async def get_device_static_info(self) -> Dict[str, str]:
         return {
             "serial": self._device.sn,
             "model": "ABS96",
+            "version": "1.0",
         }
 
     async def get_device_information(self) -> Dict[str, str]:
@@ -270,3 +277,22 @@ async def get_single_measurement(self, wavelength: int) -> List[float]:
 
     async def get_plate_presence(self) -> AbsorbanceReaderPlatePresence:
         return AbsorbanceReaderPlatePresence.UNKNOWN
+
+    async def get_device_status(self) -> AbsorbanceReaderDeviceState:
+        status = await self._loop.run_in_executor(
+            executor=self._executor,
+            func=self._get_device_status,
+        )
+        return self.convert_device_state(status.name)
+
+    @staticmethod
+    def convert_device_state(
+        device_state: DeviceStateNames,
+    ) -> AbsorbanceReaderDeviceState:
+        state_map: Dict[DeviceStateNames, AbsorbanceReaderDeviceState] = {
+            "BYONOY_DEVICE_STATE_UNKNOWN": AbsorbanceReaderDeviceState.UNKNOWN,
+            "BYONOY_DEVICE_STATE_OK": AbsorbanceReaderDeviceState.OK,
+            "BYONOY_DEVICE_STATE_BROKEN_FW": AbsorbanceReaderDeviceState.BROKEN_FW,
+            "BYONOY_DEVICE_STATE_ERROR": AbsorbanceReaderDeviceState.ERROR,
+        }
+        return state_map[device_state]

api/src/opentrons/drivers/absorbance_reader/driver.py

@@ -3,7 +3,11 @@
 import asyncio
 from typing import Dict, Optional, List, TYPE_CHECKING
 
-from opentrons.drivers.types import AbsorbanceReaderLidStatus
+from opentrons.drivers.types import (
+    AbsorbanceReaderLidStatus,
+    AbsorbanceReaderDeviceState,
+    AbsorbanceReaderPlatePresence,
+)
 from opentrons.drivers.absorbance_reader.abstract import AbstractAbsorbanceReaderDriver
 from opentrons.drivers.rpi_drivers.types import USBPort
 
@@ -61,5 +65,8 @@ async def get_single_measurement(self, wavelength: int) -> List[float]:
     async def initialize_measurement(self, wavelength: int) -> None:
         await self._connection.initialize(wavelength)
 
-    async def get_status(self) -> None:
-        pass
+    async def get_status(self) -> AbsorbanceReaderDeviceState:
+        return await self._connection.get_device_status()
+
+    async def get_plate_presence(self) -> AbsorbanceReaderPlatePresence:
+        return await self._connection.get_plate_presence()

api/src/opentrons/drivers/absorbance_reader/hid_protocol.py

@@ -4,7 +4,6 @@
     List,
     Literal,
     Tuple,
-    ClassVar,
     runtime_checkable,
     TypeVar,
 )
@@ -36,6 +35,20 @@
     "BYONOY_ERROR_INTERNAL",
 ]
 
+SlotStateNames = Literal[
+    "BYONOY_SLOT_UNKNOWN",
+    "BYONOY_SLOT_EMPTY",
+    "BYONOY_SLOT_OCCUPIED",
+    "BYONOY_SLOT_UNDETERMINED",
+]
+
+DeviceStateNames = Literal[
+    "BYONOY_DEVICE_STATE_UNKNOWN",
+    "BYONOY_DEVICE_STATE_OK",
+    "BYONOY_DEVICE_STATE_BROKEN_FW",
+    "BYONOY_DEVICE_STATE_ERROR",
+]
+
 
 @runtime_checkable
 class AbsorbanceHidInterface(Protocol):
@@ -51,7 +64,9 @@ class ErrorCode(Protocol):
 
     @runtime_checkable
     class SlotState(Protocol):
-        __members__: ClassVar[Dict[str, int]]
+        __members__: Dict[SlotStateNames, int]
+        name: SlotStateNames
+        value: int
 
     @runtime_checkable
     class MeasurementConfig(Protocol):
@@ -65,7 +80,9 @@ class DeviceInfo(Protocol):
 
     @runtime_checkable
     class DeviceState(Protocol):
-        __members__: ClassVar[Dict[str, int]]
+        __members__: Dict[DeviceStateNames, int]
+        name: DeviceStateNames
+        value: int
 
     def ByonoyAbs96SingleMeasurementConfig(self) -> MeasurementConfig:
         ...

api/src/opentrons/drivers/absorbance_reader/simulator.py

@@ -1,7 +1,10 @@
 from typing import Dict, List, Optional
 from opentrons.util.async_helpers import ensure_yield
 
-from opentrons.drivers.types import AbsorbanceReaderLidStatus
+from opentrons.drivers.types import (
+    AbsorbanceReaderLidStatus,
+    AbsorbanceReaderDeviceState,
+)
 
 from .abstract import AbstractAbsorbanceReaderDriver
 
@@ -51,5 +54,5 @@ async def initialize_measurement(self, wavelength: int) -> None:
         pass
 
     @ensure_yield
-    async def get_status(self) -> None:
-        pass
+    async def get_status(self) -> AbsorbanceReaderDeviceState:
+        return AbsorbanceReaderDeviceState.OK

api/src/opentrons/drivers/types.py

@@ -73,4 +73,13 @@ class AbsorbanceReaderPlatePresence(str, Enum):
 
     UNKNOWN = "unknown"
     PRESENT = "present"
-    ABSENCE = "absence"
+    ABSENT = "absent"
+
+
+class AbsorbanceReaderDeviceState(str, Enum):
+    """Absorbance reader device state."""
+
+    UNKNOWN = "unknown"
+    OK = "ok"
+    BROKEN_FW = "broken_fw"
+    ERROR = "error"

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

@@ -19,6 +19,7 @@
     TemperatureStatus,
     MagneticStatus,
     HeaterShakerStatus,
+    AbsorbanceReaderStatus,
     SpeedStatus,
     LiveData,
 )
@@ -47,4 +48,5 @@
     "SpeedStatus",
     "LiveData",
     "AbsorbanceReader",
+    "AbsorbanceReaderStatus",
 ]

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

@@ -7,7 +7,14 @@
     AbsorbanceReaderDriver,
     SimulatingDriver,
 )
+from opentrons.drivers.types import (
+    AbsorbanceReaderLidStatus,
+    AbsorbanceReaderPlatePresence,
+    AbsorbanceReaderDeviceState,
+)
+
 from opentrons.hardware_control.execution_manager import ExecutionManager
+from opentrons.hardware_control.poller import Reader
 from opentrons.hardware_control.modules import mod_abc
 from opentrons.hardware_control.modules.types import (
     ModuleType,
@@ -56,6 +63,7 @@ async def build(
             driver=driver,
             hw_control_loop=hw_control_loop,
         )
+        await module.setup()
         return module
 
     def __init__(
@@ -83,6 +91,14 @@ def status(self) -> AbsorbanceReaderStatus:
         """Return some string describing status."""
         return AbsorbanceReaderStatus.IDLE
 
+    @property
+    def lid_status(self) -> AbsorbanceReaderLidStatus:
+        return AbsorbanceReaderLidStatus.UNKNOWN
+
+    @property
+    def plate_presence(self) -> AbsorbanceReaderPlatePresence:
+        return AbsorbanceReaderPlatePresence.UNKNOWN
+
     @property
     def device_info(self) -> Mapping[str, str]:
         """Return a dict of the module's static information (serial, etc)"""
@@ -92,8 +108,10 @@ def device_info(self) -> Mapping[str, str]:
     def live_data(self) -> LiveData:
         """Return a dict of the module's dynamic information"""
         return {
-            "status": "idle",
+            "status": self.status.value,
             "data": {
+                "lidStatus": self.lid_status.value,
+                "platePresence": self.plate_presence.value,
                 "sampleWavelength": 400,
             },
         }
@@ -159,14 +177,50 @@ async def set_sample_wavelength(self, wavelength: int) -> None:
         """Set the Absorbance Reader's active wavelength."""
         await self._driver.initialize_measurement(wavelength)
 
-    async def start_measure(self, wavelength: int) -> None:
+    async def start_measure(self, wavelength: int) -> List[float]:
         """Initiate a single measurement."""
-        await self._driver.get_single_measurement(wavelength)
+        return await self._driver.get_single_measurement(wavelength)
 
-    async def get_supported_wavelengths(self) -> List[int]:
-        """Get the Absorbance Reader's supported wavelengths."""
-        return await self._driver.get_available_wavelengths()
+    async def setup(self) -> None:
+        """Setup the Absorbance Reader."""
+        is_open = await self._driver.is_connected()
+        if not is_open:
+            await self._driver.connect()
 
     async def get_current_wavelength(self) -> None:
         """Get the Absorbance Reader's current active wavelength."""
         pass
+
+
+class AbsorbanceReaderReader(Reader):
+    device_state: AbsorbanceReaderDeviceState
+    lid_status: AbsorbanceReaderLidStatus
+    plate_presence: AbsorbanceReaderPlatePresence
+    supported_wavelengths: List[int]
+
+    def __init__(self, driver: AbsorbanceReaderDriver) -> None:
+        self.device_state = AbsorbanceReaderDeviceState.UNKNOWN
+        self.lid_status = AbsorbanceReaderLidStatus.UNKNOWN
+        self.plate_presence = AbsorbanceReaderPlatePresence.UNKNOWN
+        self.supported_wavelengths = []
+        self._driver = driver
+
+    async def read(self) -> None:
+        await self.get_device_status()
+        await self.get_supported_wavelengths()
+
+    async def get_device_status(self) -> None:
+        """Get the Absorbance Reader's current status."""
+        self.device_state = await self._driver.get_status()
+
+    async def get_supported_wavelengths(self) -> None:
+        """Get the Absorbance Reader's supported wavelengths."""
+        self.supported_wavelengths = await self._driver.get_available_wavelengths()
+
+    async def get_lid_status(self) -> None:
+        """Get the Absorbance Reader's lid status."""
+        self.lid_status = await self._driver.get_lid_status()
+
+    async def get_plate_presence(self) -> None:
+        """Get the Absorbance Reader's plate presence."""
+        self.plate_presence = await self._driver.get_plate_presence()

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

@@ -80,6 +80,8 @@ def to_module_fixture_id(cls, module_type: ModuleType) -> str:
             return "heaterShakerModuleV1"
         if module_type == ModuleType.MAGNETIC_BLOCK:
             return "magneticBlockV1"
+        if module_type == ModuleType.ABSORBANCE_READER:
+            return "absorbanceReaderV1"
         else:
             raise ValueError(
                 f"Module Type {module_type} does not have a related fixture ID."
@@ -210,3 +212,10 @@ class AbsorbanceReaderStatus(str, Enum):
     IDLE = "idle"
     MEASURING = "measuring"
     ERROR = "error"
+
+
+class LidStatus(str, Enum):
+    ON = "on"
+    OFF = "off"
+    UNKNOWN = "unknown"
+    ERROR = "error"

api/src/opentrons/protocol_api/__init__.py

@@ -21,6 +21,7 @@
     TemperatureModuleContext,
     HeaterShakerContext,
     MagneticBlockContext,
+    AbsorbanceReaderContext,
 )
 from .disposal_locations import TrashBin, WasteChute
 from ._liquid import Liquid
@@ -50,6 +51,7 @@
     "ThermocyclerContext",
     "HeaterShakerContext",
     "MagneticBlockContext",
+    "AbsorbanceReaderContext",
     "ParameterContext",
     "Labware",
     "TrashBin",