autobahn.util

---

Module Contents

Classes

IdGenerator	ID generator for WAMP request IDs.
Stopwatch	Stopwatch based on walltime.
Tracker	A key-based statistics tracker.
EqualityMixin	Mixing to add equality comparison operators to a class.
ObservableMixin	Internal utility for enabling event-listeners on particular objects

Functions

public(obj)	The public user API of Autobahn is marked using this decorator.
encode_truncate(text, limit[, encoding, return_encoded])	Given a string, return a truncated version of the string such that
xor(→ bytes)	XOR two binary strings of arbitrary (equal) length.
utcstr([ts])	Format UTC timestamp in ISO 8601 format.
utcnow()	Get current time in UTC as ISO 8601 string.
rid()	Generate a new random integer ID from range [0, 2**53].
id()	Generate a new random integer ID from range [0, 2**53].
newid([length])	Generate a new random string ID.
generate_token(→ str)	Generate cryptographically strong tokens, which are strings like M6X5-YO5W-T5IK.
generate_activation_code()	Generate a one-time activation code or token of the form 'W97F-96MJ-YGJL'.
generate_user_password()	Generate a secure, random user password of the form 'kgojzi61dn5dtb6d'.
generate_serial_number()	Generate a globally unique serial / product code of the form 'YRAC-EL4X-FQQE-AW4T-WNUV-VN6T'.
rtime()	Precise, fast wallclock time.
machine_id(→ str)	For informational purposes, get a unique ID or serial for this machine (device).
hl(text[, bold, color])
hltype(obj)
hlid(oid)
hluserid(oid)
hlval(val[, color, bold])
hlcontract(oid)
with_0x(address)
without_0x(address)
write_keyfile(filepath, tags, msg)	Internal helper, write the given tags to the given file-
parse_keyfile(→ collections.OrderedDict)	Internal helper. This parses a node.pub or node.priv file and

autobahn.util.public(obj)

The public user API of Autobahn is marked using this decorator. Everything that is not decorated @public is library internal, can change at any time and should not be used in user program code.

autobahn.util.encode_truncate(text, limit, encoding='utf8', return_encoded=True)

Given a string, return a truncated version of the string such that the UTF8 encoding of the string is smaller than the given limit.

This function correctly truncates even in the presence of Unicode code points that encode to multi-byte encodings which must not be truncated in the middle.

Parameters:

• text (str) – The (Unicode) string to truncate.

• limit (int) – The number of bytes to limit the UTF8 encoding to.

• encoding (str) – Truncate the string in this encoding (default is utf-8).

• return_encoded (bool) – If True, return the string encoded into bytes according to the specified encoding, else return the string as a string.

Returns:

The truncated string.

Return type:

str or bytes

autobahn.util.xor(d1: bytes, d2: bytes) → bytes

XOR two binary strings of arbitrary (equal) length.

Parameters:

• d1 – The first binary string.

• d2 – The second binary string.

Returns:

XOR of the binary strings (XOR(d1, d2))

autobahn.util.utcstr(ts=None)

Format UTC timestamp in ISO 8601 format.

Note: to parse an ISO 8601 formatted string, use the iso8601 module instead (e.g. iso8601.parse_date("2014-05-23T13:03:44.123Z")).

>>> txaio.time_ns()
1641121311914026419
>>> int(iso8601.parse_date(utcnow()).timestamp() * 1000000000.)
1641121313209000192

Parameters:

ts (instance of datetime.datetime or None) – The timestamp to format.

Returns:

Timestamp formatted in ISO 8601 format.

Return type:

str

autobahn.util.utcnow()

Get current time in UTC as ISO 8601 string.

Returns:

Current time as string in ISO 8601 format.

Return type:

str

class autobahn.util.IdGenerator

Bases: object

ID generator for WAMP request IDs.

WAMP request IDs are sequential per WAMP session, starting at 1 and wrapping around at 2**53 (both value are inclusive [1, 2**53]).

The upper bound 2**53 is chosen since it is the maximum integer that can be represented as a IEEE double such that all smaller integers are representable as well.

