Communication principale de ROS2

cd workspace/src
ros2 pkg create pkg_helloworld_py --build-type ament_python --dependencies rclpy --node-name helloworld

import rclpy
from rclpy.node import Node
import time

class HelloWorldNode(Node):
  def __init__(self, name):
  super().__init__(name)
  while rclpy.ok():
  self.get_logger().info("Hello World")
  time.sleep(0.5)

def main(args=None):
  rclpy.init(args=args)
  node = HelloWorldNode("helloworld")
  rclpy.spin(node)
  node.destroy_node()
  rclpy.shutdown()

colcon build --packages-select pkg_helloworld_py
source install/setup.bash
ros2 run pkg_helloworld_py helloworld

cd ~/workspaces/src
ros2 pkg create pkg_topic --build-type ament_python --dependencies rclpy --node-name publisher_demo

import rclpy
from rclpy.node import Node
from std_msgs.msg import String
class Topic_Pub(Node):
  def __init__(self,name):
  super().__init__(name)
  self.pub = self.create_publisher(String,"/topic_demo",1)
  self.timer = self.create_timer(1,self.pub_msg)
  def pub_msg(self):
  msg = String()
  msg.data = "Hi,I send a message."
  self.pub.publish(msg)

def main():
  rclpy.init()
  pub_demo = Topic_Pub("publisher_node")
  rclpy.spin(pub_demo)
  pub_demo.destroy_node()
  rclpy.shutdown()

cd ~/workspace
colcon build --packages-select pkg_topic
source install/setup.bash

ros2 run pkg_topic publisher_demo
ros2 topic list
ros2 topic echo /topic_demo

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class Topic_Sub(Node):
  def __init__(self,name):
  super().__init__(name)
  self.sub = self.create_subscription(String,"/topic_demo",self.sub_callback,1)
  def sub_callback(self,msg):
  self.get_logger().info(msg.data)

def main():
  rclpy.init()
  sub_demo = Topic_Sub("subscriber_node")
  rclpy.spin(sub_demo)
  sub_demo.destroy_node()
  rclpy.shutdown()

cd ~/workspace
colcon build --packages-select pkg_topic
source install/setup.bash

ros2 run pkg_topic publisher_demo
ros2 run pkg_topic subscriber_demo

# File: example interfaces/srv/AddTwoInts.srv
int64 a
int64 b
---
int64 sum

ros2 pkg create pkg_service --build-type ament_python --dependencies rclpy --node-name server_demo

import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts

class Service_Server(Node):
  def __init__(self,name):
  super().__init__(name)
  self.srv = self.create_service(AddTwoInts, '/add_two_ints', self.Add2Ints_callback)
  def Add2Ints_callback(self,request,response):
  response.sum = request.a + request.b
  print("response.sum = ",response.sum)
  return response
def main():
  rclpy.init()
  server_demo = Service_Server("publisher_node")
  rclpy.spin(server_demo)
  server_demo.destroy_node()
  rclpy.shutdown()

ros2 interface show example_interfaces/srv/AddTwoInts

'server_demo = pkg_service.server_demo:main',

colcon build --packages-select pkg_service
source install/setup.bash
ros2 run pkg_service server_demo

ros2 service list
ros2 service call /add_two_ints example_interfaces/srv/AddTwoInts "{a: 1,b: 4}"

import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts

class Service_Client(Node):
  def __init__(self,name):
  super().__init__(name)
  self.client = self.create_client(AddTwoInts,'/add_two_ints')
  while not self.client.wait_for_service(timeout_sec=1.0):
  print("service not available, waiting again...")
  self.request = AddTwoInts.Request()

  def send_request(self):
  self.request.a = 10
  self.request.b = 90
  self.future = self.client.call_async(self.request)

def main():
  rclpy.init()
  service_client = Service_Client("client_node")
  service_client.send_request()
  while rclpy.ok():
  rclpy.spin_once(service_client)
  if service_client.future.done():
  try:
  response = service_client.future.result()
  print("Result = ",response.sum)
  except Exception as e:
  service_client.get_logger().info('Service call failed %r' % (e,))
  break
  service_client.destroy_node()
  rclpy.shutdown()

'client_demo = pkg_service.client_demo:main'

cd ~/workspace
colcon build --packages-select pkg_service
source install/setup.bash
ros2 run pkg_service server_demo

source install/setup.bash
ros2 run pkg_service client_demo

ros2 pkg create --build-type ament_cmake pkg_interfaces

int64 num
---
int64 sum
---
float64 progress

<buildtool_depend>rosidl_default_generators</buildtool_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<depend>action_msgs</depend>
<member_of_group>rosidl_interface_packages</member_of_group>

find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "action/Progress.action")

cd ~/workspace
colcon build --packages-select pkg_interfaces

ros2 interface show pkg_interfaces/action/Progress

ros2 pkg create pkg_action --build-type ament_python --dependencies rclpy pkg_interfaces --node-name action_server_demo

import time
import rclpy
from rclpy.action import ActionServer
from rclpy.node import Node

from pkg_interfaces.action import Progress

class Action_Server(Node):
  def __init__(self):
  super().__init__('progress_action_server')
  self._action_server = ActionServer(
  self,
  Progress,
  'get_sum',
  self.execute_callback)
  self.get_logger().info('The action server has started!')

  def execute_callback(self, goal_handle):
  self.get_logger().info('Starting task execution...')
  feedback_msg = Progress.Feedback()
  total = 0
  for i in range(1, goal_handle.request.num + 1):
  total += i
  feedback_msg.progress = i / goal_handle.request.num
  self.get_logger().info('Continuous feedback: %.2f' % feedback_msg.progress)
  goal_handle.publish_feedback(feedback_msg)
  time.sleep(1)

  goal_handle.succeed()
  result = Progress.Result()
  result.sum = total
  self.get_logger().info('Task completed!')
  return result

def main(args=None):
  rclpy.init(args=args)
  Progress_action_server = Action_Server()
  rclpy.spin(Progress_action_server)
  Progress_action_server.destroy_node()
  rclpy.shutdown()

'action_server_demo = pkg_action.action_server_demo:main',

cd ~/workspace
colcon build --packages-select pkg_action
source install/setup.bash
ros2 run pkg_action action_server_demo

ros2 action list
ros2 action send_goal /get_sum pkg_interfaces/action/Progress "{num: 10}"