osbrain.common — osBrain common logic

Miscellaneous utilities.

class osbrain.common.LogLevel

Bases: str

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

Methods

capitalize() Return a capitalized version of S, i.e.
casefold() Return a version of S suitable for caseless comparisons.
center(width[, fillchar]) Return S centered in a string of length width.
count(sub[, start[, end]]) Return the number of non-overlapping occurrences of substring sub in string S[start:end].
encode([encoding, errors]) Encode S using the codec registered for encoding.
endswith(suffix[, start[, end]]) Return True if S ends with the specified suffix, False otherwise.
expandtabs([tabsize]) Return a copy of S where all tab characters are expanded using spaces.
find(sub[, start[, end]]) Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end].
format(*args, **kwargs) Return a formatted version of S, using substitutions from args and kwargs.
format_map(mapping) Return a formatted version of S, using substitutions from mapping.
index(sub[, start[, end]]) Like S.find() but raise ValueError when the substring is not found.
isalnum() Return True if all characters in S are alphanumeric and there is at least one character in S, False otherwise.
isalpha() Return True if all characters in S are alphabetic and there is at least one character in S, False otherwise.
isdecimal() Return True if there are only decimal characters in S, False otherwise.
isdigit() Return True if all characters in S are digits and there is at least one character in S, False otherwise.
isidentifier() Return True if S is a valid identifier according to the language definition.
islower() Return True if all cased characters in S are lowercase and there is at least one cased character in S, False otherwise.
isnumeric() Return True if there are only numeric characters in S, False otherwise.
isprintable() Return True if all characters in S are considered printable in repr() or S is empty, False otherwise.
isspace() Return True if all characters in S are whitespace and there is at least one character in S, False otherwise.
istitle() Return True if S is a titlecased string and there is at least one character in S, i.e.
isupper() Return True if all cased characters in S are uppercase and there is at least one cased character in S, False otherwise.
join(iterable) Return a string which is the concatenation of the strings in the iterable.
ljust(width[, fillchar]) Return S left-justified in a Unicode string of length width.
lower() Return a copy of the string S converted to lowercase.
lstrip([chars]) Return a copy of the string S with leading whitespace removed.
maketrans(x[, y, z]) Return a translation table usable for str.translate().
partition(sep) Search for the separator sep in S, and return the part before it, the separator itself, and the part after it.
replace(old, new[, count]) Return a copy of S with all occurrences of substring old replaced by new.
rfind(sub[, start[, end]]) Return the highest index in S where substring sub is found, such that sub is contained within S[start:end].
rindex(sub[, start[, end]]) Like S.rfind() but raise ValueError when the substring is not found.
rjust(width[, fillchar]) Return S right-justified in a string of length width.
rpartition(sep) Search for the separator sep in S, starting at the end of S, and return the part before it, the separator itself, and the part after it.
rsplit([sep, maxsplit]) Return a list of the words in S, using sep as the delimiter string, starting at the end of the string and working to the front.
rstrip([chars]) Return a copy of the string S with trailing whitespace removed.
split([sep, maxsplit]) Return a list of the words in S, using sep as the delimiter string.
splitlines([keepends]) Return a list of the lines in S, breaking at line boundaries.
startswith(prefix[, start[, end]]) Return True if S starts with the specified prefix, False otherwise.
strip([chars]) Return a copy of the string S with leading and trailing whitespace removed.
swapcase() Return a copy of S with uppercase characters converted to lowercase and vice versa.
title() Return a titlecased version of S, i.e.
translate(table) Return a copy of the string S in which each character has been mapped through the given translation table.
upper() Return a copy of S converted to uppercase.
zfill(width) Pad a numeric string S with zeros on the left, to fill a field of the specified width.
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:
Event

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

osbrain.common.format_exception()

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

Useful for differenciating remote from local exceptions and exceptions that where sileced.

Returns:
str

A formatted string conaining an exception traceback information.

osbrain.common.format_method_exception(error, method, args, kwargs)
osbrain.common.get_linger()

Wrapper to get the linger option from the environment variable.

Returns:
int

Number of seconds to linger. Note that -1 means linger forever.

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:
Event

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

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:
Dict[bytes, Any]
osbrain.common.unbound_method(method)
Returns:
function

Unbounded 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.TOPIC_SEPARATOR character).

osbrain.common.validate_handler(handler, required)

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