Hands-on Demo

This demo shows how to create a new package and work with the MC-ONE stack.

Prepare

Make sure ROS2 is installed.

Clone mc_core and mc_monitor to your machine. For each repo, run build.sh to build the corresponding docker images.

Each repo has scripts to start the individual nodes or tools.

From mc_core:

• run_core.sh: run the core node
• run_audio_play.sh: play audio sample through the robot speakers
• run_audio_record.sh: record 5 seconds of audio from the robot microphone
• run_video_view.sh: view the camera feed from the robot's head camera
• run_feetech_serial_scan.sh: scan all connected Fee-URT1 bus converters for Feetech servos
• run_feetech_serial_setbaud.sh: with a single servo connected, set the baud rate for that servo
• run_feetech_serial_setid.sh: with a single servo connected, set the ID for that servo
• launch_rviz_current.sh: run RViz2 to visualize the current physical state of the robot as read from the servos (does not work without a physical robot connected)
• launch_rviz_goal.sh: run RViz2 to visualize the goal state of the robot (also works without a physical robot connected)

From mc_monitor:

• serve.sh: serve the monitor HTTP server at localhost:8000

As explained in the ROS2 documentation, create a workspace directory:

mkdir ~/ros2_ws

And add subdirectories for the logs and the packages:

mkdir ~/ros2_ws/logs
mkdir ~/ros2_ws/src

And add the following lines to your .bashrc:

source /opt/ros/jazzy/setup.sh
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp
export USER_CONTEXT_PATH=$HOME/ros2_ws

Now, to run the robot stack, open a new terminal, go to mc_core and run run_core.sh. Open another terminal, go to mc_monitor and run serve.sh.

It is possible to keep everything running in the background while working on your node.

Create Package and Node

Create a new ROS2 package with test node:

ros2 pkg create --build-type ament_python --node-name my_node my_package

Build Package

Make sure you're in the workspace directory, and run colcon build:

cd ~/ros2_ws
colcon build

And run the local setup script:

source install/setup.sh

Run the Node

To run your node, simply access it with ros2, like so:

ros2 run my_package my_node

Logging

Open the monitor (open a browser and go to localhost:8000). On the log page you should see the current log messages pass by.

Let's look at the code for the node. Remove all the code and replace it with:

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

This is a very basic node. After running colcon build, and ros2 run my_package my_node, the node should output a message to the log page of the monitor.

You can stop the node with CTRL-C.

Parameters

Let's add a simple parameter to this node.

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor,ParameterType,IntegerRange,FloatingPointRange

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()

        # declare the parameter
        self.declare_parameter('my_parameter',False,ParameterDescriptor(
            type=ParameterType.PARAMETER_BOOL,
            description='My new parameter.'
        ))

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Now, again, colcon build the node, and ros2 run it. You can see the parameter appear in the Monitor on the Parameters page. It's a boolean value that can be either true or false (checked or unchecked).

Adding a Handler

Now let's add a parameter handler to respond if a user changes the parameter:

#!/usr/bin/env python3
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor,ParameterType,IntegerRange,FloatingPointRange
from rclpy.parameter_event_handler import ParameterEventHandler

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()
        self.logger.info('Hello, World!')

        # declare the parameter
        self.declare_parameter('my_parameter',False,ParameterDescriptor(
            type=ParameterType.PARAMETER_BOOL,
            description='My new parameter.'
        ))

        # keep a local copy
        self.my_parameter = self.get_parameter('my_parameter').value

        # start the parameter handler and callback
        self.parameter_event_handler = ParameterEventHandler(self)
        self.parameter_event_handler.add_parameter_callback('my_parameter',self.get_name(),self.update_parameters)

    def update_parameters(self,event):

        # update the local copy
        self.my_parameter = self.get_parameter('my_parameter').value

        self.logger.info(f'new parameter: {self.my_parameter}')

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Again colcon build it. When you ros2 run the node now, messages should show up when switching the parameter on and off.

More Parameters

Let's add two more parameters. One will be an integer value and one will be a language specifier.

