fix(hardware,api): remove hardware import from simulate

api/src/opentrons/hardware_control/backends/flex_protocol.py

@@ -36,10 +36,9 @@
     HepaFanState,
     HepaUVState,
     StatusBarState,
+    PipetteSensorResponseQueue,
 )
 from opentrons.hardware_control.module_control import AttachedModulesControl
-from opentrons_hardware.firmware_bindings.constants import SensorId
-from opentrons_hardware.sensors.types import SensorDataType
 from ..dev_types import OT3AttachedInstruments
 from .types import HWStopCondition
 
@@ -168,9 +167,7 @@ async def liquid_probe(
         num_baseline_reads: int,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
         force_both_sensors: bool = False,
-        response_queue: Optional[
-            asyncio.Queue[Dict[SensorId, List[SensorDataType]]]
-        ] = None,
+        response_queue: Optional[PipetteSensorResponseQueue] = None,
     ) -> float:
         ...
 

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -104,7 +104,6 @@
     ErrorCode,
     SensorId,
 )
-from opentrons_hardware.sensors.types import SensorDataType
 from opentrons_hardware.firmware_bindings.messages.message_definitions import (
     StopRequest,
 )
@@ -142,6 +141,10 @@
     EstopState,
     HardwareEventHandler,
     HardwareEventUnsubscriber,
