osbrain.common — osBrain common logic

Miscellaneous utilities.

class osbrain.common.LogLevel

Bases: str

Identifies the log level: ERROR, WARNING, INFO, DEBUG.

osbrain.common.after(delay, action, *args)

Execute an action after a given number of seconds.

This function is executed in a separate thread.

Parameters:
  • delay (float) – Number of seconds to delay the action.
  • action – To be taken after the interval.
  • args (tuple, default is ()) – Arguments for the action.
Returns:

A timer object that can be terminated using the stop() method.
Return type:

Event
osbrain.common.format_exception()

Represent a traceback exception as a string in which all lines start with a | character.

Useful for differentiating remote from local exceptions and exceptions that where silenced.

Returns:A formatted string containing an exception traceback information.
Return type:str
osbrain.common.format_method_exception(error, method, args, kwargs)

Represent an exception as a formatted string that includes the name and arguments of the method where it occurred, followed by the output of format_exception.

Parameters:
  • error (Error) – The exception that was raised.
  • method (function) – The method where the exception was raised.
  • args – The arguments of the method call.
  • kwargs (dict) – The keyword arguments of the method call.
Returns:

The formatted string with the method call and traceback information.
Return type:

str
osbrain.common.get_linger(seconds=None)

Wrapper to get the linger option from the environment variable.

Parameters:seconds (float, default is None.) – Linger seconds, in seconds.
Returns:Number of seconds to linger. Note that -1 means linger forever.
Return type:int
osbrain.common.repeat(interval, action, *args)

Repeat an action forever after a given number of seconds.

If a sequence of events takes longer to run than the time available before the next event, the repeater will simply fall behind.

This function is executed in a separate thread.

Parameters:
  • interval (float) – Number of seconds between executions.
  • action – To be taken after the interval.
  • args (tuple, default is ()) – Arguments for the action.
Returns:

A timer object that can be terminated using the stop() method.
Return type:

Event
osbrain.common.topic_to_bytes(topic: Union[bytes, str]) → bytes

Return the passed topic as a bytes object.

osbrain.common.topics_to_bytes(handlers: Dict[Union[bytes, str], Any], uuid: bytes = b'')

Given some pairs topic/handler, leaves them prepared for making the actual ZeroMQ subscription.

Parameters:
  • handlers – Contains pairs “topic - handler”.
  • uuid – uuid of the SYNC_PUB/SYNC_SUB channel (if applies). For normal PUB/SUB communication, this should be b''.
Returns:
Return type:

Dict[bytes, Any]
osbrain.common.unbound_method(method)
Returns:Unbounded function.
Return type:function
osbrain.common.unique_identifier() → bytes
Returns:
  • A unique identifier that is safe to use in PUB-SUB communication
  • patterns (i.e. (does not contain the)
  • osbrain.agent.TOPIC_SEPARATOR character).
osbrain.common.validate_handler(handler, required)

Raises a ValueError exception when a required is handler but not present.