Add ROS 2 task implementations and configuration files

ros2_tasks/setup.py

@@ -0,0 +1,30 @@
+from setuptools import setup
+
+package_name = 'ros2_tasks'
+
+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='user',
+    maintainer_email='dinith.22@cse.mrt.ac.lk',
+    description='ROS 2 tasks solutions',
+    license='TODO: License declaration',
+    tests_require=['pytest'],
+    entry_points={
+        'console_scripts': [
+            'pose_re_pub = ros2_tasks.task1.pose_re_pub:main',
+            'params_setter = ros2_tasks.task2.params_setter:main',
+            'twist_from_database = ros2_tasks.task3.twist_from_database:main',
+            'zero_twist = ros2_tasks.task4.zero_twist:main',
+            'param_reader = ros2_tasks.task6.param_reader:main',
+        ],
+    },
+)

ros2_tasks/task1/package.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>ros2_tasks</name>
+  <version>0.0.0</version>
+  <description>ROS 2 tasks solutions</description>
+  <maintainer email="dinith.22@cse.mrt.ac.lk">user</maintainer>
+  <license>TODO: License declaration</license>
+
+  <depend>rclpy</depend>
+  <depend>geometry_msgs</depend>
+  <depend>std_msgs</depend>
+
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>

ros2_tasks/task1/pose_re_pub.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+
+import rclpy
+from rclpy.node import Node
+from geometry_msgs.msg import PoseWithCovarianceStamped, Pose
+
+class PoseRepublisher(Node):
+    def __init__(self):
+        super().__init__('pose_re_pub')
+
+        # Subscriber to PoseWithCovarianceStamped
+        self.subscription = self.create_subscription(
+            PoseWithCovarianceStamped,
+            'pose_with_covariance_stamped',
+            self.pose_callback,
+            10
+        )
+
+        # Publisher for Pose
+        self.publisher = self.create_publisher(Pose, 'pose', 10)
+
+        self.get_logger().info('Pose republisher node started')
+
+    def pose_callback(self, msg):
+        # Extract pose from PoseWithCovarianceStamped and republish
+        pose_msg = Pose()
+        pose_msg.position = msg.pose.pose.position
+        pose_msg.orientation = msg.pose.pose.orientation
+
+        self.publisher.publish(pose_msg)
+        self.get_logger().info('Republished pose message')
+
+def main(args=None):
+    rclpy.init(args=args)
+    node = PoseRepublisher()
+
+    try:
+        rclpy.spin(node)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        node.destroy_node()
+        rclpy.shutdown()
+
+if __name__ == '__main__':
+    main()

ros2_tasks/task2/package.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>ros2_tasks</name>
+  <version>0.0.0</version>
+  <description>ROS 2 tasks solutions</description>
+  <maintainer email="dinith.22@cse.mrt.ac.lk">user</maintainer>
+  <license>TODO: License declaration</license>
+
+  <depend>rclpy</depend>
+  <depend>geometry_msgs</depend>
+  <depend>std_msgs</depend>
+
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>

ros2_tasks/task2/params_setter.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+
+import rclpy
+from rclpy.node import Node
+
+class ParamsSetter(Node):
+    def __init__(self):
+        super().__init__('params_setter')
+
+        # Define parameter name and value
+        param = 'global_parameter'
+        value = 'default_value'
+
+        # Set global parameter
+        self.declare_parameter(param, value)
+        self.set_parameters([rclpy.parameter.Parameter(param, rclpy.Parameter.Type.STRING, value)])
+
+        self.get_logger().info(f'Set parameter "{param}" with value "{value}"')
+
+def main(args=None):
+    rclpy.init(args=args)
+    node = ParamsSetter()
+
+    try:
+        rclpy.spin(node)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        node.destroy_node()
+        rclpy.shutdown()
+
+if __name__ == '__main__':
+    main()

ros2_tasks/task3/package.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>ros2_tasks</name>
+  <version>0.0.0</version>
+  <description>ROS 2 tasks solutions</description>
+  <maintainer email="dinith.22@cse.mrt.ac.lk">user</maintainer>
+  <license>TODO: License declaration</license>
+
+  <depend>rclpy</depend>
+  <depend>geometry_msgs</depend>
+  <depend>std_msgs</depend>
+
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>

