chainlet.protolink module

Helpers for creating ChainLinks from standard protocols of objects

Tools of this module allow writing simpler code by reusing functionality of existing protocol interfaces and builtins. The interface to other chainlet objects is automatically built around the objects. Using protocol interfaces in chains allows to easily create chainlets from existing code.

Every protolink represents a specific Python protocol or builtin. For example, the iterlet() protolink maps to iter(iterable). This allows pulling chunks from iterables to a chain:

from examples import windowed_average

fixed_iterable = [1, 2, 4, 3]
chain = iterlet(fixed_iterable) >> windowed_average(size=2)
for value in chain:
    print(value)  # prints 1.0, 1.5, 3.0, 3.5

The protolinks exist mostly for convenience - they are thin wrappers using chainlet primitives. As such, they are most useful to adjust existing code and objects for pipelines.

Any protolink that works on iterables supports two modes of operation:

pull: iterable provided at instantiation

Pull data chunks directly from an iterable, work on them, and send them along a chain. These are usually equivalent to a corresponding builtin, but support chaining.

push: no iterable provided at instantiation

Wait for data chunks to be pushed in, work on them, and send them along a chain. These are usually equivalent to wrapping a chain in the corresponding builtin, but preserve chain features.

class chainlet.protolink.callet(*slave_args, **slave_kwargs)

Bases: chainlet.genlink.GeneratorLink

Pull chunks from an object using individual calls

Parameters:callee (callable) – object supporting callee()

import random

chain = callet(random.random) >> windowed_average(size=200)
for _ in range(50):
    print(next(chain))  # prints series converging to 0.5

callet._slave_factory(callee)

Pull chunks from an object using individual calls

Parameters:callee (callable) – object supporting callee()

import random

chain = callet(random.random) >> windowed_average(size=200)
for _ in range(50):
    print(next(chain))  # prints series converging to 0.5

chainlet.protolink.enumeratelet(iterable=None, start=0)

Enumerate chunks of data from an iterable or a chain

Parameters:

• iterable (iterable, None or int) – object supporting iteration, or an index
• start (int) – an index to start counting from

Raises:

TypeError – if both parameters are set and iterable does not support iteration

In pull mode, enumeratelet() works similar to the builtin enumerate() but is chainable:

chain = enumeratelet(['Paul', 'Thomas', 'Brian']) >> printlet(sep=':\t')
for value in chain:
    pass  # prints `0:  Paul`, `1:      Thomas`, `2:    Brian`

By default, enumeratelet() enumerates chunks passed in from a pipeline. To use a different starting index, either set the start keyword parameter or set the first positional parameter.

chain = iteratelet(['Paul', 'Thomas', 'Brian']) >> enumeratelet() >> printlet(sep=':\t')
for value in chain:
    pass  # prints `0:  Paul`, `1:      Thomas`, `2:    Brian`

chainlet.protolink.filterlet(function=<type 'bool'>, iterable=None)

Filter chunks of data from an iterable or a chain

Parameters:

• function (callable) – callable selecting valid elements
• iterable (iterable or None) – object providing chunks via iteration

For any chunk in iterable or the chain, it is passed on only if function(chunk) returns true.

chain = iterlet(range(10)) >> filterlet(lambda chunk: chunk % 2 == 0)
for value in chain:
    print(value)  # prints 0, 2, 4, 6, 8

class chainlet.protolink.iterlet(*slave_args, **slave_kwargs)

Bases: chainlet.genlink.GeneratorLink

Pull chunks from an object using iteration

Parameters:iterable (iterable) – object supporting iteration

chain = iterlet([1, 2, 3, 4, 5, 5, 6, 6]) >> filterlet(lambda chunk: chunk % 2 == 0)
for element in chain:
    print(element)  # prints 2, 4, 6, 6

iterlet._slave_factory(iterable)

Pull chunks from an object using iteration

Parameters:iterable (iterable) – object supporting iteration

chain = iterlet([1, 2, 3, 4, 5, 5, 6, 6]) >> filterlet(lambda chunk: chunk % 2 == 0)
for element in chain:
    print(element)  # prints 2, 4, 6, 6

class chainlet.protolink.printlet(*slave_args, **slave_kwargs)

Bases: chainlet.genlink.GeneratorLink

Print chunks of data from a chain

Parameters:

• flatten – whether to flatten data chunks
• kwargs – keyword arguments as for print()

If flatten is True, every chunk received is unpacked. This is useful when passing around connected data, e.g. from enumeratelet().

Keyword arguments via kwargs are equivalent to those of print(). For example, passing file=sys.stderr is a simple way of creating a debugging element in a chain:

debug_chain = chain[:i] >> printlet(file=sys.stderr) >> chain[i:]

printlet._slave_factory(flatten=False, **kwargs)

Print chunks of data from a chain

Parameters:

• flatten – whether to flatten data chunks
• kwargs – keyword arguments as for print()

If flatten is True, every chunk received is unpacked. This is useful when passing around connected data, e.g. from enumeratelet().

Keyword arguments via kwargs are equivalent to those of print(). For example, passing file=sys.stderr is a simple way of creating a debugging element in a chain:

debug_chain = chain[:i] >> printlet(file=sys.stderr) >> chain[i:]

chainlet.protolink.reverselet(iterable)

Pull chunks from an object using reverse iteration

Parameters:iterable (iterable) – object supporting reverse iteration

See iterlet() for an example.