#!/usr/bin/env python3
import time
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor,ParameterType,IntegerRange,FloatingPointRange
from rclpy.parameter_event_handler import ParameterEventHandler

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()

        # declare the parameter
        self.declare_parameter('my_parameter',False,ParameterDescriptor(
            type=ParameterType.PARAMETER_BOOL,
            description='My new parameter.'
        ))

        # keep a local copy
        self.my_parameter = self.get_parameter('my_parameter').value

        # declare the parameter
        self.declare_parameter('my_integer',0,ParameterDescriptor(
            type=ParameterType.PARAMETER_INTEGER,
            description='My integer value.',
            integer_range=[IntegerRange(from_value=0,to_value=10,step=2)]
        ))

        # keep a local copy
        self.my_integer = self.get_parameter('my_integer').value

        # declare the parameter
        self.declare_parameter('my_language','en_US',ParameterDescriptor(
            type=ParameterType.PARAMETER_STRING,
            description='My language specification.',
        ))

        # keep a local copy
        self.my_language = self.get_parameter('my_language').value

        # start the parameter handler and callbacks
        self.parameter_event_handler = ParameterEventHandler(self)
        self.parameter_event_handler.add_parameter_callback('my_parameter',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_integer',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_language',self.get_name(),self.update_parameters)

    def update_parameters(self,event):

        # update the local copies
        self.my_parameter = self.get_parameter('my_parameter').value
        self.my_integer = self.get_parameter('my_integer').value
        self.my_language = self.get_parameter('my_language').value

        self.logger.info(f'new parameters: {self.my_parameter},{self.my_integer},{self.my_language}')

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

colcon build and ros2 run, and now you can see a slider and an input field for the two new parameters appear in the Monitor.

String Options

But what should the user type into that field? Maybe it's better to give the user a few options.

#!/usr/bin/env python3
import time
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor,ParameterType,IntegerRange,FloatingPointRange
from rclpy.parameter_event_handler import ParameterEventHandler

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()

        # declare the parameter
        self.declare_parameter('my_parameter',False,ParameterDescriptor(
            type=ParameterType.PARAMETER_BOOL,
            description='My new parameter.'
        ))

        # keep a local copy
        self.my_parameter = self.get_parameter('my_parameter').value

        # declare the parameter
        self.declare_parameter('my_integer',0,ParameterDescriptor(
            type=ParameterType.PARAMETER_INTEGER,
            description='My integer value.',
            integer_range=[IntegerRange(from_value=0,to_value=10,step=2)]
        ))

        # keep a local copy
        self.my_integer = self.get_parameter('my_integer').value

        # declare the parameter
        self.declare_parameter('my_language','en_US',ParameterDescriptor(
            type=ParameterType.PARAMETER_STRING,
            description='My language specification.',
            additional_constraints='{"options":["en_US","zh_CN","ko_KR","es_ES"]}',
        ))

        # keep a local copy
        self.my_language = self.get_parameter('my_language').value

        # start the parameter handler and callbacks
        self.parameter_event_handler = ParameterEventHandler(self)
        self.parameter_event_handler.add_parameter_callback('my_parameter',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_integer',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_language',self.get_name(),self.update_parameters)

    def update_parameters(self,event):

        # update the local copies
        self.my_parameter = self.get_parameter('my_parameter').value
        self.my_integer = self.get_parameter('my_integer').value
        self.my_language = self.get_parameter('my_language').value

        self.logger.info(f'new parameters: {self.my_parameter},{self.my_integer},{self.my_language}')

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Notice the additional_constraints in the declaration. This is extra information for things like options and a few other special settings.

Starting and Stopping Services

Now, changing parameters means reconfiguring or restarting other services in the node, like cloud APIs, local models, etc. The easiest way to do this, is to make a start() and a stop() function. The start() function configures and starts the services, and stop() stops and clears the services. This looks like:

