read z-stage

Author: andy

devices/amsamotion_sensor.py

@@ -49,6 +49,8 @@ def get_distance_multi(self, device_addr: int, channel_num=8) -> dict:
         else:
             # use serial driver
             ret = self.serial.read_buffer()
+            if ret == "":
+                raise ValueError("read sensor fail")
             ret = ret.split('\r\n')
             ret = ret[1]
 

dist/Productions.exe

@@ -1,4 +1,6 @@
 import os.path
+import time
+import winsound
 
 from playsound import playsound
 from utils import Utils
@@ -10,6 +12,8 @@
 Alarm_2 = os.path.join('assets', 'sounds', 'mixkit-alarm-tone-996.wav')
 voice = 'assets/'
 
+value = 1
+
 
 def play_dog_barking():
     """
@@ -19,24 +23,67 @@ def play_dog_barking():
     playsound(Dog_Barking)
 
 
-def play_alarm_1():
+def play_alarm_1(project_path=None):
     """
     voice2
     :return:
     """
-    playsound(Alarm_1)
+    if project_path is not None:
+        voice = os.path.join(project_path, Alarm_1)
+    else:
+        voice = Alarm_1
+    playsound(voice)
 
 
-def play_alarm_2(project_path):
+def play_alarm_2(project_path=None):
     """
     voice2
     :return:
     """
-    voice = os.path.join(project_path, Alarm_2)
+    if project_path is not None:
+        voice = os.path.join(project_path, Alarm_2)
+    else:
+        voice = Alarm_2
     playsound(voice)
 
 
+def play_alarm_3(frequency, duration):
+    winsound.Beep(frequency, duration)
+
+
 if __name__ == '__main__':
-    # play_dog_barking()
-    # play_alarm_2(
-    pass
+    print(root_path)
+    # play_alarm_3(3000, 500)
+
+    import threading
+
+
+    def play_value(f, d):
+        play_alarm_3(f, d)
+
+
+    def judge_value():
+        while True:
+            # if value > 0.2:
+            #     play_value(500, 1000)
+            # elif 0.1 <value<= 0.2:
+            #     play_value(500, 1000)
+            # elif value
+            # elif 0.05 <= value <= 0.1:
+            #     play_value(500, 500)
+            # elif 0.03 < value < 0.05:
+            #     play_value(1000, 500)
+            # else:
+            #     play_value(1000, 300)
+            _value = int(-value * 1500 + 1000)
+            if _value < 0:
+                _value = 1500
+            play_alarm_3(_value, 500)
+
+
+
+    th = threading.Thread(target=judge_value)
+    th.start()
+    # th.join()
+    while True:
+        value = float(input("value:"))

drivers/play_sound.py

@@ -150,5 +150,5 @@ async def jog(hc: HardwareControl, mount: Mount):
 
 
 if __name__ == '__main__':
-    hc = HardwareControl("192.168.6.44")
+    hc = HardwareControl("192.168.6.33")
     asyncio.run(jog(hc, Mount.RIGHT))

ot3_testing/hardware_control/__pycache__/jog.cpython-37.pyc

@@ -14,17 +14,17 @@ class CalibrateMethod(enum.Enum):
                     "C3-X": {"Point": Point(382, 91, 300), "compensation": {"left": 0, "right": 0}},
                     "A2-X": {"Point": Point(213, 305, 300), "compensation": {"left": 0, "right": 0}},
                     "D1-Z": {"Point": Point(51, 99, 318),
-                             "compensation": {"rear_left": -0.56664, "rear_right": -0.6685, "front_left": -0.43876,
+                             "compensation": {"rear_left": 0.09, "rear_right": 0.01, "front_left": 0.11,
                                               "front_right": 0}},
                     # "B2-Z": Point(213, 324, 317),
                     "D3-Z": {"Point": Point(377, 99, 318),
-                             "compensation": {"rear_left": -0.56664, "rear_right": -0.70712, "front_left": -0.41324,
+                             "compensation": {"rear_left": 0.09, "rear_right": 0.01, "front_left": 0.11,
                                               "front_right": 0}},
                     "C2-Z": {"Point": Point(214, 210, 318),
-                             "compensation": {"rear_left": -0.60076, "rear_right": -0.679, "front_left": -0.32926,
+                             "compensation": {"rear_left": 0.09, "rear_right": 0.01, "front_left": 0.11,
                                               "front_right": 0}},
                     "A2-Z": {"Point": Point(218, 424, 390.5),
-                             "compensation": {"rear_left": -0.57076, "rear_right": -0.694, "front_left": -0.40876,
+                             "compensation": {"rear_left": 0.09, "rear_right": 0.01, "front_left": 0.11,
                                               "front_right": 0}},
                     "UninstallPos": {"Point": Point(223, 203, 500)}
                     }
@@ -48,9 +48,9 @@ class CalibrateMethod(enum.Enum):
                         "right_rear": {"device_addr": 1, "channel": 5, "offset": 0},
                         }
 
-SlotLocationCH8 = {"Y-C1-Left": {"Point": Point(198.42, 198.33, 299.16), "compensation": -0.015},
-                   "Y-C1-Right": {"Point": Point(172.07, 197.18, 299.16), "compensation": -0.03},
-                   "Y-C3-Right": {"Point": Point(499.83, 197.18, 299.16), "compensation": -0.03},
-                   "Y-A2-Right": {"Point": Point(335.94, 412.22, 299.16), "compensation": -0.03},
+SlotLocationCH8 = {"Y-C1-Left": {"Point": Point(215.42, 198.33, 299.16), "compensation": {"rear": 0, "front": 0}},
+                   "Y-C1-Right": {"Point": Point(172.07, 197.18, 299.16), "compensation": {"rear": 0, "front": 0}},
+                   "Y-C3-Right": {"Point": Point(499.83, 197.18, 299.16), "compensation": {"rear": 0, "front": 0}},
+                   "Y-A2-Right": {"Point": Point(335.94, 412.22, 299.16), "compensation": {"rear": 0, "front": 0}},
                    "UninstallPos": {"Point": Point(223, 203, 500), "compensation": 0}
                    }

ot3_testing/hardware_control/jog.py

@@ -1,13 +1,58 @@
 from ot3_testing.ot_type import Mount, Point
+import enum
+
+
+class CalibrateMethod(enum.Enum):
+    Approach = "approach"
+    Dichotomy = "dichotomy"
 
 
 ZStagePoint = {
 
     Mount.LEFT: {
-        "C2": { "point": Point(214, 210, 318), "compensation": {}}
+        "Z-A1": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-A2": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-A3": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-B1": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-B2": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-B3": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-C1": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-C2": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-C3": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-D1": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-D2": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+        "Z-D3": {"point": Point(214, 210, 418), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]},
+    },
+    Mount.RIGHT: {
+        "Z-C2": {"point": Point(177, 197, 357), "compensation": {"rear": 0, "front": 0},
+                 "channel_definition": ["below_rear", "below_front"]}
+    }
+}
 
+ZStageChannel = {
+    Mount.LEFT: {
+        "below_rear": {"device_addr": 1, "channel": 3},
+        "below_front": {"device_addr": 1, "channel": 2},
+        "left_rear": {"device_addr": 1, "channel": 1},
+        "left_front": {"device_addr": 1, "channel": 0}
     },
+
     Mount.RIGHT: {
-        "C2": Point(214, 210, 318)
+        "below_rear": {"device_addr": 1, "channel": 7},
+        "below_front": {"device_addr": 1, "channel": 6},
+        "left_rear": {"device_addr": 1, "channel": 5},
+        "left_front": {"device_addr": 1, "channel": 4}
     }
-}
+}

ot3_testing/test_config/pipette_leveling_config.py

@@ -1,6 +1,7 @@
 from ot3_testing.protocol.protocol_context import ProtocolContext
 from ot3_testing.hardware_control.hardware_control import HardwareControl
-from typing import Union
+from typing import Union, List
+from devices.amsamotion_sensor import LaserSensor
 import time
 
 
@@ -34,7 +35,7 @@ def delay_s(self, second):
         :return:
         """
         for i in range(second):
