1、ROS1:teleop.py

#!/usr/bin/env python  
# -*- coding: utf-8 -* 

from curses import KEY_LEFT, KEY_RIGHT
import  os  
import  sys  
import  tty, termios
import rospy  
from std_msgs.msg import Int8  
from lys_pkg.msg import A1Motor_msg

# publish根据按键,发送不同的legstate给QuadrupedTask.cpp进行处理  
pub = rospy.Publisher('Quadruped_insrtuction', Int8)  

def keyboardLoop():  
    #初始化  
    rospy.init_node('Quadruped_teleop')  
    rate = rospy.Rate(rospy.get_param('~hz', 5))  

    #读取按键循环  
    while not rospy.is_shutdown():  
        fd = sys.stdin.fileno()  #终端的文件描述符
        old_settings = termios.tcgetattr(fd)  #获取与终端相关的参数,返回的结果保存在termios结构体中
        #不产生回显效果  
        old_settings[3] = old_settings[3] & ~termios.ICANON & ~termios.ECHO  

        try :  
            tty.setraw( fd )  #将终端模式更改为原始模式
            ch = sys.stdin.read( 1 )  

        finally :  
            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)  

        if ch == 'c' :    #biao ding 
           leg_state = 3
        elif ch == 's' :    #stand
            leg_state = 1
        elif ch == 'y' :     #trot
            leg_state = 2
        elif ch == 'u' :     #up slope
            leg_state =4
        elif ch == 'd' :    #die
            leg_state = 0
        elif ch == 'q':    #stair 1
            leg_state=5
        elif ch == 'w':    #stair 2\3
            leg_state=6
        elif ch == 'e':     #stair 4
            leg_state=7   
        elif ch == 'r':     #stair last
            leg_state=8  
        elif ch == 'v':  #imu 
            leg_state=9  
        elif ch == 'l':    #yuandi  revise
            leg_state=10
        elif ch == 'k':     #stair jiao zheng
            leg_state=11   
        elif ch == ' ':      #stand calibration
            leg_state=12     
        elif ch == 'j':       #stair stand
            leg_state=13  
        elif ch == 'n':       #turn left
            leg_state=14 
        elif ch =='m':     #turn right
            leg_state=15
        elif ch=='z':     #slow trot
            leg_state=16
        elif ch=='x':     #up down
            leg_state=17
        elif ch=='b':     #roll
            leg_state=18
        elif ch=='a':     #front back
            leg_state=19

            
        else: 
             speed = -1
            
        #print (sys.stdin.encoding)
        #发送消息  
        print(ch)
        pub.publish(leg_state)  
        rate.sleep()  

if __name__ == '__main__':  

    try:  

        keyboardLoop()  

    except rospy.ROSInterruptException:  

        pass

2、ROS2:teleop.py

# #!/usr/bin/env python  
# # -*- coding: utf-8 -*  

from curses import KEY_LEFT, KEY_RIGHT
import  os  
import yaml
import  sys  
import  tty, termios
import rclpy  
from rclpy.node import Node
from std_msgs.msg import Int8  
from test_msgs.msg import Motormsg

# publish根据按键,发送不同的legstate给QuadrupedTask.cpp进行处理  
# pub = rclpy.Publisher('Quadruped_insrtuction', Int8)  

