Best Practices

Guidelines and recommendations for developing robust applications with xTerra robots.

Code Organization

Package Structure

my_robot_app/
├── config/
│   ├── params.yaml
│   └── robot_config.yaml
├── launch/
│   └── app.launch.py
├── src/
│   └── my_robot_app/
│       ├── __init__.py
│       ├── controllers/
│       ├── perception/
│       └── navigation/
├── package.xml
└── setup.py

Naming Conventions

  • Nodes: Use descriptive names with underscores (e.g., obstacle_detector)
  • Topics: Use hierarchical namespaces (e.g., /robot/sensors/camera/front)
  • Parameters: Use dot notation (e.g., controller.max_speed)
  • Services: Use verb-noun format (e.g., set_mode, get_status)

Error Handling

import rclpy
from rclpy.exceptions import ROSInterruptException

class RobustController:
    def __init__(self):
        self.node = rclpy.create_node('robust_controller')
        self.setup_publishers()
        self.setup_subscribers()

    def run(self):
        try:
            rclpy.spin(self.node)
        except ROSInterruptException:
            self.node.get_logger().info('Interrupted')
        except Exception as e:
            self.node.get_logger().error(f'Error: {e}')
        finally:
            self.cleanup()

    def cleanup(self):
        """Ensure proper cleanup on shutdown"""
        # Stop the robot
        self.send_zero_velocity()
        # Destroy node
        self.node.destroy_node()
        rclpy.shutdown()

Performance Optimization

Message Publishing

  • Use appropriate QoS settings for each topic
  • Avoid publishing at rates higher than necessary
  • Use transient local for infrequent updates
  • Implement throttling for sensor data when needed
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy

# For sensor data - best effort, keep last
sensor_qos = QoSProfile(
    reliability=ReliabilityPolicy.BEST_EFFORT,
    history=HistoryPolicy.KEEP_LAST,
    depth=1
)

# For commands - reliable, keep last 10
command_qos = QoSProfile(
    reliability=ReliabilityPolicy.RELIABLE,
    history=HistoryPolicy.KEEP_LAST,
    depth=10
)

Safety Practices

Always Implement Safety Checks

class SafeController:
    def __init__(self):
        self.max_speed = 1.0  # m/s
        self.max_angular = 1.5  # rad/s
        self.timeout = 0.5  # seconds
        self.last_command_time = time.time()

    def send_velocity(self, linear, angular):
        # Limit velocities
        linear = max(min(linear, self.max_speed), -self.max_speed)
        angular = max(min(angular, self.max_angular), -self.max_angular)

        # Check timeout
        if time.time() - self.last_command_time > self.timeout:
            self.emergency_stop()
            return

        # Send command
        msg = Twist()
        msg.linear.x = linear
        msg.angular.z = angular
        self.vel_pub.publish(msg)
        self.last_command_time = time.time()

Watchdog Timer

# Implement a watchdog to stop robot if no commands received
self.watchdog_timer = self.node.create_timer(
    0.1,  # 100ms
    self.watchdog_callback
)

def watchdog_callback(self):
    elapsed = time.time() - self.last_command_time
    if elapsed > self.COMMAND_TIMEOUT:
        self.send_zero_velocity()
        self.node.get_logger().warn('Command timeout - stopping robot')

Testing

Unit Testing

import unittest
from my_robot_app.controllers import VelocityController

class TestVelocityController(unittest.TestCase):
    def setUp(self):
        rclpy.init()
        self.controller = VelocityController()

    def tearDown(self):
        self.controller.destroy_node()
        rclpy.shutdown()

    def test_velocity_limits(self):
        # Test that velocities are properly limited
        vel = self.controller.limit_velocity(10.0)
        self.assertLessEqual(vel, self.controller.max_speed)

Integration Testing

# Use launch testing for integration tests
import launch_testing
import launch_testing.actions

def generate_test_description():
    return launch.LaunchDescription([
        launch.actions.Node(
            package='my_robot_app',
            executable='controller',
            name='test_controller'
        ),
        launch_testing.actions.ReadyToTest()
    ])

Logging

# Use appropriate log levels
self.node.get_logger().debug('Detailed debug information')
self.node.get_logger().info('Normal operation message')
self.node.get_logger().warn('Warning message')
self.node.get_logger().error('Error occurred')
self.node.get_logger().fatal('Critical failure')

# Include context in logs
self.node.get_logger().info(
    f'Moving to waypoint: x={x:.2f}, y={y:.2f}'
)

Documentation

  • Document all public APIs with docstrings
  • Include usage examples in README
  • Document parameter configurations
  • Maintain a changelog
  • Create diagrams for complex systems