rsb.util

Various helper classes and methods.

Code author: jwienke

Functions

Condition(*args, **kwargs) Factory function that returns a new condition variable object.
Lock allocate_lock() -> lock object
getLoggerByClass(klass) Get a python logger instance based on a class instance.
prefix() Tries to return the prefix that this code was installed into by guessing the install location from some rules.
timeToUnixMicroseconds(time) Converts a floating point, seconds based time to a unix timestamp in microseconds precision.
unixMicrosecondsToTime(value)

Classes

Enum(name, keys[, values]) Generates enum-like classes in python with proper printing support.
OrderedQueueDispatcherPool(threadPoolSize, ...) A thread pool that dispatches messages to a list of receivers.
Queue([maxsize]) Create a queue object with a given maximum size.
Thread([group, target, name, args, kwargs, ...]) A class that represents a thread of control.

Exceptions

InterruptedError

Code author: jwienke
exception rsb.util.InterruptedError

Bases: exceptions.RuntimeError

Code author: jwienke

args
message
class rsb.util.Enum(name, keys, values=None)

Bases: object

Generates enum-like classes in python with proper printing support.

Code author: jwienke

Generates a new enum.

Parameters:
  • name – name of the enum to create. Will normally be the name of the variable this constructor call is assigned to. For Used for printing.
  • keys – list of enum keys to generate
class EnumValue(name, value=None)

Bases: object

fromString(string)
class rsb.util.OrderedQueueDispatcherPool(threadPoolSize, delFunc, filterFunc=None)

Bases: object

A thread pool that dispatches messages to a list of receivers. The number of threads is usually smaller than the number of receivers and for each receiver it is guaranteed that messages arrive in the order they were published. No guarantees are given between different receivers. All methods except #start and #stop are reentrant.

The pool can be stopped and restarted at any time during the processing but these calls must be single-threaded.

Assumptions:
  • same subscriptions for multiple receivers unlikely, hence filtering done per receiver thread

Code author: jwienke

Constructs a new pool.

Parameters:
  • threadPoolSize (int >= 1) – number of threads for this pool
  • delFunc (callable) – the strategy used to deliver messages of type M to receivers of type R. This will most likely be a simple delegate function mapping to a concrete method call. Must be reentrant. callable with two arguments. First is the receiver of a message, second is the message to deliver
  • filterFunc (callable) – Reentrant function used to filter messages per receiver. Default accepts every message. callable with two arguments. First is the receiver of a message, second is the message to filter. Must return a bool, true means to deliver the message, false rejects it.
push(message)

Pushes a new message to be dispatched to all receivers in this pool.

Parameters:message – message to dispatch
registerReceiver(receiver)

Registers a new receiver at the pool. Multiple registrations of the same receiver are possible resulting in being called multiple times for the same message (but effectively this destroys the guarantee about ordering given above because multiple message queues are used for every subscription).

Parameters:receiver – new receiver
start()

Non-blocking start.

Raises:RuntimeError – if the pool was already started and is running
stop()

Blocking until every thread has stopped working.

unregisterReceiver(receiver)

Unregisters all registration of one receiver.

Parameters:receiver – receiver to unregister
Returns:True if one or more receivers were unregistered, else False
rsb.util.getLoggerByClass(klass)

Get a python logger instance based on a class instance. The logger name will be a dotted string containing python module and class name.

Parameters:klass – class instance
Returns:logger instance
rsb.util.prefix()

Tries to return the prefix that this code was installed into by guessing the install location from some rules.

Adapted from http://ttboj.wordpress.com/2012/09/20/finding-your-software-install-prefix-from-inside-python/

Returns:string path with the install prefix or empty string if not known
rsb.util.timeToUnixMicroseconds(time)

Converts a floating point, seconds based time to a unix timestamp in microseconds precision.

Parameters:time – time since epoch in seconds + fractional part.
Returns:time as integer with microseconds precision.
rsb.util.unixMicrosecondsToTime(value)