feat(hardware): add csv file logging capability to capacitive_probe method

.gitignore

@@ -160,4 +160,5 @@ opentrons-robot-app.tar.gz
 # asdf versions file
 .tool-versions
 mock_dir
+.npm-cache/
 .eslintcache

api/src/opentrons/config/defaults_ot3.py

@@ -44,6 +44,7 @@
             max_overrun_distance_mm=5.0,
             speed_mm_per_s=1.0,
             sensor_threshold_pf=3.0,
+            output_option=OutputOptions.sync_only,
         ),
     ),
     edge_sense=EdgeSenseSettings(
@@ -54,6 +55,7 @@
             max_overrun_distance_mm=0.5,
             speed_mm_per_s=1,
             sensor_threshold_pf=3.0,
+            output_option=OutputOptions.sync_only,
         ),
         search_initial_tolerance_mm=12.0,
         search_iteration_limit=8,
@@ -315,6 +317,7 @@ def _build_default_cap_pass(
         sensor_threshold_pf=from_conf.get(
             "sensor_threshold_pf", default.sensor_threshold_pf
         ),
+        output_option=from_conf.get("output_option", default.output_option),
     )
 
 

api/src/opentrons/config/types.py

@@ -103,29 +103,32 @@ def by_gantry_load(
         )
 
 
+class OutputOptions(int, Enum):
+    """Specifies where we should report sensor data to during a sensor pass."""
+
+    stream_to_csv = 0x1  # compile sensor data stream into a csv file, in addition to can_bus_only behavior
+    sync_buffer_to_csv = 0x2  # collect sensor data on pipette mcu, then stream to robot server and compile into a csv file, in addition to can_bus_only behavior
+    can_bus_only = (
+        0x4  # stream sensor data over CAN bus, in addition to sync_only behavior
+    )
+    sync_only = 0x8  # trigger pipette sync line upon sensor's detection of something
+
+
 @dataclass(frozen=True)
 class CapacitivePassSettings:
     prep_distance_mm: float
     max_overrun_distance_mm: float
     speed_mm_per_s: float
     sensor_threshold_pf: float
+    output_option: OutputOptions
+    data_files: Optional[Dict[InstrumentProbeType, str]] = None
 
 
 @dataclass(frozen=True)
 class ZSenseSettings:
     pass_settings: CapacitivePassSettings
 
 
-# str enum so it can be json serializable
-class OutputOptions(int, Enum):
-    """Specifies where we should report sensor data to during a sensor pass."""
-
-    stream_to_csv = 0x1
-    sync_buffer_to_csv = 0x2
-    can_bus_only = 0x4
-    sync_only = 0x8
-
-
 @dataclass
 class LiquidProbeSettings:
     starting_mount_height: float

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

@@ -365,7 +365,9 @@ async def capacitive_probe(
         distance_mm: float,
         speed_mm_per_s: float,
         sensor_threshold_pf: float,
-        probe: InstrumentProbeType,
+        probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
+        output_format: OutputOptions = OutputOptions.sync_only,
+        data_files: Optional[Dict[InstrumentProbeType, str]] = None,
     ) -> bool:
         ...
 

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

@@ -1430,15 +1430,44 @@ async def capacitive_probe(
         distance_mm: float,
         speed_mm_per_s: float,
         sensor_threshold_pf: float,
-        probe: InstrumentProbeType,
+        probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
+        output_option: OutputOptions = OutputOptions.sync_only,
+        data_files: Optional[Dict[InstrumentProbeType, str]] = None,
     ) -> bool:
