Merge tag 'mp.pipette.assembly_qc_25.06.09' into mp.96ch.p200

Author: ryanthecoder

hardware-testing/hardware_testing/data/ui.py

@@ -6,7 +6,7 @@
 PRINT_HEADER_DASHES = "-" * PRINT_HEADER_NUM_SPACES
 PRINT_TITLE_POUNDS = "#" * PRINT_HEADER_NUM_SPACES
 PRINT_HEADER_SPACES = " " * (PRINT_HEADER_NUM_SPACES - 1)
-PRINT_HEADER_ASTERISK = "*"
+PRINT_HEADER_ASTERISK = "*" * PRINT_HEADER_NUM_SPACES
 
 
 def get_user_answer(question: str) -> bool:
@@ -72,3 +72,43 @@ def print_warning(message: str) -> None:
 def print_info(message: str) -> None:
     """Print information."""
     print(message)
+
+def print_fail  (message: str) -> None:
+    """print fail"""
+    length = len(message)
+    dashes = PRINT_HEADER_DASHES + ("-" * length) + PRINT_HEADER_DASHES
+    middle = f"|{PRINT_HEADER_SPACES}{message}{PRINT_HEADER_SPACES}|"
+    #print(f"\n{dashes}\n{middle}\n{dashes}\n")
+    print(f'\033[1;31m\n -FAIL- {dashes} \n{middle}\n{dashes}\n\033[0m')
+
+def print_test_results  (message: str,passval:bool) -> None:
+    """print fail"""
+    PRINT_HEADER_ASTERISK = "*" * PRINT_HEADER_NUM_SPACES
+    length = len(message)
+    dashes = PRINT_HEADER_ASTERISK + ("*" * length) + PRINT_HEADER_ASTERISK
+    middle = f"|{PRINT_HEADER_SPACES}{message}{PRINT_HEADER_SPACES}|"
+    #print(f"\n{dashes}\n{middle}\n{dashes}\n")
+    if passval:
+        print(f'\033[4;32m\n 测试结果 {dashes} \n{middle}\n{dashes}\n\033[0m')
+    else:
+        print(f'\033[1;31m\n 测试结果 {dashes} \n{middle}\n{dashes}\n\033[0m')
+
+def print_results  (message: str,passval:bool) -> None:
+    """"test results list"""
+    if passval:
+        print(f'\033[4;32m\n 测试结果PASS  ')
+        print(f'{dashes}\n\033[0m')
+    else:
+        max_length = max(len(item) for item in message)
+
+        PRINT_HEADER_ASTERISK = "*" * PRINT_HEADER_NUM_SPACES
+        dashes = PRINT_HEADER_ASTERISK + ("*" * max_length) + PRINT_HEADER_ASTERISK
+
+        middle = [
+            f"|{PRINT_HEADER_SPACES}{item.center(max_length)}{PRINT_HEADER_SPACES}|"
+            for item in message
+        ]
+        print(f'\033[1;31m\n 测试结果FAIL {dashes} ')
+        for line in middle:
+            print(line)
+        print(f'{dashes}\n\033[0m')

hardware-testing/hardware_testing/drivers/sealed_pressure_fixture.py

