feat(hardware): add a new can message to batch read sensor data (#16370)

<!--
Thanks for taking the time to open a Pull Request (PR)! Please make sure
you've read the "Opening Pull Requests" section of our Contributing
Guide:


https://github.com/Opentrons/opentrons/blob/edge/CONTRIBUTING.md#opening-pull-requests

GitHub provides robust markdown to format your PR. Links, diagrams,
pictures, and videos along with text formatting make it possible to
create a rich and informative PR. For more information on GitHub
markdown, see:


https://docs.github.com/en/get-started/writing-on-github/getting-started-with-writing-and-formatting-on-github/basic-writing-and-formatting-syntax

To ensure your code is reviewed quickly and thoroughly, please fill out
the sections below to the best of your ability!
-->

# Overview

In a first step to streamline tool_sensors.py we want to batch read
sensor data, this will speed up the process and let us stream alot
better
<!--
Describe your PR at a high level. State acceptance criteria and how this
PR fits into other work. Link issues, PRs, and other relevant resources.
-->

## Test Plan and Hands on Testing

<!--
Describe your testing of the PR. Emphasize testing not reflected in the
code. Attach protocols, logs, screenshots and any other assets that
support your testing.
-->

## Changelog

<!--
List changes introduced by this PR considering future developers and the
end user. Give careful thought and clear documentation to breaking
changes.
-->

## Review requests

<!--
- What do you need from reviewers to feel confident this PR is ready to
merge?
- Ask questions.
-->

## Risk assessment

<!--
- Indicate the level of attention this PR needs.
- Provide context to guide reviewers.
- Discuss trade-offs, coupling, and side effects.
- Look for the possibility, even if you think it's small, that your
change may affect some other part of the system.
- For instance, changing return tip behavior may also change the
behavior of labware calibration.
- How do your unit tests and on hands on testing mitigate this PR's
risks and the risk of future regressions?
- Especially in high risk PRs, explain how you know your testing is
enough.
-->

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

@@ -54,6 +54,10 @@ async def get_serial_number(self) -> Optional[str]:
     def restore_system_constraints(self) -> AsyncIterator[None]:
         ...
 
+    @asynccontextmanager
+    def grab_pressure(self, channels: int, mount: OT3Mount) -> AsyncIterator[None]:
+        ...
+
     def update_constraints_for_gantry_load(self, gantry_load: GantryLoad) -> None:
         ...
 

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

@@ -161,6 +161,7 @@
     capacitive_pass,
     liquid_probe,
     check_overpressure,
+    grab_pressure,
 )
 from opentrons_hardware.hardware_control.rear_panel_settings import (
     get_door_state,
@@ -369,6 +370,14 @@ async def restore_system_constraints(self) -> AsyncIterator[None]:
             self._move_manager.update_constraints(old_system_constraints)
             log.debug(f"Restore previous system constraints: {old_system_constraints}")
 
+    @asynccontextmanager
+    async def grab_pressure(
+        self, channels: int, mount: OT3Mount
+    ) -> AsyncIterator[None]:
+        tool = axis_to_node(Axis.of_main_tool_actuator(mount))
+        async with grab_pressure(channels, tool, self._messenger):
+            yield
+
     def update_constraints_for_calibration_with_gantry_load(
         self,
         gantry_load: GantryLoad,

api/src/opentrons/hardware_control/ot3api.py

@@ -307,6 +307,12 @@ async def restore_system_constrants(self) -> AsyncIterator[None]:
         async with self._backend.restore_system_constraints():
             yield
 
+    @contextlib.asynccontextmanager
+    async def grab_pressure(self, mount: OT3Mount) -> AsyncIterator[None]:
+        instrument = self._pipette_handler.get_pipette(mount)
+        async with self._backend.grab_pressure(instrument.channels, mount):
+            yield
+
     def _update_door_state(self, door_state: DoorState) -> None:
         mod_log.info(f"Updating the window switch status: {door_state}")
         self.door_state = door_state

hardware/opentrons_hardware/firmware_bindings/constants.py

@@ -244,6 +244,7 @@ class MessageId(int, Enum):
     max_sensor_value_request = 0x70
     max_sensor_value_response = 0x71
 
+    batch_read_sensor_response = 0x81
     read_sensor_request = 0x82
     write_sensor_request = 0x83
     baseline_sensor_request = 0x84

hardware/opentrons_hardware/firmware_bindings/messages/fields.py

@@ -206,6 +206,14 @@ class FirmwareUpdateDataField(utils.BinaryFieldBase[bytes]):
     FORMAT = f"{NUM_BYTES}s"
 
 
+class BatchSensorDataField(utils.BinaryFieldBase[bytes]):
+    """The data field for many sensor data results."""
+
+    # 14 4-byte data points
+    NUM_BYTES = 56
+    FORMAT = f"{NUM_BYTES}s"
+
+
 class ErrorSeverityField(utils.UInt16Field):
     """A field for error severity."""
 

hardware/opentrons_hardware/firmware_bindings/messages/message_definitions.py

@@ -593,6 +593,17 @@ class ReadFromSensorResponse(BaseMessage):  # noqa: D101
     message_id: Literal[MessageId.read_sensor_response] = MessageId.read_sensor_response
 
 
+@dataclass
+class BatchReadFromSensorResponse(BaseMessage):  # noqa: D101
+    payload: payloads.BatchReadFromSensorResponsePayload
+    payload_type: Type[
+        payloads.BatchReadFromSensorResponsePayload
+    ] = payloads.BatchReadFromSensorResponsePayload
+    message_id: Literal[
+        MessageId.batch_read_sensor_response
+    ] = MessageId.batch_read_sensor_response
+
+
 @dataclass
 class SetSensorThresholdRequest(BaseMessage):  # noqa: D101
     payload: payloads.SetSensorThresholdRequestPayload

hardware/opentrons_hardware/firmware_bindings/messages/payloads.py

@@ -31,6 +31,7 @@
     GearMotorIdField,
     OptionalRevisionField,
     MotorUsageTypeField,
+    BatchSensorDataField,
 )
 from .. import utils
 
@@ -444,6 +445,14 @@ class ReadFromSensorResponsePayload(SensorPayload):
     sensor_data: utils.Int32Field
 
 
+@dataclass(eq=False)
+class BatchReadFromSensorResponsePayload(SensorPayload):
+    """A response for a batch of sensor responses."""
+
+    data_length: utils.UInt8Field
+    sensor_data: BatchSensorDataField
+
+
 @dataclass(eq=False)
 class SetSensorThresholdRequestPayload(SensorPayload):
     """A request to set the threshold value of a sensor."""

hardware/opentrons_hardware/hardware_control/tool_sensors.py

@@ -10,12 +10,13 @@
     Callable,
     AsyncContextManager,
     Optional,
+    AsyncIterator,
 )
 from logging import getLogger
 from numpy import float64
 from math import copysign
 from typing_extensions import Literal
