Python API

The python API is designed to take away some of the boilerplate around ROS. There are two versions of the API: mc_one and mc_one_async. mc_one is direct/blocking, and mc_one_async is designed with asyncio semantics.

mc_one_async is still in development.

General Structure

The general structure of a node tends to be:

direct/blocking:

from mc_one import Node

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

def main(args=None):
    rclpy.init(args=args)
    run_rclpy(MyNode(),args)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

async:

from mc_one_async import Node

@asyncinit
class MyNode(Node):
    async def __init__(self):
        super().__init__('my_node',namespace='/my_package')
        self.logger = self.get_logger()
        self.parameters = self.use_parameters()
        # ...more initialization

async def async_main(args=None):
    node = await MyNode()
    await asyncio.Event().wait() # just do nothing here for now
    node.destroy_node()

def main(args=None):
    rclpy.init(args=args)
    run_async_rclpy(async_main,args)
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Node

Node is a subclass of rclpy.Node with some added methods.

Parameters

Call use_parameters to access the parameters API (usually in __init__):

self.parameters = self.use_parameters()

Declaring Parameters

Parameters should be declared first. Each different type/style of parameter has a declare method.

Boolean

value = self.parameters.declare_bool(
    'name', # name of the parameter
    True, # default value
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Integer

value = self.parameters.declare_int(
    'name', # name of the parameter
    0, # default value
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Ranged Integer

value = self.parameters.declare_ranged_int(
    'name', # name of the parameter
    0, # default value
    0, # from-value
    10, # to-value
    2, # step
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Options Integer

value = self.parameters.declare_options_int(
    'name', # name of the parameter
    0, # default value
    [0,1,2], # options to choose from
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Double

value = self.parameters.declare_double(
    'name', # name of the parameter
    0.0, # default value
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Ranged Double

value = self.parameters.declare_ranged_double(
    'name', # name of the parameter
    0.0, # default value
    0.0, # from-value
    10.0, # to-value
    2.0, # step
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Options Double

value = self.parameters.declare_options_double(
    'name', # name of the parameter
    0.0, # default value
    [0.0,1.0,2.0], # options to choose from
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

String

value = self.parameters.declare_string(
    'name', # name of the parameter
    'hello', # default value
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    lines=1, # (optional) number of lines for the string
    private=True, # (optional) private indication
)

Options String

value = self.parameters.declare_options_string(
    'name', # name of the parameter
    'hello', # default value
    ['hello','world'], # options to choose from
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    lines=1, # (optional) number of lines for the string
    private=True, # (optional) private indication
)

Boolean Array

value = self.parameters.declare_bool_array(
    'name', # name of the parameter
    [True,False,True,False], # default values
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Integer Array

value = self.parameters.declare_int_array(
    'name', # name of the parameter
    [0,0,0,0], # default values
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Ranged Integer Array

value = self.parameters.declare_ranged_int_array(
    'name', # name of the parameter
    [0,0,0,0], # default values
    0, # from-value
    10, # to-value
    2, # step
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Options Integer Array

value = self.parameters.declare_options_int_array(
    'name', # name of the parameter
    [0,0,0,0], # default values
    [0,1,2], # options to choose from
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Double Array

value = self.parameters.declare_double_array(
    'name', # name of the parameter
    [0.0,0.0,0.0,0.0], # default values
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Ranged Double Array

value = self.parameters.declare_ranged_double(
    'name', # name of the parameter
    [0.0,0.0,0.0,0.0], # default values
    0.0, # from-value
    10.0, # to-value
    2.0, # step
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

Options Double Array

value = self.parameters.declare_options_double(
    'name', # name of the parameter
    [0.0,0.0,0.0,0.0], # default values
    [0.0,1.0,2.0], # options to choose from
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    private=True, # (optional) private indication
)

String Array

value = self.parameters.declare_string_array(
    'name', # name of the parameter
    ['hello','hello','hello','hello'], # default values
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    lines=1, # (optional) number of lines for the strings
    private=True, # (optional) private indication
)

Options String Array

value = self.parameters.declare_options_string_array(
    'name', # name of the parameter
    ['hello','hello','hello','hello'], # default value
    ['hello','world'], # options to choose from
    description='description', # (optional) description
    read_only=True, # (optional) read-only indication
    lines=1, # (optional) number of lines for the strings
    private=True, # (optional) private indication
)

Foreign Parameters

value = self.parameters.declare_foreign(
    'name', # name of the parameter
    'other_node', # name of the other node
    namespace='namespace', # (optional) namespace of the other node
)

Getting and Setting Parameters

To get or set the current value of a parameter, use get and set, or get_foreign and set_foreign for foreign parameters.

value = self.parameters.get('name')

self.parameters.set('name',value)

value = self.parameters.get_foreign('name','other_node',namespace='/other_package')

self.parameters.set_foreign('name',value,'other_node',namespace='/other_package')

Handling Parameter Updates

To handle changes in the declared parameters (including foreign parameters), register an update callback:

self.parameters.on_update(callback)

The callback takes the parameter name and the new value as parameter:

def callback(name,value):
    pass

Subscribing

Under Construction

For now use the ROS API like so:

self.create_subscription(
    message_type, # the message type (i.e. std_msgs.msg.String)
    topic, # the topic name
    callback, # the message callback
    10
)

The callback takes a raw ROS message:

def callback(message):
    pass

Publishing

Under Construction

For now use the ROS API like so:

publisher = self.create_publisher(
    message_type, # the message type (i.e. std_msgs.msg.String)
    topic, # the topic name
    10
)

And then to publish a message:

publisher.publish(message)

Services

Under Construction

Actions

Under Construction

Audio In

Call use_audio_in to access the audio in API (usually in __init__):

self.audio_in = self.use_audio_in(callback)

Audio samples will now stream in through the callback:

def callback(audio):
    # ...process audio.data
    pass

The message type is mc_core_interfaces.msg.AudioStamped:

std_msgs/Header header
int16[] data

Audio Out

Call use_audio_out to access the audio out API (usually in __init__):

self.audio_out = self.use_audio_out(status_callback)

The sample rate of the audio output is currently fixed, and can be retrieved by:

rate = self.audio_out.sample_rate()

Now,to play an audio sample:

self.audio_out.play(
    id, # unique ID to identify this sample
    data, # sample data (either raw bytes, or int16 np.array, or float np.array)
)

The audio samples get queued and played one by one. Once a sample is complete, the status callback is called:

def status_callback(status,id,offset):
    # status = 0: not played, 1: partially played until offset, 2: fully played
    # id = sample ID as passed to play() method
    # offset = (when status = 1) number of frames played (between 0 and len(data))
    pass

To cancel all enqueued audio right now:

self.audio_out.stop_all()

This will stop the currently playing sample and flush the queue, reporting back the status of each sample.

Video In

Call use_video_in to access the video in API (usually in __init__):

self.video_in = self.use_video_in(callback)

Video frames will now stream in through the callback. The message type is sensor_msgs.msg.CompressedImage. To use the image with OpenCV, decode the message like so:

def callback(message):
    buffer = np.frombuffer(message.data,np.uint8)
    image = cv2.imdecode(buffer,cv2.IMREAD_COLOR)
    pass

Monitor

Under Construction