ros2_tasks/task3/twist_from_database.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+
+import rclpy
+from rclpy.node import Node
+from geometry_msgs.msg import Twist
+import csv
+import os
+
+class TwistFromDatabase(Node):
+    def __init__(self):
+        super().__init__('twist_from_database')
+
+        # Publisher with queue_size=20
+        self.publisher = self.create_publisher(Twist, 'twist_from_database', 20)
+
+        # Timer for 10Hz rate
+        self.timer = self.create_timer(0.1, self.timer_callback)
+
+        # Load CSV data
+        self.twist_data = []
+        self.current_index = 0
+        self.load_csv_data()
+
+        self.get_logger().info('Twist from database node started')
+
+    def load_csv_data(self):
+        try:
+            csv_file_path = os.path.join(os.path.dirname(__file__), 'values.csv')
+            with open(csv_file_path, 'r') as file:
+                csv_reader = csv.reader(file)
+                for row in csv_reader:
+                    if len(row) == 6:
+                        # Parse linear x,y,z and angular x,y,z
+                        linear_x, linear_y, linear_z, angular_x, angular_y, angular_z = map(float, row)
+                        self.twist_data.append({
+                            'linear': [linear_x, linear_y, linear_z],
+                            'angular': [angular_x, angular_y, angular_z]
+                        })
+            self.get_logger().info(f'Loaded {len(self.twist_data)} twist messages from CSV')
+        except FileNotFoundError:
+            self.get_logger().error('values.csv file not found')
+        except Exception as e:
+            self.get_logger().error(f'Error reading CSV: {str(e)}')
+
+    def timer_callback(self):
+        if self.twist_data and self.current_index < len(self.twist_data):
+            twist_msg = Twist()
+            data = self.twist_data[self.current_index]
+
+            twist_msg.linear.x = data['linear'][0]
+            twist_msg.linear.y = data['linear'][1]
+            twist_msg.linear.z = data['linear'][2]
+            twist_msg.angular.x = data['angular'][0]
+            twist_msg.angular.y = data['angular'][1]
+            twist_msg.angular.z = data['angular'][2]
+
+            self.publisher.publish(twist_msg)
+            self.get_logger().info(f'Published twist message {self.current_index + 1}')
+
+            self.current_index += 1
+        else:
+            # Reset to beginning or stop
+            self.current_index = 0
+
+def main(args=None):
+    rclpy.init(args=args)
+    node = TwistFromDatabase()
+
+    try:
+        rclpy.spin(node)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        node.destroy_node()
+        rclpy.shutdown()
+
+if __name__ == '__main__':
+    main()

ros2_tasks/task3/values.csv

@@ -0,0 +1,3 @@
+0.4,0.2,0.0,0.0,0.0,0.1
+0.5,0.3,0.0,0.0,0.0,0.2
+0.0,0.0,0.0,0.0,0.0,0.0

ros2_tasks/task4/package.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>ros2_tasks</name>
+  <version>0.0.0</version>
+  <description>ROS 2 tasks solutions</description>
+  <maintainer email="dinith.22@cse.mrt.ac.lk">user</maintainer>
+  <license>TODO: License declaration</license>
+
+  <depend>rclpy</depend>
+  <depend>geometry_msgs</depend>
+  <depend>std_msgs</depend>
+
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>

ros2_tasks/task4/zero_twist.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+
+import rclpy
+from rclpy.node import Node
+from geometry_msgs.msg import Twist
+from std_msgs.msg import String
+
+class ZeroTwist(Node):
+    def __init__(self):
+        super().__init__('zero_twist')
+
+        # Subscriber to is_stopped topic
+        self.subscription = self.create_subscription(
+            String,
+            'is_stopped',
+            self.is_stopped_callback,
+            10
+        )
+
+        # Publisher for twist with queue_size=10
+        self.publisher = self.create_publisher(Twist, 'twist', 10)
+
+        self.get_logger().info('Zero twist node started')
+
+    def is_stopped_callback(self, msg):
+        if msg.data.lower() == 'true':
+            # Publish zero twist message
+            zero_twist = Twist()
+            zero_twist.linear.x = 0.0
+            zero_twist.linear.y = 0.0
+            zero_twist.linear.z = 0.0
+            zero_twist.angular.x = 0.0
+            zero_twist.angular.y = 0.0
+            zero_twist.angular.z = 0.0
+
+            self.publisher.publish(zero_twist)
+            self.get_logger().info('Published zero twist message')
+        else:
+            self.get_logger().info('Received false, not publishing')
+
+def main(args=None):
+    rclpy.init(args=args)
+    node = ZeroTwist()
+
+    try:
+        rclpy.spin(node)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        node.destroy_node()
+        rclpy.shutdown()
+
+if __name__ == '__main__':
+    main()

ros2_tasks/task5/package.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>ros2_tasks</name>
+  <version>0.0.0</version>
+  <description>ROS 2 tasks solutions</description>
+  <maintainer email="dinith.22@cse.mrt.ac.lk">user</maintainer>
+  <license>TODO: License declaration</license>
+
+  <depend>rclpy</depend>
+  <depend>geometry_msgs</depend>
+  <depend>std_msgs</depend>
+
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>