Hence, IDs can be safely used with languages that use IEEE double as their main (or only) number type (JavaScript, Lua, etc).

See https://github.com/wamp-proto/wamp-proto/blob/master/spec/basic.md#ids

next()

Returns next ID.

Returns:

The next ID.

Return type:

int

__next__()

autobahn.util.rid()

Generate a new random integer ID from range [0, 2**53].

The generated ID is uniformly distributed over the whole range, doesn’t have a period (no pseudo-random generator is used) and cryptographically strong.

The upper bound 2**53 is chosen since it is the maximum integer that can be represented as a IEEE double such that all smaller integers are representable as well.

Hence, IDs can be safely used with languages that use IEEE double as their main (or only) number type (JavaScript, Lua, etc).

Returns:

A random integer ID.

Return type:

int

autobahn.util.id()

Generate a new random integer ID from range [0, 2**53].

The generated ID is based on a pseudo-random number generator (Mersenne Twister, which has a period of 2**19937-1). It is NOT cryptographically strong, and hence NOT suitable to generate e.g. secret keys or access tokens.

The upper bound 2**53 is chosen since it is the maximum integer that can be represented as a IEEE double such that all smaller integers are representable as well.

Hence, IDs can be safely used with languages that use IEEE double as their main (or only) number type (JavaScript, Lua, etc).

Returns:

A random integer ID.

Return type:

int

autobahn.util.newid(length=16)

Generate a new random string ID.

The generated ID is uniformly distributed and cryptographically strong. It is hence usable for things like secret keys and access tokens.

Parameters:

length (int) – The length (in chars) of the ID to generate.

Returns:

A random string ID.

Return type:

str

autobahn.util.generate_token(char_groups: int, chars_per_group: int, chars: str | None = None, sep: str | None = None, lower_case: bool | None = False) → str

Generate cryptographically strong tokens, which are strings like M6X5-YO5W-T5IK. These can be used e.g. for used-only-once activation tokens or the like.

The returned token has an entropy of math.log(len(chars), 2.) * chars_per_group * char_groups bits.

With the default charset and 4 characters per group, generate_token() produces strings with the following entropy:

character groups	entropy (at least)	recommended use
2	38 bits
3	57 bits	one-time activation or pairing code
4	76 bits	secure user password
5	95 bits
6	114 bits	globally unique serial / product code
7	133 bits

Here are some examples:

• token(3): 9QXT-UXJW-7R4H

• token(4): LPNN-JMET-KWEP-YK45

• token(6): NXW9-74LU-6NUH-VLPV-X6AG-QUE3

Parameters:

• char_groups – Number of character groups (or characters if chars_per_group == 1).

• chars_per_group – Number of characters per character group (or 1 to return a token with no grouping).

• chars – Characters to choose from. Default is 27 character subset of the ISO basic Latin alphabet (see: DEFAULT_TOKEN_CHARS).

• sep – When separating groups in the token, the separater string.

• lower_case – If True, generate token in lower-case.

Returns:

The generated token.

autobahn.util.generate_activation_code()

Generate a one-time activation code or token of the form 'W97F-96MJ-YGJL'. The generated value is cryptographically strong and has (at least) 57 bits of entropy.

Returns:

The generated activation code.

Return type:

str

autobahn.util.generate_user_password()

Generate a secure, random user password of the form 'kgojzi61dn5dtb6d'. The generated value is cryptographically strong and has (at least) 76 bits of entropy.

Returns:

The generated password.

Return type:

str

autobahn.util.generate_serial_number()

Generate a globally unique serial / product code of the form 'YRAC-EL4X-FQQE-AW4T-WNUV-VN6T'. The generated value is cryptographically strong and has (at least) 114 bits of entropy.

Returns:

The generated serial number / product code.

Return type:

str

autobahn.util.rtime()

Precise, fast wallclock time.

Returns:

The current wallclock in seconds. Returned values are only guaranteed to be meaningful relative to each other.

Return type:

float

class autobahn.util.Stopwatch(start=True)

Bases: object

Stopwatch based on walltime.