-
+from contextlib import asynccontextmanager
 from opentrons_hardware.firmware_bindings.constants import (
     NodeId,
     SensorId,
@@ -663,3 +664,55 @@ def _drain() -> Iterator[float]:
                 break
 
     return list(_drain())
+
+
+@asynccontextmanager
+async def grab_pressure(
+    channels: int, tool: NodeId, messenger: CanMessenger
+) -> AsyncIterator[None]:
+    """Run some task and log the pressure."""
+    sensor_driver = SensorDriver()
+    sensor_id = SensorId.BOTH if channels > 1 else SensorId.S0
+    sensors: List[SensorId] = []
+    if sensor_id == SensorId.BOTH:
+        sensors.append(SensorId.S0)
+        sensors.append(SensorId.S1)
+    else:
+        sensors.append(sensor_id)
+
+    for sensor in sensors:
+        pressure_sensor = PressureSensor.build(
+            sensor_id=sensor,
+            node_id=tool,
+        )
+        num_baseline_reads = 10
+        # TODO: RH log this baseline and remove noqa
+        pressure_baseline = await sensor_driver.get_baseline(  # noqa: F841
+            messenger, pressure_sensor, num_baseline_reads
+        )
+        await messenger.ensure_send(
+            node_id=tool,
+            message=BindSensorOutputRequest(
+                payload=BindSensorOutputRequestPayload(
+                    sensor=SensorTypeField(SensorType.pressure),
+                    sensor_id=SensorIdField(sensor),
+                    binding=SensorOutputBindingField(SensorOutputBinding.report),
+                )
+            ),
+            expected_nodes=[tool],
+        )
+    try:
+        yield
+    finally:
+        for sensor in sensors:
+            await messenger.ensure_send(
+                node_id=tool,
+                message=BindSensorOutputRequest(
+                    payload=BindSensorOutputRequestPayload(
+                        sensor=SensorTypeField(SensorType.pressure),
+                        sensor_id=SensorIdField(sensor),
+                        binding=SensorOutputBindingField(SensorOutputBinding.none),
+                    )
+                ),
+                expected_nodes=[tool],
+            )

hardware/opentrons_hardware/sensors/sensor_driver.py

@@ -278,6 +278,20 @@ def __call__(
             self.response_queue.put_nowait(data)
             current_time = round((time.time() - self.start_time), 3)
             self.csv_writer.writerow([current_time, data])  # type: ignore
+        if isinstance(message, message_definitions.BatchReadFromSensorResponse):
+            data_length = message.payload.data_length.value
+            data_bytes = message.payload.sensor_data.value
+            data_ints = [
+                int.from_bytes(data_bytes[i * 4 : i * 4 + 4])
+                for i in range(data_length)
+            ]
+            for d in data_ints:
+                data = sensor_types.SensorDataType.build(
+                    d, message.payload.sensor
+                ).to_float()
+                self.response_queue.put_nowait(data)
+                current_time = round((time.time() - self.start_time), 3)
+                self.csv_writer.writerow([current_time, data])
         if isinstance(message, message_definitions.Acknowledgement):
             if (
                 self.event is not None