#!/usr/bin/env python3
import time
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor,ParameterType,IntegerRange,FloatingPointRange
from rclpy.parameter_event_handler import ParameterEventHandler

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()

        # declare the parameter
        self.declare_parameter('my_parameter',False,ParameterDescriptor(
            type=ParameterType.PARAMETER_BOOL,
            description='My new parameter.'
        ))

        # keep a local copy
        self.my_parameter = self.get_parameter('my_parameter').value

        # declare the parameter
        self.declare_parameter('my_integer',0,ParameterDescriptor(
            type=ParameterType.PARAMETER_INTEGER,
            description='My integer value.',
            integer_range=[IntegerRange(from_value=0,to_value=10,step=2)]
        ))

        # keep a local copy
        self.my_integer = self.get_parameter('my_integer').value

        # declare the parameter
        self.declare_parameter('my_language','en_US',ParameterDescriptor(
            type=ParameterType.PARAMETER_STRING,
            description='My language specification.',
            additional_constraints='{"options":["en_US","zh_CN","ko_KR","es_ES"]}',
        ))

        # keep a local copy
        self.my_language = self.get_parameter('my_language').value

        # start the services
        self.start()

        # start the parameter handler and callbacks
        self.parameter_event_handler = ParameterEventHandler(self)
        self.parameter_event_handler.add_parameter_callback('my_parameter',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_integer',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_language',self.get_name(),self.update_parameters)

    def update_parameters(self,event):

        # stop the services
        self.stop()

        # update the local copies
        self.my_parameter = self.get_parameter('my_parameter').value
        self.my_integer = self.get_parameter('my_integer').value
        self.my_language = self.get_parameter('my_language').value

        # start the services
        self.start()

    def start(self):
        self.logger.info(f'Starting services with {self.my_parameter},{self.my_integer},{self.my_language}')

    def stop(self):
        self.logger.info('Stopping services')

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Responding to Foreign Parameters

Sometimes, a node needs to react to parameters from other nodes. For instance, the node might need to respond to the user switching debug mode in the Monitor. To do this, add another parameter callback to the list:

#!/usr/bin/env python3
import time
import rclpy
from rclpy.node import Node
from rcl_interfaces.msg import ParameterDescriptor,ParameterType,IntegerRange,FloatingPointRange
from rclpy.parameter_event_handler import ParameterEventHandler

class MyNode(Node):
    def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()

        # declare the parameter
        self.declare_parameter('my_parameter',False,ParameterDescriptor(
            type=ParameterType.PARAMETER_BOOL,
            description='My new parameter.'
        ))

        # keep a local copy
        self.my_parameter = self.get_parameter('my_parameter').value

        # declare the parameter
        self.declare_parameter('my_integer',0,ParameterDescriptor(
            type=ParameterType.PARAMETER_INTEGER,
            description='My integer value.',
            integer_range=[IntegerRange(from_value=0,to_value=10,step=2)]
        ))

        # keep a local copy
        self.my_integer = self.get_parameter('my_integer').value

        # declare the parameter
        self.declare_parameter('my_language','en_US',ParameterDescriptor(
            type=ParameterType.PARAMETER_STRING,
            description='My language specification.',
            additional_constraints='{"options":["en_US","zh_CN","ko_KR","es_ES"]}',
        ))

        # keep a local copy
        self.my_language = self.get_parameter('my_language').value

        # see further...
        self.debug = True

        self.start()

        self.parameter_event_handler = ParameterEventHandler(self)
        self.parameter_event_handler.add_parameter_callback('my_parameter',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_integer',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('my_language',self.get_name(),self.update_parameters)
        self.parameter_event_handler.add_parameter_callback('debug','/mc_core/core',self.update_debug_parameter)

    def update_parameters(self,event):
        self.stop()
        self.my_parameter = self.get_parameter('my_parameter').value
        self.my_integer = self.get_parameter('my_integer').value
        self.my_language = self.get_parameter('my_language').value
        self.start()

    def update_debug_parameter(self,event):
        self.stop()
        self.debug = event.value.bool_value
        self.start()

    def start(self):
        self.logger.info(f'Starting services with {self.my_parameter},{self.my_integer},{self.my_language}. Debug: {self.debug}')

    def stop(self):
        self.logger.info('Stopping services')

def main(args=None):
    rclpy.init(args=args)
    node = MyNode()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

At this moment, the self.debug initial value is hardcoded to False. This is not strictly correct, as the debug parameter might start out as True. In order to fix this, you'd have to actually read the parameter from the core, and not only respond to the update messages. This involves a bit of boilerplate code, which should be hidden away in a small API. For now we'll leave it at this, but be aware of this issue.