-            print(f"delay {(i+1)}/({second})s")
+            print(f"delay {(i + 1)}/({second})s")
             time.sleep(1)
 
     def get_address(self):
@@ -53,3 +54,55 @@ def print_info(self, info: str):
         """
         input("=======Info======\n"
               + info)
+
+    async def read_distance_mm_from_code_value(self, code_value: int, get_voltage=False, k=-2, b=35):
+        """
+        read real value
+        :param code_value:
+        :param get_voltage: return voltage
+        :param k:
+        :param b:
+        :return:
+        """
+        # if self.test_name == "96ch":
+        #     voltage = round(float((code_value / 1600) / 2), 3)  # /V
+        # else:
+        voltage = round(float(code_value), 3)
+        if get_voltage:
+            return voltage
+        else:
+            distance = k * voltage + b  # /mm
+            return round(float(distance), 3)
+
+    async def read_definition_distance(self, definition: List, channel_definition, laser: LaserSensor, mount,
+                                       only_code=False,
+                                       send=False, add_compensation=True) -> dict:
+        """
+        read distance, using one device id (please use same device_id in the positions)
+        :param definition:
+        :param channel_definition:
+        :param laser:
+        :param mount:
+        :param only_code:
+        :param send: 是否需要发送再接收，区分两种传感器
+        :param add_compensation: 添加补偿
+        :return:
+        """
+        print("Reading Sensor...")
+        result = {}
+        time.sleep(1)
+        _channel_definition = channel_definition[mount]
+        device_addr = _channel_definition[definition[0]]["device_addr"]
+        code_value_list = laser.get_distance_multi(device_addr)
+
+        if only_code:
+            for item in definition:
+                result.update({item: code_value_list[_channel_definition[item]["channel"]]})
+
+        else:
+            for item in definition:
+                code_value = code_value_list[channel_definition[item]["channel"]]
+                distance_value = await self.read_distance_mm_from_code_value(code_value)
+                result.update({item: distance_value})
+
+        return result

ot3_testing/test_config/zstage_leveling_config.py

@@ -56,7 +56,7 @@ async def move_to_test_slot(self, slot_name: str):
         """
         await self.move_to_test_point(self.slot_location[slot_name]["Point"], MountDefinition)
 
