feat(api): Validate plate reader status using live data hookups for engine and introduce lid status to the PAPI

api/docs/v2/conf.py

@@ -445,7 +445,7 @@
     ("py:class", r".*protocol_api\.config.*"),
     ("py:class", r".*opentrons_shared_data.*"),
     ("py:class", r".*protocol_api._parameters.Parameters.*"),
-    ("py:class", r".*AbsorbanceReaderContext"),  # shh it's a secret (for now)
+    ("py:class", r".*AbsorbanceReaderContext"), 
 	("py:class", r".*RobotContext"),  # shh it's a secret (for now)
     ("py:class", r'.*AbstractLabware|APIVersion|LabwareLike|LoadedCoreMap|ModuleTypes|NoneType|OffDeckType|ProtocolCore|WellCore'),  # laundry list of not fully qualified things
 ]

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

@@ -61,6 +61,7 @@ async def get_available_wavelengths(self) -> List[int]:
         return await self._connection.get_supported_wavelengths()
 
     async def get_single_measurement(self, wavelength: int) -> List[float]:
+        # TODO (cb, 08-02-2024): The list of measurements for 96 wells is rotated 180 degrees (well A1 is where well H12 should be) this must be corrected
         return await self._connection.get_single_measurement(wavelength)
 
     async def initialize_measurement(self, wavelength: int) -> None:

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

@@ -204,11 +204,11 @@ def status(self) -> AbsorbanceReaderStatus:
 
     @property
     def lid_status(self) -> AbsorbanceReaderLidStatus:
-        return AbsorbanceReaderLidStatus.UNKNOWN
+        return self._reader.lid_status
 
     @property
     def plate_presence(self) -> AbsorbanceReaderPlatePresence:
-        return AbsorbanceReaderPlatePresence.UNKNOWN
+        return self._reader.plate_presence
 
     @property
     def device_info(self) -> Mapping[str, str]:
@@ -321,6 +321,11 @@ async def get_current_wavelength(self) -> None:
         """Get the Absorbance Reader's current active wavelength."""
         pass  # TODO: implement
 
+    async def get_current_lid_status(self) -> AbsorbanceReaderLidStatus:
+        """Get the Absorbance Reader's current lid status."""
+        await self._reader.get_lid_status()
+        return self._reader.lid_status
+
     def _enter_error_state(self, error: Exception) -> None:
         self._error = str(error)
         if isinstance(error, AbsorbanceReaderDisconnectedError):

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

@@ -562,3 +562,10 @@ def open_lid(self) -> None:
                 moduleId=self.module_id,
             )
         )
+
+    def is_lid_on(self) -> bool:
+        """Returns True if the Absorbance Reader's lid is currently on the Reader slot."""
+        abs_state = self._engine_client.state.modules.get_absorbance_reader_substate(
+            self.module_id
+        )
+        return abs_state.is_lid_on

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

@@ -369,3 +369,7 @@ def close_lid(self) -> None:
     @abstractmethod
     def open_lid(self) -> None:
         """Open the Absorbance Reader's lid."""
+
+    @abstractmethod
+    def is_lid_on(self) -> bool:
+        """Return True if the Absorbance Reader's lid is currently closed."""

api/src/opentrons/protocol_api/module_contexts.py

@@ -994,9 +994,14 @@ def close_lid(self) -> None:
 
     @requires_version(2, 21)
     def open_lid(self) -> None:
-        """Close the lid of the Absorbance Reader."""
+        """Open the lid of the Absorbance Reader."""
         self._core.open_lid()
 
+    @requires_version(2, 21)
+    def is_lid_on(self) -> bool:
+        """Return True if the Absorbance Reader's lid is currently closed."""
+        return self._core.is_lid_on()
+
     @requires_version(2, 21)
     def initialize(self, wavelength: int) -> None:
         """Initialize the Absorbance Reader by taking zero reading."""

api/src/opentrons/protocol_api/protocol_context.py

@@ -42,6 +42,7 @@
     RobotTypeError,
     UnsupportedAPIError,
 )
+from opentrons_shared_data.errors.exceptions import CommandPreconditionViolated
 
 from ._types import OffDeckType
 from .core.common import ModuleCore, LabwareCore, ProtocolCore
@@ -707,6 +708,13 @@ def move_labware(
                 f"Expected labware of type 'Labware' but got {type(labware)}."
             )
 
