Resource Management

Threads

mil_tools.thread_lock(lock: allocate_lock)

A decortor for using an existing thread lock to thread-lock a function. This prevents the function from being executed by multiple threads at once.

import threading
lock = threading.Lock()

@thread_lock(lock)
def my_function(a, b, c):
    print(a, b, c)

ROS Resources

mil_tools.wait_for_param(param_name: str, timeout: Optional[float] = None, poll_rate: float = 0.1) → Optional[Any]

Blocking wait for a parameter named to exist. Polls at the frequency of poll_rate. Once the parameter exists, return get and return it.

This function intentionally leaves failure logging duties to the developer.

Parameters

• param_name (str) – The name of the parameter to watch.

• timeout (Optional[float]) – The number of seconds to wait for the parameter to exist.

• poll_rate (float) – The Hz rate to poll at.

Returns

If found, the value of the parameter. Returns None if the parameter never came to exist.

Return type

Optional[Any]

mil_tools.wait_for_subscriber(node_name: str, topic: str, timeout: float = 5.0) → bool

Blocks until a node with the name node_name subscribes to a topic. Useful in integration tests.

Parameters

• node_name (str) – The node name to check the subscription status of. If a local node name, then the name is resolved to be the global name.

• topic (str) – The topic to check whether the node is subscribed to. If a local topic name, then the name is resolved to be the global name.

• timeout (float) – The amount of time to wait (in seconds) to attempt to estalish a connection.

Returns

Whether the node with the given name has subscribed to the given topic.

Return type

bool

mil_tools.wait_for_service(service, warn_time: float = 1.0, warn_msg: str = 'Waiting for service..', timeout: Optional[float] = None) → None

A fancy extension of wait for service that will warn with a message if it is taking a while.

Parameters

• warn_time (float) – float in seconds, how long to wait before logging warn_msg

• warn_msg (str) – msg logged with rospy.logwarn if warn_time passes without service connected

• timeout (Optional[float]) – overall timeout. If None, does nothing. If a float, will raise exception if many TOTAL seconds has passed without connecting.