+        if output_option == OutputOptions.sync_buffer_to_csv:
+            assert (
+                self._subsystem_manager.device_info[
+                    SubSystem.of_mount(mount)
+                ].revision.tertiary
+                == "1"
+            )
+        csv_output = bool(output_option.value & OutputOptions.stream_to_csv.value)
+        sync_buffer_output = bool(
+            output_option.value & OutputOptions.sync_buffer_to_csv.value
+        )
+        can_bus_only_output = bool(
+            output_option.value & OutputOptions.can_bus_only.value
+        )
+        data_files_transposed = (
+            None
+            if data_files is None
+            else {
+                sensor_id_for_instrument(probe): data_files[probe]
+                for probe in data_files.keys()
+            }
+        )
         status = await capacitive_probe(
-            self._messenger,
-            sensor_node_for_mount(mount),
-            axis_to_node(moving),
-            distance_mm,
-            speed_mm_per_s,
-            sensor_id_for_instrument(probe),
+            messenger=self._messenger,
+            tool=sensor_node_for_mount(mount),
+            mover=axis_to_node(moving),
+            distance=distance_mm,
+            plunger_speed=speed_mm_per_s,
+            mount_speed=speed_mm_per_s,
+            csv_output=csv_output,
+            sync_buffer_output=sync_buffer_output,
+            can_bus_only_output=can_bus_only_output,
+            data_files=data_files_transposed,
+            sensor_id=sensor_id_for_instrument(probe),
             relative_threshold_pf=sensor_threshold_pf,
         )
 

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

@@ -749,7 +749,9 @@ async def capacitive_probe(
         distance_mm: float,
         speed_mm_per_s: float,
         sensor_threshold_pf: float,
-        probe: InstrumentProbeType,
+        probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
+        output_format: OutputOptions = OutputOptions.sync_only,
+        data_files: Optional[Dict[InstrumentProbeType, str]] = None,
     ) -> bool:
         self._position[moving] += distance_mm
         return True

api/src/opentrons/hardware_control/ot3api.py

@@ -2660,7 +2660,9 @@ async def capacitive_probe(
             machine_pass_distance,
             pass_settings.speed_mm_per_s,
             pass_settings.sensor_threshold_pf,
-            probe=probe,
+            probe,
+            pass_settings.output_option,
+            pass_settings.data_files,
         )
         end_pos = await self.gantry_position(mount, refresh=True)
         if retract_after:

api/tests/opentrons/config/ot3_settings.py

@@ -138,6 +138,8 @@
                 "max_overrun_distance_mm": 2,
                 "speed_mm_per_s": 3,
                 "sensor_threshold_pf": 4,
+                "output_option": OutputOptions.sync_only,
+                "data_files": None,
             },
         },
         "edge_sense": {
@@ -148,6 +150,8 @@
                 "max_overrun_distance_mm": 5,
                 "speed_mm_per_s": 6,
                 "sensor_threshold_pf": 7,
+                "output_option": OutputOptions.sync_only,
+                "data_files": None,
             },
             "search_initial_tolerance_mm": 18,
             "search_iteration_limit": 3,

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -107,6 +107,7 @@ def fake_settings() -> CapacitivePassSettings:
         max_overrun_distance_mm=2,
         speed_mm_per_s=4,
         sensor_threshold_pf=1.0,
+        output_option=OutputOptions.sync_only,
     )
 
 
@@ -486,6 +487,8 @@ def _update_position(
             speed_mm_per_s: float,
             threshold_pf: float,
             probe: InstrumentProbeType,
+            output_option: OutputOptions = OutputOptions.sync_only,
+            data_file: Optional[str] = None,
         ) -> None:
             hardware_backend._position[moving] += distance_mm / 2
 
@@ -865,7 +868,14 @@ async def test_capacitive_probe(
     # This is a negative probe because the current position is the home position
     # which is very large.
     mock_backend_capacitive_probe.assert_called_once_with(
-        mount, moving, 3, 4, 1.0, InstrumentProbeType.PRIMARY
+        mount,
+        moving,
+        3,
+        4,
+        1.0,
+        InstrumentProbeType.PRIMARY,
+        fake_settings.output_option,
+        fake_settings.data_files,
     )
 
     original = moving.set_in_point(here, 0)

hardware-testing/hardware_testing/examples/capacitive_probe_ot3.py

@@ -2,7 +2,7 @@
 import argparse
 import asyncio
 
-from opentrons.config.types import CapacitivePassSettings
+from opentrons.config.types import CapacitivePassSettings, OutputOptions
 from opentrons.hardware_control.ot3api import OT3API
 
 from hardware_testing.opentrons_api import types
@@ -44,12 +44,14 @@
     max_overrun_distance_mm=3,
     speed_mm_per_s=1,
     sensor_threshold_pf=STABLE_CAP_PF,
+    output_option=OutputOptions.sync_only,
 )
 PROBE_SETTINGS_XY_AXIS = CapacitivePassSettings(
     prep_distance_mm=CUTOUT_SIZE / 2,
     max_overrun_distance_mm=3,
     speed_mm_per_s=1,
     sensor_threshold_pf=STABLE_CAP_PF,
+    output_option=OutputOptions.sync_only,
 )
 
 

