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

import rclpy  # ROS 2 Python client library
from rclpy.node import Node  # ROS 2 node class
import time

"""
Create a HelloWorld node. It prints the "hello world" log during initialization.
"""
class HelloWorldNode(Node):
  def __init__(self, name):
  super().__init__(name)  # initialize the ROS 2 node base class
  while rclpy.ok():  # Check whether ROS 2 is running normally
  self.get_logger().info("Hello World")  # ROS 2 log output
  time.sleep(0.5)  # Sleep to control the loop interval

def main(args=None):  # main entry function of the ROS 2 node
  rclpy.init(args=args)  # initialize the ROS 2 Python interface
  node = HelloWorldNode("helloworld")  # create and initialize the ROS 2 node object
  rclpy.spin(node)  # keep spinning until ROS 2 exits
  node.destroy_node()  # destroy the node object
  rclpy.shutdown()  # Shut down the ROS 2 Python client library

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 the `rclpy` library
import rclpy
from rclpy.node import Node
# Import the `String` message type
from std_msgs.msg import String
# Create a `Topic_Pub` subclass of `Node` and pass the node name as an argument
class Topic_Pub(Node):
  def __init__(self,name):
  super().__init__(name)
  # Create a publisher with `create_publisher()`. The arguments are:
  # Topic data type, topic name, and message queue depth
  self.pub = self.create_publisher(String,"/topic_demo",1)
  # Create a timer that triggers the callback every 1 second. The arguments are:
  # Callback interval and callback function
  self.timer = self.create_timer(1,self.pub_msg)
  # Define the callback function
  def pub_msg(self):
  msg = String()  #Create a `String` message variable named `msg`
  msg.data = "Hi,I send a message." #Assign a value to `msg.data`
  self.pub.publish(msg) #Publish the topic data

# Main function
def main():
  rclpy.init() #Initialize
  pub_demo = Topic_Pub("publisher_node") #Create a `Topic_Pub` object and pass in the node name
  rclpy.spin(pub_demo)  #Call `rclpy.spin()` and pass the `Topic_Pub` object created above
  pub_demo.destroy_node()  #destroy the node object
  rclpy.shutdown()  #Shut down the ROS 2 Python client library

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 related libraries
import rclpy
from rclpy.node import Node
from std_msgs.msg import String

class Topic_Sub(Node):
  def __init__(self,name):
  super().__init__(name)
  #Create a subscriber with `create_subscription()`. The arguments are the topic data type, topic name, callback function, and queue depth
  self.sub = self.create_subscription(String,"/topic_demo",self.sub_callback,1)
  #Callback function: print the received message
  def sub_callback(self,msg):
  self.get_logger().info(msg.data)



def main():
  rclpy.init() #initialize the ROS 2 Python interface
  sub_demo = Topic_Sub("subscriber_node") # Create and initialize the object
  rclpy.spin(sub_demo)
  sub_demo.destroy_node()  #destroy the node object
  rclpy.shutdown()  #Shut down the ROS 2 Python client library

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

ros2 run pkg_topic publisher_demo
ros2 run pkg_topic subscriber_demo

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

# Import related library files
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts

class Service_Server(Node):
  def __init__(self,name):
  super().__init__(name)
  # Create a service server with `create_service()`. The arguments are:
  # Service data type, service name, and service callback function
  self.srv = self.create_service(AddTwoInts, '/add_two_ints', self.Add2Ints_callback)
  # This service callback adds two integers and returns the result
  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()  # destroy the node object
  rclpy.shutdown()  # Shut down the ROS 2 Python client library

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 related libraries
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts

class Service_Client(Node):
  def __init__(self,name):
  super().__init__(name)
  # Create a client with `create_client()`. The arguments are the service data type and service topic name
  self.client = self.create_client(AddTwoInts,'/add_two_ints')
  # Wait in a loop until the service server starts successfully
  while not self.client.wait_for_service(timeout_sec=1.0):
  print("service not available, waiting again...")
  # Create the service request data object
  self.request = AddTwoInts.Request()

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

def main():
  rclpy.init() # Initialize the node
  service_client = Service_Client("client_node") #Create the object
  service_client.send_request() #send the service request
  while rclpy.ok():
  rclpy.spin_once(service_client)
  #Check whether the data processing is complete
  if service_client.future.done():
  try:
  #Get the service response and print it
  response = service_client.future.result()
  print("service_client.request.a = ",service_client.request.a)
  print("service_client.request.b = ",service_client.request.b)
  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')
  # Create the 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...')

  # Generate continuous feedback.
  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)

  # Generate the final response.
  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}"