ready for replace laser reader

Author: andy

devices/amsamotion_sensor.py

@@ -52,7 +52,7 @@ def get_distance_multi(self, device_addr: int, channel_num=8) -> dict:
             if ret == "":
                 raise ValueError("read sensor fail")
             ret = ret.split('\r\n')
-            ret = ret[1]
+            ret = ret[-2]
 
             for index, item in enumerate(ret.split(',')):
                 multi_value.update({index: (float(item.split(':')[1].strip()) / 1000)})

devices/laser_stj_10_m0.py

@@ -0,0 +1,68 @@
+"""
+@description: high accuracy laser sensor
+"""
+
+import codecs
+from drivers.serial_driver import SerialDriver
+
+MEASURE_CODE = bytes.fromhex("024D45415355524503")
+LOW_MEASURE_CODE = "GetVolt"
+
+
+class LaserSensor:
+    def __init__(self, send=True):
+        self.serial = SerialDriver()
+        self.send = send
+        self.accuracy = "high"
+
+    def init_device(self, select_default=False):
+        """
+        get device
+        :return:
+        """
+        if self.accuracy == "high":
+            self.serial.init(115200, select_default=select_default)
+        else:
+            self.serial.init(9600, select_default=select_default)
+
+    def close(self):
+        self.serial.close()
+
+    def read_sensor_high(self):
+        result = self.serial.write_and_get_buffer(MEASURE_CODE)
+        data = codecs.encode(result, 'hex')
+        if len(data) <= 0:
+            raise ValueError("Read Sensor Fail")
+        data = data.decode('utf-8')
+        # analyze data
+        _bai = int(data[3:4])
+        _shi = int(data[5:6])
+        _wei = int(data[7:8])
+        primary_value = _bai * 100 + _shi * 10 + _wei
+        _qian = int(data[11:12])
+        _bai = int(data[13:14])
+        _shi = int(data[15:16])
+        _wei = int(data[17:18])
+        secondary_value = _qian * 0.1 + _bai * 0.01 + _shi * 0.001 + _wei * 0.0001
+        value = primary_value + secondary_value
+        return value
+
+    def read_sensor_low(self):
+        multi_value = {}
+        result = self.serial.write_and_get_buffer(LOW_MEASURE_CODE)
+        result = result.decode('utf-8')
+        print(f"Result: {result}")
+        for index, item in enumerate(result.split(',')):
+            _value = float(item.split(':')[1].strip()) / 1000
+            _value = round(_value, 3)
+            multi_value.update({index: _value})
+        return multi_value
+
+
+if __name__ == '__main__':
+    ls = LaserSensor()
+    ls.init_device()
+    # ls.accuracy = "low"
+    while True:
+        ret = ls.read_sensor_high()
+        print(ret)

drivers/serial_driver.py

@@ -134,6 +134,4 @@ def read_buffer2(self, read_length, hex_flag=None):
 if __name__ == '__main__':
     s = SerialDriver()
     s.init(9600)
-    for i in range(100):
-        data = s.read_buffer()
-        print(data)
+    s.write_and_get_buffer("GetVolt")

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

@@ -47,7 +47,7 @@ async def jog(hc: HardwareControl, mount: Mount):
     else:
         select = PositionSel.MOUNT_RIGHT
     # useful_pos = await hc.require_useful_pos(select)
-    useful_pos = Point(50, 100, 500)
+    useful_pos = Point(60, 50, 400)
     print("useful_pos: ", useful_pos)
     await hc.move_to(mount, useful_pos, target="mount")
 
@@ -151,4 +151,4 @@ async def jog(hc: HardwareControl, mount: Mount):
 
 if __name__ == '__main__':
     hc = HardwareControl("192.168.6.33")
-    asyncio.run(jog(hc, Mount.LEFT))
+    asyncio.run(jog(hc, Mount.RIGHT))

ot3_testing/hardware_control/jog.py

@@ -84,7 +84,7 @@ async def read_distance_mm_from_code_value(self, code_value: int, get_voltage=Fa
 
     async def read_definition_distance(self, definition: List, channel_definition, laser: LaserSensor, mount,
                                        only_code=False,
-                                       send=False, add_compensation=True) -> dict:
+                                       send=False, add_compensation=True, wait_time= 1) -> dict:
         """
         read distance, using one device id (please use same device_id in the positions)
         :param definition:
@@ -98,7 +98,9 @@ async def read_definition_distance(self, definition: List, channel_definition, l
         """
         print("Reading Sensor...")
         result = {}
-        time.sleep(1)
+        for i in range(wait_time):
+            time.sleep(1)
+            print(f"wait ({wait_time})/{i+1}...")
         _channel_definition = channel_definition[mount]
         device_addr = _channel_definition[definition[0]]["device_addr"]
         code_value_list = laser.get_distance_multi(device_addr)

ot3_testing/tests/__pycache__/base_init.cpython-37.pyc

@@ -98,7 +98,7 @@ async def run_target_point(self, target: Point, target_name: str):
         # step 2 : move to target
         await self.move_to_test_point(target, ThisMount)
         # 等待读数稳定
-        time.sleep(1)
+        time.sleep(5)
         distance = self.sanliang.read_distance_n_times(5)  # 读取示数
         while abs(distance) >= FirstThreshold:
             play_alarm_2()

ot3_testing/tests/base_init.py

@@ -15,6 +15,8 @@
 ApplyCompensationFlag = True
 
 TEST_SPEC = 0.3
+DefaultPort = True
+AdjustBeforeTest = False
 
 
 class ZStageLeveling(TestBase):
@@ -33,7 +35,7 @@ def init_laser_sensor(self, send=False):
         :return:
         """
         self.laser_sensor = LaserSensor(send=send)
-        self.laser_sensor.init_device()
+        self.laser_sensor.init_device(select_default=DefaultPort)
 
     async def move_to_test_point(self, p: Point):
         """
@@ -142,7 +144,8 @@ async def run_test_slot(self, point: Point, slot_name: str, read_definition: Lis
             await self.calibrate_to_zero(slot_name, 0.1, read_definition, ZStageChannel,
                                          method=CalibrateMethod.Approach)
 
-        ret_dict = await self.read_definition_distance(read_definition, ZStageChannel, self.laser_sensor, self.mount)
+        ret_dict = await self.read_definition_distance(read_definition, ZStageChannel, self.laser_sensor, self.mount,
+                                                       wait_time=5)
         for key, value in ret_dict.items():
             print(f"{slot_name}-{key}: {value}")
         self.judge_test_result(list(ret_dict.values()), TEST_SPEC)
@@ -199,9 +202,10 @@ async def run_z_stage_test(self, flex_name, project_path=None):
         self.init_laser_sensor(send=False)
 
         # adjust
-        await self.adjust_leveling('Z-C2', Mount.RIGHT)
-        input("Run Test (开始测试)？")
-        # run test
+        if AdjustBeforeTest:
+            await self.adjust_leveling('Z-C2', Mount.RIGHT)
+            input("Run Test (开始测试)？")
+            # run test
         await self.api.home()
 
         csv_title = []
@@ -265,5 +269,6 @@ def save_csv(self, file_path, title, content):
 
 
 if __name__ == '__main__':
-    obj = ZStageLeveling(ZStagePoint, robot_ip="192.168.6.33")
-    asyncio.run(obj.run_z_stage_test("xxx"))
+    obj = ZStageLeveling(ZStagePoint, robot_ip="192.168.6.45")
+    for i in range(6):
+        asyncio.run(obj.run_z_stage_test(f"xxx_{i + 1}"))