refactor(main.py, task_3.py, task_test.py): update task execution flow and enhance control logic
- Commented out run_task_3 in main.py to prevent execution during testing. - Modified turn_degree function to set precision to True for improved control. - Removed the unnecessary task_3/main copy.py file. - Updated run_task_3 in task_3.py to improve gait file handling and control logic. - Adjusted parameters for stability detection and added checks for descending phases. - Updated task_test.py to focus on straight movement while commenting out previous commands.
This commit is contained in:
		
							parent
							
								
									0af26d8c70
								
							
						
					
					
						commit
						d89aca3b7a
					
				@ -9,7 +9,7 @@ from utils.log_helper import LogHelper, get_logger, section, info, debug, warnin
 | 
			
		||||
# 创建本模块特定的日志记录器
 | 
			
		||||
logger = get_logger("旋转控制")
 | 
			
		||||
 | 
			
		||||
def turn_degree(ctrl, msg, degree=90, absolute=False, precision=False):
 | 
			
		||||
def turn_degree(ctrl, msg, degree=90, absolute=False, precision=True):
 | 
			
		||||
    """
 | 
			
		||||
    结合里程计实现精确稳定的旋转指定角度
 | 
			
		||||
    
 | 
			
		||||
 | 
			
		||||
							
								
								
									
										4
									
								
								main.py
									
									
									
									
									
								
							
							
						
						
									
										4
									
								
								main.py
									
									
									
									
									
								
							@ -44,9 +44,9 @@ def main():
 | 
			
		||||
        # run_task_2_5(Ctrl, msg)
 | 
			
		||||
 | 
			
		||||
        # 执行任务3
 | 
			
		||||
        run_task_3(Ctrl, msg)
 | 
			
		||||
        # run_task_3(Ctrl, msg)
 | 
			
		||||
 | 
			
		||||
        # run_task_test(Ctrl, msg)
 | 
			
		||||
        run_task_test(Ctrl, msg)
 | 
			
		||||
 | 
			
		||||
        # time.sleep(100)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
							
								
								
									
										4981
									
								
								task_3/Gait_Params_up_full.toml
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										4981
									
								
								task_3/Gait_Params_up_full.toml
									
									
									
									
									
										Normal file
									
								
							
										
											
												File diff suppressed because it is too large
												Load Diff
											
										
									
								
							
							
								
								
									
										142
									
								
								task_3/task_3.py
									
									
									
									
									
								
							
							
						
						
									
										142
									
								
								task_3/task_3.py
									
									
									
									
									
								
							@ -33,15 +33,6 @@ robot_cmd = {
 | 
			
		||||
    'value':0,  'duration':0
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
def load_gait_file(file_path):
 | 
			
		||||
    """加载步态文件"""
 | 
			
		||||
    try:
 | 
			
		||||
        with open(file_path, 'r') as f:
 | 
			
		||||
            return f.read()
 | 
			
		||||
    except Exception as e:
 | 
			
		||||
        error(f"加载步态文件失败: {e}", "文件")
 | 
			
		||||
        return None
 | 
			
		||||
 | 
			
		||||
def run_task_3(ctrl, msg):
 | 
			
		||||
    section('任务3:步态切换', "启动")
 | 
			
		||||
    info('开始执行任务3...', "启动")
 | 
			
		||||
@ -85,6 +76,18 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
        f.writelines(toml.dumps(full_steps))
 | 
			
		||||
        f.close()
 | 
			
		||||
 | 
			
		||||
        # pre
 | 
			
		||||
        file_obj_gait_def = open("./task_3/Gait_Def_up.toml",'r')
 | 
			
		||||
        file_obj_gait_params = open("./task_3/Gait_Params_up_full.toml",'r')
 | 
			
		||||
        usergait_msg.data = file_obj_gait_def.read()
 | 
			
		||||
        lcm_usergait.publish("user_gait_file",usergait_msg.encode())
 | 
			
		||||
        time.sleep(0.5)
 | 
			
		||||
        usergait_msg.data = file_obj_gait_params.read()
 | 
			
		||||
        lcm_usergait.publish("user_gait_file",usergait_msg.encode())
 | 
			
		||||
        time.sleep(0.1)
 | 
			
		||||
        file_obj_gait_def.close()
 | 
			
		||||
        file_obj_gait_params.close()
 | 
			
		||||
 | 
			
		||||
        file_obj_gait_params = open("./task_3/Gait_Params_up_full.toml",'r')
 | 
			
		||||
        usergait_msg.data = file_obj_gait_params.read()
 | 
			
		||||
        lcm_usergait.publish("user_gait_file", usergait_msg.encode())
 | 
			
		||||
@ -100,15 +103,15 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
        
 | 
			
		||||
        # 参数设置
 | 
			
		||||
        stable_count = 0  # 用于计数z轴稳定的次数
 | 
			
		||||
        stable_threshold = 10  # 连续5次检测z轴不再增加则认为已经停止
 | 
			
		||||
        stable_threshold = 10  # 连续15次检测z轴不再增加则认为已经停止
 | 
			
		||||
        z_speed_threshold = 0.01  # z轴速度阈值,小于这个值认为已经停止爬升
 | 
			
		||||
        climb_speed_threshold = 0.05  # 检测到开始爬坡的速度阈值
 | 
			
		||||
        max_iterations = 600  # 最大循环次数,作为安全保障
 | 
			
		||||
        min_iterations = 200  # 最小循环次数,作为安全保障
 | 
			
		||||
        start_height = ctrl.odo_msg.xyz[2]  # 记录起始高度
 | 
			
		||||
        
 | 
			
		||||
        # 阶段控制
 | 
			
		||||
        climbing_detected = False  # 是否检测到正在爬坡
 | 
			
		||||
        initial_flat_stage = True  # 是否在初始平路阶段
 | 
			
		||||
        
 | 
			
		||||
        info(f"开始监测里程计Z轴速度,初始高度: {start_height}", "监测")
 | 
			
		||||
        
 | 
			
		||||
@ -128,11 +131,10 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
            # 检测是否开始爬坡阶段 - 使用z轴速度判断
 | 
			
		||||
            if not climbing_detected and vz > climb_speed_threshold:
 | 
			
		||||
                climbing_detected = True
 | 
			
		||||
                initial_flat_stage = False
 | 
			
		||||
                info(f"检测到开始爬坡,Z轴速度: {vz:.3f}, 当前高度: {ctrl.odo_msg.xyz[2]:.3f}", "监测")
 | 
			
		||||
            
 | 
			
		||||
            # 只有在检测到爬坡后,才开始监控Z轴是否停止增加
 | 
			
		||||
            if climbing_detected:
 | 
			
		||||
            if i > min_iterations and climbing_detected:
 | 
			
		||||
                # 如果Z轴速度接近于0或者为负,表示已经停止爬升或开始下降
 | 
			
		||||
                if abs(vz) < z_speed_threshold or vz < 0:
 | 
			
		||||
                    stable_count += 1
 | 
			
		||||
@ -145,7 +147,6 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
                    stable_count = 0
 | 
			
		||||
            
 | 
			
		||||
            time.sleep(0.2)
 | 
			
		||||
        
 | 
			
		||||
    except KeyboardInterrupt:
 | 
			
		||||
        msg.mode = 7 #PureDamper before KeyboardInterrupt:
 | 
			
		||||
        msg.gait_id = 0
 | 
			
		||||
@ -155,10 +156,68 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
        pass
 | 
			
		||||
 | 
			
		||||
    section('任务3-2:直线行走', "开始")
 | 
			
		||||
    go_straight(ctrl, msg, distance=1)
 | 
			
		||||
    # go_straight(ctrl, msg, distance=1)
 | 
			
		||||
    msg.mode = 11  # Locomotion模式
 | 
			
		||||
    msg.gait_id = 26  # 自变频步态
 | 
			
		||||
    msg.duration = 0  # wait next cmd
 | 
			
		||||
    msg.step_height = [0.06, 0.06]  # 抬腿高度
 | 
			
		||||
    msg.vel_des = [0, 0.2, 0]  # [前进速度, 侧向速度, 角速度]
 | 
			
		||||
    msg.life_count += 1
 | 
			
		||||
    ctrl.Send_cmd(msg)
 | 
			
		||||
    time.sleep(0.3)
 | 
			
		||||
 | 
			
		||||
    section('任务3-3:down', "完成")
 | 
			
		||||
    try:
 | 
			
		||||
        steps = toml.load("./task_3/Gait_Params_up.toml")
 | 
			
		||||
        full_steps = {'step':[robot_cmd]}
 | 
			
		||||
        k = 0
 | 
			
		||||
        for i in steps['step']:
 | 
			
		||||
            cmd = copy.deepcopy(robot_cmd)
 | 
			
		||||
            cmd['duration'] = i['duration']
 | 
			
		||||
            if i['type'] == 'usergait':                
 | 
			
		||||
                cmd['mode'] = 11 # LOCOMOTION
 | 
			
		||||
                cmd['gait_id'] = 110 # USERGAIT
 | 
			
		||||
                cmd['vel_des'] = i['body_vel_des']
 | 
			
		||||
                cmd['rpy_des'] = i['body_pos_des'][0:3]
 | 
			
		||||
                cmd['pos_des'] = i['body_pos_des'][3:6]
 | 
			
		||||
                cmd['foot_pose'][0:2] = i['landing_pos_des'][0:2]
 | 
			
		||||
                cmd['foot_pose'][2:4] = i['landing_pos_des'][3:5]
 | 
			
		||||
                cmd['foot_pose'][4:6] = i['landing_pos_des'][6:8]
 | 
			
		||||
                cmd['ctrl_point'][0:2] = i['landing_pos_des'][9:11]
 | 
			
		||||
                cmd['step_height'][0] = math.ceil(i['step_height'][0] * 1e3) + math.ceil(i['step_height'][1] * 1e3) * 1e3
 | 
			
		||||
                cmd['step_height'][1] = math.ceil(i['step_height'][2] * 1e3) + math.ceil(i['step_height'][3] * 1e3) * 1e3
 | 
			
		||||
                cmd['acc_des'] = i['weight']
 | 
			
		||||
                cmd['value'] = i['use_mpc_traj']
 | 
			
		||||
                cmd['contact'] = math.floor(i['landing_gain'] * 1e1)
 | 
			
		||||
                cmd['ctrl_point'][2] =  i['mu']
 | 
			
		||||
            if k == 0:
 | 
			
		||||
                full_steps['step'] = [cmd]
 | 
			
		||||
            else:
 | 
			
		||||
                full_steps['step'].append(cmd)
 | 
			
		||||
            k=k+1
 | 
			
		||||
        f = open("./task_3/Gait_Params_up_full.toml", 'w')
 | 
			
		||||
        f.write("# Gait Params\n")
 | 
			
		||||
        f.writelines(toml.dumps(full_steps))
 | 
			
		||||
        f.close()
 | 
			
		||||
 | 
			
		||||
        # pre
 | 
			
		||||
        file_obj_gait_def = open("./task_3/Gait_Def_up.toml",'r')
 | 
			
		||||
        file_obj_gait_params = open("./task_3/Gait_Params_up_full.toml",'r')
 | 
			
		||||
        usergait_msg.data = file_obj_gait_def.read()
 | 
			
		||||
        lcm_usergait.publish("user_gait_file",usergait_msg.encode())
 | 
			
		||||
        time.sleep(0.5)
 | 
			
		||||
        usergait_msg.data = file_obj_gait_params.read()
 | 
			
		||||
        lcm_usergait.publish("user_gait_file",usergait_msg.encode())
 | 
			
		||||
        time.sleep(0.1)
 | 
			
		||||
        file_obj_gait_def.close()
 | 
			
		||||
        file_obj_gait_params.close()
 | 
			
		||||
 | 
			
		||||
        file_obj_gait_params = open("./task_3/Gait_Params_up_full.toml",'r')
 | 
			
		||||
        usergait_msg.data = file_obj_gait_params.read()
 | 
			
		||||
        lcm_usergait.publish("user_gait_file", usergait_msg.encode())
 | 
			
		||||
        time.sleep(0.5)
 | 
			
		||||
        file_obj_gait_params.close()
 | 
			
		||||
 | 
			
		||||
        msg.mode = 62
 | 
			
		||||
        msg.value = 0
 | 
			
		||||
        msg.contact = 15
 | 
			
		||||
@ -168,51 +227,53 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
        
 | 
			
		||||
        # 参数设置
 | 
			
		||||
        stable_count = 0  # 用于计数z轴稳定的次数
 | 
			
		||||
        stable_threshold = 10  # 连续5次检测z轴不再增加则认为已经停止
 | 
			
		||||
        z_speed_threshold = 0.01  # z轴速度阈值,小于这个值认为已经停止爬升
 | 
			
		||||
        climb_speed_threshold = 0.05  # 检测到开始爬坡的速度阈值
 | 
			
		||||
        stable_threshold = 8  # 连续10次检测z轴速度接近零则认为已经到达平地
 | 
			
		||||
        z_speed_threshold = 0.01  # z轴速度阈值,小于这个值认为已经停止下降
 | 
			
		||||
        descent_speed_threshold = -0.05  # 检测到开始下坡的速度阈值(负值表示下降)
 | 
			
		||||
        max_iterations = 600  # 最大循环次数,作为安全保障
 | 
			
		||||
        min_iterations = 100  # 最小循环次数,确保有足够的时间开始动作
 | 
			
		||||
        start_height = ctrl.odo_msg.xyz[2]  # 记录起始高度
 | 
			
		||||
        
 | 
			
		||||
        # 阶段控制
 | 
			
		||||
        climbing_detected = False  # 是否检测到正在爬坡
 | 
			
		||||
        initial_flat_stage = True  # 是否在初始平路阶段
 | 
			
		||||
        descending_detected = False  # 是否检测到正在下坡
 | 
			
		||||
        flat_ground_detected = False  # 是否检测到已到达平地
 | 
			
		||||
        
 | 
			
		||||
        info(f"开始监测里程计Z轴速度,初始高度: {start_height}", "监测")
 | 
			
		||||
        info(f"开始监测下坡过程,初始高度: {start_height}", "监测")
 | 
			
		||||
        
 | 
			
		||||
        for i in range(max_iterations):
 | 
			
		||||
            # 发送控制命令维持心跳
 | 
			
		||||
            ctrl.Send_cmd(msg)
 | 
			
		||||
            
 | 
			
		||||
            # 每10次迭代打印一次当前信息
 | 
			
		||||
            if i % 10 == 0:
 | 
			
		||||
                # 获取当前Z轴位置和速度
 | 
			
		||||
                current_vz = ctrl.odo_msg.vxyz[2]  # z轴速度
 | 
			
		||||
                info(f"当前Z轴速度={current_vz:.3f}", "监测")
 | 
			
		||||
            
 | 
			
		||||
            # 获取z轴速度
 | 
			
		||||
            # 获取z轴速度和当前高度
 | 
			
		||||
            vz = ctrl.odo_msg.vxyz[2]
 | 
			
		||||
            current_height = ctrl.odo_msg.xyz[2]
 | 
			
		||||
            
 | 
			
		||||
            # 检测是否开始爬坡阶段 - 使用z轴速度判断
 | 
			
		||||
            if not climbing_detected and vz < -climb_speed_threshold:
 | 
			
		||||
                climbing_detected = True
 | 
			
		||||
                initial_flat_stage = False
 | 
			
		||||
                info(f"检测到开始爬坡,Z轴速度: {vz:.3f}, 当前高度: {ctrl.odo_msg.xyz[2]:.3f}", "监测")
 | 
			
		||||
            # 每10次迭代打印一次当前信息
 | 
			
		||||
            if observe and i % 10 == 0:
 | 
			
		||||
                info(f"当前Z轴速度={vz:.3f}, 当前高度={current_height:.3f}", "监测")
 | 
			
		||||
            
 | 
			
		||||
            # 只有在检测到爬坡后,才开始监控Z轴是否停止增加
 | 
			
		||||
            if climbing_detected:
 | 
			
		||||
                # 如果Z轴速度接近于0或者为正,表示已经停止爬升或开始下降
 | 
			
		||||
                if abs(vz) < z_speed_threshold or vz > 0:
 | 
			
		||||
            # 检测是否开始下坡阶段 - 使用z轴速度判断(负值表示下降)
 | 
			
		||||
            if not descending_detected and vz < descent_speed_threshold:
 | 
			
		||||
                descending_detected = True
 | 
			
		||||
                info(f"检测到开始下坡,Z轴速度: {vz:.3f}, 当前高度: {current_height:.3f}", "监测")
 | 
			
		||||
            
 | 
			
		||||
            # 只有在检测到下坡后,才开始监控是否到达平地
 | 
			
		||||
            if i > min_iterations and descending_detected:
 | 
			
		||||
                # 如果Z轴速度接近于0,表示已经停止下降(到达平地)
 | 
			
		||||
                if abs(vz) < z_speed_threshold:
 | 
			
		||||
                    stable_count += 1
 | 
			
		||||
                    if stable_count >= stable_threshold:
 | 
			
		||||
                        current_height = ctrl.odo_msg.xyz[2]
 | 
			
		||||
                        info(f"Z轴速度趋近于0,停止循环。当前速度: {vz:.3f}, 当前高度: {current_height:.3f}", "监测")
 | 
			
		||||
                        info(f"检测到已到达平地,Z轴速度趋近于0,停止循环。当前速度: {vz:.3f}, 当前高度: {current_height:.3f}, 下降了: {start_height - current_height:.3f}米", "监测")
 | 
			
		||||
                        flat_ground_detected = True
 | 
			
		||||
                        break
 | 
			
		||||
                else:
 | 
			
		||||
                    # 如果Z轴仍有明显上升速度,重置稳定计数
 | 
			
		||||
                    # 如果Z轴仍有明显下降速度,重置稳定计数
 | 
			
		||||
                    stable_count = 0
 | 
			
		||||
            
 | 
			
		||||
            time.sleep(0.2)
 | 
			
		||||
        
 | 
			
		||||
        if not flat_ground_detected:
 | 
			
		||||
            info(f"达到最大循环次数,未能明确检测到到达平地。当前高度: {ctrl.odo_msg.xyz[2]:.3f}", "警告")
 | 
			
		||||
    except KeyboardInterrupt:
 | 
			
		||||
        msg.mode = 7 #PureDamper before KeyboardInterrupt:
 | 
			
		||||
        msg.gait_id = 0
 | 
			
		||||
@ -220,4 +281,3 @@ def run_task_3(ctrl, msg):
 | 
			
		||||
        msg.life_count += 1
 | 
			
		||||
        ctrl.Send_cmd(msg)
 | 
			
		||||
        pass
 | 
			
		||||
                
 | 
			
		||||
@ -7,12 +7,15 @@ sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 | 
			
		||||
 | 
			
		||||
from base_move.turn_degree import turn_degree
 | 
			
		||||
from base_move.move_base_hori_line import follow_left_side_track
 | 
			
		||||
from base_move.go_straight import go_straight
 | 
			
		||||
from utils.log_helper import LogHelper, get_logger, section, info, debug, warning, error, success, timing
 | 
			
		||||
 | 
			
		||||
# 创建本模块特定的日志记录器
 | 
			
		||||
logger = get_logger("任务-test")
 | 
			
		||||
 | 
			
		||||
def run_task_test(ctrl, msg):
 | 
			
		||||
    turn_degree(ctrl, msg, 90, absolute=True)
 | 
			
		||||
    follow_left_side_track(ctrl, msg)
 | 
			
		||||
    # turn_degree(ctrl, msg, 90, absolute=True)
 | 
			
		||||
    # turn_degree(ctrl, msg, 90, absolute=True)
 | 
			
		||||
    go_straight(ctrl, msg, distance=0.3, speed=0.1)
 | 
			
		||||
    # follow_left_side_track(ctrl, msg)
 | 
			
		||||
    # time.sleep(100)
 | 
			
		||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user