feat(hardware-testing, api): CNC firmware and scripts to measure labware (#20253)

# Overview

PR for touch probe proof of concept. Adds firmware, drivers, and
hardware testing scripts for the touch probe.


https://docs.google.com/document/d/151cAolVlUmFbdYdXOCe5GIgDmzycKI_iPKSS7zWbS0I/edit?tab=t.0

## Changelog

-added touch probe firmware + drivers
-added touch_probe mock scripts to hardware testing

---------

Co-authored-by: Ryan Howard <[email protected]>
Co-authored-by: Rhyann Clarke <[email protected]>
Co-authored-by: Rhyann Clarke <[email protected]>

Author: 3 people

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

@@ -172,6 +172,7 @@
     liquid_probe,
     check_overpressure,
     grab_pressure,
+    touch_probe,
 )
 from opentrons_hardware.hardware_control.rear_panel_settings import (
     get_door_state,
@@ -1524,6 +1525,22 @@ def _pop_queue() -> Optional[Tuple[NodeId, ErrorCode]]:
         else:
             yield
 
+    async def touch_probe(self, axis: Axis, speed: float, distance: float) -> float:
+
+        status = await touch_probe(
+            messenger=self._messenger,
+            mover=axis_to_node(axis),
+            distance=distance,
+            speed=speed,
+        )
+
+        self._position[axis_to_node(axis)] = status.motor_position
+
+        if status.move_ack != MoveCompleteAck.stopped_by_condition:
+            raise Exception("Booo")
+
+        return self._position[axis_to_node(axis)]
+
     async def liquid_probe(
         self,
         mount: OT3Mount,

api/src/opentrons/hardware_control/ot3api.py

@@ -3302,3 +3302,26 @@ async def read_stem_capacitance(
         realmount = OT3Mount.from_mount(mount)
         s_data = await self._backend.read_capacitive_sensor(realmount, primary)
         return s_data if s_data else 0.0
+
+    async def touch_probe(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        axis: Axis,
+        speed: float,
+        distance: float,
+    ) -> top_types.Point:
+        await self._backend.touch_probe(axis, speed, distance)  # type: ignore[attr-defined]
+        realmount = OT3Mount.from_mount(mount)
+        machine_pos = await self._backend.update_position()
+        end_pos = self.get_deck_from_machine(machine_pos)
+        end_point = top_types.Point(
+            end_pos[Axis.X], end_pos[Axis.Y], end_pos[Axis.by_mount(realmount)]
+        )
+        offset = offset_for_mount(
+            realmount,
+            top_types.Point(*self._config.left_mount_offset),
+            top_types.Point(*self._config.right_mount_offset),
+            top_types.Point(*self._config.gripper_mount_offset),
+        )
+        cp = self.critical_point_for(realmount, None)
+        return end_point + offset + cp

hardware-testing/hardware_testing/drivers/asair_sensor.py

@@ -164,22 +164,24 @@ def get_reading(self, retries: int = 5) -> Reading:
         log.debug(f"sending {data_packet}")
         command_bytes = codecs.decode(data_packet.encode(), "hex")
         try:
-            self._th_sensor.flushInput()
-            self._th_sensor.flushOutput()
+            self._th_sensor.flushInput()  # type: ignore[attr-defined]
+            self._th_sensor.flushOutput()  # type: ignore[attr-defined]
+
             self._th_sensor.write(command_bytes)
             time.sleep(0.1)
 
-            length = self._th_sensor.inWaiting()
+            length = self._th_sensor.inWaiting()  # type: ignore[attr-defined]
             res = self._th_sensor.read(length)
-            log.debug(f"received {res}")
+            log.debug(f"received {res!r}")
 
-            res = codecs.encode(res, "hex")
-            relative_hum = res[6:10]
-            temp = res[10:14]
-            log.info(f"Temp: {temp}, RelativeHum: {relative_hum}")
+            res_hex = codecs.encode(res, "hex").decode("ascii")
+            relative_hum_str = res_hex[6:10]
+            temp_str = res_hex[10:14]
+            log.info(f"Temp: {temp_str!r}, RelativeHum: {relative_hum_str!r}")
+
+            temp = float(int(temp_str, 16)) / 10
+            relative_hum = float(int(relative_hum_str, 16)) / 10
 
-            temp = float(int(temp, 16)) / 10
-            relative_hum = float(int(relative_hum, 16)) / 10
             return Reading(temperature=temp, relative_humidity=relative_hum)
 
         except (IndexError, ValueError) as e:
@@ -202,15 +204,15 @@ def get_serial(self) -> str:
         log.debug(f"sending {data_packet}")
         command_bytes = codecs.decode(data_packet.encode(), "hex")
         try:
-            self._th_sensor.flushInput()
-            self._th_sensor.flushOutput()
+            self._th_sensor.flushInput()  # type: ignore[attr-defined]
+            self._th_sensor.flushOutput()  # type: ignore[attr-defined]
             self._th_sensor.write(command_bytes)
             time.sleep(0.1)
 
-            length = self._th_sensor.inWaiting()
+            length = self._th_sensor.inWaiting()  # type: ignore[attr-defined]
             res = self._th_sensor.read(length)
             res = codecs.encode(res, "hex")
-            log.debug(f"received {res}")
+            log.debug(f"received {res!r}")
             dev_id = res[6:14]
             return dev_id.decode()
 

hardware-testing/hardware_testing/drivers/mark10/mark10_fg.py

@@ -103,7 +103,9 @@ def read_force(self, timeout: float = 1.0) -> float:
             try:
                 force_val, units = line.split(" ")
                 if units != "N":
-                    self._force_guage.write("N\r\n")  # Set force gauge units to Newtons
+                    self._force_guage.write(
+                        "N\r\n".encode("utf-8")
+                    )  # Set force gauge units to Newtons
                     print(f'Setting gauge units from {units} to "N" (newtons)')
                     continue
                 else:

hardware-testing/hardware_testing/drivers/pressure_fixture.py

@@ -158,7 +158,7 @@ def create(cls, port: str, slot_side: str = "left") -> "PressureFixture":
     def connect(self) -> None:
         """Connect."""
         self._port.open()
-        self._port.flushInput()
+        self._port.flushInput()  # type: ignore[attr-defined]
         # NOTE: device might take a few seconds to boot up
         sleep(FIXTURE_REBOOT_TIME)
         fw_version = self.firmware_version()

hardware-testing/hardware_testing/drivers/radwag/driver.py

@@ -121,14 +121,14 @@ def _write_command(self, cmd: str) -> None:
     def _read_response(
         self, command: RadwagCommand, timeout: Optional[float] = None
     ) -> RadwagResponse:
+        prev_timeout: float = float(self._connection.timeout or 0.0)
         if timeout is not None:
-            prev_timeout = float(self._connection.timeout)
             self._connection.timeout = timeout
             response = self._connection.readline()
             self._connection.timeout = prev_timeout
         else:
             response = self._connection.readline()
-        self._raw_log.write(f"{datetime.datetime.now()} <-- {response}\n")
+        self._raw_log.write(f"{datetime.datetime.now()} <-- {response!r}\n")
         data = radwag_response_parse(response.decode("utf-8"), command)
         return data
 

hardware-testing/hardware_testing/drivers/sealed_pressure_fixture.py

@@ -3,8 +3,8 @@
 
 import time
 from typing import Union, Optional, Any, List
-import serial  # type: ignore[import-untyped]
-from serial.tools.list_ports import comports  # type: ignore[import-untyped]
+import serial  # type: ignore[import]
+from serial.tools.list_ports import comports  # type: ignore[import]
 
 ReceiveBuffer = 100
 
@@ -42,7 +42,7 @@ def init_serial(self, baud: int) -> None:
             bytesize=serial.EIGHTBITS,
             timeout=1,
         )
-        if self.com.isOpen():
+        if self.com.isOpen():  # type: ignore[attr-defined]
             print(f"{self.device} Opened! \n")
         # settings
         self.com.bytesize = serial.EIGHTBITS  # 数据位 8
@@ -75,9 +75,9 @@ def write_and_get_buffer(
             return None
         if not isinstance(send, bytes):
             send = (str(send) + "\r\n").encode("utf-8")
-        self.com.flushInput()  # type: ignore [union-attr]
-        self.com.flushOutput()  # type: ignore [union-attr]
-        self.com.write(send)  # type: ignore [union-attr]
+        self.com.flushInput()  # type: ignore[attr-defined]
+        self.com.flushOutput()  # type: ignore[attr-defined]
+        self.com.write(send)  # type: ignore[attr-defined]
         time.sleep(0.1)
         if delay is None:
             pass
@@ -86,7 +86,7 @@ def write_and_get_buffer(
         if only_write is True:
             return None
         for i in range(times):
-            data = self.com.read(ReceiveBuffer)  # type: ignore [union-attr]
+            data = self.com.read(ReceiveBuffer)  # type: ignore[attr-defined]
             if type(data) is not bytes:
                 if "OK" not in data.decode("utf-8") or "busy" in data.decode("utf-8"):
                     time.sleep(1)

hardware-testing/hardware_testing/drivers/touch_probe/__init__.py

@@ -0,0 +1,5 @@
+"""CNC Driver."""
+from .dimensions import ProbeTarget
+from .touch_probe import TouchProbe, ProbeConfig
+
+__all__ = ["ProbeConfig", "ProbeTarget", "TouchProbe"]

hardware-testing/hardware_testing/drivers/touch_probe/demo.py

@@ -0,0 +1,104 @@
+"""Proof of concept for use of touch probe."""
+import argparse
+import asyncio
+from hardware_testing.opentrons_api import helpers_ot3
+from hardware_testing.opentrons_api.types import OT3Mount, Axis, Point
+from hardware_testing.drivers.touch_probe import TouchProbe, ProbeConfig
+
+# ============================================================================
+# Main Function
+# ============================================================================
+
+
+async def _main(simulating: bool, mount: OT3Mount, num_wells: int, slot: int) -> None:
+    config = ProbeConfig()
+    api = await helpers_ot3.build_async_ot3_hardware_api(
+        is_simulating=simulating, use_defaults=True
+    )
+    await api.home()
+    await api.cache_instruments()
+    tp = TouchProbe(api, mount, config)
+    # probe deck surface
+    print(f"\nProbing slot center #{6}")
+    deck_pos = await tp.get_deck_z(6)
+    print(f"z deck pos: {deck_pos.z}")
+    deck_square_center_radius = await tp.search_hole_and_center(deck_pos)
+    if deck_square_center_radius:
+        deck_square_center, deck_square_radius = deck_square_center_radius
+        print(
+            f"Deck square center: {deck_square_center} length: {deck_square_radius * 2}"
+        )
+
+    # probe gauge block
+    print(f"\nProbing Gauge Block on slot #{7}")
+    block_dims = await tp.calibrate_labware(7, deck_pos)
+    if not block_dims:
+        print("Calibration failed.")
+        return
+
+    print("\nCalibration Block Dimensions:")
+    print(f"  Width:  {block_dims.width:.3f} mm")
+    print(f"  Length: {block_dims.length:.3f} mm")
+    print(f"  Height: {block_dims.height:.3f} mm")
+
+    # probe module
+    print(f"\nProbing Module on slot #{4}")
+    module_pos = await tp.get_deck_z(4)
+    print(f"z module pos: {module_pos.z}")
+    await tp.search_hole_and_center(module_pos)
+
+    # Probe labware
+    print(f"\nStarting Calibration on slot #{slot}")
+    dimensions = await tp.calibrate_labware(slot, deck_pos)
+    if not dimensions:
+        print("Calibration failed.")
+        return
+
+    dimensions.set_num_wells(num_wells)
+
+    # calculate A1 pos
+    first_well_xy = Point(
+        x=dimensions.x_min.x + dimensions.x_offset,
+        y=dimensions.y_max.y - dimensions.y_offset,
+        z=dimensions.z_max.z,
+    )
+
+    # Detecting A1 + center
+    well_center_and_radius = await tp.search_hole_and_center(first_well_xy)
+    if well_center_and_radius:
+        well_center, dimensions.radius = well_center_and_radius
+        brim_z = await tp.get_brim_height(well_center, dimensions.radius)
+        if brim_z is not None:
+            dimensions.z_max = dimensions.z_max._replace(z=brim_z)
+            well_center = well_center._replace(z=brim_z)
+            dimensions.well_bottom = await tp.get_bottom(well_center)
+        else:
+            print("Unable to determine well dimensions.")
+
+    # Print final dimensions using dataclass properties
+    print("\nLabware Dimensions:")
+    print(f"  Deck Radius: {deck_square_radius:.3f} mm")
+    print(f"  Deck Center: {deck_square_center} mm")
+    print(f"  Width:  {dimensions.width:.3f} mm")
+    print(f"  Length: {dimensions.length:.3f} mm")
+    print(f"  Height: {dimensions.height:.3f} mm")
+    print(f"  Depth: {dimensions.depth:.3f} mm")
+    print(f"  Bottom Offset: {dimensions.bottom_offset:.3f} mm")
+    print(f"  Well Radius: {dimensions.radius:.3f} mm")
+    print(f"  Well Center: {well_center} mm")
+    print(f"  radius: {dimensions.radius} mm")
+    print(f"  spacing: {dimensions.spacing}")
+    await asyncio.sleep(3.0)
+    await api.home([Axis.Z, Axis.X, Axis.Y])
+    return
+
+
+if __name__ == "__main__":
+    print("\nTouch Probe Test\n")
+    arg_parser = argparse.ArgumentParser(description="Touch Probe Test")
+    arg_parser.add_argument("--simulate", action="store_true")
+    arg_parser.add_argument("--num_wells", type=int, default=96)
+    arg_parser.add_argument("--slot", type=int, default=5)
+    args = arg_parser.parse_args()
+    mount = OT3Mount.LEFT
+    asyncio.run(_main(args.simulate, mount, args.num_wells, args.slot))

hardware-testing/hardware_testing/drivers/touch_probe/dimensions.py

@@ -0,0 +1,76 @@
+"""Data and configuration ."""
+from hardware_testing.opentrons_api.types import Point
+from typing import Optional
+from dataclasses import dataclass
+
+# ============================================================================
+# Data
+# ============================================================================
+
+
+@dataclass
+class ProbeTarget:
+    """Global configuration of dimensions found."""
+
+    deck_pos: Point
+    x_min: Point
+    x_max: Point
+    y_min: Point
+    y_max: Point
+    z_max: Point  # top of object
+    radius: Optional[float]
+
+    @property
+    def width(self) -> float:
+        """Width of the probed area (X dimension)."""
+        return abs(self.x_max.x - self.x_min.x)
+
+    @property
+    def length(self) -> float:
+        """Length of the probed area (Y dimension)."""
+        return abs(self.y_max.y - self.y_min.y)
+
+    @property
+    def height(self) -> float:
+        """Height of the probed area (Z dimension from deck to top)."""
+        return self.z_max.z - self.deck_pos.z
+
+
+@dataclass
+class LabwareDims(ProbeTarget):
+    """Container for probe results with computed properties."""
+
+    well_bottom: Optional[Point] = None
+    num_wells: Optional[int] = 96
+    x_offset: float = 0.0
+    y_offset: float = 0.0
+    spacing: float = 0.0
+
+    def set_num_wells(self, num_wells: int) -> None:
+        """Determines dimensions based on well number."""
+        self.num_wells = num_wells
+        match num_wells:
+            case 96:
+                self.x_offset = 14.38
+                self.y_offset = 11.23
+                self.spacing = 9.0
+            case 384:
+                self.x_offset = 11.3
+                self.y_offset = 8.5
+                self.spacing = 3.4
+            case _:
+                raise ValueError(f"Unsupported num_wells: {num_wells}")
+
+    @property
+    def depth(self) -> Optional[float]:
+        """Return the vertical distance from the top to the bottom of the hole, or None if unknown."""
+        if self.well_bottom is None:
+            return None
+        return self.z_max.z - self.well_bottom.z
+
+    @property
+    def bottom_offset(self) -> Optional[float]:
+        """Height of the bottom offset (Z dimension from hole bottom to deck)."""
+        if self.well_bottom is None:
+            return None
+        return self.well_bottom.z - self.deck_pos.z