> built 2026-09-15 10:07 UTC from 7a19d20 (master) · creek 0.1.35. Details: build_info.json
# index.html.md
# creek
A simple facade for streams: wrap an iterable, layer transformations on it, and read unbounded streams as if they were lists.
To install: `pip install creek`
The smallest complete example: a `Creek` subclass that parses each line of a file-like stream.
```python
>>> from io import StringIO
>>> from creek import Creek
>>> class CsvCreek(Creek):
... def data_to_obj(self, line):
... return [x.strip() for x in line.strip().split(',')]
>>> stream = CsvCreek(StringIO('a, b, c\n1,2, 3\n'))
>>> list(stream)
[['a', 'b', 'c'], ['1', '2', '3']]
```
## For AI agents
`creek` ships tooling for coding agents. If you are one, start here.
**The documentation, machine-readable**: [`llms.txt`](https://i2mint.github.io/creek/llms.txt) indexes every page; [`creek.md`](https://i2mint.github.io/creek/creek.md) is the whole documentation in one file; every page has a `.md` twin; [`objects.inv`](https://i2mint.github.io/creek/objects.inv) maps symbols to URLs.
If you are a control freak, the rest of this README is written for you, starting at [The Creek layers]().
## The Creek layers
The `Creek` base class offers a layer-able wrap of the stream interface. There are three layering methods, `pre_iter`, `data_to_obj` and `post_iter`, each the identity by default. Iterating over a creek is exactly:
```python
yield from self.post_iter(map(self.data_to_obj, self.pre_iter(self.stream)))
```
- `pre_iter(stream)`: prepare and/or filter the raw stream (skip a header, enumerate lines, …)
- `data_to_obj(item)`: transform each item the stream yields
- `post_iter(objs)`: further process or filter the transformed objects
Everything else (`seek`, `close`, the context manager protocol, …) is delegated to the wrapped stream:
```python
>>> src = StringIO(
... '''a, b, c
... 1,2, 3
... 4, 5,6
... '''
... )
>>> class MyCreek(Creek):
... def data_to_obj(self, line):
... return [x.strip() for x in line.strip().split(',')]
...
>>> stream = MyCreek(src)
>>> list(stream)
[['a', 'b', 'c'], ['1', '2', '3'], ['4', '5', '6']]
>>> stream.seek(0) # oh!... but we consumed the stream already, so let's go back to the beginning
0
>>> list(stream)
[['a', 'b', 'c'], ['1', '2', '3'], ['4', '5', '6']]
>>> stream.seek(0) # rewind again
0
>>> next(stream)
['a', 'b', 'c']
>>> next(stream)
['1', '2', '3']
```
Let’s add a filter! There are two kinds you can use: one applied to the line before the data is transformed by `data_to_obj` (in `pre_iter`), and one applied after, to the objects (in `post_iter`).
```python
>>> src = StringIO(
... '''a, b, c
... 1,2, 3
... 4, 5,6
... ''')
>>> class MyFilteredCreek(MyCreek):
... def post_iter(self, objs):
... yield from filter(lambda obj: str.isnumeric(obj[0]), objs)
>>>
>>> s = MyFilteredCreek(src)
>>> list(s)
[['1', '2', '3'], ['4', '5', '6']]
>>> s.seek(0)
0
>>> list(s)
[['1', '2', '3'], ['4', '5', '6']]
>>> s.seek(0)
0
>>> next(s)
['1', '2', '3']
```
Recipes:
- `pre_iter`: `itertools.islice` to skip header lines
- `pre_iter`: `enumerate` to get line indices in the stream iterator
- `pre_iter = functools.partial(map, line_pre_proc_func)` to preprocess all lines with `line_pre_proc_func`
- `pre_iter`: `functools.partial(filter, condition)` to filter before `data_to_obj`
- `post_iter`: `itertools.chain.from_iterable` to flatten a chunked/segmented stream
- `post_iter`: `functools.partial(filter, condition)` to filter the yielded objects
## Beyond Creek
- `InfiniteSeq` and `IndexedBuffer` (`creek.infinite_sequence`): list-like `s[i:j]` access to an unbounded stream, through a bounded buffer.
- `BufferStats` and `Segmenter` (`creek.tools`): rolling-window statistics and segmentation of a stream.
- `dynamically_index` and `filter_and_index_stream` (`creek.tools`): `enumerate` with a custom index rule, and filtered indexing.
## Documentation
- Rendered docs: https://i2mint.github.io/creek/
- The whole API in one Markdown file: https://i2mint.github.io/creek/creek.md
This documentation as a single file: creek.md (Markdown, for agents).
# _autosummary/creek.automatas.html.md
# creek.automatas
Automatas (finite state machines): step a state through a stream of symbols.
Main entry points:
- `mapping_to_transition_func`: a transition function from a `{(state, symbol): next_state}` mapping
- `basic_automata`: `basic_automata(transition_func, state)(symbols)` yields the state after each symbol
- `BasicAutomata`: the same as a stateful, resettable object
```pycon
>>> from creek.automatas import mapping_to_transition_func, basic_automata
>>> step = mapping_to_transition_func({('a', 1): 'b', ('b', 1): 'a'})
>>> list(basic_automata(step, 'a')([1, 1, 1]))
['b', 'a', 'b']
```
### Functions
| [`automata`](_autosummary/creek.automatas.html.md#creek.automatas.automata)(transition_func, state) | Make an automata: `basic_automata(f, state)(symbols)` yields the state after each symbol. |
|------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------|
| [`basic_automata`](_autosummary/creek.automatas.html.md#creek.automatas.basic_automata)(transition_func, state) | Make an automata: `basic_automata(f, state)(symbols)` yields the state after each symbol. |
| [`mapping_to_transition_func`](_autosummary/creek.automatas.html.md#creek.automatas.mapping_to_transition_func)(mapping[, strict]) | Make a transition function from a `{(state, symbol): next_state}` mapping. |
### Classes
| [`BasicAutomata`](_autosummary/creek.automatas.html.md#creek.automatas.BasicAutomata)(transition_func[, state]) | Stateful automata: `automata(state, symbols)` yields the state after each symbol; `reset` restores the initial one. |
|--------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------|
| [`MappingTransitionFunc`](_autosummary/creek.automatas.html.md#creek.automatas.MappingTransitionFunc)(mapping[, strict]) | Class form of `mapping_to_transition_func`: a callable `(state, symbol) -> next_state`. |
### *class* creek.automatas.BasicAutomata(transition_func, state=None)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
Stateful automata: `automata(state, symbols)` yields the state after each symbol; `reset` restores the initial one.
#### reset()
Reset state to initial state and return self.
#### transition(symbol)
Apply one `symbol` to the current state and return the new state.
* **Return type:**
[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`)
### *class* creek.automatas.MappingTransitionFunc(mapping, strict=True)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
Class form of `mapping_to_transition_func`: a callable `(state, symbol) -> next_state`.
#### map_states(state_mapper)
Return a new MappingTransitionFunc with the same mapping but with
state_mapper applied to the states.
* **Return type:**
[`MappingTransitionFunc`](_autosummary/creek.automatas.html.md#creek.automatas.MappingTransitionFunc)
### creek.automatas.automata(transition_func, state)
Make an automata: `basic_automata(f, state)(symbols)` yields the state after each symbol.
* **Return type:**
[`Callable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Callable)[[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`), [`Iterable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Iterable)[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`Symbol`)]], [`Iterable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Iterable)[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`)]]
### creek.automatas.basic_automata(transition_func, state)
Make an automata: `basic_automata(f, state)(symbols)` yields the state after each symbol.
* **Return type:**
[`Callable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Callable)[[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`), [`Iterable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Iterable)[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`Symbol`)]], [`Iterable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Iterable)[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`)]]
### creek.automatas.mapping_to_transition_func(mapping, strict=True)
Make a transition function from a `{(state, symbol): next_state}` mapping.
* **Parameters:**
* **mapping** ([`Mapping`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Mapping)[[`tuple`](https://docs.python.org/3/builtins/stdtypes.html#tuple)[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`), [`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`Symbol`)], [`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`)]) – The transitions.
* **strict** ([`bool`](https://docs.python.org/3/builtins/functions.html#bool)) – If true (the default), an unmapped `(state, symbol)` pair is a
lookup error (`KeyError` for a `dict`); if false, the state is left
unchanged.
* **Return type:**
[`Callable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Callable)[[[`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`), [`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`Symbol`)], [`TypeVar`](https://docs.python.org/3/library/typing.html#typing.TypeVar)(`State`)]
# _autosummary/creek.base.html.md
# creek.base
The `Creek` base class: a layer-able wrapper around a stream.
Main entry points:
- `Creek`: subclass it and override `pre_iter`, `data_to_obj` and/or `post_iter`
- `Creek.wrap`: wrap a stream class (or instance) so that you get creeks out of it
```pycon
>>> from creek.base import Creek
>>> class Doubler(Creek):
... def data_to_obj(self, x):
... return x * 2
>>> list(Doubler([1, 2, 3]))
[2, 4, 6]
```
### Classes
| [`Creek`](_autosummary/creek.base.html.md#creek.base.Creek)(stream) | A layer-able version of the stream interface. |
|------------------------------------------------------------------|-------------------------------------------------|
### *class* creek.base.Creek(stream)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
A layer-able version of the stream interface.
A `Creek` wraps a stream (any iterable, typically a file-like object) and
delegates attribute access to it, so it can be used where the stream was.
Iteration goes through three layering methods, each the identity by default:
- `pre_iter(stream)`: prepare and/or filter the raw stream
- `data_to_obj(item)`: transform each item the stream yields
- `post_iter(objs)`: further process or filter the transformed objects
That is, `iter(creek)` is `post_iter(map(data_to_obj, pre_iter(stream)))`.
* **Parameters:**
**stream** – The wrapped stream. Attributes not defined on the creek
(`seek`, `close`…) are looked up on it.
```pycon
>>> from io import StringIO
>>>
>>> src = StringIO(
... '''a, b, c
... 1,2, 3
... 4, 5,6
... '''
... )
>>>
>>> from creek.base import Creek
>>>
>>> class MyCreek(Creek):
... def data_to_obj(self, line):
... return [x.strip() for x in line.strip().split(',')]
...
>>> stream = MyCreek(src)
>>>
>>> list(stream)
[['a', 'b', 'c'], ['1', '2', '3'], ['4', '5', '6']]
```
If we try that again, we’ll get an empty list since the cursor is at the end.
```pycon
>>> list(stream)
[]
```
But if the underlying stream has a seek, so does the creek, so we can “rewind”
```pycon
>>> stream.seek(0)
0
```
```pycon
>>> list(stream)
[['a', 'b', 'c'], ['1', '2', '3'], ['4', '5', '6']]
```
You can also use `next` to get stream items one by one
```pycon
>>> stream.seek(0) # rewind again to get back to the beginning
0
>>> next(stream)
['a', 'b', 'c']
>>> next(stream)
['1', '2', '3']
```
Let’s add a filter! There’s two kinds you can use.
One that is applied to the line before the data is transformed by data_to_obj,
and the other that is applied after (to the obj).
```pycon
>>> from creek.base import Creek
>>> from io import StringIO
>>>
>>> src = StringIO(
... '''a, b, c
... 1,2, 3
... 4, 5,6
... ''')
>>> class MyFilteredCreek(MyCreek):
... def post_iter(self, objs):
... yield from filter(lambda obj: str.isnumeric(obj[0]), objs)
>>>
>>> s = MyFilteredCreek(src)
>>>
```
```pycon
>>> list(s)
[['1', '2', '3'], ['4', '5', '6']]
>>> s.seek(0)
0
>>> next(s)
['1', '2', '3']
>>> next(s)
['4', '5', '6']
```
Recipes:
- `pre_iter`: involving `itertools.islice` to skip header lines
- `pre_iter`: involving enumerate to get line indices in stream iterator
- `pre_iter = functools.partial(map, pre_proc_func)` to preprocess all stream
items with `pre_proc_func`
- `pre_iter`: include filter before obj
- `post_iter`: `chain.from_iterable` to flatten a chunked/segmented stream
- `post_iter`: `functools.partial(filter, condition)` to filter yielded objs
#### *static* data_to_obj(x)
Return `x` unchanged.
#### *static* post_iter(x)
Return `x` unchanged.
#### *static* pre_iter(x)
Return `x` unchanged.
#### *classmethod* wrap(obj)
Wrap `obj` in `cls`: a subclass wrapping instances at construction if `obj` is a class, else `cls(obj)`.
# _autosummary/creek.html.md
# creek
creek: a simple facade for streams.
Wrap a stream in a `Creek` and override `pre_iter`, `data_to_obj` or
`post_iter` to layer transformations on it; read an unbounded stream as if it
were a list with `InfiniteSeq`; index, window and segment streams with the
functions and classes of `creek.tools`.
Main entry points:
- `Creek`: the layer-able stream wrapper (`creek.base`)
- `InfiniteSeq, IndexedBuffer`: list-like views of unbounded streams (`creek.infinite_sequence`)
- `BufferStats, Segmenter`: rolling-window statistics and segmentation (`creek.tools`)
- `dynamically_index, filter_and_index_stream`: indexing streams (`creek.tools`)
```pycon
>>> from creek import Creek
>>> class Doubler(Creek):
... def data_to_obj(self, x):
... return x * 2
>>> list(Doubler([1, 2, 3]))
[2, 4, 6]
```
### Modules
| [`automatas`](_autosummary/creek.automatas.html.md#module-creek.automatas) | Automatas (finite state machines): step a state through a stream of symbols. |
|---------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------|
| [`base`](_autosummary/creek.base.html.md#module-creek.base) | The `Creek` base class: a layer-able wrapper around a stream. |
| [`infinite_sequence`](_autosummary/creek.infinite_sequence.html.md#module-creek.infinite_sequence) | List-like read access to an unbounded stream, through a bounded buffer. |
| [`labeling`](_autosummary/creek.labeling.html.md#module-creek.labeling) | Tools to label/annotate stream elements |
| [`multi_streams`](_autosummary/creek.multi_streams.html.md#module-creek.multi_streams) | Merge several sorted streams into one stream of `(stream_id, item)` pairs. |
| [`tools`](_autosummary/creek.tools.html.md#module-creek.tools) | Tools to index, window and segment streams. |
| [`util`](_autosummary/creek.util.html.md#module-creek.util) | Iterator, cursor and function-composition utilities used across creek. |
# _autosummary/creek.infinite_sequence.html.md
# creek.infinite_sequence
List-like read access to an unbounded stream, through a bounded buffer.
An `IndexedBuffer` keeps the last `buffer_len` items of a stream and lets you
index them with the positions they had in the stream. An `InfiniteSeq` pairs such
a buffer with an iterator, pulling items on demand, so that `s[i:j]` behaves as
if the whole stream were a list, as long as queries move forward and fit in the
buffer.
Main entry points:
- `InfiniteSeq`: an iterator plus a buffer, sliced like a list
- `IndexedBuffer`: the buffer alone, fed with `append` / `extend`
- `BufferedGetter`: a buffer queried with a filter function instead of indices
```pycon
>>> from itertools import count
>>> from creek.infinite_sequence import InfiniteSeq
>>> s = InfiniteSeq(count(), buffer_len=5)
>>> s[3:6]
[3, 4, 5]
```
### Functions
| [`absolute_item`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.absolute_item)(item, max_idx) | Returns an item with absolute references: i.e. with negative indices idx resolved to max_idx + idx. |
|--------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------|
| [`asis`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.asis)(obj) | Return `obj` unchanged (the default transform). |
| [`consume`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.consume)(gen, n) | Consume n iterations of generator (without returning elements) |
| [`new_type`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.new_type)(name, typ[, doc]) | Make a `typing.NewType` called `name` with `doc` as its docstring (`typ` is currently ignored). |
| [`none_safe_addition`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.none_safe_addition)(x, y) | Adds the two numbers if x is not None, or return None if not |
| [`shift_slice`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.shift_slice)(slice_obj, shift) | Return `slice_obj` with `start` and `stop` shifted by `shift` (`None` bounds stay `None`). |
| [`simple_interval_relationship`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.simple_interval_relationship)(x, y[, ...]) | Get the simple relationship between intervals x and y. |
| [`slice_args`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.slice_args)(slice_obj) | Return the `(start, stop, step)` of a slice. |
| [`validate_interval`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.validate_interval)(interval) | Asserts that input is a valid interval, raising a ValueError if not |
### Classes
| [`BufferedGetter`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.BufferedGetter)(buffer_len[, prefill, ...]) | `BufferedGetter` is intended to be a more general (but not optimized) class that offers a query-interface to a buffer, intended to be used when the buffer is being filled by a (possibly live) stream of data items. |
|---------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| [`ExceptionRaiserCallbackMixin`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.ExceptionRaiserCallbackMixin)(\*args, \*\*kwargs) | Make the instance callable and have the effect of raising the instance. |
| [`IndexedBuffer`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.IndexedBuffer)(buffer_len[, prefill, ...]) | A list-like object that gives a limited-past read view of an unbounded stream. |
| [`InfiniteSeq`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.InfiniteSeq)(iterator, buffer_len) | A list-like (read) view of an unbounded sequence/stream. |
| [`Relations`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.Relations)(\*values) | Point-interval and interval-interval relations. |
### Exceptions
| [`NotDuringError`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.NotDuringError)(\*args, \*\*kwargs) | IndexError that indicates that there was an attempt to index some data that is not contained in the buffer (i.e. is that a part of the request is NO LONGER, or NOT YET covered by the buffer). |
|------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| [`OverlapsFutureError`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.OverlapsFutureError)(\*args, \*\*kwargs) | IndexError that indicates that there was an attempt to index some data that is in the FUTURE (i.e. is NOT YET completely covered by the buffer). |
| [`OverlapsPastError`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.OverlapsPastError)(\*args, \*\*kwargs) | IndexError that indicates that there was an attempt to index some data that is in the PAST (i.e. is NO LONGER completely covered by the buffer). |
| [`RelationNotHandledError`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.RelationNotHandledError) | TypeError that indicates that a relation is either not a valid one, or not handled by conditional clause. |
### *class* creek.infinite_sequence.BufferedGetter(buffer_len, prefill=(), input_data_trans=, query_trans=, slice_get_postproc=)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
`BufferedGetter` is intended to be a more general (but not optimized) class that
offers a query-interface to a buffer, intended to be used when the buffer is
being filled by a (possibly live) stream of data items.
By contrast…
The `IndexedBuffer` is a particular case where the queries are slices and the index
that is sliced on is an enumeration one.
The `InfiniteSeq` is a class combining `IndexedBuffer` with a data source it can
pull data from (according to the demands of the query).
```pycon
>>> from creek.infinite_sequence import BufferedGetter
>>>
>>>
>>> b = BufferedGetter(20)
>>> b.extend([
... (1, 3, 'completely before'),
... (2, 4, 'still completely before (upper bounds are strict)'),
... (3, 6, 'partially before, but overlaps bottom'),
... (4, 5, 'totally', 'inside'), # <- note this tuple has 4 elements
... (5, 8), # <- note this tuple has only the minimum (2) elements,
... (7, 10, 'partially after, but overlaps top'),
... (8, 11, 'completely after (strict upper bound)'),
... (100, 101, 'completely after (obviously)')
... ])
>>> b[lambda x: 3 < x[0] < 8]
[(4, 5, 'totally', 'inside'),
(5, 8),
(7, 10, 'partially after, but overlaps top')]
```
#### append(x)
Transform `x` with `input_data_trans` and add it to the buffer.
* **Return type:**
[`None`](https://docs.python.org/3/builtins/constants.html#None)
#### clear()
Empty the buffer.
#### extend(iterable)
Append each item of `iterable`.
#### filter(filt)
Iterate over the buffer items for which `filt` is true.
#### ingress(x)
Return `x` unchanged.
* **Return type:**
`BufferItem` ([`type`](https://docs.python.org/3/builtins/functions.html#type))
### *class* creek.infinite_sequence.ExceptionRaiserCallbackMixin(\*args, \*\*kwargs)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
Make the instance callable and have the effect of raising the instance.
Meant to add to an exception class so that instances of this class can be used as callbacks that raise the error
### *class* creek.infinite_sequence.IndexedBuffer(buffer_len, prefill=(), if_overlaps_past=OverlapsPastError('Some of the data requested was in the past or in the future'), if_overlaps_future=OverlapsFutureError('Some of the data requested was in the past or in the future'), slice_get_postproc=)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
A list-like object that gives a limited-past read view of an unbounded stream.
For example, say we had the stream of increasing integers 0, 1, 2, …
that is being fed to an `IndexedBuffer`.
What `IndexedBuffer(buffer_len=4)` offers is access to the buffer’s contents,
but using the indices that the stream (if it were one big list in memory) would
use, instead of the buffer’s own indices:
```default
0 1 2 3 [4 5 6 7] 8 9
```
`IndexedBuffer` uses `collections.deque`, exposing the `append`, `extend`
and `clear` methods, updating the index reference under a lock.
* **Parameters:**
* **buffer_len** – How many of the most recent items are kept.
* **prefill** – Items the buffer starts with (`max_idx` still starts at 0).
* **if_overlaps_past** – Stored as an attribute; not consulted by item access,
which always raises `OverlapsPastError`.
* **if_overlaps_future** – Stored as an attribute; not consulted by item access,
which always raises `OverlapsFutureError`.
* **slice_get_postproc** ([`Callable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Callable)) – Applied to the `islice` of the buffer that a slice
request selects; `list` by default.
* **Raises:**
* [**OverlapsPastError**](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.OverlapsPastError) – On `s[item]` when some of the requested range is no
longer in the buffer.
* [**OverlapsFutureError**](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.OverlapsFutureError) – On `s[item]` when some of the requested range is not
yet in the buffer.
```pycon
>>> s = IndexedBuffer(buffer_len=4)
>>> s.extend(range(4)) # adding 4 elements in bulk (filling the buffer completely)
>>> list(s)
[0, 1, 2, 3]
>>> s[2]
2
>>> s[1:2]
[1]
>>> s[1:1]
[]
```
Let’s add two more elements (using append this time), making the buffer “shift”
```pycon
>>> s.append(4)
>>> s.append(5)
>>> list(s)
[2, 3, 4, 5]
>>> s[2]
2
>>> s[5]
5
>>> s[2:5]
[2, 3, 4]
>>> s[3:6]
[3, 4, 5]
>>> assert s[2:6] == list(range(2, 6))
```
You can slice with step:
```pycon
>>> s[2:6:2]
[2, 4]
```
You can slice with negatives
```pycon
>>> s[2:-2]
[2, 3]
```
On the other hand, if you ask for something that is not in the buffer (anymore, or yet), you’ll get an
error that tells you so:
```pycon
>>> # element for idx 1 is missing in [2, 3, 4, 5]
>>> s[1:4]
Traceback (most recent call last):
...
OverlapsPastError: You asked for slice(1, 4, None), but the buffer only contains the index range: 2:6
```
```pycon
>>> # elements for 0:2 are missing (as well as 6:9, but OverlapsPastError trumps OverlapsFutureError
>>> s[0:9]
Traceback (most recent call last):
...
OverlapsPastError: You asked for slice(0, 9, None), but the buffer only contains the index range: 2:6
```
```pycon
>>> # element for 6:9 are missing in [2, 3, 4, 5]
>>> s[4:9]
Traceback (most recent call last):
...
OverlapsFutureError: You asked for slice(4, 9, None), but the buffer only contains the index range: 2:6
```
#### append(x)
Add one item, advancing `max_idx` by 1.
* **Return type:**
[`None`](https://docs.python.org/3/builtins/constants.html#None)
#### clear()
Empty the buffer and reset `max_idx` to 0.
#### extend(iterable)
Extend buffer with an iterable of items
* **Return type:**
[`None`](https://docs.python.org/3/builtins/constants.html#None)
#### *property* min_idx
The stream index of the oldest item still in the buffer.
#### outer_to_buffer_idx(idx)
Translate a stream index (int, slice, or iterable of ints) into the buffer’s own index.
### *class* creek.infinite_sequence.InfiniteSeq(iterator, buffer_len)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
A list-like (read) view of an unbounded sequence/stream.
It is the combination of `IndexedBuffer` and an iterator that will be used to
source the buffer according to the slices that are requested.
If a slice is requested whose data is “in the future”, the iterator will be
consumed until the buffer can satisfy that request.
If the requested slice has any part of it that is “in the past”, that is,
has already been iterated through and is not in the buffer anymore, a
`OverlapsPastError` will be raised.
Therefore, `InfiniteSeq` is meant for ordered slice queries of size no more than
the buffer size.
If these conditions are satisfied, an `InfiniteSeq` will behave (with `i:j`
queries) as if it were one long list in memory.
Can be used with a live stream of data as long as the buffer size is big enough
to handle the data production and query rates.
* **Parameters:**
* **iterator** ([`Iterator`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Iterator)) – The source of items; consumed forward only, as far as queries
require.
* **buffer_len** ([`int`](https://docs.python.org/3/builtins/functions.html#int)) – How many of the most recent items stay available.
* **Raises:**
[**OverlapsPastError**](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.OverlapsPastError) – On `s[item]` if some of the requested range has already
left the buffer.
For example, take an iterator that cycles from 0 to 99 forever:
```pycon
>>> from itertools import cycle
>>> iterator = cycle(range(100))
```
Let’s make an `InfiniteSeq` instance for this stream, accomodating for a view of
up to 11 items.
```pycon
>>> s = InfiniteSeq(iterator, buffer_len=11)
```
Let’s ask for element 15 (which is the (15 + 1)th element (and should have a value of 15).
```pycon
>>> s[15]
15
```
Now, to get this value, the iterator will move forward up to that point;
that is, until the buffer’s head (i.e. most recent) item contains that requested (15 + 1)th element.
But the buffer is of size 11, so we still have access to a few previous elements:
```pycon
>>> s[11]
11
>>> s[5:15]
[5, 6, 7, 8, 9, 10, 11, 12, 13, 14]
```
But if we asked for anything before index 5…
```pycon
>>> s[2:7]
Traceback (most recent call last):
...
OverlapsPastError: You asked for slice(2, 7, None), but the buffer only contains the index range: 5:16
```
So we can’t go backwards. But we can always go forwards:
```pycon
>>> s[95:105]
[95, 96, 97, 98, 99, 0, 1, 2, 3, 4]
```
You can also use slices with step and with negative integers (referencing the head of the buffer)
```pycon
>>> s[120:130:2]
[20, 22, 24, 26, 28]
>>> s[120:130]
[20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
>>> s[-8:-2]
[22, 23, 24, 25, 26, 27]
```
but you cannot slice farther back than the buffer
```pycon
>>> try:
... s[-20:-2]
... except OverlapsPastError as e:
... msg_text = str(e)
>>> print(msg_text)
You asked for slice(110, 128, None), but the buffer only contains the index range: 119:130
```
Sometimes the source provides data in chunks. Sometimes these chunks are not even of fixed size.
In those situations, you can use `itertools.chain` to “flatten” the iterator as in the following example:
```pycon
>>> from creek.infinite_sequence import InfiniteSeq
>>> from typing import Mapping
>>>
>>> class Source(Mapping):
... n = 100
...
... __len__ = lambda self: self.n
...
... def __iter__(self):
... yield from range(self.n)
...
... def __getitem__(self, k):
... print(f"Asking for {k}")
... return list(range(k * 10, (k + 1) * 10))
...
>>>
>>> source = Source()
>>>
```
See that when we ask for a chunk of data, there’s a print notification about it.
```pycon
>>> assert source[3] == [30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
Asking for 3
```
Now let’s make an iterator of the data and an InfiniteSeq (with buffer length 10) on top of it.
```pycon
>>> from itertools import chain
>>> iterator = chain.from_iterable(source.values())
>>> s = InfiniteSeq(iterator, 10)
```
See that when you ask for :5, you see that chunk 0 is requested.
```pycon
>>> s[:5]
Asking for 0
[0, 1, 2, 3, 4]
```
If you ask for something that’s already in the buffer, you won’t see the print notification though.
```pycon
>>> s[4:8]
[4, 5, 6, 7]
```
The following shows you how InfiniteSeq “hits” the data source as it’s getting the data it needs for the request.
```pycon
>>> s[8:12]
Asking for 1
[8, 9, 10, 11]
>>>
>>> s[40:42]
Asking for 2
Asking for 3
Asking for 4
[40, 41]
```
### *exception* creek.infinite_sequence.NotDuringError(\*args, \*\*kwargs)
Bases: [`ExceptionRaiserCallbackMixin`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.ExceptionRaiserCallbackMixin), [`IndexError`](https://docs.python.org/3/builtins/exceptions.html#IndexError)
IndexError that indicates that there was an attempt to index some data that is not contained in the buffer
(i.e. is that a part of the request is NO LONGER, or NOT YET covered by the buffer)
### *exception* creek.infinite_sequence.OverlapsFutureError(\*args, \*\*kwargs)
Bases: [`NotDuringError`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.NotDuringError)
IndexError that indicates that there was an attempt to index some data that is in the FUTURE
(i.e. is NOT YET completely covered by the buffer)
### *exception* creek.infinite_sequence.OverlapsPastError(\*args, \*\*kwargs)
Bases: [`NotDuringError`](_autosummary/creek.infinite_sequence.html.md#creek.infinite_sequence.NotDuringError)
IndexError that indicates that there was an attempt to index some data that is in the PAST
(i.e. is NO LONGER completely covered by the buffer)
### *exception* creek.infinite_sequence.RelationNotHandledError
Bases: [`TypeError`](https://docs.python.org/3/builtins/exceptions.html#TypeError)
TypeError that indicates that a relation is either not a valid one, or not handled by conditional clause.
### *class* creek.infinite_sequence.Relations(\*values)
Bases: [`Enum`](https://docs.python.org/3/library/enum.html#enum.Enum)
Point-interval and interval-interval relations.
See Allen’s interval algebra for (some of the) interval relations
([https://en.wikipedia.org/wiki/Allen%27s_interval_algebra](https://en.wikipedia.org/wiki/Allen%27s_interval_algebra)).
### creek.infinite_sequence.absolute_item(item, max_idx)
Returns an item with absolute references: i.e. with negative indices idx
resolved to max_idx + idx
```pycon
>>> absolute_item(-1, 10)
9
>>> absolute_item(slice(-4, -2, 2), 10)
slice(6, 8, 2)
```
But anything else that’s not a slice or int will be left untouched
(and will probably result in errors if you use with IndexedBuffer)
```pycon
>>> absolute_item((-7, -2), 10)
(-7, -2)
```
### creek.infinite_sequence.asis(obj)
Return `obj` unchanged (the default transform).
### creek.infinite_sequence.consume(gen, n)
Consume n iterations of generator (without returning elements)
### creek.infinite_sequence.new_type(name, typ, doc=None)
Make a `typing.NewType` called `name` with `doc` as its docstring (`typ` is currently ignored).
### creek.infinite_sequence.none_safe_addition(x, y)
Adds the two numbers if x is not None, or return None if not
### creek.infinite_sequence.shift_slice(slice_obj, shift)
Return `slice_obj` with `start` and `stop` shifted by `shift` (`None` bounds stay `None`).
### creek.infinite_sequence.simple_interval_relationship(x, y, above_bt=, below_tt=)
Get the simple relationship between intervals x and y.
* **Parameters:**
* **x** ([`tuple`](https://docs.python.org/3/builtins/stdtypes.html#tuple)[`Union`[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)], `Union`[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)]]) – A point (a number), an interval (a 2-tuple of numbers), or a slice.
* **y** ([`tuple`](https://docs.python.org/3/builtins/stdtypes.html#tuple)[`Union`[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)], `Union`[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)]]) – An interval; a 2-tuple of numbers.
* **above_bt** ([`Callable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Callable)) – `above_bt(x_bt, y_bt)` boolean function (`ge` or `gt`)
deciding if x starts after y does.
* **below_tt** ([`Callable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Callable)) – `below_tt(x_tt, y_tt)` boolean function (`lt` or `le`)
deciding if x ends before y does.
* **Returns:**
One of three relations:
`Relations.BEFORE` if some of x is below y,
`Relations.AFTER` if some of x is after y,
`Relations.DURING` if x is entirely within y
* **Raises:**
[**ValueError**](https://docs.python.org/3/builtins/exceptions.html#ValueError) – If `y` (or an interval `x`) is not a 2-tuple with `bt <= tt`.
The target `y` interval is expressed only by its bounds, but we don’t know if
these are inclusive or not. The `above_bt` and `below_tt` arguments let us
express that, by defining what “below the lowest (bt) bound” and “above the
highest (tt) bound” mean.
The function is meant to be curried (partial), for example:
```pycon
>>> from functools import partial
>>> from operator import le, lt, ge, gt
>>> default = simple_interval_relationship # uses above_bt=ge, below_tt=lt
>>> including_bounds = partial(simple_interval_relationship, above_bt=ge, below_tt=le)
>>> excluding_bounds = partial(simple_interval_relationship, above_bt=gt, below_tt=lt)
```
Take `(4, 8)` as the target interval, and want to query the relationship of other
points and intervals with it.
No matter what the function is, they will always agree on any intervals that don’t
share any bounds.
```pycon
>>> for relation_func in (default, including_bounds, excluding_bounds):
... print (
... relation_func(3, (4, 8)),
... relation_func(5, (4, 8)),
... relation_func(9, (4, 8)),
... relation_func((3, 7), (4, 8)),
... relation_func((5, 7), (4, 8)),
... relation_func((7, 9), (4, 8))
... )
Relations.BEFORE Relations.DURING Relations.AFTER Relations.BEFORE Relations.DURING Relations.AFTER
Relations.BEFORE Relations.DURING Relations.AFTER Relations.BEFORE Relations.DURING Relations.AFTER
Relations.BEFORE Relations.DURING Relations.AFTER Relations.BEFORE Relations.DURING Relations.AFTER
```
But if the two intervals share some bounds, these functions will diverge.
```pycon
>>> for relation_func in (default, including_bounds, excluding_bounds):
... print (
... relation_func(4, (4, 8)),
... relation_func(8, (4, 8)),
... relation_func((4, 7), (4, 8)),
... relation_func((4, 8), (4, 8)),
... relation_func((5, 8), (4, 8))
... )
Relations.DURING Relations.AFTER Relations.DURING Relations.AFTER Relations.AFTER
Relations.DURING Relations.DURING Relations.DURING Relations.DURING Relations.DURING
Relations.BEFORE Relations.AFTER Relations.BEFORE Relations.BEFORE Relations.AFTER
```
The function can be used with the FIRST argument being a slice object as well.
This can then be used to enable [i:j] access.
```pycon
>>> for relation_func in (default, including_bounds, excluding_bounds):
... print (
... relation_func(slice(4, 7), (4, 8)),
... relation_func(slice(4, 8), (4, 8)),
... relation_func(slice(5, 8), (4, 8))
... )
Relations.DURING Relations.AFTER Relations.AFTER
Relations.DURING Relations.DURING Relations.DURING
Relations.BEFORE Relations.BEFORE Relations.AFTER
```
### creek.infinite_sequence.slice_args(slice_obj)
Return the `(start, stop, step)` of a slice.
### creek.infinite_sequence.validate_interval(interval)
Asserts that input is a valid interval, raising a ValueError if not
# _autosummary/creek.labeling.html.md
# creek.labeling
Tools to label/annotate stream elements
The motivating example is the case of an incoming stream
that we need to segment, according to the detection of an event.
For example, take a stream of integers and detect the event “multiple of 5”:
```python
1->2->3->4->'multiple of 5'->6->7->...
```
When the stream is “live”, we don’t want to process it immediately, but instead
we prefer to annotate it on the fly, by adding some metadata to it.
The simplest addition of metadata information could look like:
```python
3->4->('multiple of 5', 5) -> 6 -> ...
```
One critique of using tuples to contain both annotated (here, `5`) and annotation (here, `'multiple of 5'`)
is that the semantics aren’t explicit.
The fact that the original element was annotated the distinction of annotation and annotated is based on an
implicit convention.
This is not too much of a problem here, but becomes unwieldy in more complex situations, for example,
if we want to accommodate for multiple labels.
A `LabeledElement` `x` has an attribute `x.element`,
and a container of labels `x.labels` (list, set or dict).
Multiple labels can be used to segment streams into overlapping segments.
```python
(group0)->(group0)->(group0, group1)->(group0, group1)-> (group1)->(group1)->...
```
### Functions
| [`label_element`](_autosummary/creek.labeling.html.md#creek.labeling.label_element)(elem, label, labeled_element_cls) | Label `element` with `label` (or add this label to the existing labels). |
|----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------|
### Classes
| [`DictLabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.DictLabeledElement)(element) | A LabeledElement that uses a `dict` as the labels container. |
|--------------------------------------------------------------------------------|-------------------------------------------------------------------------------------|
| [`LabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.LabeledElement)(element) | Abstract class to label elements -- that is, associate some metadata to an element. |
| [`ListLabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.ListLabeledElement)(element) | A LabeledElement that uses a `list` as the labels container. |
| [`SetLabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.SetLabeledElement)(element) | A LabeledElement that uses a `set` as the labels container. |
### *class* creek.labeling.DictLabeledElement(element)
Bases: [`LabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.LabeledElement)
A LabeledElement that uses a `dict` as the labels container.
Use this when you need to keep labels classified and have quick access to the
a specific class of labels.
Note that when adding a label, you need to specify it as a `{key: val, ...}`
`dict`, the keys being the (hashable) label kinds,
and the vals being the values for those kinds.
```pycon
>>> x = DictLabeledElement(42).add_label({'string': 'forty-two'})
>>> x.element
42
>>> x.labels
{'string': 'forty-two'}
>>> x.add_label({'type': 'number', 'prime': False})
DictLabeledElement(42)
>>> x.element
42
>>> assert x.labels == {'string': 'forty-two', 'type': 'number', 'prime': False}
```
#### *static* add_new_label(labels, label)
Merge the `label` dict into `labels`.
#### mk_new_labels_container
alias of [`dict`](https://docs.python.org/3/builtins/stdtypes.html#dict)
### *class* creek.labeling.LabeledElement(element)
Bases: [`ABC`](https://docs.python.org/3/library/abc.html#abc.ABC)
Abstract class to label elements – that is, associate some metadata to an element.
To make a concrete LabeledElement, one must subclass LabeledElement and provide
- a `mk_new_labels_container`, a `LabelFactory`, which is a callable that takes
no input and returns a new empty labels container
- an `add_new_label`, an `AddLabel`, a `(Labels, Label)` callable that adds
a single label to the labels container.
#### add_label(label)
Add `label` to this element’s labels and return `self`.
#### *abstractmethod static* add_new_label(labels, label)
Add `label` to `labels` in place.
#### *abstractmethod static* mk_new_labels_container(self)
Return a new, empty labels container.
* **Return type:**
[`Iterable`](https://docs.python.org/3/library/collections.abc.html#collections.abc.Iterable)[`Label` ([`Any`](https://docs.python.org/3/library/typing.html#typing.Any))]
### *class* creek.labeling.ListLabeledElement(element)
Bases: [`LabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.LabeledElement)
A LabeledElement that uses a `list` as the labels container.
Use this when you need to use unhashable labels, or label insertion order matters,
or don’t need fast `label in labels` checks or label deduplication.
```pycon
>>> x = ListLabeledElement(42).add_label('forty-two')
>>> x.element
42
>>> x.labels
['forty-two']
>>> x.add_label('number')
ListLabeledElement(42)
>>> x.element
42
>>> assert x.labels == ['forty-two', 'number']
```
#### *static* add_new_label(self, object,)
Append object to the end of the list.
#### mk_new_labels_container
alias of [`list`](https://docs.python.org/3/builtins/stdtypes.html#list)
### *class* creek.labeling.SetLabeledElement(element)
Bases: [`LabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.LabeledElement)
A LabeledElement that uses a `set` as the labels container.
Use this when you want to get fast `label in labels` check and/or maintain the
labels unduplicated.
Note that since `set` is the container, the labels will have to be hashable.
```pycon
>>> x = SetLabeledElement(42).add_label('forty-two')
>>> x.element
42
>>> x.labels
{'forty-two'}
>>> x.add_label('number')
SetLabeledElement(42)
>>> x.element
42
>>> assert x.labels == {'forty-two', 'number'}
```
#### *static* add_new_label()
Add an element to a set.
This has no effect if the element is already present.
#### mk_new_labels_container
alias of [`set`](https://docs.python.org/3/builtins/stdtypes.html#set)
### creek.labeling.label_element(elem, label, labeled_element_cls)
Label `element` with `label` (or add this label to the existing labels).
The `labeled_element_cls`, the `LabeledElement` class to use to label the element,
is meant to be “partialized out”, like this:
```pycon
>>> from functools import partial
>>> from creek.labeling import DictLabeledElement
>>> my_label_element = partial(label_element, labeled_element_cls=DictLabeledElement)
>>> # and then just use my_label_element(elem, label) to label elem
```
You’ll probably often want to use `DictLabeledElement`, because, for example:
```text
{'n_channels': 2, 'phase', 2, 'session': 16987485}
```
is a lot easier (and less dangerous) to use then, say:
```text
[2, 2, 16987485]
```
But there are cases where, say:
```pycon
>>> from creek.labeling import SetLabeledElement
>>> my_label_element = partial(label_element, labeled_element_cls=SetLabeledElement)
>>> x = my_label_element(42, 'divisible_by_seven')
>>> _ = my_label_element(x, 'is_a_number')
>>> 'divisible_by_seven' in x # equivalent to 'divisible_by_seven' in x.labels
True
>>> x.labels.issuperset({'is_a_number', 'divisible_by_seven'})
True
```
is more convenient to use then using a dict with boolean values to do the same
* **Parameters:**
* **elem** (`Union`[`Element` ([`Any`](https://docs.python.org/3/library/typing.html#typing.Any)), [`LabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.LabeledElement)]) – The element that is being labeled
* **label** (`Label` ([`Any`](https://docs.python.org/3/library/typing.html#typing.Any))) – The label to add to the element
* **labeled_element_cls** – The `LabeledElement` class to use to label the element
* **Return type:**
[`LabeledElement`](_autosummary/creek.labeling.html.md#creek.labeling.LabeledElement)
* **Returns:**
The labeled element: a new `labeled_element_cls` wrapping `elem`, or
`elem` itself (with the label added) if it already is a `labeled_element_cls`.
# _autosummary/creek.multi_streams.html.md
# creek.multi_streams
Merge several sorted streams into one stream of `(stream_id, item)` pairs.
Main entry points:
- `MergedStreams`: the merge, driven by `heapq.merge` on a sort key of the items
- `multi_stream_items`: tag each stream’s items with its id, one iterable per stream
```pycon
>>> from creek.multi_streams import MergedStreams
>>> list(MergedStreams({'a': [1, 3], 'b': [2]}))
[('a', 1), ('b', 2), ('a', 3)]
```
### Functions
| [`multi_stream_items`](_autosummary/creek.multi_streams.html.md#creek.multi_streams.multi_stream_items)(streams_map) | Provides a iterable of (k1, v1_1), (k1, v1_2), ... |
|-----------------------------------------------------------------------------------------|------------------------------------------------------|
| [`transform_methods`](_autosummary/creek.multi_streams.html.md#creek.multi_streams.transform_methods)(cls[, method_trans]) | Applies method_trans to all the methods of `cls` |
### Classes
| [`MergedStreams`](_autosummary/creek.multi_streams.html.md#creek.multi_streams.MergedStreams)(streams_map[, sort_key]) | Creates an iterable of `(stream_id, stream_item)` pairs from a stream Mapping, that is, `{stream_id: stream, ...}`: A sort of "flattening" of the Mapping. |
|-------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------|
| [`SortKeys`](_autosummary/creek.multi_streams.html.md#creek.multi_streams.SortKeys)() | A few `itemgetter` sort keys, exposed as staticmethods. |
### *class* creek.multi_streams.MergedStreams(streams_map, sort_key=None)
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
Creates an iterable of `(stream_id, stream_item)` pairs from a stream Mapping,
that is, `{stream_id: stream, ...}`: A sort of “flattening” of the Mapping.
This can be useful, for instance, if you want to make “slabs” of data, gathering
together all the data for a given time period, from multiple streams.
The `stream_item` will be yielded in sorted order.
Sort behavior can be modified by the `sort_key` argument which behaves like `key`
arguments of built-in like `sorted`, `heapq.merge`, `itertools.groupby`, etc.
If given, the `sort_key` function applies to `stream_item` (not to `stream_id`).
#### IMPORTANT
To function as expected, the streams should be already sorted (according
to the `sort_key` order).
The cannonical use case of this function is to “flatten”, or “weave together”
multiple streams of timestamped data. We’re given several streams that provide
`(timestamp, data)` items (where timestamps arrive in order within each stream)
and we get a single stream of `(stream_id, (timestamp, data))` items where
the `timestamp` values are yielded in sorted order.
The following example uses a dict pointing to a fixed-size list as the `stream_map`
but in general the `stream_map` will be a `Mapping` (not necessarily a dict)
whose values are potentially bound-less streams.
```pycon
>>> streams_map = {
... 'hello': [(2, 'two'), (3, 'three'), (5, 'five')],
... 'world': [(0, 'zero'), (1, 'one'), (3, 'three'), (6, 'six')]
... }
>>> streams_items = MergedStreams(streams_map)
>>> it = iter(streams_items)
>>> list(it)
[('world', (0, 'zero')),
('world', (1, 'one')),
('hello', (2, 'two')),
('hello', (3, 'three')),
('world', (3, 'three')),
('hello', (5, 'five')),
('world', (6, 'six'))]
```
### *class* creek.multi_streams.SortKeys
Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)
A few `itemgetter` sort keys, exposed as staticmethods.
### creek.multi_streams.multi_stream_items(streams_map)
Provides a iterable of (k1, v1_1), (k1, v1_2), …
```pycon
>>> streams_map = {'hello': 'abc', 'world': [1, 2]}
>>> hello_items, world_items = multi_stream_items(streams_map)
>>> list(hello_items)
[('hello', 'a'), ('hello', 'b'), ('hello', 'c')]
>>> list(world_items)
[('world', 1), ('world', 2)]
```
### creek.multi_streams.staticmethods(cls, \*, method_trans=)
Applies method_trans to all the methods of `cls`
```pycon
>>> from functools import partial
>>> staticmethods = partial(transform_methods, method_trans=staticmethod)
```
Now staticmethods is a class decorator that can be used to make all methods
be defined as staticmethods in bulk
```pycon
>>> @staticmethods
... class C:
... foo = lambda x: x + 1
... bar = lambda y: y * 2
>>> c = C()
>>> c.foo(6)
7
>>> c.bar(6)
12
```
### creek.multi_streams.transform_methods(cls, method_trans=)
Applies method_trans to all the methods of `cls`
```pycon
>>> from functools import partial
>>> staticmethods = partial(transform_methods, method_trans=staticmethod)
```
Now staticmethods is a class decorator that can be used to make all methods
be defined as staticmethods in bulk
```pycon
>>> @staticmethods
... class C:
... foo = lambda x: x + 1
... bar = lambda y: y * 2
>>> c = C()
>>> c.foo(6)
7
>>> c.bar(6)
12
```
# _autosummary/creek.tools.html.md
# creek.tools
Tools to index, window and segment streams.
Everything here is a plain function or a small callable class that you compose
with `map`, `filter` and `functools.partial`; nothing depends on `Creek`.
Main entry points:
- `filter_and_index_stream`: index a stream (`enumerate` by default) and filter on the items
- `dynamically_index / DynamicIndexer`: `enumerate` where you choose how the index advances
- `BufferStats`: a fixed-size buffer that returns a statistic of its contents each time it is fed
- `Segmenter`: feed a `BufferStats` and hand its contents to a callback after each value
```pycon
>>> from creek.tools import BufferStats
>>> list(map(BufferStats(maxlen=3, func=sum), range(5)))
[0, 1, 3, 6, 9]
```
### Functions
| [`alt_dynamically_index`](_autosummary/creek.tools.html.md#creek.tools.alt_dynamically_index)([idx_updater, start]) | Alternative to dynamically_index using itertools and partial |
|----------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| [`always_true`](_autosummary/creek.tools.html.md#creek.tools.always_true)(x) | Return `True`, whatever `x` is. |
| [`apply_and_fanout`](_autosummary/creek.tools.html.md#creek.tools.apply_and_fanout)(seq, func, idx) | Apply function (that returns an Iterable) to an element of a sequence and fanout (broadcast) the resulting items, to produce an iterable of tuples each containing one of these items along with 'a copy' of the other tuple elements |
| [`apply_func_to_index`](_autosummary/creek.tools.html.md#creek.tools.apply_func_to_index)(seq, apply_to_idx, func) | Apply `func` to the element at `apply_to_idx` of `seq`, returning a tuple. |
| [`count_increments`](_autosummary/creek.tools.html.md#creek.tools.count_increments)(current_idx, obj[, step]) | Index updater that ignores `obj` and advances the index by `step`. |
| [`current_time`](_autosummary/creek.tools.html.md#creek.tools.current_time)(current_idx, obj) | Doesn't even look at current_idx or obj. |
| [`dynamically_index`](_autosummary/creek.tools.html.md#creek.tools.dynamically_index)(iterable[, start, idx_updater]) | Generalization of `enumerate(iterable)` that allows one to specify how the indices should be updated. |
| [`fanout_and_flatten`](_autosummary/creek.tools.html.md#creek.tools.fanout_and_flatten)(iterable_of_seqs, func, idx) | Apply apply_and_fanout to an iterable of sequences. |
| [`fanout_and_flatten_dicts`](_autosummary/creek.tools.html.md#creek.tools.fanout_and_flatten_dicts)(iterable_of_dicts, ...) | Apply apply_and_fanout to an iterable of dicts. |
| [`filter_and_index_stream`](_autosummary/creek.tools.html.md#creek.tools.filter_and_index_stream)(stream, data_item_filt) | Index a stream and filter it (based only on the data items). |
| [`is_not_none`](_autosummary/creek.tools.html.md#creek.tools.is_not_none)(x) | Return `True` if `x` is not `None`. |
| [`return_buffer_on_stats_condition`](_autosummary/creek.tools.html.md#creek.tools.return_buffer_on_stats_condition)(stats, buffer) | Return `buffer` if `cond(stats)` holds, else `else_val`. |
| [`segment_overlaps`](_autosummary/creek.tools.html.md#creek.tools.segment_overlaps)(bt_tt_segment, query_bt, ...) | Returns True if, and only if, bt_tt_segment overlaps query interval. |
| [`size_increments`](_autosummary/creek.tools.html.md#creek.tools.size_increments)(current_idx, obj[, size_func]) | Index updater that advances the index by `size_func(obj)`, `len` by default. |
### Classes
| [`BufferStats`](_autosummary/creek.tools.html.md#creek.tools.BufferStats)([values, maxlen, func, add_new_val]) | A callable, fixed-size (fifo) buffer returning a statistic of its contents on every call. |
|-----------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------|
| [`DynamicIndexer`](_autosummary/creek.tools.html.md#creek.tools.DynamicIndexer)([start, idx_updater]) | Turn data items into `(index, item)` pairs, with a pluggable index update rule. |
| [`Segmenter`](_autosummary/creek.tools.html.md#creek.tools.Segmenter)(buffer[, stats_buffer_callback]) | Feed values to a `BufferStats` and hand its contents to a callback after each one. |
### *class* creek.tools.BufferStats(values=(), maxlen=