debug

Author: wweiye

api/src/opentrons/hardware_control/ot3api.py

@@ -1892,6 +1892,7 @@ async def tip_pickup_moves(
         # the volume chamber and the drop-tip sleeve on p1000.
         # This extra shake ensures those tips are removed
         for rel_point, speed in spec.shake_off_moves:
+            
             await self.move_rel(realmount, rel_point, speed=speed)
 
         if isinstance(self._backend, OT3Simulator):
@@ -1900,9 +1901,62 @@ async def tip_pickup_moves(
         # fixme: really only need this during labware position check so user
         # can verify if a tip is properly attached
         if spec.ending_z_retract_distance:
+            print("spec.ending_z_retract_distance",spec.ending_z_retract_distance)
             await self.move_rel(
                 realmount, top_types.Point(z=spec.ending_z_retract_distance)
             )
+        
+    async def tip_pickup_moves_96(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        presses: Optional[int] = None,
+        increment: Optional[float] = None,
+        ) -> None:
+        """This is a slightly more barebones variation of pick_up_tip. This is only the motor routine
+        directly involved in tip pickup, and leaves any state updates and plunger moves to the caller.
+        """
+        realmount = OT3Mount.from_mount(mount)
+        instrument = self._pipette_handler.get_pipette(realmount)
+
+        if (
+            self.gantry_load == GantryLoad.HIGH_THROUGHPUT
+            and instrument.nozzle_manager.current_configuration.configuration
+            == top_types.NozzleConfigurationType.FULL
+        ):  
+            spec = self._pipette_handler.plan_ht_pick_up_tip(
+                instrument.nozzle_manager.current_configuration.tip_count
+            )
+            if spec.z_distance_to_tiprack:
+                await self.move_rel(
+                    realmount, top_types.Point(z=spec.z_distance_to_tiprack)
+                )
+            await self._tip_motor_action(realmount, spec.tip_action_moves)
+        else:
+            spec = self._pipette_handler.plan_lt_pick_up_tip(
+                realmount,
+                instrument.nozzle_manager.current_configuration.tip_count,
+                presses,
+                increment,
+            )
+            await self._force_pick_up_tip(realmount, spec)
+
+        # neighboring tips tend to get stuck in the space between
+        # the volume chamber and the drop-tip sleeve on p1000.
+        # This extra shake ensures those tips are removed
+        for rel_point, speed in spec.shake_off_moves:
+            await self.move_rel(realmount, rel_point, speed=speed)
+
+        if isinstance(self._backend, OT3Simulator):
+            self._backend._update_tip_state(realmount, True)
+
+        # fixme: really only need this during labware position check so user
+        # can verify if a tip is properly attached
+        print("spec.ending_z_retract_distance",spec.ending_z_retract_distance)
+        # if spec.ending_z_retract_distance:
+        #     print("spec.ending_z_retract_distance",spec.ending_z_retract_distance)
+        #     await self.move_rel(
+        #         realmount, top_types.Point(z=spec.ending_z_retract_distance)
+        #     )
 
     async def _move_to_plunger_bottom(
         self,
@@ -2303,6 +2357,31 @@ def add_tip_to_instr() -> None:
 
         if prep_after:
             await self.prepare_for_aspirate(realmount)
+    
+    async def pick_up_tip_96_fixture(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        tip_length: float,
+        presses: Optional[int] = None,
+        increment: Optional[float] = None,
+        prep_after: bool = True,
+    ) -> None:
+        """Pick up tip from current location."""
+        realmount = OT3Mount.from_mount(mount)
+        instrument = self._pipette_handler.get_pipette(realmount)
+
+        def add_tip_to_instr() -> None:
+            instrument.add_tip(tip_length=tip_length)
+            instrument.set_current_volume(0)
+
+        await self._move_to_plunger_bottom(realmount, rate=1.0)
+
+        await self.tip_pickup_moves_96(mount, presses, increment)
+
+        add_tip_to_instr()
+
+        if prep_after:
+            await self.prepare_for_aspirate(realmount)
 
     def set_current_tiprack_diameter(
         self, mount: Union[top_types.Mount, OT3Mount], tiprack_diameter: float

hardware-testing/hardware_testing/drivers/pressure_fixture.py

@@ -15,12 +15,15 @@
 
 FIXTURE_REBOOT_TIME = 2
 FIXTURE_NUM_CHANNELS: Final[int] = 8
+FIXTURE_NUM_CHANNELS_96: Final[int] = 96
 FIXTURE_BAUD_RATE: Final[int] = 115200
 FIXTURE_VERSION_REQUIRED = "1.0.0"
+FIXTURE_VERSION_REQUIRED2 = "0.0.1"
 
 FIXTURE_CMD_TERMINATOR = "\r\n"
 FIXTURE_CMD_GET_VERSION = "VERSION"
 FIXTURE_CMD_GET_ALL_PRESSURE = "GETPRESSURE:15"
+FIXTURE_CMD_GET_ALL_PRESSURE_96 = "GETPRESSURE:255"
 
 LOCATION_A1_LEFT = Point(x=14.4, y=74.5, z=71.2)
 LOCATION_A1_RIGHT = LOCATION_A1_LEFT._replace(x=128 - 14.4)
@@ -40,6 +43,11 @@ def connect(self) -> None:
         """Connect to the Mark10 Force Gauge."""
         ...
 
+    @abstractmethod
+    def connect_96(self) -> None:
+        """Connect to the Mark10 Force Gauge."""
+        ...
+
     @abstractmethod
     def disconnect(self) -> None:
         """Disconnect from the Mark10 Force Gauge."""
@@ -54,6 +62,11 @@ def firmware_version(self) -> str:
     def read_all_pressure_channel(self) -> List[float]:
         """Read all pressure channels on fixture in Pascals."""
         ...
+    @abstractmethod
+    def read_all_pressure_channel_96(self) -> List[float]:
+        """Read all pressure channels on fixture in Pascals."""
+        ...
+
 
     def position_in_slot(self, side: Literal["left", "right"] = "left") -> Point:
         """Position in slot."""
@@ -101,6 +114,10 @@ def read_all_pressure_channel(self) -> List[float]:
         """Read Pressure for all channels."""
         pressure = [random.uniform(2.5, 2) for _ in range(FIXTURE_NUM_CHANNELS)]
         return pressure
+    def read_all_pressure_channel_96(self) -> List[float]:
+        """Read Pressure for all channels."""
+        pressure = [random.uniform(2.5, 2) for _ in range(FIXTURE_NUM_CHANNELS_96)]
+        return pressure
 
 
 def connect_to_fixture(
@@ -136,6 +153,38 @@ def connect_to_fixture(
     ui.print_info("no fixture found returning simulator")
     return SimPressureFixture()
 
+def connect_to_fixture96(
+    simulate: bool, side: str = "left", autosearch: bool = True
+) -> PressureFixtureBase:
+    """Try to find and return an presure fixture, if not found return a simulator."""
+    ui.print_title("Connecting to presure fixture")
+    if not simulate:
+        if not autosearch:
+            port = list_ports_and_select(device_name="Pressure fixture")
+            fixture = PressureFixture.create(port=port, slot_side=side)
+            fixture.connect_96()
+            ui.print_info(f"Found fixture on port {port}")
+            return fixture
+        else:
+            ports = comports()
+            assert ports
+            for _port in ports:
+                port = _port.device  # type: ignore[attr-defined]
+                try:
+                    ui.print_info(
+                        f"Trying to connect to Pressure fixture on port {port}"
+                    )
+                    fixture = PressureFixture.create(port=port, slot_side=side)
+                    fixture.connect_96()
+                    ui.print_info(f"Found fixture on port {port}")
+                    return fixture
+                except:  # noqa: E722
+                    pass
+            use_sim = ui.get_user_answer("No pressure sensor found, use simulator?")
+            if not use_sim:
+                raise SerialException("No sensor found")
+    ui.print_info("no fixture found returning simulator")
+    return SimPressureFixture()
 
 class PressureFixture(PressureFixtureBase):
     """OT3 Pressure Fixture Driver."""
@@ -162,9 +211,22 @@ def connect(self) -> None:
         # NOTE: device might take a few seconds to boot up
         sleep(FIXTURE_REBOOT_TIME)
         fw_version = self.firmware_version()
+        print(f"unexpected pressure-fixture version: {fw_version}")
         assert (
             fw_version == FIXTURE_VERSION_REQUIRED
         ), f"unexpected pressure-fixture version: {fw_version}"
+    
+    def connect_96(self) -> None:
+        """Connect."""
+        self._port.open()
+        self._port.flushInput()
+        # NOTE: device might take a few seconds to boot up
+        sleep(FIXTURE_REBOOT_TIME)
+        fw_version = self.firmware_version()
+        print(f"unexpected pressure-fixture version: {fw_version}")
+        assert (
+            fw_version == FIXTURE_VERSION_REQUIRED2
+        ), f"unexpected pressure-fixture version: {fw_version}"
 
     def disconnect(self) -> None:
         """Disconnect."""
@@ -190,14 +252,57 @@ def read_all_pressure_channel(self) -> List[float]:
         if self._slot_side == "left":
             data.reverse()  # reverse order, so pipette channel 1 is at index 0
         return data
+    
+    def read_all_pressure_channel_96(self) -> List[float]:
+        """Reads from all the channels from the fixture."""
+        cmd_str = f"{FIXTURE_CMD_GET_ALL_PRESSURE_96}{FIXTURE_CMD_TERMINATOR}"
+        self._port.write(cmd_str.encode("utf-8"))
+        response = self._port.readlines()#.decode("utf-8")
+        datalist = []
+        for res in response:
+            res_list = res.decode("utf-8").split(",")[:-1]  # ignore the last comma
+            data_str = [d.split("=")[-1].strip() for d in res_list]  # remove PRESSURE=
+            # for i in range(len(data_str)):  # replace all -0.00 with 0.00
+            #     if data_str[i] == "-0.00":
+            #         data_str[i] = "0.00"
+            data = [float(d) for d in data_str]  # convert to float
+            data.reverse
+            #print("yuans",data)
+            # datalist.extend(data)
+            datalist = data + datalist
+        # if self._slot_side == "left":
+        #     data.reverse()  # reverse order, so pipette channel 1 is at index 0
+        #print("datalist",datalist)
+        #datalist.reverse()
+        return datalist
+    def print_pressure_datas(self,data_list):
+        row_labels = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H']
+        col_labels = list(range(1, 13))
+        cell_width = 9
+
+        # 打印表头
+        print(" " * 3 + "".join(f"{col:>{cell_width}}" for col in col_labels))
+
+        # 打印每一行
+        for i in range(8):
+            row_data = data_list[i * 12:(i + 1) * 12]
+            row_str = f"{row_labels[i]}: "
+            for val in row_data:
+                try:
+                    row_str += f"{float(val):>{cell_width}.2f}"
+                except:
+                    row_str += f"{str(val):>{cell_width}}"
+            print(row_str)
 
 
 if __name__ == "__main__":
-    port_name = input("type the port of the device (eg: COM1): ")
+    port_name = list_ports_and_select(device_name="Pressure fixture")
+    #port_name = input("type the port of the device (eg: COM1): ")
     fixture = PressureFixture.create(port=port_name, slot_side="left")
-    fixture.connect()
+    fixture.connect_96()
     print(f"Device firmware version: {fixture.firmware_version()}")
     while True:
-        readings = fixture.read_all_pressure_channel()
-        print(readings)
+        readings = fixture.read_all_pressure_channel_96()
+        print("zuihoujieguo:",readings)
+        fixture.print_pressure_datas(readings)
         sleep(0.1)

hardware-testing/hardware_testing/opentrons_api/helpers_ot3.py

@@ -842,7 +842,7 @@ async def move_to_arched_ot3(
     abs_position: Point,
     speed: Optional[float] = None,
     safe_height: float = -100.0,
-) -> None:
+) -> Point:
     """Move OT3 gantry in an arched path."""
     z_ax = Axis.by_mount(mount)
     max_z = get_endstop_position_ot3(api, mount)[z_ax]
@@ -855,6 +855,7 @@ async def move_to_arched_ot3(
     ]
     for p in points:
         await api.move_to(mount=mount, abs_position=p, speed=speed)
+    return points
 
 
 class SensorResponseBad(Exception):