+        # Ensure that when moving to an absorbance reader than the lid is open
+        if isinstance(new_location, AbsorbanceReaderContext):
+            if new_location.is_lid_on():
+                raise CommandPreconditionViolated(
+                    f"Cannot move {labware.name} onto the Absorbance Reader Module when Lid is Closed."
+                )
+
         location: Union[
             ModuleCore,
             LabwareCore,

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

@@ -7,10 +7,14 @@
 
 from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
 from ...errors.error_occurrence import ErrorOccurrence
+from ...errors import CannotPerformModuleAction
 from opentrons.protocol_engine.types import AddressableAreaLocation
+
 from opentrons.protocol_engine.resources import labware_validation
 from .types import MoveLidResult
 
+from opentrons.drivers.types import AbsorbanceReaderLidStatus
+
 if TYPE_CHECKING:
     from opentrons.protocol_engine.state import StateView
     from opentrons.protocol_engine.execution import (
@@ -55,17 +59,30 @@ async def execute(
         mod_substate = self._state_view.modules.get_absorbance_reader_substate(
             module_id=params.moduleId
         )
-        # Make sure the lid is open
-        mod_substate.raise_if_lid_status_not_expected(lid_on_expected=False)
 
-        # Allow propagation of ModuleNotAttachedError.
-        _ = self._equipment.get_module_hardware_api(mod_substate.module_id)
-
-        # lid should currently be docked
+        # lid should currently be on the module
         assert mod_substate.lid_id is not None
         loaded_lid = self._state_view.labware.get(mod_substate.lid_id)
         assert labware_validation.is_absorbance_reader_lid(loaded_lid.loadName)
 
+        # If the lid is already Closed, No-op out
+        if mod_substate.is_lid_on:
+            current_offset_id = self._equipment.find_applicable_labware_offset_id(
+                labware_definition_uri=loaded_lid.definitionUri,
+                labware_location=loaded_lid.location,
+            )
+            return SuccessData(
+                public=CloseLidResult(
+                    lidId=loaded_lid.id,
+                    newLocation=loaded_lid.location,
+                    offsetId=current_offset_id,
+                ),
+                private=None,
+            )
+
+        # Allow propagation of ModuleNotAttachedError.
+        _ = self._equipment.get_module_hardware_api(mod_substate.module_id)
+
         current_location = loaded_lid.location
         validated_current_location = (
             self._state_view.geometry.ensure_valid_gripper_location(current_location)
@@ -108,6 +125,21 @@ async def execute(
             labware_definition_uri=loaded_lid.definitionUri,
             labware_location=new_location,
         )
+
+        if not self._state_view.config.use_virtual_modules:
+            abs_reader = self._equipment.get_module_hardware_api(mod_substate.module_id)
+
+            if abs_reader is not None:
+                result = await abs_reader.get_current_lid_status()
+                if result is not AbsorbanceReaderLidStatus.ON:
+                    raise CannotPerformModuleAction(
+                        "The Opentrons Plate Reader lid mechanicaly position did not match expected Closed state."
+                    )
+            else:
+                raise CannotPerformModuleAction(
+                    "Could not reach the Hardware API for Opentrons Plate Reader Module."
+                )
+
         return SuccessData(
             public=CloseLidResult(
                 lidId=loaded_lid.id, newLocation=new_location, offsetId=new_offset_id

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

@@ -7,9 +7,13 @@
 
 from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
 from ...errors.error_occurrence import ErrorOccurrence
+from ...errors import CannotPerformModuleAction
+
 from .types import MoveLidResult
 from opentrons.protocol_engine.resources import labware_validation
 
+from opentrons.drivers.types import AbsorbanceReaderLidStatus
+
 if TYPE_CHECKING:
     from opentrons.protocol_engine.state import StateView
     from opentrons.protocol_engine.execution import (
@@ -50,17 +54,29 @@ async def execute(self, params: OpenLidParams) -> SuccessData[OpenLidResult, Non
         mod_substate = self._state_view.modules.get_absorbance_reader_substate(
             module_id=params.moduleId
         )
-        # Make sure the lid is closed
-        mod_substate.raise_if_lid_status_not_expected(lid_on_expected=True)
-
-        # Allow propagation of ModuleNotAttachedError.
-        _ = self._equipment.get_module_hardware_api(mod_substate.module_id)
-
         # lid should currently be on the module
         assert mod_substate.lid_id is not None
         loaded_lid = self._state_view.labware.get(mod_substate.lid_id)
         assert labware_validation.is_absorbance_reader_lid(loaded_lid.loadName)
 
+        # If the lid is already Open, No-op out
+        if not mod_substate.is_lid_on:
+            current_offset_id = self._equipment.find_applicable_labware_offset_id(
+                labware_definition_uri=loaded_lid.definitionUri,
+                labware_location=loaded_lid.location,
+            )
+            return SuccessData(
+                public=OpenLidResult(
+                    lidId=loaded_lid.id,
+                    newLocation=loaded_lid.location,
+                    offsetId=current_offset_id,
+                ),
+                private=None,
+            )
+
+        # Allow propagation of ModuleNotAttachedError.
+        _ = self._equipment.get_module_hardware_api(mod_substate.module_id)
+
         current_location = loaded_lid.location
         validated_current_location = (
             self._state_view.geometry.ensure_valid_gripper_location(current_location)
@@ -94,6 +110,21 @@ async def execute(self, params: OpenLidParams) -> SuccessData[OpenLidResult, Non
             labware_definition_uri=loaded_lid.definitionUri,
             labware_location=new_location,
         )
+
+        if not self._state_view.config.use_virtual_modules:
+            abs_reader = self._equipment.get_module_hardware_api(mod_substate.module_id)
+
+            if abs_reader is not None:
+                result = await abs_reader.get_current_lid_status()
+                if result is not AbsorbanceReaderLidStatus.OFF:
+                    raise CannotPerformModuleAction(
+                        "The Opentrons Plate Reader lid mechanicaly position did not match expected Open state."
+                    )
+            else:
+                raise CannotPerformModuleAction(
+                    "Could not reach the Hardware API for Opentrons Plate Reader Module."
+                )
+
         return SuccessData(
             public=OpenLidResult(
                 lidId=loaded_lid.id,

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

@@ -16,6 +16,8 @@
 
 from opentrons.protocol_engine.resources import deck_configuration_provider
 
+from opentrons.drivers.types import AbsorbanceReaderLidStatus
+
 if TYPE_CHECKING:
     from ..state import StateView
     from ..execution import EquipmentHandler
@@ -151,6 +153,39 @@ async def execute(
                 module_id=params.moduleId,
             )
 
+        # Handle lid position update for loaded Plate Reader module on deck
+        if (
+            not self._state_view.config.use_virtual_modules
+            and params.model == ModuleModel.ABSORBANCE_READER_V1
+            and params.moduleId is not None
+        ):
+            abs_reader = self._equipment.get_module_hardware_api(
+                self._state_view.modules.get_absorbance_reader_substate(
+                    params.moduleId
+                ).module_id
+            )
+
+            if abs_reader is not None:
+                result = await abs_reader.get_current_lid_status()
+                if (
+                    isinstance(result, AbsorbanceReaderLidStatus)
+                    and result is not AbsorbanceReaderLidStatus.ON
+                ):
+                    reader_area = self._state_view.modules.ensure_and_convert_module_fixture_location(
+                        params.location.slotName,
+                        self._state_view.config.deck_type,
+                        params.model,
+                    )
+                    lid_labware = self._state_view.labware.get_by_addressable_area(
+                        reader_area
+                    )
+                    if lid_labware is not None:
+                        lid_labware.location = (
+                            self._state_view.modules.absorbance_reader_dock_location(
+                                params.moduleId
+                            )
+                        )
+
         return SuccessData(
             public=LoadModuleResult(
                 moduleId=loaded_module.module_id,

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

@@ -322,6 +322,22 @@ def get_by_slot(
 
         return None
 
+    def get_by_addressable_area(
+        self,
+        addressable_area: str,
+    ) -> Optional[LoadedLabware]:
+        """Get the labware located in a given addressable area, if any."""
+        loaded_labware = list(self._state.labware_by_id.values())
+
+        for labware in loaded_labware:
+            if (
+                isinstance(labware.location, AddressableAreaLocation)
+                and labware.location.addressableAreaName == addressable_area
+            ):
+                return labware
+
+        return None
+
     def get_definition(self, labware_id: str) -> LabwareDefinition:
         """Get labware definition by the labware's unique identifier."""
         return self.get_definition_by_uri(