@@ -0,0 +1,139 @@
+# encoding:utf-8
+
+import time
+from typing import Union
+import serial
+import serial.tools.list_ports
+
+ReceiveBuffer = 100
+
+
+class SerialDriver:
+
+    @classmethod
+    def get_com_list(cls):
+        port_list = serial.tools.list_ports.comports()
+        return port_list
+
+    def __init__(self):
+        self.device = None
+        self.com = None
+
+    def get_device(self):
+        """
+        select device
+        :return:
+        """
+        port_list = SerialDriver.get_com_list()
+        print("=" * 5 + "PORT LIST" + "=" * 5)
+        for index, p in enumerate(port_list):
+            print(f"{index + 1} >>{p.device}")
+        select = input("Select Port Number(输入串口号对应的数字):")
+        self.device = port_list[int(select.strip()) - 1].device
+
+    def init_serial(self, baud):
+
+        """
+        init connection
+        :param baud:
+        :return:
+        """
+        self.com = serial.Serial(self.device, baud, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE,
+                                 bytesize=serial.EIGHTBITS, timeout=1)
+        if self.com.isOpen():
+            print(f"{self.device} Opened! \n")
+        # settings
+        self.com.bytesize = serial.EIGHTBITS  # 数据位 8
+        self.com.parity = serial.PARITY_NONE  # 无校验
+        self.com.stopbits = serial.STOPBITS_ONE  # 停止位 1
+
+    def close(self):
+        """
+        close com
+        :return:
+        """
+        self.com.close()
+        print(f"{self.device} Closed! \n")
+
+    def init(self, baud):
+        """
+        main
+        :return:
+        """
+        self.get_device()
+        try:
+            self.init_serial(baud)
+        except:
+            print("Can't find device")
+
+    def write_and_get_buffer(self, send: Union[str, int, bytes], only_write=False, delay=None, times=30):
+        """
+        send cmd
+        :return:
+        """
+        if self.com is None:
+            return
+        if type(send) is not bytes:
+            send = (send + "\r\n").encode('utf-8')
+        self.com.flushInput()
+        self.com.flushOutput()
+        self.com.write(send)
+        time.sleep(0.1)
+        if delay is None:
+            pass
+        else:
+            time.sleep(delay)
+        if only_write is True:
+            return
+        for i in range(times):
+            data = self.com.read(ReceiveBuffer)
+            if type(data) is not bytes:
+                if "OK" not in data.decode('utf-8') or "busy" in data.decode('utf-8'):
+                    time.sleep(1)
+                    continue
+            else:
+                return data
+            return data.decode('utf-8')
+
+    def read_buffer(self):
+        """
+        读取缓存
+        :return:
+        """
+        try:
+            self.com.flushInput()
+            self.com.flushOutput()
+        except:
+            pass
+        time.sleep(3)
+        length = ReceiveBuffer
+        data = self.com.read(length)
+        self.com.flushInput()
+        self.com.flushOutput()
+        return data.decode('utf-8')
+    
+    def get_pressure(self):
+        """
+        analyze pressure value
+        """
+        for _i in range(5):
+            try:
+                respond = self.read_buffer()
+                respond_list = respond.split('|')
+                respond_value = respond_list[1]
+                
+                average_value = respond_value.split('\r\n')[0].split('\t')[1].strip()
+                average_value = float(average_value)
+                return average_value
+            except:
+                print(f"get pressure fail at {_i} times")
+                pass
+
+
+
+if __name__ == '__main__':
+    s = SerialDriver()
+    s.init(9600)
+    for i in range(100):
+        result = s.get_pressure()
+        print(result)

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

@@ -205,20 +205,20 @@ async def _find_reservoir_pos() -> None:
     for trial in range(2):
         ui.print_header("JOG to 96-Tip RACK")
         if trial == 0:
-            tip_rack = str(pipette) + "ul"
+            tip_rack = pipette
             test_volume: int = pipette
         else:
-            tip_rack = "50ul"
+            tip_rack = 50
             test_volume = 1 if pipette == 200 else 5
         if not api.is_simulator:
-            ui.get_user_ready(f"ADD 96 tip-rack-{tip_rack} to slot #{TIP_RACK_96_SLOT}")
+            ui.get_user_ready(f"ADD 96 tip-rack-{tip_rack}ul to slot #{TIP_RACK_96_SLOT}")
         await helpers_ot3.move_to_arched_ot3(
             api, OT3Mount.LEFT, tip_rack_96_a1_nominal + Point(z=30)
         )
         await helpers_ot3.jog_mount_ot3(api, OT3Mount.LEFT)
         print("picking up tips")
         await api.pick_up_tip(
-            OT3Mount.LEFT, helpers_ot3.get_default_tip_length(pipette)
+            OT3Mount.LEFT, helpers_ot3.get_default_tip_length(tip_rack)
         )
         await api.home_z(OT3Mount.LEFT)
         if reservoir_a1_actual is None: