The util module

Various utility classes and functions.

This module only depends on the Python standard library.

class Dispatcher(default_func=None)[source]

Bases: object

Dispatches calls via an instance to methods based on the first argument.

A Dispatcher is used as a decorator when defining a class, and then called via an instance to select a method based on the first argument (which must be hashable).

If you override methods in a subclass that were dispatched, the dispatcher automatically dispatches to the new method. If you want to add new keywords in a subclass, just create a new Dispatcher with the same name. It will automatically inherit the references stored in the old dispatcher.

Usage:

class MyClass:
    dispatch = Dispatcher()

    @dispatch(1)
    def call_one(self, value):
        print("One called", value)

    @dispatch(2)
    def call_two(self, value):
        print("Two called", value)

    def handle_input(self, number, value):
        self.dispatch(number, value)

>>> i = MyClass()
>>> i.handle_input(2, 3)
Two called 3

Values of the first argument that are not handled are normally silently ignored, but you can also specify a default function, that is called when a value is not handled:

class MyClass:
    @Dispatcher
    def dispatch(self, number, value):
        print("Default function called:", number, value)

    @dispatch(1)
    def call_one(self, value):
        print("One called", value)

    @dispatch(2)
    def call_two(self, value):
        print("Two called", value)

    def handle_input(self, number, value):
        self.dispatch(number, value)

>>> i = MyClass()
>>> i.handle_input(3, 10)
Default function called: 3 10
class Observable[source]

Bases: object

Simple base class for objects that need to announce events.

Use connect() to add a callable to be called when a certain event occurs.

To announce an event from inside methods, use emit(). In your documentation you should specify which arguments are used for which events; in order to keep this class simple and fast, no checking is performed whatsoever.

Example:

>>> o = Observable()
>>>
>>> def slot(arg):
...     print("slot called:", arg)
...
>>> o.connect('test', slot)
>>>
>>> o.emit('test', 1)   # in a method of your Observable subclass
slot called: 1

Is is also possible to use emit() in a with context. In that case the return values of the connected functions are collected and if they are a context manager, they are entered as well. An example:

>>> import contextlib
>>>
>>> @contextlib.contextmanager
... def f():
...     print("one")
...     yield
...     print("two")
...
>>> o=Observable()
>>> o.connect('test', f)
>>>
>>> with o.emit('test'):
...     print("Yo!!!")
...
one
Yo!!!
two

This enables you to announce events, and connected objects can perform a task before the event’s context starts and another task when the event’s context exits.

connect(event, func, once=False, prepend_self=False, priority=0)[source]

Register a function to be called when a certain event occurs.

The event should be a string or any hashable object that identifies the event. The priority determines the order the functions are called. Lower numbers are called first. If once is set to True, the function is called once and then removed from the list of callbacks. If prepend_self is True, the callback is called with the treebuilder itself as first argument.

disconnect(event, func)[source]

Remove a previously registered callback function.

disconnect_all(event=None)[source]

Disconnect all functions (from the event).

If event is None, disconnects all connected functions from all events.

has_connections(event)[source]

Return True when there is at least one callback registered for the event.

This can be used before performing some task, the task maybe then can be optimized because we know nobody needs the events.

is_connected(event, func)[source]

Return True if func is connected to event.

emit(event, *args, **kwargs)[source]

Call all callbacks for the event.

Returns a list of the return values of all callbacks. When using this list in a with context, all return values that are context managers, are entered. (The list is an _EmitResult object that extends Python’s list builtin, so that it can be used as a context manager.)

class Symbol[source]

Bases: object

An unique object that has a name.

cached_method(func)[source]

Wrap a method and caches its return value.

The method argument tuple should be hashable. Keyword arguments are not supported. The cache is thread-safe. Does not keep a reference to the instance.

cached_property(func)[source]

Like property, but caches the computed value.

merge_adjacent(stream, factory=<class 'tuple'>)[source]

Yield items from a stream of tuples.

The first two items of each tuple are regarded as pos and end. If they are adjacent, and the rest of the tuples compares the same, the items are merged.

Instead of the default factory tuple, you can give a named tuple or any other type to wrap the streams items in.

merge_adjacent_actions(tokens)[source]

Yield three-tuples (pos, end, action).

Adjacent actions that are the same are merged into one range.

merge_adjacent_actions_with_language(tokens)[source]

Yield four-tuples (pos, end, action, language).

Adjacent actions that are the same and occurred in the same language are merged into one range.

get_bom_encoding(data)[source]

Get the BOM (Byte Order Mark) of data, if any.

A two-tuple is returned (encoding, data). If the data starts with a BOM mark, its encoding is determined and the BOM mark is stripped off. Otherwise, the returned encoding is None and the data is returned unchanged.

split_list(l, separator)[source]

Split list on items that compare equal to separator.

Yields result lists that may be empty.

quote(s)[source]

Like repr, but return s with double quotes, escaping ” and .

tokens(nodes)[source]

Helper to yield tokens from the iterable of nodes.

tokens_bw(nodes)[source]

Helper to yield tokens from the iterable in backward direction.

Make sure nodes is already in backward direction.