hardware-testing/hardware_testing/examples/capacitive_probe_ot3_tunable.py

@@ -0,0 +1,109 @@
+"""Capacitive probe OT3."""
+import argparse
+import asyncio
+
+from opentrons.config.types import CapacitivePassSettings, OutputOptions
+from opentrons.hardware_control.ot3api import OT3API
+from opentrons.hardware_control.types import InstrumentProbeType
+
+from hardware_testing.opentrons_api import types
+from hardware_testing.opentrons_api import helpers_ot3
+
+# distance added to the pipette shaft
+# when the calibration probe is attached
+PROBE_LENGTH = 34.5
+
+# the capacitive readings need to be stable <0.1
+# before probing anything
+STABLE_CAP_PF = 0.1
+
+# capacitance relative threshold in picofarads
+CAP_REL_THRESHOLD_PF = 10.0
+
+# ideally these values come from either:
+# 1) the API config file
+# 2) or, found through manually jogging
+# The Z is different from the XY probing location
+# because the pipette cannot reach the bottom of the
+# cutout, so we cannot probe the Z inside the cutout
+ASSUMED_Z_LOCATION = types.Point(x=228, y=150, z=80)  # C2 slot center
+ASSUMED_XY_LOCATION = types.Point(x=228, y=150, z=ASSUMED_Z_LOCATION.z)
+
+# configure how the probing motion behaves
+# capacitive_probe will always automatically do the following:
+# 1) move to the "prep" distance away from the assumed location
+# 2) set the capacitive threshold
+#   a) the value is sent over CAN to the pipette's MCU
+#   b) the pipette will trigger the SYNC line when the threshold is reached
+# 3) move along the specified axis, at the specified speed
+#   a) the max distance probed = prep + max_overrun
+# 4) movement will stop when (either/or):
+#   a) the sensor is triggered
+#   b) or, the max distance is reached
+# 5) move to the "prep" distance away from the assumed location
+PROBE_SETTINGS_Z_AXIS = CapacitivePassSettings(
+    prep_distance_mm=10,
+    max_overrun_distance_mm=3,
+    speed_mm_per_s=1,
+    sensor_threshold_pf=CAP_REL_THRESHOLD_PF,
+    output_option=OutputOptions.sync_only,
+)
+PROBE_SETTINGS_Z_AXIS_OUTPUT = CapacitivePassSettings(
+    prep_distance_mm=10,
+    max_overrun_distance_mm=3,
+    speed_mm_per_s=1,
+    sensor_threshold_pf=CAP_REL_THRESHOLD_PF,
+    output_option=OutputOptions.sync_buffer_to_csv,
+    data_files={InstrumentProbeType.PRIMARY: "/data/capacitive_sensor_data.csv"},
+)
+
+
+async def _probe_sequence(api: OT3API, mount: types.OT3Mount, stable: bool) -> float:
+    z_ax = types.Axis.by_mount(mount)
+
+    print("Align the XY axes above Z probe location...")
+    home_pos_z = helpers_ot3.get_endstop_position_ot3(api, mount)[z_ax]
+    await api.move_to(mount, ASSUMED_Z_LOCATION._replace(z=home_pos_z))
+
+    if stable:
+        await helpers_ot3.wait_for_stable_capacitance_ot3(
+            api, mount, threshold_pf=STABLE_CAP_PF, duration=1.0
+        )
+    found_z, _ = await api.capacitive_probe(
+        mount, z_ax, ASSUMED_Z_LOCATION.z, PROBE_SETTINGS_Z_AXIS
+    )
+    print(f"Found deck Z location = {found_z} mm")
+    return found_z
+
+
+async def _main(is_simulating: bool, cycles: int, stable: bool) -> None:
+    api = await helpers_ot3.build_async_ot3_hardware_api(
+        is_simulating=is_simulating, pipette_left="p1000_single_v3.3"
+    )
+    mount = types.OT3Mount.LEFT
+    if not api.hardware_pipettes[mount.to_mount()]:
+        raise RuntimeError("No pipette attached")
+
+    # add length to the pipette, to account for the attached probe
+    await api.add_tip(mount, PROBE_LENGTH)
+
+    await helpers_ot3.home_ot3(api)
+    for c in range(cycles):
+        print(f"Cycle {c + 1}/{cycles}")
+        await _probe_sequence(api, mount, stable)
+
+    # move up, "remove" the probe, then disengage the XY motors when done
+    z_ax = types.Axis.by_mount(mount)
+    top_z = helpers_ot3.get_endstop_position_ot3(api, mount)[z_ax]
+    await api.move_to(mount, ASSUMED_XY_LOCATION._replace(z=top_z))
+    await api.remove_tip(mount)
+    await api.disengage_axes([types.Axis.X, types.Axis.Y])
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--simulate", action="store_true")
+    parser.add_argument("--cycles", type=int, default=1)
+    parser.add_argument("--stable", type=bool, default=True)
+    args = parser.parse_args()
+    asyncio.run(_main(args.simulate, args.cycles, args.stable))