This can be used to do code timing and uses the most precise walltime measurement available on the platform. This is a very light-weight object, so create/dispose is very cheap.

elapsed()

Return total time elapsed in seconds during which the stopwatch was running.

Returns:

The elapsed time in seconds.

Return type:

float

pause()

Pauses the stopwatch and returns total time elapsed in seconds during which the stopwatch was running.

Returns:

The elapsed time in seconds.

Return type:

float

resume()

Resumes a paused stopwatch and returns total elapsed time in seconds during which the stopwatch was running.

Returns:

The elapsed time in seconds.

Return type:

float

stop()

Stops the stopwatch and returns total time elapsed in seconds during which the stopwatch was (previously) running.

Returns:

The elapsed time in seconds.

Return type:

float

class autobahn.util.Tracker(tracker, tracked)

Bases: object

A key-based statistics tracker.

track(key)

Track elapsed for key.

Parameters:

key (str) – Key under which to track the timing.

diff(start_key, end_key, formatted=True)

Get elapsed difference between two previously tracked keys.

Parameters:

• start_key (str) – First key for interval (older timestamp).

• end_key (str) – Second key for interval (younger timestamp).

• formatted (bool) – If True, format computed time period and return string.

Returns:

Computed time period in seconds (or formatted string).

Return type:

float or str

absolute(key)

Return the UTC wall-clock time at which a tracked event occurred.

Parameters:

key (str) – The key

Returns:

Timezone-naive datetime.

Return type:

instance of datetime.datetime

__getitem__(key)

__iter__()

__str__()

Return str(self).

class autobahn.util.EqualityMixin

Bases: object

Mixing to add equality comparison operators to a class.

Two objects are identical under this mixin, if and only if:

1. both object have the same class

2. all non-private object attributes are equal

__eq__(other)

Compare this object to another object for equality.

Parameters:

other (obj) – The other object to compare with.

Returns:

True iff the objects are equal.

Return type:

bool

__ne__(other)

Compare this object to another object for inequality.

Parameters:

other (obj) – The other object to compare with.

Returns:

True iff the objects are not equal.

Return type:

bool

class autobahn.util.ObservableMixin

Bases: object

Internal utility for enabling event-listeners on particular objects

_parent

_valid_events

_listeners

_results

set_valid_events(valid_events=None)

Parameters:

valid_events – if non-None, .on() or .fire() with an event not listed in valid_events raises an exception.

_check_event(event)

Internal helper. Throws RuntimeError if we have a valid_events list, and the given event isnt’ in it. Does nothing otherwise.

on(event, handler)

Add a handler for an event.

Parameters:

• event – the name of the event

• handler – a callable thats invoked when .fire() is called for this events. Arguments will be whatever are given to .fire()

off(event=None, handler=None)

Stop listening for a single event, or all events.

Parameters:

• event – if None, remove all listeners. Otherwise, remove listeners for the single named event.

• handler – if None, remove all handlers for the named event; otherwise remove just the given handler.

fire(event, *args, **kwargs)

Fire a particular event.

Parameters:

event – the event to fire. All other args and kwargs are passed on to the handler(s) for the event.

Returns:

a Deferred/Future gathering all async results from all handlers and/or parent handlers.

autobahn.util.machine_id() → str

For informational purposes, get a unique ID or serial for this machine (device).

Returns:

Unique machine (device) ID (serial), e.g. 81655b901e334fc1ad59cbf2719806b7.

autobahn.util.hl(text, bold=False, color='yellow')

autobahn.util.hltype(obj)

autobahn.util.hlid(oid)

autobahn.util.hluserid(oid)

autobahn.util.hlval(val, color='white', bold=True)

autobahn.util.hlcontract(oid)

autobahn.util.with_0x(address)

autobahn.util.without_0x(address)

autobahn.util.write_keyfile(filepath, tags, msg)

Internal helper, write the given tags to the given file-

autobahn.util.parse_keyfile(key_path: str, private: bool = True) → collections.OrderedDict

Internal helper. This parses a node.pub or node.priv file and returns a dict mapping tags -> values.