add flex stacker data as rebase

Opentrons · Jan 9, 2025 · 5d65076 · 5d65076
1 parent 70a3461
commit 5d65076
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 10 deletions.
diff --git a/api/src/opentrons/hardware_control/modules/flex_stacker.py b/api/src/opentrons/hardware_control/modules/flex_stacker.py
@@ -30,6 +30,7 @@
     StackerAxisState,
     UploadFunction,
     LiveData,
+    FlexStackerData,
 )
 
 log = logging.getLogger(__name__)
@@ -189,17 +190,15 @@ def is_simulated(self) -> bool:
 
     @property
     def live_data(self) -> LiveData:
-        return {
-            "status": self.status.value,
-            "data": {
-                "latchState": self.latch_state.value,
-                "platformState": self.platform_state.value,
-                "hopperDoorState": self.hopper_door_state.value,
-                "axisStateX": self.limit_switch_status[StackerAxis.X].value,
-                "axisStateZ": self.limit_switch_status[StackerAxis.Z].value,
-                "errorDetails": self._reader.error,
-            },
+        data: FlexStackerData = {
+            "latchState": self.latch_state.value,
+            "platformState": self.platform_state.value,
+            "hopperDoorState": self.hopper_door_state.value,
+            "axisStateX": self.limit_switch_status[StackerAxis.X].value,
+            "axisStateZ": self.limit_switch_status[StackerAxis.Z].value,
+            "errorDetails": self._reader.error,
         }
+        return {"status": self.status.value, "data": data}
 
     async def prep_for_update(self) -> str:
         await self._poller.stop()

diff --git a/api/src/opentrons/hardware_control/modules/types.py b/api/src/opentrons/hardware_control/modules/types.py
@@ -103,13 +103,23 @@ class AbsorbanceReaderData(TypedDict):
     referenceWavelength: int
 
 
+class FlexStackerData(TypedDict):
+    latchState: str
+    platformState: str
+    hopperDoorState: str
+    axisStateX: str
+    axisStateZ: str
+    errorDetails: str | None
+
+
 ModuleData = Union[
     Dict[Any, Any],  # This allows an empty dict as module data
     MagneticModuleData,
     TemperatureModuleData,
     HeaterShakerData,
     ThermocyclerData,
     AbsorbanceReaderData,
+    FlexStackerData,
 ]
 
 
@@ -144,6 +154,12 @@ def is_absorbance_reader_data(
     ) -> TypeGuard[AbsorbanceReaderData]:
         return data is not None and "uptime" in data.keys()
 
+    @classmethod
+    def is_flex_stacker_data(
+        cls, data: ModuleData | None
+    ) -> TypeGuard[FlexStackerData]:
+        return data is not None and "platformState" in data.keys()
+
 
 class LiveData(TypedDict):
     status: str

diff --git a/robot-server/robot_server/modules/module_data_mapper.py b/robot-server/robot_server/modules/module_data_mapper.py
@@ -168,6 +168,7 @@ def map_data(
             )
         elif module_type == ModuleType.FLEX_STACKER:
             module_cls = FlexStackerModule
+            assert ModuleDataValidator.is_flex_stacker_data(live_data["data"])
             module_data = FlexStackerModuleData(
                 status=FlexStackerStatus(live_data["status"]),
                 latchState=cast(LatchState, live_data["data"].get("latchState")),