add absorbance reader command to protocol engine

protocol eng commands

draft

initialize and initiate_read to protocol_api

add initialize to module control

add absorbance_reader/initialize absorbance_reader/measure

draft2

add absorbance reader module context

fix lint

update protocol commands

add addresseable areas for abs plate reader

save abs reader in deck config

add types

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,13 +3,16 @@ 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'
+import { A } from 'vitest/dist/reporters-1evA5lom'
 
 export * from './api-types'
 
@@ -44,11 +47,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/async_byonoy.py

@@ -5,7 +5,6 @@
 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
@@ -36,6 +35,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 +233,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]:

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,6 +7,11 @@
     AbsorbanceReaderDriver,
     SimulatingDriver,
 )
+from opentrons.drivers.types import (
+    AbsorbanceReaderLidStatus,
+    AbsorbanceReaderPlatePresence,
+)
+
 from opentrons.hardware_control.execution_manager import ExecutionManager
 from opentrons.hardware_control.modules import mod_abc
 from opentrons.hardware_control.modules.types import (
@@ -83,6 +88,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 +105,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,9 +174,10 @@ 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)
+        measurement = await self._driver.get_single_measurement(wavelength)
+        return measurement
 
     async def get_supported_wavelengths(self) -> List[int]:
         """Get the Absorbance Reader's supported wavelengths."""
@@ -170,3 +186,7 @@ async def get_supported_wavelengths(self) -> List[int]:
     async def get_current_wavelength(self) -> None:
         """Get the Absorbance Reader's current active wavelength."""
         pass
+
+    async def get_lid_status(self) -> AbsorbanceReaderLidStatus:
+        """Get the Absorbance Reader's lid status."""
+        return await self._driver.get_lid_status()

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/core/common.py

@@ -10,6 +10,7 @@
     AbstractThermocyclerCore,
     AbstractHeaterShakerCore,
     AbstractMagneticBlockCore,
+    AbstractAbsorbanceReaderCore,
 )
 from .protocol import AbstractProtocol
 from .well import AbstractWellCore
@@ -24,4 +25,5 @@
 ThermocyclerCore = AbstractThermocyclerCore
 HeaterShakerCore = AbstractHeaterShakerCore
 MagneticBlockCore = AbstractMagneticBlockCore
+AbsorbanceReaderCore = AbstractAbsorbanceReaderCore
 ProtocolCore = AbstractProtocol[InstrumentCore, LabwareCore, ModuleCore]

api/src/opentrons/protocol_api/core/engine/module_core.py

@@ -33,6 +33,7 @@
     AbstractThermocyclerCore,
     AbstractHeaterShakerCore,
     AbstractMagneticBlockCore,
+    AbstractAbsorbanceReaderCore,
 )
 from .exceptions import InvalidMagnetEngageHeightError
 
@@ -467,3 +468,23 @@ def get_labware_latch_status(self) -> HeaterShakerLabwareLatchStatus:
 
 class MagneticBlockCore(NonConnectedModuleCore, AbstractMagneticBlockCore):
     """Magnetic Block control interface via a ProtocolEngine."""
+
+
+class AbsorbanceReaderCore(ModuleCore, AbstractAbsorbanceReaderCore):
+    """Absorbance Reader core logic implementation for Python protocols."""
+
+    _sync_module_hardware: SynchronousAdapter[hw_modules.AbsorbanceReader]
+    _initialized_value: Optional[int] = None
+
+    def initialize(self, wavelength: int) -> None:
+        """Initialize the Absorbance Reader by taking zero reading."""
+        self._engine_client.absorbance_reader_initialize(
+            module_id=self.module_id, wavelength=wavelength
+        )
+
+    def initiate_read(self) -> None:
+        """Initiate read on the Absorbance Reader."""
+        if self._initialized_value:
+            self._engine_client.absorbance_reader_measure(
+                module_id=self.module_id, wavelength=self._initialized_value
+            )

api/src/opentrons/protocol_api/core/engine/protocol.py

@@ -64,6 +64,7 @@
     HeaterShakerModuleCore,
     NonConnectedModuleCore,
     MagneticBlockCore,
+    AbsorbanceReaderCore,
 )
 from .exceptions import InvalidModuleLocationError
 from . import load_labware_params
@@ -455,6 +456,7 @@ def _create_module_core(
             ModuleType.MAGNETIC: MagneticModuleCore,
             ModuleType.THERMOCYCLER: ThermocyclerModuleCore,
             ModuleType.HEATER_SHAKER: HeaterShakerModuleCore,
+            ModuleType.ABSORBANCE_READER: AbsorbanceReaderCore,
         }
 
         module_type = load_module_result.model.as_type()

api/src/opentrons/protocol_api/core/module.py

@@ -343,3 +343,21 @@ class AbstractMagneticBlockCore(AbstractModuleCore):
     """Core control interface for an attached Magnetic Block."""
 
     MODULE_TYPE: ClassVar = ModuleType.MAGNETIC_BLOCK
