完善制动性能测试

code/aio.py

@@ -22,6 +22,7 @@ class App:
         # ========================================================================
         self.root = ctk.CTk()
         self.__set_root()
+        self.__t = datetime.now().strftime("%Y%m%d%H%M%S")
         # ========================================================================
         self.style = ttk.Style()
         self.style.configure("tv.Treeview", font=self.f_treeview, rowheight=25)
@@ -596,8 +597,7 @@ class App:
     @clibs.db_lock
     def __quit_preparation(self):
         os.chdir(clibs.log_path)
-        t = datetime.now().strftime("%Y%m%d%H%M%S")
+        disk_conn = sqlite3.connect(f"log_{self.__t}.db", isolation_level=None, check_same_thread=False, cached_statements=256)
-        disk_conn = sqlite3.connect(f"log_{t}.db", isolation_level=None, check_same_thread=False, cached_statements=256)
         disk_cursor = disk_conn.cursor()
 
         if clibs.db_state == "readonly":

code/automatic_test/do_brake.py

@@ -87,14 +87,17 @@ def initialization(path, sub, data_dirs, data_files, hr, w2t):
     return _config_file, _prj_file, _result_dirs, _avs
 
 
-@clibs.db_lock
+def gen_result_file(path, axis, t_end, reach, load, speed, speed_target, rounds, w2t):
-def gen_result_file(path, axis, t_end, reach, load, speed, rounds):
+    d_vel, d_trq, d_stop, threshold = [], [], [], 0.95
-    d_vel, d_trq, d_stop = [], [], []
 
     start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t_end-12))
     end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t_end))
+    try:
+        clibs.lock.acquire(True)
         clibs.cursor.execute(f"select content from logs where time between '{start_time}' and '{end_time}' and content like '%diagnosis.result%' order by id asc")
         records = clibs.cursor.fetchall()
+    finally:
+        clibs.lock.release()
 
     for record in records:  # 保留最后12s的数据
         data = eval(record[0])["data"]
@@ -107,12 +110,19 @@ def gen_result_file(path, axis, t_end, reach, load, speed, rounds):
             elif item.get("channel", None) == 0 and item.get("name", None) == "device_safety_estop":
                 d_stop.extend(d_item)
 
+    idx = d_stop.index(0)
+    av_estop = sum(d_vel[idx - 20:idx])/20 * clibs.RADIAN
+    if av_estop / speed_target < threshold:
+        w2t(f"[av_estop: {av_estop:.2f} | shouldbe: {speed_target:.2f}] 本次触发 ESTOP 时未采集到指定百分比的最大速度，即将重试！\n", "#8A2BE2")
+        return False
+
     df1 = pandas.DataFrame.from_dict({"hw_joint_vel_feedback": d_vel})
     df2 = pandas.DataFrame.from_dict({"device_servo_trq_feedback": d_trq})
     df3 = pandas.DataFrame.from_dict({"device_safety_estop": d_stop})
     df = pandas.concat([df1, df2, df3], axis=1)
     filename = f"{path}/j{axis}/reach{reach}_load{load}_speed{speed}/reach{reach}_load{load}_speed{speed}_{rounds}.data"
     df.to_csv(filename, sep="\t", index=False)
+    return True
 
 
 def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
@@ -244,7 +254,7 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
                     speed_target = avs[axis-1] * float(speed) / 100
                     clibs.insert_logdb("INFO", "do_brake", f"axis = {axis}, direction = {pon}, max speed = {speed_max}")
                     if speed_max < speed_target*0.95 or speed_max > speed_target*1.05:
-                        w2t(f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}", "indigo")
+                        w2t(f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}\n", "indigo")
                         clibs.insert_logdb("WARNING", "do_brake", f"Axis: {axis}-{count} | Speed: {speed_max} | Shouldbe: {speed_target}")
                     md.write_speed_max(speed_max)
 
@@ -255,6 +265,11 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
                     else:
                         break
 
+                while 1:
+                    clibs.c_ec.setdo_value(io_name, "true")
+                    md.r_reset_estop()
+                    md.r_clear_alarm()
+                    md.write_act(0)
                     # 5. 重新运行程序，发送继续运动信号，当速度达到最大值时，通过DO触发急停
                     hr.execution("rl_task.pp_to_main", tasks=["brake"])
                     hr.execution("state.switch_auto")
@@ -274,7 +289,7 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
                     def exec_brake():
                         flag, start, data, record = True, time.time(), None, None
                         while flag:
-                        time.sleep(0.2)
+                            time.sleep(0.1)
                             if time.time() - start > 20:
                                 w2t("20s 内未触发急停，需排查......", "red", "BrakeTimeoutError")
 
@@ -294,14 +309,17 @@ def run_rl(path, sub, hr, md, config_file, prj_file, result_dirs, avs, w2t):
                                 if (pon == "positive" and speed_moment > 0) or (pon == "negative" and speed_moment < 0):
                                     if abs(speed_moment) > speed_target * 0.95:
                                         clibs.c_ec.setdo_value(io_name, "false")
-                                    time.sleep(5)
+                                        time.sleep(3)
                                         flag = False
                                         break
                         return time.time()
 
                     t_end = exec_brake()
                     # 6. 保留数据并处理输出
-                gen_result_file(path, axis, t_end, reach, load, speed, rounds)
+                    ret = gen_result_file(path, axis, t_end, reach, load, speed, speed_target, rounds, w2t)
+                    if ret:
+                        break
+
     else:
         w2t(f"\n{sub.removeprefix("tool")}%负载的制动性能测试执行完毕，如需采集其他负载，须切换负载类型，并更换其他负载，重新执行。\n", "green")
         hr.execution("diagnosis.open", open=False, display_open=False, overrun=True, turn_area=True, delay_motion=False)