+    PipetteSensorId,
+    PipetteSensorType,
+    PipetteSensorData,
+    PipetteSensorResponseQueue,
 )
 from opentrons.hardware_control.errors import (
     InvalidPipetteName,
@@ -212,6 +215,7 @@
 from .types import HWStopCondition
 from .flex_protocol import FlexBackend
 from .status_bar_state import StatusBarStateController
+from opentrons_hardware.sensors.types import SensorDataType
 
 log = logging.getLogger(__name__)
 
@@ -967,7 +971,6 @@ def _build_tip_action_group(
     async def tip_action(
         self, origin: float, targets: List[Tuple[float, float]]
     ) -> None:
-
         move_group = self._build_tip_action_group(origin, targets)
         runner = MoveGroupRunner(
             move_groups=[move_group],
@@ -1492,9 +1495,7 @@ async def liquid_probe(
         num_baseline_reads: int,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
         force_both_sensors: bool = False,
-        response_queue: Optional[
-            asyncio.Queue[Dict[SensorId, List[SensorDataType]]]
-        ] = None,
+        response_queue: Optional[PipetteSensorResponseQueue] = None,
     ) -> float:
         head_node = axis_to_node(Axis.by_mount(mount))
         tool = sensor_node_for_pipette(OT3Mount(mount.value))
@@ -1503,6 +1504,27 @@ async def liquid_probe(
                 "Liquid Presence Detection not available on this pipette."
             )
 
+        if response_queue is None:
+            response_capture: Optional[
+                Callable[[Dict[SensorId, List[SensorDataType]]], None]
+            ] = None
+        else:
+
+            def response_capture(data: Dict[SensorId, List[SensorDataType]]) -> None:
+                response_queue.put_nowait(
+                    {
+                        PipetteSensorId(sensor_id.value): [
+                            PipetteSensorData(
+                                sensor_type=PipetteSensorType(packet.sensor_type.value),
+                                _as_int=packet.to_int,
+                                _as_float=packet.to_float(),
+                            )
+                            for packet in packets
+                        ]
+                        for sensor_id, packets in data.items()
+                    }
+                )
+
         positions = await liquid_probe(
             messenger=self._messenger,
             tool=tool,
@@ -1515,7 +1537,7 @@ async def liquid_probe(
             num_baseline_reads=num_baseline_reads,
             sensor_id=sensor_id_for_instrument(probe),
             force_both_sensors=force_both_sensors,
-            response_queue=response_queue,
+            emplace_data=response_capture,
         )
         for node, point in positions.items():
             self._position.update({node: point.motor_position})

api/src/opentrons/hardware_control/backends/ot3simulator.py

@@ -45,6 +45,7 @@
     EstopPhysicalStatus,
     HardwareEventHandler,
     HardwareEventUnsubscriber,
+    PipetteSensorResponseQueue,
 )
 
 from opentrons_shared_data.pipette.types import PipetteName, PipetteModel
@@ -62,9 +63,9 @@
 )
 from opentrons.util.async_helpers import ensure_yield
 from .types import HWStopCondition
-from .flex_protocol import FlexBackend
-from opentrons_hardware.firmware_bindings.constants import SensorId
-from opentrons_hardware.sensors.types import SensorDataType
+from .flex_protocol import (
+    FlexBackend,
+)
 
 log = logging.getLogger(__name__)
 
@@ -354,9 +355,7 @@ async def liquid_probe(
         num_baseline_reads: int,
         probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
         force_both_sensors: bool = False,
-        response_queue: Optional[
-            asyncio.Queue[Dict[SensorId, List[SensorDataType]]]
-        ] = None,
+        response_queue: Optional[PipetteSensorResponseQueue] = None,
     ) -> float:
         z_axis = Axis.by_mount(mount)
         pos = self._position

api/src/opentrons/hardware_control/ot3api.py

@@ -98,6 +98,7 @@
     EstopState,
     HardwareFeatureFlags,
     FailedTipStateCheck,
+    PipetteSensorResponseQueue,
 )
 from .errors import (
     UpdateOngoingError,
@@ -143,8 +144,6 @@
 from .backends.flex_protocol import FlexBackend
 from .backends.ot3simulator import OT3Simulator
 from .backends.errors import SubsystemUpdating
-from opentrons_hardware.firmware_bindings.constants import SensorId
-from opentrons_hardware.sensors.types import SensorDataType
 
 mod_log = logging.getLogger(__name__)
 
@@ -2679,9 +2678,7 @@ async def _liquid_probe_pass(
         probe: InstrumentProbeType,
         p_travel: float,
         force_both_sensors: bool = False,
-        response_queue: Optional[
-            asyncio.Queue[Dict[SensorId, List[SensorDataType]]]
-        ] = None,
+        response_queue: Optional[PipetteSensorResponseQueue] = None,
     ) -> float:
         plunger_direction = -1 if probe_settings.aspirate_while_sensing else 1
         end_z = await self._backend.liquid_probe(
@@ -2715,9 +2712,7 @@ async def liquid_probe(  # noqa: C901
         probe_settings: Optional[LiquidProbeSettings] = None,
         probe: Optional[InstrumentProbeType] = None,
         force_both_sensors: bool = False,
-        response_queue: Optional[
-            asyncio.Queue[Dict[SensorId, List[SensorDataType]]]
-        ] = None,
+        response_queue: Optional[PipetteSensorResponseQueue] = None,
     ) -> float:
         """Search for and return liquid level height.
 

api/src/opentrons/hardware_control/types.py

@@ -1,3 +1,4 @@
+from asyncio import Queue
 import enum
 import logging
 from dataclasses import dataclass
@@ -712,3 +713,63 @@ def __init__(
         super().__init__(
             f"Expected tip state {expected_state}, but received {actual_state}."
         )
+
+
+@enum.unique
+class PipetteSensorId(int, enum.Enum):
+    """Sensor IDs available.
+
+    Not to be confused with SensorType. This is the ID value that separate
+    two or more of the same type of sensor within a system.
+
+    Note that this is a copy of an enum defined in opentrons_hardware.firmware_bindings.constants. That version
+    is authoritative; this version is here because this data is exposed above the hardware control layer and
+    therefore needs a typing source here so that we don't create a dependency on the internal hardware package.
+    """
+
+    S0 = 0x0
+    S1 = 0x1
+    UNUSED = 0x2
+    BOTH = 0x3
+
+
+@enum.unique
+class PipetteSensorType(int, enum.Enum):
+    """Sensor types available.
+
+    Note that this is a copy of an enum defined in opentrons_hardware.firmware_bindings.constants. That version
+    is authoritative; this version is here because this data is exposed above the hardware control layer and
+    therefore needs a typing source here so that we don't create a dependency on the internal hardware package.
+    """
+
+    tip = 0x00
+    capacitive = 0x01
+    environment = 0x02
+    pressure = 0x03
+    pressure_temperature = 0x04
+    humidity = 0x05
+    temperature = 0x06
+
+
+@dataclass(frozen=True)
+class PipetteSensorData:
+    """Sensor data from a monitored sensor.
+
+    Note that this is a copy of an enum defined in opentrons_hardware.firmware_bindings.constants. That version
+    is authoritative; this version is here because this data is exposed above the hardware control layer and
+    therefore needs a typing source here so that we don't create a dependency on the internal hardware package.
+    """
+
+    sensor_type: PipetteSensorType
+    _as_int: int
+    _as_float: float
+
+    def to_float(self) -> float:
+        return self._as_float
+
+    @property
+    def to_int(self) -> int:
+        return self._as_int
+
+
+PipetteSensorResponseQueue = Queue[Dict[PipetteSensorId, List[PipetteSensorData]]]

hardware/opentrons_hardware/hardware_control/tool_sensors.py

@@ -271,8 +271,8 @@ async def liquid_probe(
     num_baseline_reads: int,
     sensor_id: SensorId = SensorId.S0,
     force_both_sensors: bool = False,
-    response_queue: Optional[
-        asyncio.Queue[Dict[SensorId, List[SensorDataType]]]
+    emplace_data: Optional[
+        Callable[[Dict[SensorId, List[SensorDataType]]], None]
     ] = None,
 ) -> Dict[NodeId, MotorPositionStatus]:
     """Move the mount and pipette simultaneously while reading from the pressure sensor."""
@@ -360,12 +360,12 @@ async def liquid_probe(
         await finalize_logs(messenger, tool, listeners, pressure_sensors)
 
     # give response data to any consumer that wants it
-    if response_queue:
+    if emplace_data:
         for s_id in listeners.keys():
             data = listeners[s_id].get_data()
             if data:
                 for d in data:
-                    response_queue.put_nowait({s_id: data})
+                    emplace_data({s_id: data})
 
     return positions