def keyboardLoop():  
    #初始化  
    # rclpy.init_node('Quadruped_teleop')  
    # rclpy.Node('Quadruped_teleop')
    # rclpy.init(args=args)
    # node = rclpy.create_node('ros2_1_test')
    # qos = rclpy.qos.QoSProfile(durability=rclpy.qos.QoSDurabilityPolicy.RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, depth=1)
    # publisher = node.create_publisher(String, '/ros2_to_1', qos_profile=qos)

    rclpy.init()
    pubnode = rclpy.create_node('Quadruped_teleop')
    qos = rclpy.qos.QoSProfile(durability=rclpy.qos.QoSDurabilityPolicy.RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, depth=1)

    publisher = pubnode.create_publisher(Int8,'/Quadruped_insrtuction',qos_profile=qos)
    # rate = rclpy.Rate(rclpy.get_param('~hz', 5))  
    rate = pubnode.create_rate(5)
    
    #读取按键循环  
    # while not rclpy.is_shutdown():  
    while rclpy.ok():
        fd = sys.stdin.fileno()  #终端的文件描述符
        old_settings = termios.tcgetattr(fd)  #获取与终端相关的参数,返回的结果保存在termios结构体中
        #不产生回显效果  
        old_settings[3] = old_settings[3] & ~termios.ICANON & ~termios.ECHO  

        try :  
            tty.setraw( fd )  #将终端模式更改为原始模式
            ch = sys.stdin.read( 1 )  

        finally :  
            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)  

        if ch == 'c' :    #biao ding 
           leg_state = 3
        elif ch == 's' :    #stand
            leg_state = 1
        elif ch == 'y' :     #trot
            leg_state = 2
        elif ch == 'u' :     #up slope
            leg_state =4
        elif ch == 'd' :    #die
            leg_state = 0
        elif ch == 'q':    #stair 1
            leg_state=5
        elif ch == 'w':    #stair 2\3
            leg_state=6
        elif ch == 'e':     #stair 4
            leg_state=7   
        elif ch == 'r':     #stair last
            leg_state=8  
        elif ch == 'v':  #imu 
            leg_state=9  
        elif ch == 'l':    #yuandi  revise
            leg_state=10
        elif ch == 'k':     #stair jiao zheng
            leg_state=11   
        elif ch == ' ':      #stand calibration
            leg_state=12     
        elif ch == 'j':       #stair stand
            leg_state=13  
        elif ch == 'n':       #turn left
            leg_state=14 
        elif ch =='m':     #turn right
            leg_state=15
        elif ch=='z':     #slow trot
            leg_state=16
        elif ch=='x':     #up down
            leg_state=17
        elif ch=='b':     #roll
            leg_state=18
        elif ch=='a':     #front back
            leg_state=19

            
        else: 
             speed = -1
            
        #print (sys.stdin.encoding)
        #发送消息  
        print(ch)
        msg = Int8()

        # dictionary = {
        # 'topic_name': '/Quadruped_teleop',
        # 'latch' : True,
        # 'reliability': 'reliable',
        # 'durability': 'transient_local',
        # 'history': 'keep_last',
        # 'depth': 1,
        # 'queue_size': 1,
        # 'type': 'std_msgs/Int8'
        # }

        # msg.data = str(yaml.dump(dictionary, Dumper=yaml.dumper.SafeDumper))

        msg.data = leg_state
        publisher.publish(msg)
        # msg=Int8
        # msg = leg_state
        # teleoppublisher.publish(msg)
        # teleoppublisher.publish(leg_state)  
        # rate.sleep()  
        # pubnode.destroy_node()
    
        # rclpy.shutdown()

if __name__ == '__main__':  

    try:  

        keyboardLoop()  

    except rclpy.ROSInterruptException:  

        pass

setup.py

from setuptools import setup

package_name = 'teleop'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='hahalim',
    maintainer_email='1171274147@qq.com',
    description='teleop',
    license='Apache License 2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
             "Quadruped_teleop = teleop.teleop:keyboardLoop"//可执行文件=包名.XXX(py文件名).函数名 
            #  "Quadruped_teleop = teleop.teleop:main"
        ],
    },
)

3、修改语句说明

rospy.init_node('Quadruped_teleop')改为rclpy.init()和pubnode = rclpy.create_node('Quadruped_teleop')

加上qos = rclpy.qos.QoSProfile(durability=rclpy.qos.QoSDurabilityPolicy.RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, depth=1)

pub = rospy.Publisher('Quadruped_insrtuction', Int8) 改为publisher = pubnode.create_publisher(Int8,'/Quadruped_insrtuction',qos_profile=qos)

rate = rospy.Rate(rospy.get_param('~hz', 5)) 改为rate = pubnode.create_rate(5)

pub.publish(leg_state) 改为msg = Int8()和msg.data = leg_state和publisher.publish(msg)

Logo

魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。

更多推荐