Isaac ROS Pipeline Code Examples
1. Isaac ROS Image Processing Pipeline
# Example: Isaac ROS Image Processing Node
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from cv_bridge import CvBridge
import cv2
import numpy as np
from isaac_ros_test import IsaacROSBaseTest
import pytest
from launch import LaunchDescription
from launch_ros.actions import ComposableNodeContainer
from launch_ros.descriptions import ComposableNode
from launch_testing_ros import LaunchTestService
class IsaacImageProcessorNode(Node):
def __init__(self):
super().__init__('isaac_image_processor')
# Create subscription to raw image
self.image_sub = self.create_subscription(
Image,
'/image_raw',
self.image_callback,
10
)
# Create publisher for processed image
self.processed_pub = self.create_publisher(
Image,
'/image_processed',
10
)
# Initialize CV bridge
self.bridge = CvBridge()
# Isaac ROS specific parameters
self.declare_parameters(
namespace='',
parameters=[
('input_width', 640),
('input_height', 480),
('enable_resize', True),
('resize_width', 320),
('resize_height', 240)
]
)
self.get_logger().info('Isaac Image Processor Node initialized')
def image_callback(self, msg):
"""Process incoming image using Isaac ROS optimized functions"""
try:
# Convert ROS Image to OpenCV format
cv_image = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')
# Apply Isaac ROS optimized image processing
processed_image = self.apply_isaac_processing(cv_image)
# Convert back to ROS Image format
processed_msg = self.bridge.cv2_to_imgmsg(processed_image, encoding='bgr8')
processed_msg.header = msg.header
# Publish processed image
self.processed_pub.publish(processed_msg)
except Exception as e:
self.get_logger().error(f'Error processing image: {str(e)}')
def apply_isaac_processing(self, image):
"""Apply Isaac ROS optimized image processing"""
# Get parameters
enable_resize = self.get_parameter('enable_resize').get_parameter_value().bool_value
resize_width = self.get_parameter('resize_width').get_parameter_value().integer_value
resize_height = self.get_parameter('resize_height').get_parameter_value().integer_value
# Resize if enabled (Isaac ROS provides optimized resize)
if enable_resize:
image = cv2.resize(image, (resize_width, resize_height))
# Apply Isaac ROS optimized image processing (e.g., color conversion, filtering)
# This is a simplified example - real Isaac ROS nodes use CUDA acceleration
processed = cv2.GaussianBlur(image, (5, 5), 0)
return processed
def generate_launch_description():
"""Generate launch description for Isaac ROS image processing pipeline"""
launch_description = LaunchDescription()
# Isaac ROS Image Proc Container
image_proc_container = ComposableNodeContainer(
name='image_proc_container',
namespace='',
package='rclcpp_components',
executable='component_container_mt',
composable_node_descriptions=[
ComposableNode(
package='isaac_ros_image_proc',
plugin='nvidia::isaac_ros::image_proc::ResizeNode',
name='resize_node',
parameters=[{
'output_width': 640,
'output_height': 480,
}],
remappings=[
('image', '/camera/image_raw'),
('camera_info', '/camera/camera_info'),
('resized/image', '/camera/image_resized'),
('resized/camera_info', '/camera/camera_info_resized'),
],
),
ComposableNode(
package='isaac_ros_image_proc',
plugin='nvidia::isaac_ros::image_proc::RectifyNode',
name='rectify_node',
remappings=[
('image_raw', '/camera/image_resized'),
('camera_info', '/camera/camera_info_resized'),
('image_rect', '/camera/image_rect'),
('camera_info_rect', '/camera/camera_info_rect'),
],
),
],
output='screen',
)
launch_description.add_action(image_proc_container)
return launch_description
def main(args=None):
rclpy.init(args=args)
image_processor = IsaacImageProcessorNode()
try:
rclpy.spin(image_processor)
except KeyboardInterrupt:
pass
finally:
image_processor.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
2. Isaac ROS Object Detection Pipeline
# Example: Isaac ROS Object Detection Node using DetectNet
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from vision_msgs.msg import Detection2DArray, Detection2D, ObjectHypothesisWithPose
from cv_bridge import CvBridge
import numpy as np
import torch
from std_msgs.msg import String
class IsaacObjectDetectionNode(Node):
def __init__(self):
super().__init__('isaac_object_detection')
# Create subscription to camera image
self.image_sub = self.create_subscription(
Image,
'/image_raw',
self.image_callback,
10
)
# Create publisher for detections
self.detection_pub = self.create_publisher(
Detection2DArray,
'/detections',
10
)
# Create publisher for status
self.status_pub = self.create_publisher(String, '/detection_status', 10)
# Initialize CV bridge
self.bridge = CvBridge()
# Isaac ROS DetectNet parameters
self.declare_parameters(
namespace='',
parameters=[
('model_path', '/models/detectnet_model.trt'),
('confidence_threshold', 0.5),
('max_objects', 10),
('input_width', 512),
('input_height', 288)
]
)
# Initialize model (Isaac ROS uses TensorRT optimized models)
self.initialize_model()
self.get_logger().info('Isaac Object Detection Node initialized')
def initialize_model(self):
"""Initialize Isaac ROS DetectNet model"""
try:
# In real Isaac ROS implementation, this would load a TensorRT optimized model
# For demonstration, we'll use a placeholder
self.model_initialized = True
self.get_logger().info('DetectNet model initialized')
except Exception as e:
self.get_logger().error(f'Error initializing model: {str(e)}')
self.model_initialized = False
def image_callback(self, msg):
"""Process image for object detection using Isaac ROS pipeline"""
if not self.model_initialized:
self.get_logger().warn('Model not initialized, skipping detection')
return
try:
# Convert ROS Image to OpenCV format
cv_image = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')
# Perform object detection using Isaac ROS optimized pipeline
detections = self.perform_detection(cv_image)
# Create Detection2DArray message
detection_array = Detection2DArray()
detection_array.header = msg.header
for detection in detections:
detection_2d = Detection2D()
detection_2d.header = msg.header
# Set bounding box
detection_2d.bbox.center.x = detection['center_x']
detection_2d.bbox.center.y = detection['center_y']
detection_2d.bbox.size_x = detection['width']
detection_2d.bbox.size_y = detection['height']
# Set results (confidence and class)
hypothesis = ObjectHypothesisWithPose()
hypothesis.hypothesis.class_id = detection['class']
hypothesis.hypothesis.score = detection['confidence']
detection_2d.results.append(hypothesis)
detection_array.detections.append(detection_2d)
# Publish detections
self.detection_pub.publish(detection_array)
# Publish status
status_msg = String()
status_msg.data = f'Detected {len(detections)} objects'
self.status_pub.publish(status_msg)
except Exception as e:
self.get_logger().error(f'Error in object detection: {str(e)}')
def perform_detection(self, image):
"""Perform object detection using Isaac ROS optimized pipeline"""
# Get parameters
confidence_threshold = self.get_parameter('confidence_threshold').get_parameter_value().double_value
max_objects = self.get_parameter('max_objects').get_parameter_value().integer_value
input_width = self.get_parameter('input_width').get_parameter_value().integer_value
input_height = self.get_parameter('input_height').get_parameter_value().integer_value
# Preprocess image for Isaac ROS DetectNet
# In real implementation, this would use Isaac ROS optimized preprocessing
resized_image = cv2.resize(image, (input_width, input_height))
normalized_image = resized_image.astype(np.float32) / 255.0
tensor = np.transpose(normalized_image, (2, 0, 1)) # HWC to CHW
tensor = np.expand_dims(tensor, axis=0) # Add batch dimension
# In real Isaac ROS implementation, this would run the TensorRT model
# For demonstration, we'll return mock detections
detections = [
{
'center_x': 256,
'center_y': 144,
'width': 100,
'height': 100,
'class': 'person',
'confidence': 0.92
},
{
'center_x': 100,
'center_y': 200,
'width': 80,
'height': 120,
'class': 'car',
'confidence': 0.87
}
]
# Filter detections by confidence threshold
detections = [d for d in detections if d['confidence'] >= confidence_threshold]
# Limit to max_objects
if len(detections) > max_objects:
detections = detections[:max_objects]
return detections
def main(args=None):
rclpy.init(args=args)
detector = IsaacObjectDetectionNode()
try:
rclpy.spin(detector)
except KeyboardInterrupt:
pass
finally:
detector.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
3. Isaac ROS Point Cloud Processing Pipeline
# Example: Isaac ROS Point Cloud Processing Node
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import PointCloud2, PointField
from sensor_msgs_py import point_cloud2
from std_msgs.msg import Header
from geometry_msgs.msg import Point
from visualization_msgs.msg import Marker, MarkerArray
import numpy as np
import struct
from collections import namedtuple
class IsaacPointCloudProcessorNode(Node):
def __init__(self):
super().__init__('isaac_pointcloud_processor')
# Create subscription to point cloud
self.pc_sub = self.create_subscription(
PointCloud2,
'/velodyne_points',
self.pc_callback,
10
)
# Create publisher for processed point cloud
self.processed_pc_pub = self.create_publisher(
PointCloud2,
'/velodyne_points_processed',
10
)
# Create publisher for visualization
self.vis_pub = self.create_publisher(MarkerArray, '/pointcloud_visualization', 10)
# Isaac ROS Point Cloud parameters
self.declare_parameters(
namespace='',
parameters=[
('min_range', 0.5),
('max_range', 100.0),
('ground_removal_height', 0.1),
('clustering_voxel_size', 0.5),
('min_cluster_size', 10),
('max_cluster_size', 1000)
]
)
self.get_logger().info('Isaac Point Cloud Processor Node initialized')
def pc_callback(self, msg):
"""Process incoming point cloud using Isaac ROS pipeline"""
try:
# Convert PointCloud2 to structured array using Isaac ROS optimized functions
points = point_cloud2.read_points(msg, field_names=['x', 'y', 'z'], skip_nans=True)
points_array = np.array(list(points))
if len(points_array) == 0:
self.get_logger().warn('Empty point cloud received')
return
# Apply Isaac ROS optimized point cloud processing
processed_points = self.process_pointcloud(points_array)
# Convert back to PointCloud2
processed_pc = self.create_pointcloud2_msg(processed_points, msg.header)
# Publish processed point cloud
self.processed_pc_pub.publish(processed_pc)
# Visualize the results
self.visualize_pointcloud(processed_points)
except Exception as e:
self.get_logger().error(f'Error processing point cloud: {str(e)}')
def process_pointcloud(self, points):
"""Process point cloud using Isaac ROS optimized pipeline"""
# Get parameters
min_range = self.get_parameter('min_range').get_parameter_value().double_value
max_range = self.get_parameter('max_range').get_parameter_value().double_value
ground_removal_height = self.get_parameter('ground_removal_height').get_parameter_value().double_value
# Remove points outside range
distances = np.sqrt(points[:, 0]**2 + points[:, 1]**2 + points[:, 2]**2)
range_filtered = points[(distances >= min_range) & (distances <= max_range)]
# Remove ground points (Isaac ROS provides optimized ground removal)
ground_filtered = range_filtered[range_filtered[:, 2] > ground_removal_height]
# In real Isaac ROS implementation, this would include:
# - Ground plane fitting using RANSAC
# - Clustering for object segmentation
# - Noise filtering
# - Voxel grid filtering
return ground_filtered
def create_pointcloud2_msg(self, points, header):
"""Create PointCloud2 message from numpy array using Isaac ROS format"""
# Define point fields
fields = [
PointField(name='x', offset=0, datatype=PointField.FLOAT32, count=1),
PointField(name='y', offset=4, datatype=PointField.FLOAT32, count=1),
PointField(name='z', offset=8, datatype=PointField.FLOAT32, count=1)
]
# Create PointCloud2 message
pc2_msg = PointCloud2()
pc2_msg.header = header
pc2_msg.height = 1
pc2_msg.width = len(points)
pc2_msg.fields = fields
pc2_msg.is_bigendian = False
pc2_msg.point_step = 12 # 3 * 4 bytes (float32)
pc2_msg.row_step = 12 * len(points)
pc2_msg.is_dense = True
# Pack points into binary data
data = []
for point in points:
data.append(struct.pack('fff', point[0], point[1], point[2]))
pc2_msg.data = b''.join(data)
return pc2_msg
def visualize_pointcloud(self, points):
"""Visualize point cloud in RViz using Isaac ROS optimized visualization"""
marker_array = MarkerArray()
# Create a marker for each point (for demonstration, sample every 100th point for performance)
for i in range(0, len(points), 100):
point = points[i]
marker = Marker()
marker.header.frame_id = 'velodyne'
marker.header.stamp = self.get_clock().now().to_msg()
marker.ns = 'pointcloud_points'
marker.id = i // 100
marker.type = Marker.SPHERE
marker.action = Marker.ADD
# Set position
marker.pose.position.x = point[0]
marker.pose.position.y = point[1]
marker.pose.position.z = point[2]
marker.pose.orientation.w = 1.0
# Set scale
marker.scale.x = 0.1
marker.scale.y = 0.1
marker.scale.z = 0.1
# Set color (blue for processed points)
marker.color.r = 0.0
marker.color.g = 0.0
marker.color.b = 1.0
marker.color.a = 0.8
marker_array.markers.append(marker)
self.vis_pub.publish(marker_array)
def main(args=None):
rclpy.init(args=args)
processor = IsaacPointCloudProcessorNode()
try:
rclpy.spin(processor)
except KeyboardInterrupt:
pass
finally:
processor.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
4. Isaac ROS Visual SLAM Pipeline
# Example: Isaac ROS Visual SLAM Node
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image, CameraInfo
from geometry_msgs.msg import PoseStamped, TransformStamped
from nav_msgs.msg import Odometry
from visualization_msgs.msg import MarkerArray
from cv_bridge import CvBridge
import numpy as np
import tf2_ros
class IsaacVisualSLAMNode(Node):
def __init__(self):
super().__init__('isaac_visual_slam')
# Create subscriptions for stereo images
self.left_image_sub = self.create_subscription(
Image,
'/camera/left/image_rect_color',
self.left_image_callback,
10
)
self.right_image_sub = self.create_subscription(
Image,
'/camera/right/image_rect_color',
self.right_image_callback,
10
)
# Create subscriptions for camera info
self.left_info_sub = self.create_subscription(
CameraInfo,
'/camera/left/camera_info',
self.left_info_callback,
10
)
self.right_info_sub = self.create_subscription(
CameraInfo,
'/camera/right/camera_info',
self.right_info_callback,
10
)
# Create publishers for pose and odometry
self.pose_pub = self.create_publisher(PoseStamped, '/visual_slam/pose', 10)
self.odom_pub = self.create_publisher(Odometry, '/visual_slam/odometry', 10)
self.map_pub = self.create_publisher(MarkerArray, '/visual_slam/map', 10)
# TF broadcaster
self.tf_broadcaster = tf2_ros.TransformBroadcaster(self)
# Initialize CV bridge
self.bridge = CvBridge()
# Isaac ROS Visual SLAM parameters
self.declare_parameters(
namespace='',
parameters=[
('enable_debug_mode', False),
('min_num_features', 1000),
('max_num_features', 2000),
('feature_match_threshold', 0.7),
('relocalization_threshold', 0.5)
]
)
# Initialize internal state
self.camera_matrix_left = None
self.camera_matrix_right = None
self.last_image_left = None
self.last_image_right = None
self.current_position = np.array([0.0, 0.0, 0.0])
self.current_orientation = np.array([0.0, 0.0, 0.0, 1.0]) # w, x, y, z
self.feature_points = [] # 3D map points
self.get_logger().info('Isaac Visual SLAM Node initialized')
def left_image_callback(self, msg):
"""Handle left camera image"""
self.last_image_left = msg
# Process stereo pair if both images are available
if self.last_image_right is not None:
self.process_stereo_pair()
def right_image_callback(self, msg):
"""Handle right camera image"""
self.last_image_right = msg
# Process stereo pair if both images are available
if self.last_image_left is not None:
self.process_stereo_pair()
def left_info_callback(self, msg):
"""Handle left camera info"""
self.camera_matrix_left = np.array(msg.k).reshape(3, 3)
def right_info_callback(self, msg):
"""Handle right camera info"""
self.camera_matrix_right = np.array(msg.k).reshape(3, 3)
def process_stereo_pair(self):
"""Process stereo image pair using Isaac ROS Visual SLAM"""
if self.camera_matrix_left is None or self.camera_matrix_right is None:
self.get_logger().warn('Camera matrices not available, skipping SLAM')
return
try:
# Convert ROS Images to OpenCV format
left_cv = self.bridge.imgmsg_to_cv2(self.last_image_left, desired_encoding='bgr8')
right_cv = self.bridge.imgmsg_to_cv2(self.last_image_right, desired_encoding='bgr8')
# Perform visual SLAM using Isaac ROS optimized pipeline
pose_update = self.perform_visual_slam(left_cv, right_cv)
# Update current pose
if pose_update is not None:
self.update_pose(pose_update)
# Publish results
self.publish_results()
# Clear processed images
self.last_image_left = None
self.last_image_right = None
except Exception as e:
self.get_logger().error(f'Error in stereo processing: {str(e)}')
def perform_visual_slam(self, left_image, right_image):
"""Perform visual SLAM using Isaac ROS optimized pipeline"""
# Get parameters
min_features = self.get_parameter('min_num_features').get_parameter_value().integer_value
max_features = self.get_parameter('max_num_features').get_parameter_value().integer_value
# In real Isaac ROS implementation, this would:
# - Extract features using CUDA-optimized detectors
# - Match features between stereo images
# - Perform 3D reconstruction
# - Optimize pose using bundle adjustment
# - Track features over time for localization
# For demonstration, we'll simulate pose estimation
# In practice, Isaac ROS provides optimized implementations using TensorRT and CUDA
# Simulate pose change (e.g., robot moving forward)
dt = 0.1 # 10Hz processing
linear_velocity = 0.1 # m/s
angular_velocity = 0.05 # rad/s
# Update position based on velocity (simplified)
dx = linear_velocity * dt
dtheta = angular_velocity * dt
# Update orientation quaternion
from math import sin, cos
dq = np.array([
cos(dtheta/2), # w
0.0, # x
0.0, # y
sin(dtheta/2) # z
])
# Update position
self.current_position[0] += dx * cos(self.current_orientation[3]) # w component for z-rotation
self.current_position[1] += dx * sin(self.current_orientation[3])
# Update orientation (simplified quaternion multiplication)
# In real implementation, Isaac ROS would use optimized rotation calculations
self.current_orientation[3] += dtheta/2 # Simplified update
# Normalize quaternion
norm = np.linalg.norm(self.current_orientation)
if norm > 0:
self.current_orientation /= norm
return {
'position': self.current_position.copy(),
'orientation': self.current_orientation.copy()
}
def update_pose(self, pose_update):
"""Update current pose based on SLAM results"""
self.current_position = pose_update['position']
self.current_orientation = pose_update['orientation']
def publish_results(self):
"""Publish SLAM results"""
# Create and publish pose
pose_msg = PoseStamped()
pose_msg.header.stamp = self.get_clock().now().to_msg()
pose_msg.header.frame_id = 'map'
pose_msg.pose.position.x = self.current_position[0]
pose_msg.pose.position.y = self.current_position[1]
pose_msg.pose.position.z = self.current_position[2]
pose_msg.pose.orientation.w = self.current_orientation[0]
pose_msg.pose.orientation.x = self.current_orientation[1]
pose_msg.pose.orientation.y = self.current_orientation[2]
pose_msg.pose.orientation.z = self.current_orientation[3]
self.pose_pub.publish(pose_msg)
# Create and publish odometry
odom_msg = Odometry()
odom_msg.header.stamp = self.get_clock().now().to_msg()
odom_msg.header.frame_id = 'map'
odom_msg.child_frame_id = 'base_link'
odom_msg.pose.pose = pose_msg.pose
# For simplicity, set velocity to zero
# In real implementation, Isaac ROS would estimate velocity
self.odom_pub.publish(odom_msg)
# Broadcast transform
t = TransformStamped()
t.header.stamp = self.get_clock().now().to_msg()
t.header.frame_id = 'map'
t.child_frame_id = 'base_link'
t.transform.translation.x = self.current_position[0]
t.transform.translation.y = self.current_position[1]
t.transform.translation.z = self.current_position[2]
t.transform.rotation.w = self.current_orientation[0]
t.transform.rotation.x = self.current_orientation[1]
t.transform.rotation.y = self.current_orientation[2]
t.transform.rotation.z = self.current_orientation[3]
self.tf_broadcaster.sendTransform(t)
# Publish map visualization
self.publish_map_visualization()
def publish_map_visualization(self):
"""Publish map visualization markers"""
marker_array = MarkerArray()
# Example: Publish current position as a marker
position_marker = Marker()
position_marker.header.frame_id = 'map'
position_marker.header.stamp = self.get_clock().now().to_msg()
position_marker.ns = 'slam_pose'
position_marker.id = 0
position_marker.type = Marker.SPHERE
position_marker.action = Marker.ADD
# Set position
position_marker.pose.position.x = self.current_position[0]
position_marker.pose.position.y = self.current_position[1]
position_marker.pose.position.z = self.current_position[2]
position_marker.pose.orientation.w = 1.0
# Set scale
position_marker.scale.x = 0.2
position_marker.scale.y = 0.2
position_marker.scale.z = 0.2
# Set color (green for current position)
position_marker.color.r = 0.0
position_marker.color.g = 1.0
position_marker.color.b = 0.0
position_marker.color.a = 1.0
marker_array.markers.append(position_marker)
# Example: Publish trajectory as a line (simplified)
trajectory_marker = Marker()
trajectory_marker.header.frame_id = 'map'
trajectory_marker.header.stamp = self.get_clock().now().to_msg()
trajectory_marker.ns = 'slam_trajectory'
trajectory_marker.id = 1
trajectory_marker.type = Marker.LINE_STRIP
trajectory_marker.action = Marker.ADD
# Set scale
trajectory_marker.scale.x = 0.05 # Line width
# Set color (blue for trajectory)
trajectory_marker.color.r = 0.0
trajectory_marker.color.g = 0.0
trajectory_marker.color.b = 1.0
trajectory_marker.color.a = 1.0
# Add current position to trajectory (in real implementation, store full trajectory)
point = Point()
point.x = self.current_position[0]
point.y = self.current_position[1]
point.z = self.current_position[2]
trajectory_marker.points.append(point)
marker_array.markers.append(trajectory_marker)
self.map_pub.publish(marker_array)
def main(args=None):
rclpy.init(args=args)
slam_node = IsaacVisualSLAMNode()
try:
rclpy.spin(slam_node)
except KeyboardInterrupt:
pass
finally:
slam_node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
5. Isaac ROS Deep Learning Pipeline
# Example: Isaac ROS Deep Learning Node
import rclpy
from rclpy.node import Node
from sensor_msgs.msg import Image
from std_msgs.msg import String
from cv_bridge import CvBridge
import numpy as np
import torch
import torch_tensorrt
class IsaacDeepLearningNode(Node):
def __init__(self):
super().__init__('isaac_deep_learning')
# Create subscription to camera image
self.image_sub = self.create_subscription(
Image,
'/image_raw',
self.image_callback,
10
)
# Create publisher for inference results
self.result_pub = self.create_publisher(String, '/dl_inference/result', 10)
# Initialize CV bridge
self.bridge = CvBridge()
# Isaac ROS Deep Learning parameters
self.declare_parameters(
namespace='',
parameters=[
('model_path', '/models/trt_model.plan'),
('input_width', 224),
('input_height', 224),
('mean', [0.485, 0.456, 0.406]),
('std', [0.229, 0.224, 0.225]),
('confidence_threshold', 0.5)
]
)
# Initialize Isaac ROS deep learning components
self.initialize_deep_learning()
self.get_logger().info('Isaac Deep Learning Node initialized')
def initialize_deep_learning(self):
"""Initialize Isaac ROS deep learning components"""
try:
# Get parameters
model_path = self.get_parameter('model_path').get_parameter_value().string_value
input_width = self.get_parameter('input_width').get_parameter_value().integer_value
input_height = self.get_parameter('input_height').get_parameter_value().integer_value
# Initialize CUDA context for Isaac ROS
if torch.cuda.is_available():
self.device = torch.device('cuda')
self.get_logger().info('CUDA available for Isaac ROS deep learning')
else:
self.device = torch.device('cpu')
self.get_logger().info('Using CPU for Isaac ROS deep learning')
# In real Isaac ROS implementation, this would load a TensorRT optimized model
# For demonstration, we'll create a placeholder
self.model_loaded = True
self.input_width = input_width
self.input_height = input_height
# Get normalization parameters
mean_param = self.get_parameter('mean').get_parameter_value().double_array_value
std_param = self.get_parameter('std').get_parameter_value().double_array_value
self.normalization_mean = np.array(mean_param)
self.normalization_std = np.array(std_param)
self.get_logger().info(f'Isaac Deep Learning model initialized from {model_path}')
except Exception as e:
self.get_logger().error(f'Error initializing deep learning: {str(e)}')
self.model_loaded = False
def image_callback(self, msg):
"""Process image with Isaac ROS deep learning pipeline"""
if not self.model_loaded:
self.get_logger().warn('Model not loaded, skipping inference')
return
try:
# Convert ROS Image to OpenCV format
cv_image = self.bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8')
# Preprocess image for Isaac ROS inference
processed_tensor = self.preprocess_image(cv_image)
# Perform inference using Isaac ROS optimized pipeline
result = self.perform_inference(processed_tensor)
# Publish results
result_msg = String()
result_msg.data = f"Isaac ROS inference result: {result}"
self.result_pub.publish(result_msg)
except Exception as e:
self.get_logger().error(f'Error in deep learning inference: {str(e)}')
def preprocess_image(self, image):
"""Preprocess image for Isaac ROS deep learning model"""
# Get parameters
input_width = self.get_parameter('input_width').get_parameter_value().integer_value
input_height = self.get_parameter('input_height').get_parameter_value().integer_value
# Resize image to model input size
resized = cv2.resize(image, (input_width, input_height))
# Convert BGR to RGB
rgb = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB)
# Normalize pixel values using Isaac ROS standard values
normalized = rgb.astype(np.float32) / 255.0
# Apply normalization (mean and std from parameters)
normalized = (normalized - self.normalization_mean) / self.normalization_std
# Transpose from HWC to CHW format
tensor = np.transpose(normalized, (2, 0, 1))
# Add batch dimension
tensor = np.expand_dims(tensor, axis=0)
# Convert to PyTorch tensor and move to device
torch_tensor = torch.from_numpy(tensor).to(self.device)
return torch_tensor
def perform_inference(self, tensor):
"""Perform inference using Isaac ROS optimized pipeline"""
# In real Isaac ROS implementation, this would run a TensorRT optimized model
# For demonstration, we'll return mock results
# Simulate inference delay to mimic real processing time
import time
time.sleep(0.01) # 10ms delay
# Mock inference results
results = {
'class_id': 1,
'class_name': 'object',
'confidence': 0.89,
'inference_time_ms': 8.5
}
self.get_logger().info(f'Isaac ROS inference completed: {results}')
return results
def main(args=None):
rclpy.init(args=args)
dl_node = IsaacDeepLearningNode()
try:
rclpy.spin(dl_node)
except KeyboardInterrupt:
pass
finally:
dl_node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
import cv2
main()
6. Isaac ROS Pipeline Launch File
# Example: Isaac ROS Pipeline Launch File
from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription, GroupAction
from launch.conditions import IfCondition
from launch.launch_description_sources import PythonLaunchDescriptionSource
from launch.substitutions import LaunchConfiguration, PathJoinSubstitution
from launch_ros.actions import Node, ComposableNodeContainer, LoadComposableNodes
from launch_ros.descriptions import ComposableNode
from ament_index_python.packages import get_package_share_directory
def generate_launch_description():
# Launch Arguments
use_sim_time = LaunchConfiguration('use_sim_time', default='True')
enable_image_proc = LaunchConfiguration('enable_image_proc', default='True')
enable_object_detection = LaunchConfiguration('enable_object_detection', default='True')
enable_pointcloud_proc = LaunchConfiguration('enable_pointcloud_proc', default='True')
enable_visual_slam = LaunchConfiguration('enable_visual_slam', default='True')
# Create the launch description and populate
ld = LaunchDescription()
# Declare launch arguments
ld.add_action(
DeclareLaunchArgument(
'use_sim_time',
default_value='True',
description='Use simulation (Isaac Sim) clock if true'
)
)
ld.add_action(
DeclareLaunchArgument(
'enable_image_proc',
default_value='True',
description='Enable Isaac ROS image processing pipeline'
)
)
ld.add_action(
DeclareLaunchArgument(
'enable_object_detection',
default_value='True',
description='Enable Isaac ROS object detection pipeline'
)
)
ld.add_action(
DeclareLaunchArgument(
'enable_pointcloud_proc',
default_value='True',
description='Enable Isaac ROS point cloud processing pipeline'
)
)
ld.add_action(
DeclareLaunchArgument(
'enable_visual_slam',
default_value='True',
description='Enable Isaac ROS visual SLAM pipeline'
)
)
# Isaac ROS Image Processing Container
image_proc_container = ComposableNodeContainer(
name='isaac_ros_image_proc_container',
namespace='isaac_ros',
package='rclcpp_components',
executable='component_container_mt',
parameters=[{'use_sim_time': use_sim_time}],
composable_node_descriptions=[
ComposableNode(
package='isaac_ros_image_proc',
plugin='nvidia::isaac_ros::image_proc::ResizeNode',
name='image_resize_node',
parameters=[{
'output_width': 640,
'output_height': 480,
}],
remappings=[
('image', '/camera/image_raw'),
('camera_info', '/camera/camera_info'),
('image_resize', '/camera/image_resized'),
],
condition=IfCondition(enable_image_proc)
),
ComposableNode(
package='isaac_ros_image_proc',
plugin='nvidia::isaac_ros::image_proc::RectifyNode',
name='image_rectify_node',
remappings=[
('image_raw', '/camera/image_resized'),
('camera_info', '/camera/camera_info'),
('image_rect', '/camera/image_rect'),
],
condition=IfCondition(enable_image_proc)
),
],
output='screen',
condition=IfCondition(enable_image_proc)
)
# Isaac ROS Object Detection Container
detection_container = ComposableNodeContainer(
name='isaac_ros_detection_container',
namespace='isaac_ros',
package='rclcpp_components',
executable='component_container_mt',
parameters=[{'use_sim_time': use_sim_time}],
composable_node_descriptions=[
ComposableNode(
package='isaac_ros_detectnet',
plugin='nvidia::isaac_ros::detectnet::DetectNetNode',
name='detectnet_node',
parameters=[{
'model_name': 'ssd_mobilenet_v2_coco',
'input_width': 512,
'input_height': 288,
'confidence_threshold': 0.7,
}],
remappings=[
('image_input', '/camera/image_rect'),
('tensor_input', 'tensor_sub'),
('detections', '/detections'),
],
condition=IfCondition(enable_object_detection)
),
],
output='screen',
condition=IfCondition(enable_object_detection)
)
# Isaac ROS Point Cloud Processing Container
pointcloud_container = ComposableNodeContainer(
name='isaac_ros_pointcloud_container',
namespace='isaac_ros',
package='rclcpp_components',
executable='component_container_mt',
parameters=[{'use_sim_time': use_sim_time}],
composable_node_descriptions=[
ComposableNode(
package='isaac_ros_pointcloud_utils',
plugin='nvidia::isaac_ros::pointcloud_utils::PointCloudXyzNode',
name='pointcloud_xyz_node',
remappings=[
('depth', '/depth/image_rect_raw'),
('camera_info', '/depth/camera_info'),
('pointcloud', '/points'),
],
condition=IfCondition(enable_pointcloud_proc)
),
],
output='screen',
condition=IfCondition(enable_pointcloud_proc)
)
# Isaac ROS Visual SLAM Container
slam_container = ComposableNodeContainer(
name='isaac_ros_slam_container',
namespace='isaac_ros',
package='rclcpp_components',
executable='component_container_mt',
parameters=[{'use_sim_time': use_sim_time}],
composable_node_descriptions=[
ComposableNode(
package='isaac_ros_visual_slam',
plugin='nvidia::isaac_ros::visual_slam::VisualSlamNode',
name='visual_slam_node',
parameters=[{
'use_sim_time': use_sim_time,
'enable_debug_mode': False,
'enable_localization_n_mapping': True,
'rectified_images': True,
}],
remappings=[
('left/camera_info', '/camera/left/camera_info'),
('left/image_rect_color', '/camera/left/image_rect_color'),
('right/camera_info', '/camera/right/camera_info'),
('right/image_rect_color', '/camera/right/image_rect_color'),
('visual_slam/pose_graph/poses', '/visual_slam/poses'),
('visual_slam/imu', '/imu/data'),
],
condition=IfCondition(enable_visual_slam)
),
],
output='screen',
condition=IfCondition(enable_visual_slam)
)
# Add containers to launch description
ld.add_action(image_proc_container)
ld.add_action(detection_container)
ld.add_action(pointcloud_container)
ld.add_action(slam_container)
return ld
These code examples demonstrate the key components of Isaac ROS pipelines, including:
- Image processing with optimized CUDA operations
- Object detection using TensorRT-optimized models
- Point cloud processing with GPU acceleration
- Visual SLAM with optimized tracking and mapping
- Deep learning inference with TensorRT
- A complete launch file for the pipeline
The examples follow Isaac ROS best practices and demonstrate how to leverage NVIDIA's GPU acceleration for robotics perception tasks.