-    async def read_definition_distance(self, position: List[str], only_code=False) -> dict:
+    async def _read_definition_distance(self, position: List[str], only_code=False) -> dict:
         """
         read distance, using one device id (please use same device_id in the positions)
         :param position:
@@ -75,12 +75,12 @@ async def read_definition_distance(self, position: List[str], only_code=False) -
         else:
             for item in position:
                 code_value = code_value_list[self.channel_definition[item]["channel"]]
-                distance_value = await self.read_distance_mm_from_code_value(code_value)
+                distance_value = await self._read_distance_mm_from_code_value(code_value)
                 result.update({item: distance_value})
 
         return result
 
-    async def read_distance_mm_from_code_value(self, code_value: int, get_voltage=False):
+    async def _read_distance_mm_from_code_value(self, code_value: int, get_voltage=False):
         """
         read real value
         :param code_value:
@@ -165,10 +165,10 @@ async def calibrate_to_zero(self, test_name: str, spec: float, read_definition:
         init_step = 2
         while True:
             # read voltage
-            ret_dict = await self.read_definition_distance(read_definition, only_code=True)
+            ret_dict = await self._read_definition_distance(read_definition, only_code=True)
             # min_voltage = min(distance_list)
             _min = list(ret_dict.values())[0]  # judge the first channel
-            min_voltage = await self.read_distance_mm_from_code_value(_min, get_voltage=True)
+            min_voltage = await self._read_distance_mm_from_code_value(_min, get_voltage=True)
             print("current min voltage is: ", min_voltage)
             if spec >= (min_voltage - target_voltage) >= 0:
                 break
@@ -215,7 +215,7 @@ async def run_test_slot(self, test_slot_name: str, test_slot_value: str, read_de
         if with_cal:
             await self.calibrate_to_zero(test_slot_value, 0.1, read_definition, method=CalibrateMethod.Approach)
 
-        ret_dict = await self.read_definition_distance(read_definition)
+        ret_dict = await self._read_definition_distance(read_definition)
         for key, value in ret_dict.items():
             print(f"{test_slot_value}-{key}: {value}")
 
@@ -267,21 +267,24 @@ async def run_8ch_test(self, flex_name: str, project_path=None):
         csv_title.append(time_str + flex_name)
         print("=" * 5 + "Test Result" + "=" * 5 + "\n")
         for key, value in test_result.items():
+            result = []
             distance_list = list(value.values())
-            difference = distance_list[0] - distance_list[1]
+            difference = round(abs(distance_list[0] - distance_list[1]), 3)
             compensation = self.slot_location[key]["compensation"]
             if ApplyCompensationFlag:
-                print(f"apply offset {compensation} to {difference}, difference -> {difference - compensation}")
-                difference = difference - compensation
-            print(f"{key} --> {value} (mm) --> difference: {round(difference, 3)}(mm)")
+                compensation_idx = 0
+                for compensation_key, compensation_value in compensation.items():
+                    print(
+                        f"apply offset {compensation_key} -> {compensation_value}  to {distance_list[compensation_idx]}")
+                    result.append(compensation_value + distance_list[compensation_idx])
+                    compensation_idx += 1
+                difference = round(abs(result[0] - result[1]), 3)
+            print(f"{key} --> {value} (mm) --> difference: {difference}(mm)")
             for item_key, item_value in value.items():
                 csv_title.append(key + " " + item_key)
-                if ApplyCompensationFlag and "front" in item_key:
-                    csv_list.append(item_value - compensation)
-                else:
-                    csv_list.append(item_value)
+                csv_list.append(item_value)
             csv_title.append(key + "-Result")
-            csv_list.append(abs(difference))
+            csv_list.append(difference)
 
         # save csv
         if project_path is not None:
@@ -374,7 +377,8 @@ async def run_96ch_test(self, flex_name: str, project_path=None):
             if ApplyCompensationFlag:
                 compensation_idx = 0
                 for compensation_key, compensation_value in compensation.items():
-                    print(f"apply offset {compensation_key} -> {compensation_value} to {distance_list[compensation_idx]}")
+                    print(
+                        f"apply offset {compensation_key} -> {compensation_value} to {distance_list[compensation_idx]}")
                     result.append(compensation_value + distance_list[compensation_idx])
                     compensation_idx += 1
                 difference = round((abs(max(result) - min(result))), 3)
@@ -413,5 +417,11 @@ async def run_96ch_test(self, flex_name: str, project_path=None):
 
 if __name__ == '__main__':
     import asyncio
-    pipette_leveling = PipetteLeveling(SlotLocationCH96, ChannelDefinitionCH96, )
-    asyncio.run(pipette_leveling.run_96ch_test("FLXA1020240513001"))
+
+    # pipette_leveling = PipetteLeveling(SlotLocationCH96, ChannelDefinitionCH96, )
+    # pipette_leveling.test_name = '96ch'
+    # asyncio.run(pipette_leveling.run_96ch_test("0518003"))
+
+    pipette_leveling = PipetteLeveling(SlotLocationCH8, ChannelDefinitionCH8, )
+    pipette_leveling.test_name = '8ch'
+    asyncio.run(pipette_leveling.run_8ch_test("0518003"))