draft

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

@@ -7,6 +7,7 @@
     ClassVar,
     runtime_checkable,
     TypeVar,
+    Literal,
 )
 
 Response = TypeVar("Response")

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

@@ -159,9 +159,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."""

api/src/opentrons/protocol_engine/commands/absorbance_reader/__init__.py

@@ -1,17 +1,17 @@
 """Command models for Absorbance Reader commands."""
 
 from .measure import (
-    AbsorbanceMeasureCommandType,
-    AbsorbanceMeasureParams,
-    AbsorbanceMeasureResult,
-    AbsorbanceMeasure,
-    AbsorbanceMeasureCreate,
+    MeasureAbsorbanceCommandType,
+    MeasureAbsorbanceParams,
+    MeasureAbsorbanceResult,
+    MeasureAbsorbance,
+    MeasureAbsorbanceCreate,
 )
 
 __all__ = [
-    "AbsorbanceMeasureCommandType",
-    "AbsorbanceMeasureParams",
-    "AbsorbanceMeasureResult",
-    "AbsorbanceMeasure",
-    "AbsorbanceMeasureCreate",
+    "MeasureAbsorbanceCommandType",
+    "MeasureAbsorbanceParams",
+    "MeasureAbsorbanceResult",
+    "MeasureAbsorbance",
+    "MeasureAbsorbanceCreate",
 ]

api/src/opentrons/protocol_engine/commands/absorbance_reader/measure.py

@@ -5,33 +5,33 @@
 
 from pydantic import BaseModel, Field
 
-from opentrons.hardware_control.modules.types import ThermocyclerStep
-
 from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate
 
 if TYPE_CHECKING:
     from opentrons.protocol_engine.state import StateView
     from opentrons.protocol_engine.execution import EquipmentHandler
 
 
-AbsorbanceMeasureCommandType = Literal["absorbanceReader/measure"]
+MeasureAbsorbanceCommandType = Literal["absorbanceReader/measure"]
 
 
-class AbsorbanceMeasureParams(BaseModel):
+class MeasureAbsorbanceParams(BaseModel):
     """Input parameters for a single absorbance reading."""
 
     moduleId: str = Field(..., description="Unique ID of the Thermocycler.")
-    sampleWavelength: float = Field(..., description="Sample wavelength in nm.")
+    sampleWavelength: int = Field(..., description="Sample wavelength in nm.")
 
 
-class AbsorbanceMeasureResult(BaseModel):
+class MeasureAbsorbanceResult(BaseModel):
     """Result data from running an aborbance reading."""
 
-    data: List[float] = Field(..., description="Absorbance data points.")
+    data: Optional[List[float]] = Field(
+        ..., min_items=96, max_items=96, description="Absorbance data points."
+    )
 
 
-class AbsorbanceMeasureImpl(
-    AbstractCommandImpl[AbsorbanceMeasureParams, AbsorbanceMeasureResult]
+class MeasureAbsorbanceImpl(
+    AbstractCommandImpl[MeasureAbsorbanceParams, MeasureAbsorbanceResult]
 ):
     """Execution implementation of a Thermocycler's run profile command."""
 
