feat(api): Plate Reader live data hookup for read behavior (#16024)

Covers PLAT-204, PLAT-467
Ensure the Plate reader .read() method utilizes live data, initiates a
read and returns the data from the plate reader module.

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

@@ -1,7 +1,7 @@
 """Protocol API module implementation logic."""
 from __future__ import annotations
 
-from typing import Optional, List
+from typing import Optional, List, Dict
 
 from opentrons.hardware_control import SynchronousAdapter, modules as hw_modules
 from opentrons.hardware_control.modules.types import (
@@ -23,6 +23,7 @@
 from opentrons.protocol_engine.errors.exceptions import (
     LabwareNotLoadedOnModuleError,
     NoMagnetEngageHeightError,
+    CannotPerformModuleAction,
 )
 
 from opentrons.protocols.api_support.types import APIVersion
@@ -536,14 +537,26 @@ def initialize(self, wavelength: int) -> None:
         )
         self._initialized_value = wavelength
 
-    def read(self) -> None:
-        """Initiate read on the Absorbance Reader."""
+    def read(self) -> Optional[Dict[str, float]]:
+        """Initiate a read on the Absorbance Reader, and return the results. During Analysis, this will return None."""
         if self._initialized_value:
             self._engine_client.execute_command(
                 cmd.absorbance_reader.ReadAbsorbanceParams(
                     moduleId=self.module_id, sampleWavelength=self._initialized_value
                 )
             )
+        if not self._engine_client.state.config.use_virtual_modules:
+            read_result = (
+                self._engine_client.state.modules.get_absorbance_reader_substate(
+                    self.module_id
+                ).data
+            )
+            if read_result is not None:
+                return read_result
+            raise CannotPerformModuleAction(
+                "Absorbance Reader failed to return expected read result."
+            )
+        return None
 
     def close_lid(
         self,

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

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import List, Optional, TypeVar, ClassVar
+from typing import List, Dict, Optional, TypeVar, ClassVar
 
 from opentrons.drivers.types import (
     HeaterShakerLabwareLatchStatus,
@@ -359,7 +359,7 @@ def initialize(self, wavelength: int) -> None:
         """Initialize the Absorbance Reader by taking zero reading."""
 
     @abstractmethod
-    def read(self) -> None:
+    def read(self) -> Optional[Dict[str, float]]:
         """Get an absorbance reading from the Absorbance Reader."""
 
     @abstractmethod

api/src/opentrons/protocol_api/module_contexts.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import List, Optional, Union, cast
+from typing import List, Dict, Optional, Union, cast
 
 from opentrons_shared_data.labware.types import LabwareDefinition
 from opentrons_shared_data.module.types import ModuleModel, ModuleType
@@ -1008,6 +1008,6 @@ def initialize(self, wavelength: int) -> None:
         self._core.initialize(wavelength)
 
     @requires_version(2, 21)
-    def read(self) -> None:
-        """Initiate read on the Absorbance Reader."""
-        self._core.read()
+    def read(self) -> Optional[Dict[str, float]]:
+        """Initiate read on the Absorbance Reader. Returns a dictionary of values ordered by well name."""
+        return self._core.read()

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

@@ -1,11 +1,12 @@
 """Command models to read absorbance."""
 from __future__ import annotations
-from typing import List, Optional, TYPE_CHECKING
+from typing import Optional, Dict, TYPE_CHECKING
 from typing_extensions import Literal, Type
 
 from pydantic import BaseModel, Field
 
 from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
+from ...errors import CannotPerformModuleAction
 from ...errors.error_occurrence import ErrorOccurrence
 
 if TYPE_CHECKING:
@@ -24,12 +25,9 @@ class ReadAbsorbanceParams(BaseModel):
 
 
 class ReadAbsorbanceResult(BaseModel):
-    """Result data from running an aborbance reading."""
+    """Result data from running an aborbance reading, returned as a dictionary map of values by well name (eg. ("A1": 0.0, ...))."""
 
-    # TODO: Transform this into a more complex model, such as a map with well names.
-    data: Optional[List[float]] = Field(
-        ..., min_items=96, max_items=96, description="Absorbance data points."
-    )
+    data: Optional[Dict[str, float]] = Field(..., description="Absorbance data points.")
 
 
 class ReadAbsorbanceImpl(
@@ -58,10 +56,20 @@ async def execute(
             abs_reader_substate.module_id
         )
 
+        if abs_reader_substate.configured is False:
+            raise CannotPerformModuleAction(
+                "Cannot perform Read action on Absorbance Reader without calling `.initialize(...)` first."
+            )
+
         if abs_reader is not None:
             result = await abs_reader.start_measure(wavelength=params.sampleWavelength)
+            converted_values = (
+                self._state_view.modules.convert_absorbance_reader_data_points(
+                    data=result
+                )
+            )
             return SuccessData(
-                public=ReadAbsorbanceResult(data=result),
+                public=ReadAbsorbanceResult(data=converted_values),
                 private=None,
             )
 

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 NewType, Optional, List
+from typing import NewType, Optional, Dict
 
 from opentrons.protocol_engine.errors import CannotPerformModuleAction
 
@@ -16,7 +16,7 @@ class AbsorbanceReaderSubState:
     configured: bool
     measured: bool
     is_lid_on: bool
-    data: Optional[List[float]]
+    data: Optional[Dict[str, float]]
     configured_wavelength: Optional[int]
     lid_id: Optional[str]
 

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

@@ -613,6 +613,7 @@ def _handle_absorbance_reader_commands(
         configured_wavelength = absorbance_reader_substate.configured_wavelength
         is_lid_on = absorbance_reader_substate.is_lid_on
         lid_id = absorbance_reader_substate.lid_id
+        data = absorbance_reader_substate.data
 
         if isinstance(command.result, absorbance_reader.InitializeResult):
             self._state.substate_by_module_id[module_id] = AbsorbanceReaderSubState(
@@ -642,7 +643,7 @@ def _handle_absorbance_reader_commands(
                 configured_wavelength=configured_wavelength,
                 is_lid_on=False,
                 measured=True,
-                data=None,
+                data=data,
                 lid_id=lid_id,
             )
 
@@ -653,7 +654,7 @@ def _handle_absorbance_reader_commands(
                 configured_wavelength=configured_wavelength,
                 is_lid_on=True,
                 measured=True,
-                data=None,
+                data=data,
                 lid_id=lid_id,
             )
 
@@ -1273,6 +1274,25 @@ def is_flex_deck_with_thermocycler(self) -> bool:
         else:
             return False
 
+    def convert_absorbance_reader_data_points(
+        self, data: List[float]
+    ) -> Dict[str, float]:
+        """Return the data from the Absorbance Reader module in a map of wells for each read value."""
+        if len(data) == 96:
+            # We have to reverse the reader values because the Opentrons Absorbance Reader is rotated 180 degrees on the deck
+            data.reverse()
+            well_map: Dict[str, float] = {}
+            for i, value in enumerate(data):
+                row = chr(ord("A") + i // 12)  # Convert index to row (A-H)
+                col = (i % 12) + 1  # Convert index to column (1-12)
+                well_key = f"{row}{col}"
+                well_map[well_key] = value
+            return well_map
+        else:
+            raise ValueError(
+                "Only readings of 96 Well labware are supported for conversion to map of values by well."
+            )
+
     def ensure_and_convert_module_fixture_location(
         self,
         deck_slot: DeckSlotName,