Fundamentals

Requesting data

With the connection created in the previous example, you can now request data from the application. For example, the position of the end-effector of the robot can be requested and printed to the console every second. You can test it by adding the following code after the connection code in the previous example:

import time

while True:
    # get the parameter value
    position = req.getParameter('root/ManipulatorControl/manipulatorToolPoseActual').get()
    if position is not None:
        print(f"EE Position: {position.value}")
    else:
        print("Failed to get parameter value")
    # wait for 1 second before requesting again
    time.sleep(1)

Move the robot around and you should see the end-effector position updating in the console output.

By changing the parameter path in the req.getParameter() function you can request any parameter that is available in the parameter tree of the connected application.

You can also request multiple parameters at once by passing a list of parameter paths to the req.getParameterList() function. Here is an example of requesting both the end-effector position and the joint positions:

paths = ['root/ManipulatorControl/manipulatorToolPoseActual',
         'root/ManipulatorControl/jointPositionsActual']
while True:
    # get the parameter values
    results = req.getParameterList(paths).get()
    if results is not None:
        ee_position = results[0].value
        joint_positions = results[1].value
        print(f"EE Position: {ee_position}, Joint Positions: {joint_positions}")
    else:
        print("Failed to get parameter values")
    # wait for 1 second before requesting again
    time.sleep(1)

Individual parameter requests are performed in parallel and responses can arrive at different times. This means separate req.getParameter(...) calls may not represent a single synchronized snapshot and their order may not match the parameter-tree order. If you need synchronized values, request parameters together with req.getParameterList(paths) (returns a single snapshot), or use subscriptions with the desired timing.

If you require higher rates for specific signals, prefer using subscriptions for those high-rate parameters (or create separate subscriptions with different group/divider settings) and only request the slower parameters as needed.

Subscribing to data

Instead of requesting data in a loop, you can also subscribe to a parameter and receive updates whenever the value changes. This is more efficient for parameters that change frequently. You can subscribe to the same end-effector position parameter as follows:

# status codes live on the protobuf namespace, not on the motorcortex module
motorcortex_msg = motorcortex_types.motorcortex()

# define a callback function that is called when new data is received
def print_value(result):
    print(f"Received new positions: {result[0].value}")

# Subscribe to parameter changes
sub_position = sub.subscribe('root/ManipulatorControl/manipulatorToolPoseActual', "Group1", frq_divider=100)

# get reply from the server
is_subscribed = sub_position.get()
# print subscription status and layout
if (is_subscribed is not None) and (is_subscribed.status == motorcortex_msg.OK):
    print(f"Subscription successful, layout: {sub_position.layout()}")
else:
    print(f"Subscription failed")
    sub.close()
    exit()

# set the callback function that handles the received data
# Note that this is a non-blocking call, starting a new thread that handles
# the messages. You should keep the application alive for a s long as you need to
# receive the messages
sub_position.notify(print_value)

time.sleep(60)  # keep the application running for 60 seconds to receive updates
sub.close()  # close the subscription when done

With this code, the print_value function is called whenever the value of the end-effector position gets updated through the subscription. This does not mean the value has changed necessarily, but that a new value is available. You can move the robot around to see the updates in the console output.

You can group multiple parameters into a single subscription. This ensures that the parameters are synchronized when received. Here is an example of subscribing to both the end-effector position and the joint positions in a single subscription:

paths = ['root/ManipulatorControl/manipulatorToolPoseActual',
         'root/ManipulatorControl/jointPositionsActual']
sub_multi = sub.subscribe(paths, group_alias="Positions", frq_divider=1000)

is_subscribed = sub_multi.get()
# print subscription status and layout
if (is_subscribed is not None) and (is_subscribed.status == motorcortex_msg.OK):
    print(f"Subscription successful, layout: {sub_multi.layout()}")
else:
    print(f"Subscription failed")
    sub.close()
    exit()

# set the callback function that handles the received data
def print_multi_value(result):
    print(f"Received new EE position: {result[0].value}, \nJoint positions: {result[1].value}\n\n")

sub_multi.notify(print_multi_value)

time.sleep(60)  # keep the application running for 60 seconds to receive updates
sub.close()  # close the subscription when done

The client can create multiple subscriptions for different groups, for instance at different data-rates (by setting the “divider” parameter differently for different groups).

The group name is shared between clients and a group is published by the server once to all clients that have subscription to the group with that name. This way the communication can easily scale across large numbers of clients.

If a group name already exists on the server and another client requests a subscription to a set of signals with the same group name, the server adds any requested signals that are not already in the group to the group. The other clients using the same group name then also receive the additional signals.

Setting parameter values

Now that you can listen to data from the Motorcortex application, you can send data to it. This is how a client application interacts with the Motorcortex application to control robots or machines.

Start by engaging the robot:

reply = req.setParameter("root/Logic/stateCommand", 2).get()
print(f"Engage: {reply}")

This code sends a command to the Logic module of the Motorcortex application to bring the robot to its engaged state. The value 2 is StateEvents.GOTO_ENGAGED_E of the stateCommand parameter; the full list of state, mode and interpreter commands is in robot_control.system_defs.

Using the same technique you can also open and close the gripper of the robot:

import time 
for i in range(5):
    reply = req.setParameter("root/UserParameters/IO/gripper", 0).get()
    time.sleep(1)
    reply = req.setParameter("root/UserParameters/IO/gripper", 1).get()
    time.sleep(1)
print("Finished gripper open/close")

This code opens and closes the gripper 5 times with a delay of 1 second in between.

Setting array values

Instead of setting single scalar values, you can also set array values. For example, to set the joint velocities of the robot, you can use the following code:

reply = req.setParameter("root/Logic/modeCommand", 3).get()
print(f"Set to manual joint mode: {reply}")
time.sleep(2)
reply = req.setParameter("root/ManipulatorControl/hostInJointVelocity", [0.1, 0.3, 0.0, 0.0, 0.0, 0.3]).get()
print(f"Set joint velocity: {reply}")

The robot switches to manual joint mode and starts moving the joints with the specified velocities. However, it only briefly moves. This is because the command is sent to root/ManipulatorControl/hostInJointVelocity which also has a watchdog running in the Motorcortex Application. If no new command is received within a certain time, the command is reset to zero for safety reasons.

The entire array does not have to be set all at once. You can also set individual elements of the array. For example, to set only the third and fourth joint velocity, you can use:

reply = req.setParameter("root/Logic/modeCommand", 3).get()
print(f"Set to manual joint mode: {reply}")
time.sleep(2)
reply = req.setParameter("root/ManipulatorControl/hostInJointVelocity", [0.1, 0.3], offset=2, length=2).get()
print(f"Set joint velocity: {reply}")

Overwriting parameters

During debugging and testing, you might want to overwrite certain parameters in the Motorcortex Application. This can be done using the overwriteParameter function. For example, to overwrite the root/ManipulatorControl/hostInJointVelocityGain parameter. You can then test out a new gain value without changing the configuration in the Motorcortex Application.

reply = req.overwriteParameter("root/ManipulatorControl/hostInJointVelocityGain", 0.0, force_activate=True).get()
print(f"Set joint position: {reply}")
time.sleep(5)
reply = req.releaseParameter("root/ManipulatorControl/hostInJointVelocityGain").get()

In Motorcortex Desk the parameter is highlighted in yellow, indicating that it is being overwritten by a client application. After 5 seconds, the parameter is released and goes back to its original value.