Codes Nodes - Galileo-CDR2020/ROS GitHub Wiki

Codes Nodes

Topic

Publisher

Syntaxe minimale Publisher :

pub = rospy.Publisher("<topic_name>", <msg_type>, queue_size = 10)
pub.publish(<data>)

Exemple simple

Envoie un message 10 fois par seconde :

#!/usr/bin/env python
import rospy
from std_msgs.msg import String

def talker():
  pub = rospy.Publisher('chatter', String, queue_size = 10)
  rospy.init_node('talker', anonymous = True)
  rate = rospy.Rate(10) # 10hz
  while not rospy.is_shutdown():
    hello_str = "hello world %s" % rospy.get_time()
    rospy.loginfo(hello_str)
    pub.publish(hello_str)
    rate.sleep()

if __name__ == '__main__':
  try:
    talker()
  except rospy.ROSInterruptException:
    pass

Subscriber

Syntaxe minimal Subscriber :

rospy.Subscriber("<topic_name>", <msg_type>, <callback>)

Exemple simple :

#!/usr/bin/env python
import rospy
from std_msgs.msg import String

def callback(data):
  rospy.loginfo(rospy.get_caller_id() + "I heard %s", data.data)
    
def listener():
  rospy.init_node('listener', anonymous=True)

  rospy.Subscriber("chatter", String, callback)

  # spin() simply keeps python from exiting until this node is stopped
  rospy.spin()

if __name__ == '__main__':
  listener()

Service

Serveur Service

Syntaxe minimale d'un Service :

from <package_name>.srv import <service_type>, <service_type>Response

def callback(req):
   ...
  return <service_type>Response(...args)

s = rospy.Service('<service_name>', <service_type>, <callback>)

Exemple simple :

#!/usr/bin/env python
from beginner_tutorials.srv import AddTwoInts, AddTwoIntsResponse
import rospy

def handle_add_two_ints(req):
    print "Returning [%s + %s = %s]"%(req.a, req.b, (req.a + req.b))
    return AddTwoIntsResponse(req.a + req.b)

def add_two_ints_server():
    rospy.init_node('add_two_ints_server')
    s = rospy.Service('add_two_ints', AddTwoInts, handle_add_two_ints)
    print "Ready to add two ints."
    rospy.spin()

if __name__ == "__main__":
    add_two_ints_server()

Service Client

Syntaxe minimale d'un client d'un service :

rospy.wait_for_service('<service_name>')
...
try:
  service = rospy.ServiceProxy('<service_name>', <service_type>)
  res = service(...args)
  return res
except rospy.ServiceException, e:
  print "Service call failed: %s" % e

Exemple simple :

#!/usr/bin/env python
import sys
import rospy
from beginner_tutorials.srv import *

def add_two_ints_client(x, y):
  rospy.wait_for_service('add_two_ints')
  try:
    add_two_ints = rospy.ServiceProxy('add_two_ints', AddTwoInts)
    resp1 = add_two_ints(x, y)
    return resp1.sum
  except rospy.ServiceException, e:
    print "Service call failed: %s"%e

def usage():
  return "%s [x y]"%sys.argv[0]

if __name__ == "__main__":
  if len(sys.argv) == 3:
    x = int(sys.argv[1])
    y = int(sys.argv[2])
  else:
    print usage()
    sys.exit(1)
  print "Requesting %s+%s"%(x, y)
  print "%s + %s = %s"%(x, y, add_two_ints_client(x, y))

Class

Class Minimale :

#!/usr/bin/env python
import rospy

class ClassName:
    """ Ros node that ...
    """

    def __init__(self):
        self.config_ros()

    def config_ros(self):
        rospy.init_node('listener', anonymous=True)


if __name__ == '__main__':
    c = ClassName()
    rospy.spin()

Class Complète :

#!/usr/bin/env python
import rospy

class ClassName:
    """ Ros node that ...
    """

    def __init__(self):
        self.config_ros()

    def config_ros(self):
        rospy.init_node('listener', anonymous=True)

        # Publisher:
        self.pub = rospy.Publisher("<topic>", <Type>, queue_size=10)
        
        # Subscriber:
        rospy.Subscriber('<topic>', <Type>, self.msg_callback)

        # Service:
        self.s = rospy.Service('<service_name>', <service_type>, self.srv_callback)


    def msg_callback(self, data):
        """ Message Callback
        """
        rospy.loginfo(rospy.get_caller_id() + "I heard %s", data.<value>)


    def srv_callback(self, data):
        """ Service Callback
        """
        a, b = data.a, data.b
        res = ...
        return <service_type>Response(res, ...)
    
    def request_service(self, a, b):
        """ Service Client
        """
        try:
            service = rospy.ServiceProxy('<service_name>', <service_type>)
            res = service(...args)
            return res
        except rospy.ServiceException, e:
            print "Service call failed: %s" % e


if __name__ == '__main__':
    c = ClassName()

    rospy.wait_for_service('<service_name>')
    c.requestService(a, b)
    rospy.spin()
⚠️ **GitHub.com Fallback** ⚠️