hardware-testing/hardware_testing/production_qc/ninety_six_assembly_qc_ot3/test_capacitance.py

@@ -173,7 +173,8 @@ async def _probe(distance: float, speed: float) -> float:
                 NodeId.pipette_left,
                 NodeId.head_l,
                 distance=distance,
-                speed=speed,
+                plunger_speed=speed,
+                mount_speed=speed,
                 sensor_id=sensor_id,
                 relative_threshold_pf=default_probe_cfg.sensor_threshold_pf,
             )

hardware-testing/hardware_testing/production_qc/robot_assembly_qc_ot3/test_instruments.py

@@ -1,7 +1,7 @@
 """Test Instruments."""
 from typing import List, Tuple, Optional, Union
 
-from opentrons.config.types import CapacitivePassSettings
+from opentrons.config.types import CapacitivePassSettings, OutputOptions
 from opentrons.hardware_control.ot3api import OT3API
 
 from hardware_testing.data.csv_report import (
@@ -30,6 +30,7 @@
     max_overrun_distance_mm=0,
     speed_mm_per_s=Z_PROBE_DISTANCE_MM / Z_PROBE_TIME_SECONDS,
     sensor_threshold_pf=1.0,
+    output_option=OutputOptions.can_bus_only,
 )
 
 RELATIVE_MOVE_FROM_HOME_DELTA = Point(x=-500, y=-300)

hardware-testing/hardware_testing/scripts/gripper_ot3.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from typing import Optional, List, Any, Dict
 
-from opentrons.config.defaults_ot3 import CapacitivePassSettings
+from opentrons.config.defaults_ot3 import CapacitivePassSettings, OutputOptions
 from opentrons.hardware_control.ot3api import OT3API
 
 from hardware_testing.opentrons_api import types
@@ -73,6 +73,7 @@
     max_overrun_distance_mm=1,
     speed_mm_per_s=1,
     sensor_threshold_pf=0.5,
+    output_option=OutputOptions.sync_only,
 )
 LABWARE_PROBE_CORNER_TOP_LEFT_XY = {
     "plate": Point(x=5, y=-5),

hardware/opentrons_hardware/firmware_bindings/constants.py

@@ -253,7 +253,7 @@ class MessageId(int, Enum):
     peripheral_status_request = 0x8C
     peripheral_status_response = 0x8D
     baseline_sensor_response = 0x8E
-    send_accumulated_pressure_data = 0x8F
+    send_accumulated_sensor_data = 0x8F
 
     set_hepa_fan_state_request = 0x90
     get_hepa_fan_state_request = 0x91