@@ -44,7 +44,7 @@ def __init__(
         self._state_view = state_view
         self._equipment = equipment
 
-    async def execute(self, params: AbsorbanceMeasureParams) -> AbsorbanceMeasureResult:
+    async def execute(self, params: MeasureAbsorbanceParams) -> MeasureAbsorbanceResult:
         """Initiate a single absorbance measurement."""
         abs_reader_substate = self._state_view.modules.get_absorbance_reader_substate(
             module_id=params.moduleId
@@ -53,23 +53,28 @@ async def execute(self, params: AbsorbanceMeasureParams) -> AbsorbanceMeasureRes
         abs_reader = self._equipment.get_module_hardware_api(
             abs_reader_substate.module_id
         )
-        return AbsorbanceMeasureResult(data=[])
+
+        if abs_reader is not None:
+            result = await abs_reader.start_measure(wavelength=params.sampleWavelength)
+            return MeasureAbsorbanceResult(data=result)
+
+        return MeasureAbsorbanceResult(data=None)
 
 
-class AbsorbanceMeasure(BaseCommand[AbsorbanceMeasureParams, AbsorbanceMeasureResult]):
+class MeasureAbsorbance(BaseCommand[MeasureAbsorbanceParams, MeasureAbsorbanceResult]):
     """A command to execute an Absorbance Reader measurement."""
 
-    commandType: AbsorbanceMeasureCommandType = "absorbanceReader/measure"
-    params: AbsorbanceMeasureParams
-    result: Optional[AbsorbanceMeasureResult]
+    commandType: MeasureAbsorbanceCommandType = "absorbanceReader/measure"
+    params: MeasureAbsorbanceParams
+    result: Optional[MeasureAbsorbanceResult]
 
-    _ImplementationCls: Type[AbsorbanceMeasureImpl] = AbsorbanceMeasureImpl
+    _ImplementationCls: Type[MeasureAbsorbanceImpl] = MeasureAbsorbanceImpl
 
 
-class AbsorbanceMeasureCreate(BaseCommandCreate[AbsorbanceMeasureParams]):
+class MeasureAbsorbanceCreate(BaseCommandCreate[MeasureAbsorbanceParams]):
     """A request to execute an Absorbance Reader measurement."""
 
-    commandType: AbsorbanceMeasureCommandType = "absorbanceReader/measure"
-    params: AbsorbanceMeasureParams
+    commandType: MeasureAbsorbanceCommandType = "absorbanceReader/measure"
+    params: MeasureAbsorbanceParams
 
-    _CommandCls: Type[AbsorbanceMeasure] = AbsorbanceMeasure
+    _CommandCls: Type[MeasureAbsorbance] = MeasureAbsorbance

api/src/opentrons/protocol_engine/commands/command_unions.py

@@ -5,6 +5,7 @@
 
 from pydantic import Field
 
+from . import absorbance_reader
 from . import heater_shaker
 from . import magnetic_module
 from . import temperature_module
@@ -353,6 +354,7 @@
         thermocycler.OpenLid,
         thermocycler.CloseLid,
         thermocycler.RunProfile,
+        absorbance_reader.MeasureAbsorbance,
         calibration.CalibrateGripper,
         calibration.CalibratePipette,
         calibration.CalibrateModule,
@@ -419,6 +421,7 @@
     thermocycler.CloseLidParams,
     thermocycler.RunProfileParams,
     thermocycler.RunProfileStepParams,
+    absorbance_reader.MeasureAbsorbanceParams,
     calibration.CalibrateGripperParams,
     calibration.CalibratePipetteParams,
     calibration.CalibrateModuleParams,
@@ -482,6 +485,7 @@
     thermocycler.OpenLidCommandType,
     thermocycler.CloseLidCommandType,
     thermocycler.RunProfileCommandType,
+    absorbance_reader.MeasureAbsorbanceCommandType,
     calibration.CalibrateGripperCommandType,
     calibration.CalibratePipetteCommandType,
     calibration.CalibrateModuleCommandType,
@@ -546,6 +550,7 @@
         thermocycler.OpenLidCreate,
         thermocycler.CloseLidCreate,
         thermocycler.RunProfileCreate,
+        absorbance_reader.MeasureAbsorbanceCreate,
         calibration.CalibrateGripperCreate,
         calibration.CalibratePipetteCreate,
         calibration.CalibrateModuleCreate,
@@ -611,6 +616,7 @@
     thermocycler.OpenLidResult,
     thermocycler.CloseLidResult,
     thermocycler.RunProfileResult,
+    absorbance_reader.MeasureAbsorbanceResult,
     calibration.CalibrateGripperResult,
     calibration.CalibratePipetteResult,
     calibration.CalibrateModuleResult,

api/src/opentrons/protocol_engine/execution/equipment.py

@@ -14,13 +14,15 @@
     HeaterShaker,
     TempDeck,
     Thermocycler,
+    AbsorbanceReader,
 )
 from opentrons.hardware_control.nozzle_manager import NozzleMap
 from opentrons.protocol_engine.state.module_substates import (
     MagneticModuleId,
     HeaterShakerModuleId,
     TemperatureModuleId,
     ThermocyclerModuleId,
+    AbsorbanceReaderId,
 )
 from ..errors import (
     FailedToLoadPipetteError,
@@ -488,6 +490,13 @@ def get_module_hardware_api(
     ) -> Optional[Thermocycler]:
         ...
 
+    @overload
+    def get_module_hardware_api(
+        self,
+        module_id: AbsorbanceReaderId,
+    ) -> Optional[AbsorbanceReader]:
+        ...
+
     def get_module_hardware_api(self, module_id: str) -> Optional[AbstractModule]:
         """Get the hardware API for a given module."""
         use_virtual_modules = self._state_store.config.use_virtual_modules

api/src/opentrons/protocol_engine/state/module_substates/absorbance_reader_substate.py

@@ -1,6 +1,6 @@
 """Heater-Shaker Module sub-state."""
 from dataclasses import dataclass
-from typing import List, NewType, Optional
+from typing import NewType
 
 
 AbsorbanceReaderId = NewType("AbsorbanceReaderId", str)
@@ -11,10 +11,3 @@ class AbsorbanceReaderSubState:
     """Absorbance-Plate-Reader-specific state."""
 
     module_id: AbsorbanceReaderId
-    initialized: bool
-    is_lid_open: bool
-    is_loaded: bool
-    is_measuring: bool
-    temperature: float
-    sample_wavelength: Optional[int]
-    supported_wavelengths: Optional[List[int]]

api/src/opentrons/protocol_engine/state/modules.py

@@ -62,10 +62,12 @@
     HeaterShakerModuleSubState,
     TemperatureModuleSubState,
     ThermocyclerModuleSubState,
+    AbsorbanceReaderSubState,
     MagneticModuleId,
     HeaterShakerModuleId,
     TemperatureModuleId,
     ThermocyclerModuleId,
+    AbsorbanceReaderId,
     MagneticBlockSubState,
     MagneticBlockId,
     ModuleSubStateType,
@@ -321,6 +323,10 @@ def _add_module_substate(
             self._state.substate_by_module_id[module_id] = MagneticBlockSubState(
                 module_id=MagneticBlockId(module_id)
             )
+        elif ModuleModel.is_absorbance_reader(actual_model):
+            self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState(
+                module_id=AbsorbanceReaderId(module_id)
+            )
 
     def _update_additional_slots_occupied_by_thermocycler(
         self,
@@ -644,6 +650,22 @@ def get_thermocycler_module_substate(
             expected_name="Thermocycler Module",
         )
 
+    def get_absorbance_reader_substate(
+        self, module_id: str
+    ) -> AbsorbanceReaderSubState:
+        """Return a `AbsorbanceReaderSubState` for the given Absorbance Reader.
+
+        Raises:
+           ModuleNotLoadedError: If module_id has not been loaded.
+           WrongModuleTypeError: If module_id has been loaded,
+               but it's not an Absorbance Reader.
+        """
+        return self._get_module_substate(
+            module_id=module_id,
+            expected_type=AbsorbanceReaderSubState,
+            expected_name="Thermocycler Module",
+        )
+
     def get_location(self, module_id: str) -> DeckSlotLocation:
         """Get the slot location of the given module."""
         location = self.get(module_id).location

api/src/opentrons/protocol_engine/types.py

@@ -306,6 +306,7 @@ class ModuleModel(str, Enum):
     THERMOCYCLER_MODULE_V2 = "thermocyclerModuleV2"
     HEATER_SHAKER_MODULE_V1 = "heaterShakerModuleV1"
     MAGNETIC_BLOCK_V1 = "magneticBlockV1"
+    ABSORBANCE_READER_V1 = "absorbanceReaderV1"
 
     def as_type(self) -> ModuleType:
         """Get the ModuleType of this model."""
@@ -319,6 +320,8 @@ def as_type(self) -> ModuleType:
             return ModuleType.HEATER_SHAKER
         elif ModuleModel.is_magnetic_block(self):
             return ModuleType.MAGNETIC_BLOCK
+        elif ModuleModel.is_absorbance_reader(self):
+            return ModuleType.ABSORBANCE_READER
 
         assert False, f"Invalid ModuleModel {self}"
 
@@ -355,6 +358,13 @@ def is_magnetic_block(cls, model: ModuleModel) -> TypeGuard[MagneticBlockModel]:
         """Whether a given model is a Magnetic block."""
         return model == cls.MAGNETIC_BLOCK_V1
 
+    @classmethod
+    def is_absorbance_reader(
+        cls, model: ModuleModel
+    ) -> TypeGuard[AbsorbanceReaderModel]:
+        """Whether a given model is a Magnetic block."""
+        return model == cls.ABSORBANCE_READER_V1
+
 
 TemperatureModuleModel = Literal[
     ModuleModel.TEMPERATURE_MODULE_V1, ModuleModel.TEMPERATURE_MODULE_V2
@@ -367,6 +377,7 @@ def is_magnetic_block(cls, model: ModuleModel) -> TypeGuard[MagneticBlockModel]:
 ]
 HeaterShakerModuleModel = Literal[ModuleModel.HEATER_SHAKER_MODULE_V1]
 MagneticBlockModel = Literal[ModuleModel.MAGNETIC_BLOCK_V1]
+AbsorbanceReaderModel = Literal[ModuleModel.ABSORBANCE_READER_V1]
 
 
 class ModuleDimensions(BaseModel):