123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
import time
|
||
import sys
|
||
import os
|
||
import toml
|
||
import copy
|
||
import math
|
||
|
||
# 添加父目录到路径，以便能够导入utils
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
# 添加当前目录到路径，确保可以找到local文件
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from utils.log_helper import LogHelper, get_logger, section, info, debug, warning, error, success, timing
|
||
from base_move.turn_degree import turn_degree
|
||
from file_send_lcmt import file_send_lcmt
|
||
|
||
# 创建本模块特定的日志记录器
|
||
logger = get_logger("任务3")
|
||
|
||
observe = True
|
||
|
||
robot_cmd = {
|
||
'mode':0, 'gait_id':0, 'contact':0, 'life_count':0,
|
||
'vel_des':[0.0, 0.0, 0.0],
|
||
'rpy_des':[0.0, 0.0, 0.0],
|
||
'pos_des':[0.0, 0.0, 0.0],
|
||
'acc_des':[0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
|
||
'ctrl_point':[0.0, 0.0, 0.0],
|
||
'foot_pose':[0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
|
||
'step_height':[0.0, 0.0],
|
||
'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...', "启动")
|
||
usergait_msg = file_send_lcmt()
|
||
lcm_usergait = ctrl.lc_s
|
||
|
||
try:
|
||
steps = toml.load("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("Gait_Params_up_full.toml", 'w')
|
||
f.write("# Gait Params\n")
|
||
f.writelines(toml.dumps(full_steps))
|
||
f.close()
|
||
|
||
file_obj_gait_def = open("Gait_Def_up.toml",'r')
|
||
file_obj_gait_params = open("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()
|
||
|
||
user_gait_list = open("Usergait_List.toml",'r')
|
||
steps = toml.load(user_gait_list)
|
||
for step in steps['step']:
|
||
msg.mode = step['mode']
|
||
msg.value = step['value']
|
||
msg.contact = step['contact']
|
||
msg.gait_id = step['gait_id']
|
||
msg.duration = step['duration']
|
||
msg.life_count += 1
|
||
for i in range(3):
|
||
msg.vel_des[i] = step['vel_des'][i]
|
||
msg.rpy_des[i] = step['rpy_des'][i]
|
||
msg.pos_des[i] = step['pos_des'][i]
|
||
msg.acc_des[i] = step['acc_des'][i]
|
||
msg.acc_des[i+3] = step['acc_des'][i+3]
|
||
msg.foot_pose[i] = step['foot_pose'][i]
|
||
msg.ctrl_point[i] = step['ctrl_point'][i]
|
||
for i in range(2):
|
||
msg.step_height[i] = step['step_height'][i]
|
||
ctrl.Send_msg()
|
||
time.sleep( 0.1 )
|
||
for i in range(325): #15s Heat beat It is used to maintain the heartbeat when life count is not updated
|
||
ctrl.Send_msg()
|
||
time.sleep( 0.2 )
|
||
except KeyboardInterrupt:
|
||
msg.mode = 7 #PureDamper before KeyboardInterrupt:
|
||
msg.gait_id = 0
|
||
msg.duration = 0
|
||
msg.life_count += 1
|
||
ctrl.Send_msg()
|
||
pass |