+
+
+class AbstractAbsorbanceReaderCore(AbstractModuleCore):
+    """Core control interface for an attached Absorbance Reader Module."""
+
+    MODULE_TYPE: ClassVar = ModuleType.ABSORBANCE_READER
+
+    @abstractmethod
+    def get_serial_number(self) -> str:
+        """Get the module's unique hardware serial number."""
+
+    @abstractmethod
+    def initialize(self, wavelength: int) -> None:
+        """Initialize the Absorbance Reader by taking zero reading."""
+
+    @abstractmethod
+    def initiate_read(self) -> None:
+        """Initiate read on the Absorbance Reader."""

api/src/opentrons/protocol_api/module_contexts.py

@@ -22,6 +22,7 @@
     ThermocyclerCore,
     HeaterShakerCore,
     MagneticBlockCore,
+    AbsorbanceReaderCore,
 )
 from .core.core_map import LoadedCoreMap
 from .core.engine import ENGINE_CORE_API_VERSION
@@ -955,3 +956,31 @@ class MagneticBlockContext(ModuleContext):
     """
 
     _core: MagneticBlockCore
+
+
+class AbsorbanceReaderContext(ModuleContext):
+    """An object representing a connected Absorbance Reader Module.
+
+    It should not be instantiated directly; instead, it should be
+    created through :py:meth:`.ProtocolContext.load_module`.
+
+    .. versionadded:: 2.17
+    """
+
+    _core: AbsorbanceReaderCore
+
+    @property
+    @requires_version(2, 17)
+    def get_serial_number(self) -> str:
+        """Get the module's unique hardware serial number."""
+        return self._core.get_serial_number()
+
+    @requires_version(2, 17)
+    def initialize(self, wavelength: int) -> None:
+        """Initialize the Absorbance Reader by taking zero reading."""
+        self._core.initialize(wavelength)
+
+    @requires_version(2, 17)
+    def initiate_read(self) -> None:
+        """Initiate read on the Absorbance Reader."""
+        self._core.initiate_read()

api/src/opentrons/protocol_api/protocol_context.py

@@ -50,6 +50,7 @@
     AbstractThermocyclerCore,
     AbstractHeaterShakerCore,
     AbstractMagneticBlockCore,
+    AbstractAbsorbanceReaderCore,
 )
 from .core.engine import ENGINE_CORE_API_VERSION
 from .core.legacy.legacy_protocol_core import LegacyProtocolCore
@@ -66,6 +67,7 @@
     ThermocyclerContext,
     HeaterShakerContext,
     MagneticBlockContext,
+    AbsorbanceReaderContext,
     ModuleContext,
 )
 from ._parameters import Parameters
@@ -80,6 +82,7 @@
     ThermocyclerContext,
     HeaterShakerContext,
     MagneticBlockContext,
+    AbsorbanceReaderContext,
 ]
 
 
@@ -1204,6 +1207,8 @@ def _create_module_context(
         module_cls = HeaterShakerContext
     elif isinstance(module_core, AbstractMagneticBlockCore):
         module_cls = MagneticBlockContext
+    elif isinstance(module_core, AbstractAbsorbanceReaderCore):
+        module_cls = AbsorbanceReaderContext
     else:
         assert False, "Unsupported module type"
 

api/src/opentrons/protocol_api/validation.py

@@ -29,6 +29,7 @@
     ThermocyclerModuleModel,
     HeaterShakerModuleModel,
     MagneticBlockModel,
+    AbsorbanceReaderModel,
     ThermocyclerStep,
 )
 
@@ -272,6 +273,7 @@ def ensure_definition_is_labware(definition: LabwareDefinition) -> None:
     "thermocyclerModuleV2": ThermocyclerModuleModel.THERMOCYCLER_V2,
     "heaterShakerModuleV1": HeaterShakerModuleModel.HEATER_SHAKER_V1,
     "magneticBlockV1": MagneticBlockModel.MAGNETIC_BLOCK_V1,
+    "absorbanceReaderV1": AbsorbanceReaderModel.ABSORBANCE_READER_V1,
 }
 
 

api/src/opentrons/protocol_engine/clients/sync_client.py

@@ -873,3 +873,31 @@ def load_liquid(
         )
         result = self._transport.execute_command(request=request)
         return cast(commands.LoadLiquidResult, result)
+
+    def absorbance_reader_initialize(
+        self,
+        module_id: str,
+        wavelength: int,
+    ) -> commands.absorbance_reader.InitializeResult:
+        """Execute a `absorbanceReader/initialize` command and return the result."""
+        request = commands.absorbance_reader.InitializeCreate(
+            params=commands.absorbance_reader.InitializeParams(
+                moduleId=module_id, sampleWavelength=wavelength
+            )
+        )
+        result = self._transport.execute_command(request=request)
+        return cast(commands.absorbance_reader.InitializeResult, result)
+
+    def absorbance_reader_measure(
+        self,
+        module_id: str,
+        wavelength: int,
+    ) -> commands.absorbance_reader.MeasureAbsorbanceResult:
+        """Execute a `absorbanceReader/measure` command and return the result."""
+        request = commands.absorbance_reader.MeasureAbsorbanceCreate(
+            params=commands.absorbance_reader.MeasureAbsorbanceParams(
+                moduleId=module_id, sampleWavelength=wavelength
+            )
+        )
+        result = self._transport.execute_command(request=request)
+        return cast(commands.absorbance_reader.MeasureAbsorbanceResult, result)