This is the top-level namespace of the rocon_python_comms ROS package. It provides additional python modules beyond rospy for working with ros python communications.
This module defines exceptions raised by the rocon_python_comms package. These exception names are all included in the main rocon_python_comms namespace. To catch one, import it this way:
from rocon_python_comms import ServicePairException
This module contains anything relating to introspection or manipulation of ros nodes.
Do a lookup to find a node with the given name. The given name is treated as the unresolved node name. Hence this lookup will find nodes with the same name, but different namespaces.
This will raise exceptions, if the node couldn’t be found or in case unique is set multiple nodes with the same name are found.
| Parameters: | wanted_node_name (str) – unresolved name of the node looked for (e.g. ‘gateway’, not ‘/concert/gateway’) |
|---|---|
| Returns: | the fully resolved name of the node (unique) or list of fully resolved names (non-unique) |
| Return type: | str |
| Raises: | NotFoundException |
| Todo: | accept resolved names -> https://github.com/robotics-in-concert/rocon_tools/issues/30 |
This module contains the client side api for communicating across a rocon service pair. A facade pattern is used here to simplify the interaction with the client side publisher and subscriber.
Bases: object
The client side of a pubsub service pair.This class provides a simplified api for handling requests/responses on the pubsub pair (client side). Although the development of this class was for non-blocking request-response behaviour, it does support legacy style blocking behavious as well.
Usage (Non-Blocking):
#!/usr/bin/env python
import rospy
from chatter.msg import ChatterRequest, ChatterResponse, ChatterPair
from rocon_python_comms import ServicePairClient
import unique_id
class ChatterClient(object):
def __init__(self):
self.response = None
self.client = ServicePairClient('chatter', ChatterPair)
self.client.wait_for_service(rospy.Duration(3.0)) # should catch some exceptions here in case of timeouts
self.request_id = self.client(ChatterRequest("Hello dude"), timeout=rospy.Duration(3.0), callback=self.callback)
rospy.loginfo("Request id %s" % unique_id.toHexString(self.request_id))
if self.response is not None:
print("Response %s" % self.response)
def callback(self, request_id, msg):
# ideally you'd check request_id against self.request_id
self.response = msg
rospy.loginfo("Got the response: %s" % msg)
if __name__ == '__main__':
rospy.init_node('chatter_client', anonymous=True)
chatter_client = ChatterClient()
rospy.spin()
Usage (Blocking):
#!/usr/bin/env python
import rospy
from chatter.msg import ChatterRequest, ChatterResponse, ChatterPair
from rocon_python_comms import ServicePairClient
import unique_id
class ChatterClient(object):
def __init__(self):
self.response = None
self.client = ServicePairClient('chatter', ChatterPair)
self.client.wait_for_service(rospy.Duration(3.0)) # should catch some exceptions here in case of timeouts
self.response = self.client(ChatterRequest("Hello dude"), timeout=rospy.Duration(3.0))
if self.response is not None:
print("Response %s" % self.response)
if __name__ == '__main__':
rospy.init_node('chatter_client', anonymous=True)
chatter_client = ChatterClient()
rospy.spin()
Request msg type for this pair <ServicePairSpec>Request.
Response msg type for this pair <ServicePairSpec>Response.
Base message type for this pair.
Initiates and executes the client request to the server. The type of arguments supplied determines whether to apply blocking or non-blocking behaviour.
If no callback is supplied, the mode is blocking, otherwise non-blocking. If no timeout is specified, then a return of None indicates that the operation timed out.
| Parameters: |
|
|---|---|
| Returns: | msg/id for blocking calls it is the response message, for non-blocking it is the unique id |
| Return type: | <name>Response/uuid_msgs.UniqueID or None (if timed out) |
This module contains the server side api for communicating across a rocon service pair. A facade pattern is used here to simplify the interaction with the server side publisher and subscriber.
Bases: object
The server side of a pubsub service pair. This class provides a simplified api for handling requests/responses on the pubsub pair. There are two modes of operation - 1) blocking and 2) threaded.
Non-Threaded
In the first, the users’ callback function directly runs whenever an incoming request is received. In this case, your callbacks should be very minimal so that incoming requests don’t get blocked and queued up.
#!/usr/bin/env python
import rospy
from chatter.msg import ChatterRequest, ChatterResponse, ChatterPair
from rocon_python_comms import ServicePairServer
class ChatterServer(object):
def __init__(self):
self.server = ServicePairServer('chatter', self.callback, ChatterPair)
def callback(self, request_id, msg):
rospy.loginfo("Server : I heard %s" % msg.babble)
response = ChatterResponse()
response.reply = "I heard %s" % msg.babble
self.server.reply(request_id, response)
if __name__ == '__main__':
rospy.init_node('chatter_server', anonymous=True)
chatter_server = ChatterServer()
rospy.spin()
Threaded
In the second, we spawn a background thread and shunt the callback into this thread. Just toggle the use_threads flag when constructing the server:
self.server = ServicePairServer('chatter', self.callback, ChatterPair, use_threads=True)
Request msg type for this pair <ServicePairSpec>Request.
Response msg type for this pair <ServicePairSpec>Response.
Base message type for this pair.
| Parameters: |
|
|---|
Send a reply to a previously received request (identified by request_id). Use this instead of writing directly to the publisher - just pass the content of the response data and the id that was issued with the request.
| Parameters: |
|
|---|
This module contains anything relating to introspection or manipulation of ros services.
Do a lookup to find services of the type specified. This will raise exceptions if it times out or returns multiple values. It can apply the additional logic of whether this should return a single unique result, or a list. Under the hood this calls out to the ros master for a list of registered services and it parses that to determine the result. If nothing is found, it loops around internally on a 10Hz loop until the result is found or the specified timeout is reached.
Usage:
from rocon_python_comms import find_service
try:
service_name = rocon_python_comms.find_service('rocon_interaction_msgs/SetInteractions',
timeout=rospy.rostime.Duration(15.0),
unique=True)
except rocon_python_comms.NotFoundException as e:
rospy.logwarn("failed to find the set_interactions service.")
| Parameters: | |
|---|---|
| Returns: | the fully resolved name of the service (unique) or list of names (non-unique) |
| Return type: | str |
| Raises: |
This module provides a means of interacting with a ros latched publisher in the same style as you would a ros service (request-response).
Works like a service proxy, but using a latched subscriber instead (regular subscribers will also work, but this is especially useful for latched subscribers since they typically always provide data).
If no timeout is specified when calling, it blocks indefinitely on a 100ms loop until a message arrives. Alternatively it will return with None if a specified timeout is reached.
Usage:
from rocon_python_comms import SubscriberProxy
gateway_info = SubscriberProxy('gateway_info', gateway_msgs.GatewayInfo)(rospy.Duration(0.5))
if gateway_info is not None:
# do something
| Todo: | upgrade to make use of python events instead of manual loops |
|---|
Returns immediately with the latest data or blocks indefinitely until the next data arrives.
| Parameters: | timeout (rospy.Duration) – time to wait for data, polling at 10Hz. |
|---|---|
| Returns: | latest data or None |
| Parameters: | |
|---|---|
| Returns: | msg type data or None |
| Return type: | same as the msg type specified in the arg or None |
Unregister the subscriber so future instantiations of this class can pull a fresh subscriber (important if the data is latched).
This module contains anything relating to introspection or manipulation of ros topics.
Do a lookup to find topics of the type specified. It can apply the additional logic of whether this should return a single unique result, or a list. Under the hood this calls out to the ros master for a list of registered topics and it parses that to determine the result. If nothing is found, it loops around internally on a 100ms loop until the result is found or the specified timeout is reached.
This will raise exceptions if it times out or returns multiple values.
Usage:
from rocon_python_comms import find_topic
try:
pairing_topic_name = find_topic('rocon_interaction_msgs/Pair', timeout=rospy.rostime.Duration(0.5), unique=True)
except rocon_python_comms.NotFoundException as e:
rospy.logwarn("support for interactions disabled")
| Parameters: | |
|---|---|
| Returns: | the fully resolved name of the topic (unique) or list of names (non-unique) |
| Return type: | str |
| Raises: | NotFoundException if no topic is found within the timeout |
This module provides a wall time rate class (missing in rospy).