> built 2026-10-20 11:28 UTC from 07dc7cf (master) · stream2py 1.0.48. Details: build_info.json

# index.html.md

<!-- generated by epythet -->

[Documentation hosted here](https://i2mint.github.io/stream2py/index.html).

To install: `pip install stream2py`

# stream2py

Bring data streams to python, with ease.

One of the goals of the suite of i2i tools is to get from idea 2 implementation without all the fuss.
We’ve got py2store to do that for the storage (reading or writing) concern,
and others (e.g. py2cli, py2ws, py2dash) to take care of exposing python functions to the world (of command line interfaces, webservices, browser dashboards, etc.).

Here, we address the stream acquisition concern.
As always, we aim at offering as-simple-as-drawing-a-simple-drawing means to get things done.

## Plugins

`stream2py` has the core functionality, and is completely dependency-free (only builtin python).

That said, to work with specific stream sources, you can install plugins that will allow you to work with them.
At the time of writing this we have:

- Audio: [audiostream2py](https://github.com/i2mint/audiostream2py)
- Keyboard inputs: [keyboardstream2py](https://github.com/i2mint/keyboardstream2py)
- PLC (Programmable Logic Controller): [plcstream2py](https://github.com/i2mint/plcstream2py)
- Computer health stats: [pchealthstream2py](https://github.com/i2mint/pchealthstream2py)
- Video: [videostream2py](https://github.com/i2mint/videostream2py)
- Flask: [flaskstream2py](https://github.com/i2mint/flaskstream2py)

# What `stream2py` is for

## Reduce vocabulary entropy

One way we do this is by reducing the vocabulary entropy:
We don’t want to have to think about how every specific source calls a read, or a size,
or a time to pause before reads,
or what format THAT particular sensor is encoding it’s data in,
having you shuffle through documentation pages before you can figure out how to start doing
the fun stuff,
which happens to be the stuff that actually produces value. And, oh, once you figure it out,
if you don’t use it for a few months or years, next time you need to do something similar, you’ll have to figure it all out again.

No. That’s just a waste of time of time. Instead, we say you do that at most once.
You don’t have to do it at all if the community (us) already provided you with the their-language-to-our-consistent-language adapter for the stream you want to hook into. And if it’s something new, well you’ll have to figure it out, but you then write the adapter once, and now you (1) can use the rest of stream2py’s tools and (2) you don’t have to do it again.

## Go back in time

We also address the problem of impermanence.

Think of the streams that different sensors such as audio, vibration, video offer, or even “industrial” signals such as wifi, can bus data, PLC, etc. They happen, and they’re gone. Sure, they usually have buffers, but these are typically just big enough to get the data from high frequency reads – not enough to have the time for some more involved analysis that smart systems require.

We address this problem by

## Multi readers

It often happens that you want to do more than one thing with a stream.
Say store it, visualize it in real time, and direct it to a analysis pipeline.
In order for this to be possible with no hiccups, some things need to be taken care of.
We did, so you don’t have to.

## Timestamp correctly

In our extensive experience with people (the write code to store stream data),
we’ve noticed that many engineers, when faced with the task to timestamp the segments of
stream that they’re saving, follow a design pattern that goes like this
(a) get the stream data
(b) ask the system what date/time it is
(c) use that (and perhaps, just to make even more likely for the timestamp to be interpreted
incorrectly, call it the “offset_date”)

The problem with this design pattern is that it’s all pattern and no design.
It is **not** the timestamp of the beginning of the segment:
That time happened **after** the **end** of the event of the end of the segment occurred,
and even more so, **after** the system that will timestamp and store.
Further, there is a lot of wiggle room in the delay accumulated between the actual event,
and the moment we ask the system what time it is.
Sometimes it doesn’t matter, but sometimes it does:
For example, if we want to align with some other timestamped data,
or use these timestamps to determine if there’s gaps or overlaps between the segments
we’ve acquired.

Point is, stream2py will give you the tools to tackle that problem properly.
It does so by having the stream2py buffers mentioned above keep data flow statistics that readers can then use to
more precisely timestamp what they read.

[Documentation here](https://i2mint.github.io/stream2py/index.html).

<p class="epythet-aggregates">This documentation as a single file: <a href="stream2py.md">stream2py.md</a> (Markdown, for agents).</p>


# _autosummary/stream2py.buffer_reader.html.md

# stream2py.buffer_reader

A BufferReader gives data access to any number of consumers and provides methods to seek data such
as next(), range(), head(), tail().  Each BufferReader instance has it’s own cursor keeping track of
what data was last seen.

### Classes

| [`BufferReader`](_autosummary/stream2py.buffer_reader.html.md#stream2py.buffer_reader.BufferReader)(buffer, source_reader_info, ...)   | Reader that is constructed from StreamBuffer.mk_reader()   |
|--------------------------------------------------------------------------------------------------|------------------------------------------------------------|

### *class* stream2py.buffer_reader.BufferReader(buffer, source_reader_info, stop_event, , blocking=False, timeout=None, read_size=1, peek=False, strict_n=False, ignore_no_item_found=False)

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)

Reader that is constructed from StreamBuffer.mk_reader()

```pycon
>>> from stream2py import StreamBuffer, BufferReader
>>> from stream2py.examples.source_reader import SimpleCounterString
>>>
>>> source_reader = SimpleCounterString(start=0, stop=100)
>>> stream_buffer = StreamBuffer(source_reader=source_reader, maxlen=100)
>>> stream_buffer.start()
>>>
>>> buffer_reader = stream_buffer.mk_reader()
>>> isinstance(buffer_reader, BufferReader)
True
>>> buffer_reader.is_stopped
False
>>> buffer_reader.last_item == None  # None until value is seen
True
>>> buffer_reader.next() # get first item
's0'
>>> buffer_reader.last_item # check last_item cursor
's0'
>>> # next will not update last_item cursor
>>> buffer_reader.next(peek=True), buffer_reader.last_item
('s1', 's0')
>>> # same as what was peeked but now cursor is updated
>>> buffer_reader.next(), buffer_reader.last_item
('s1', 's1')
>>> buffer_reader.range(start=5, stop=10)
['s5', 's6', 's7', 's8', 's9', 's10']
>>> buffer_reader.last_item
's10'
>>> buffer_reader.range(start=5, stop=10, step=3), buffer_reader.last_item
(['s5', 's8'], 's8')
>>> buffer_reader.range(start=10, stop=15, peek=True)
['s10', 's11', 's12', 's13', 's14', 's15']
>>> buffer_reader.last_item # unchanged with peek
's8'
>>> buffer_reader.last_key # key of last_item
8
>>> # only 9 is greater than last item key
>>> buffer_reader.range(start=0, stop=9, only_new_items=True)
['s9']
>>> buffer_reader.head(), buffer_reader.last_item, buffer_reader.last_key
('s0', 's0', 0)
>>> buffer_reader.tail(), buffer_reader.last_item, buffer_reader.last_key
('s99', 's99', 99)
>>>
>>> stream_buffer.stop()
>>>
>>> buffer_reader.is_stopped
True
```

#### is_same_buffer(other_buffer_reader)

Check if reader is looking at the same buffer

#### *property* is_stopped *: [bool](https://docs.python.org/3/builtins/functions.html#bool)*

Checks if stop event has been set by StreamBuffer.

* **Returns:**
  bool

#### *property* last_item

last seen item cursor

#### *property* last_key

key to last seen item cursor

#### range(start, stop, step=None, , peek=False, ignore_no_item_found=None, only_new_items=False, start_le=False, stop_ge=False, join=False)

Enables:

1. Get last n minutes
2. Give me data I don’t have

* **Parameters:**
  * **start** – starting range key of item inclusive
  * **stop** – ending range key of item inclusive
  * **step**
  * **peek** – if True, last_item cursor will not be updated
  * **ignore_no_item_found** – if True, return None when no next item
    instead of raising exception.
  * **only_new_items** – if True and no new items, raise ValueError
    or return None if ignore_no_item_found
  * **start_le** – if True, increase the search range to find start less than or equal by
    rounding down if start is in between keys,
    i.e. keys=[0, 10, 20], start=9 will include key=0
  * **stop_ge** – if True, raise ValueError when there is no key greater than or equal to
    stop in buffer, if ignore_no_item_found is also True, return None instead of ValueError
  * **join** – if True, return ONE item instead of a list: the items in range
    trimmed to exactly `[start, stop]` and joined, using the source reader’s
    `slice_item` and `join_items` (both must be implemented, otherwise
    NotImplementedError is raised before anything is read). Implies
    `start_le=True` so the chunk containing `start` is included, and is
    incompatible with `step`. The `last_item` cursor still moves to the
    last *buffer* item, not the trimmed one. With `only_new_items` the
    joined item starts after the cursor. When nothing is found the result is
    the same as without `join` (`[]`, `None` or an exception).
* **Returns:**
  list of items in range (a single joined item if `join`)

#### read(n=None, , peek=None, ignore_no_item_found=None, strict_n=None, blocking=None, timeout=None)

Finds an item with a key greater than the last returned item.
Raise ValueError if no item found with key above last item.

#### NOTE
Arguments whose values are not specified will be replaced with
self.

* **Parameters:**
  * **n** – number of items to return
  * **peek** – if True, last_item cursor will not be updated
  * **ignore_no_item_found** – if True, return None when no next item
    instead of raising exception
  * **strict_n** – if True, raise ValueError if n items are not available
  * **blocking** – if True, wait to return next item if not yet available.
    Defaults to the reader’s `blocking` (False unless set at construction).
  * **timeout** – only used when `blocking` is True. Defaults to the reader’s
    `timeout`. Maximum number of seconds
    to wait; `None` (default) waits until an item is available or the stream
    stops. When it runs out, returns None (as when the stream stops).
* **Returns:**
  next item or list of next items if n > 1

#### set_sleep_time_on_iter_none(sleep_time_s=0.1)

Set the sleep time of the iter yield loop when next data item is not yet available.

* **Parameters:**
  **sleep_time_s** ([`Union`](https://docs.python.org/3/library/typing.html#typing.Union)[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)]) – seconds to sleep

#### *property* source_reader_info *: [dict](https://docs.python.org/3/builtins/stdtypes.html#dict)*

A dict with important source info set by SourceReader.

* **Returns:**
  dict

#### tail(, peek=False, ignore_no_item_found=False, only_new_items=False)

Finds the last item in buffer. Raise ValueError if no item found.

* **Parameters:**
  * **peek** – if True, last_item cursor will not be updated
  * **ignore_no_item_found** – if True, return None when no next item
    instead of raising exception
  * **only_new_items** – if True and no new items, raise ValueError
    or return None if ignore_no_item_found
* **Returns:**
  tail item


# _autosummary/stream2py.exceptions.html.md

# stream2py.exceptions

Custom exception classes for stream2py

This module defines the exception hierarchy for stream2py, providing more
specific and informative error messages than generic Python exceptions.

### Exceptions

| [`BufferError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.BufferError)               | Base exception for buffer-related errors                                       |
|----------------------------------------------------------------------------|--------------------------------------------------------------------------------|
| [`BufferOverflowError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.BufferOverflowError)       | Raised when buffer is full and auto_drop=False                                 |
| [`ConfigurationError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.ConfigurationError)        | Raised when stream2py objects are misconfigured                                |
| [`InvalidDataError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.InvalidDataError)          | Raised when data doesn't meet expected format or constraints                   |
| [`NoDataAvailableError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.NoDataAvailableError)      | Raised when no data is available and ignore_no_item_found=False                |
| [`Stream2PyError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.Stream2PyError)            | Base exception for all stream2py errors                                        |
| [`StreamAlreadyStoppedError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.StreamAlreadyStoppedError) | Raised when operations require a running stream but stream has been stopped    |
| [`StreamNotStartedError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.StreamNotStartedError)     | Raised when operations require a started stream but stream hasn't been started |

### *exception* stream2py.exceptions.BufferError

Bases: [`Stream2PyError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.Stream2PyError)

Base exception for buffer-related errors

### *exception* stream2py.exceptions.BufferOverflowError

Bases: [`BufferError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.BufferError)

Raised when buffer is full and auto_drop=False

### *exception* stream2py.exceptions.ConfigurationError

Bases: [`Stream2PyError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.Stream2PyError)

Raised when stream2py objects are misconfigured

### *exception* stream2py.exceptions.InvalidDataError

Bases: [`Stream2PyError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.Stream2PyError)

Raised when data doesn’t meet expected format or constraints

### *exception* stream2py.exceptions.NoDataAvailableError

Bases: [`BufferError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.BufferError)

Raised when no data is available and ignore_no_item_found=False

### *exception* stream2py.exceptions.Stream2PyError

Bases: [`Exception`](https://docs.python.org/3/builtins/exceptions.html#Exception)

Base exception for all stream2py errors

### *exception* stream2py.exceptions.StreamAlreadyStoppedError

Bases: [`Stream2PyError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.Stream2PyError)

Raised when operations require a running stream but stream has been stopped

### *exception* stream2py.exceptions.StreamNotStartedError

Bases: [`Stream2PyError`](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.Stream2PyError)

Raised when operations require a started stream but stream hasn’t been started


# _autosummary/stream2py.html.md

# stream2py

## The Rationale

Stream2py enables you to create a single asynchronous Source Reader and simply distribute
thread-safe readers to one or more consumers for the same data window. Or in other words, read and
work on data without worrying about missing data.

Normal way to process audio recording:

```default
import pyaudio

CHUNK = 1024
WIDTH = 2
CHANNELS = 1
RATE = 44100
RECORD_SECONDS = 5

p = pyaudio.PyAudio()

stream = p.open(
    format=p.get_format_from_width(WIDTH),
    channels=CHANNELS,
    rate=RATE,
    input=True,
    frames_per_buffer=CHUNK
)

for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
    data = stream.read(CHUNK)
    # risk of buffer overflow before next read
    long_calculations(data)

stream.stop_stream()
stream.close()
p.terminate()
```

Stream2py way to process audio recording:

```default
from stream2py.sources.audio import PyAudioSourceReader

CHUNK = 1024
WIDTH = 2
CHANNELS = 1
RATE = 44100
RECORD_SECONDS = 5

source_reader = PyAudioSourceReader(
    rate=RATE,
    width=WIDTH,
    channels=CHANNELS,
    frames_per_buffer=CHUNK
)
with StreamBuffer(
    source_reader=source_reader, maxlen=int(RATE / CHUNK * RECORD_SECONDS)
) as stream_buffer:
    buffer_reader = stream_buffer.mk_reader()

    for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
        data = next(buffer_reader)
        long_calculations(data)
```

## How to use

There are 3 components

1. A SourceReader defines how to get data with the methods: open(), read(), and close(), and also how
   the data is ordered with the key() method and an info property describing the instance.

```default
from typing import Tuple
from stream2py import SourceReader

EnumeratedReadData = Tuple[int, str]

class LineSourceReader(SourceReader):
    def __init__(self, file, mode='rt'):
        self.file = file
        self.mode = mode
        self.line_count = 0
        self._f = None

    def open(self):
        self.line_count = 0
        self._f = open(self.file, self.mode)

    def read(self) -> EnumeratedReadData:
        read_data = (self.line_count, self._f.readline())
        self.line_count += 1
        return read_data

    def close(self):
        self._f.close()

    def key(self, data: EnumeratedReadData):
        return data[0]

    @property
    def info(self):
        return dict(file=self.file, mode=self.mode)
```

1. A StreamBuffer has 2 jobs: First, it manages the open, read, and close of a SourceReader and puts
   read data onto a thread-safe buffer.

Second, it is a factory of BufferReaders instances for multiple consumers.

```default
# ...Continued from above

# download txt from: https://www.gutenberg.org/cache/epub/1524/pg1524.txt
FILE = 'hamlet.txt'
source_reader = LineSourceReader(FILE)

with StreamBuffer(
    source_reader=source_reader,
    maxlen=None,  # None should only be used when reading objects with fixed lengths like a file
) as stream_buffer:
    buffer_reader = stream_buffer.mk_reader()
```

1. A BufferReader gives data access to any number of consumers and provides methods to seek data such
   as next(), range(), head(), tail().  Each BufferReader instance has it’s own cursor keeping track of
   what data was last seen.

```default
# ...Continued from above

    line_1 = buffer_reader.next()
    line_2 = next(buffer_reader)
    line_1_10 = buffer_reader.range(start=1, stop=10)
```

### The All-in-One Helper

Here’s a bonus helper function to shortcut the SourceReader class.

```default
from time import sleep
from stream2py import mk_stream_buffer

FILE = 'hamlet.txt'
with mk_stream_buffer(
    read_stream=lambda open_inst: next(open_inst, None),
    open_stream=lambda: open(FILE, 'rt'),
    close_stream=lambda open_inst: open_inst.close(),
    key=lambda read_data: read_data[0],
    maxlen=None,
) as stream_buffer:
    buffer_reader = stream_buffer.mk_reader()
    sleep(1)
    # do stuff with buffer_reader
    print(buffer_reader.range(start=40, stop=50))
```

### Modules

| [`buffer_reader`](_autosummary/stream2py.buffer_reader.html.md#module-stream2py.buffer_reader)   | A BufferReader gives data access to any number of consumers and provides methods to seek data such as next(), range(), head(), tail().                                                     |
|-------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| [`exceptions`](_autosummary/stream2py.exceptions.html.md#module-stream2py.exceptions)         | Custom exception classes for stream2py                                                                                                                                                     |
| [`protocols`](_autosummary/stream2py.protocols.html.md#module-stream2py.protocols)           | Defines protocols for classes to avoid circular imports                                                                                                                                    |
| [`simply`](_autosummary/stream2py.simply.html.md#module-stream2py.simply)                 | Helper Function to construct a SourceReader and StreamBuffer in one                                                                                                                        |
| [`source_reader`](_autosummary/stream2py.source_reader.html.md#module-stream2py.source_reader)   | A SourceReader defines how to get data with the methods: open(), read(), and close(), and also how the data is ordered with the key() method and an info property describing the instance. |
| [`sources`](_autosummary/stream2py.sources.html.md#module-stream2py.sources)               | A few light weight (builtin-only ) source readers you can use                                                                                                                              |
| [`stream_buffer`](_autosummary/stream2py.stream_buffer.html.md#module-stream2py.stream_buffer)   | A StreamBuffer has 2 jobs: First, it manages the open, read, and close of a SourceReader and puts read data onto a thread-safe buffer.                                                     |
| [`stream_source`](_autosummary/stream2py.stream_source.html.md#module-stream2py.stream_source)   | A BufferReader gives data access to any number of consumers and provides methods to seek data such as next(), range(), head(), tail().                                                     |
| [`util`](_autosummary/stream2py.util.html.md#module-stream2py.util)                     | Util functions.                                                                                                                                                                            |
| [`utility`](_autosummary/stream2py.utility.html.md#module-stream2py.utility)               | Low-level building blocks used by stream2py's buffers and readers.                                                                                                                         |


# _autosummary/stream2py.protocols.html.md

# stream2py.protocols

Defines protocols for classes to avoid circular imports

### Classes

| [`Source`](_autosummary/stream2py.protocols.html.md#stream2py.protocols.Source)(\*args, \*\*kwargs)   | The interface for the StreamSource class.   |
|-------------------------------------------------------------------------------|---------------------------------------------|

### *class* stream2py.protocols.Source(\*args, \*\*kwargs)

Bases: [`Protocol`](https://docs.python.org/3/library/typing.html#typing.Protocol)

The interface for the StreamSource class.


# _autosummary/stream2py.simply.html.md

# stream2py.simply

Helper Function to construct a SourceReader and StreamBuffer in one

Example of usage:

```default
from stream2py.simply import mk_stream_buffer

counter = iter(range(10))
with mk_stream_buffer(
    read_stream=lambda open_inst: next(counter),
    open_stream=lambda: print('open'),
    close_stream=lambda open_inst: print('close'),
    auto_drop=False,
    maxlen=2,
) as count_stream_buffer:
    count_reader = count_stream_buffer.mk_reader()
    print(f'start reading {count_reader.source_reader_info}')
    for i in count_reader:
        # do stuff with read data
        print(i)
        if count_stream_buffer.auto_drop is False:
            count_stream_buffer.drop()

    print('done reading')
```

### Functions

| [`mk_stream_buffer`](_autosummary/stream2py.simply.html.md#stream2py.simply.mk_stream_buffer)(read_stream[, open_stream, ...])   | Helper Function to construct a SourceReader and StreamBuffer in one   |
|------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------|

### stream2py.simply.mk_stream_buffer(read_stream, open_stream=None, close_stream=None, info=None, key=None, maxlen=100, sleep_time_on_read_none_seconds=None, auto_drop=True, enumerate_data=True)

Helper Function to construct a SourceReader and StreamBuffer in one

```default
from time import sleep
from stream2py import mk_stream_buffer

# download txt from: https://www.gutenberg.org/cache/epub/1524/pg1524.txt
FILE = 'hamlet.txt'
with mk_stream_buffer(
    read_stream=lambda open_inst: next(open_inst, None),
    open_stream=lambda: open(FILE, 'rt'),
    close_stream=lambda open_inst: open_inst.close(),
    key=lambda read_data: read_data[0],
    maxlen=None,
) as stream_buffer:
    buffer_reader = stream_buffer.mk_reader()
    sleep(1)
    # do stuff with buffer_reader
    print(buffer_reader.range(start=40, stop=50))
```

* **Parameters:**
  * **read_stream** ([`Callable`](https://docs.python.org/3/library/typing.html#typing.Callable)[[[`Optional`](https://docs.python.org/3/library/typing.html#typing.Optional)[[`Any`](https://docs.python.org/3/library/typing.html#typing.Any)]], [`Optional`](https://docs.python.org/3/library/typing.html#typing.Optional)[[`Any`](https://docs.python.org/3/library/typing.html#typing.Any)]]) – takes open instance and returns the next read item
  * **open_stream** ([`Callable`](https://docs.python.org/3/library/typing.html#typing.Callable)[[], [`Optional`](https://docs.python.org/3/library/typing.html#typing.Optional)[[`Any`](https://docs.python.org/3/library/typing.html#typing.Any)]]) – open stream and return an open instance such as a file descriptor
  * **close_stream** ([`Callable`](https://docs.python.org/3/library/typing.html#typing.Callable)[[[`Optional`](https://docs.python.org/3/library/typing.html#typing.Optional)[[`Any`](https://docs.python.org/3/library/typing.html#typing.Any)]], [`None`](https://docs.python.org/3/builtins/constants.html#None)]) – take open instance and close the instance
  * **info** ([`Callable`](https://docs.python.org/3/library/typing.html#typing.Callable)[[[`Optional`](https://docs.python.org/3/library/typing.html#typing.Optional)[[`Any`](https://docs.python.org/3/library/typing.html#typing.Any)]], [`dict`](https://docs.python.org/3/builtins/stdtypes.html#dict)]) – dict help in describing the instance
  * **key** ([`Callable`](https://docs.python.org/3/library/typing.html#typing.Callable)[[[`Any`](https://docs.python.org/3/library/typing.html#typing.Any)], [`ComparableType`](_autosummary/stream2py.utility.typing_hints.html.md#stream2py.utility.typing_hints.ComparableType)]) – sort key for read items
  * **maxlen** ([`Optional`](https://docs.python.org/3/library/typing.html#typing.Optional)[[`int`](https://docs.python.org/3/builtins/functions.html#int)]) – max read items to keep in buffer window
  * **sleep_time_on_read_none_seconds** ([`float`](https://docs.python.org/3/builtins/functions.html#float)) – time to wait if data is not ready to be read
  * **auto_drop** ([`bool`](https://docs.python.org/3/builtins/functions.html#bool)) – False is useful when data can be read independent of time and will quickly
    overflow the buffer maxlen before data is consumed. StreamBuffer.drop() must be called to
    continue reading after buffer has filled up. Default True.
  * **enumerate_data** ([`bool`](https://docs.python.org/3/builtins/functions.html#bool)) – wrap read data in an enumerated tuple. Default True.
* **Returns:**


# _autosummary/stream2py.source_reader.html.md

# stream2py.source_reader

A SourceReader defines how to get data with the methods: open(), read(), and close(), and also how
the data is ordered with the key() method and an info property describing the instance.

### Classes

| [`SourceReader`](_autosummary/stream2py.source_reader.html.md#stream2py.source_reader.SourceReader)(\*args, \*\*kwargs)   | (deprecated) Abstract class interface to be used by StreamBuffer.   |
|-------------------------------------------------------------------------------------|---------------------------------------------------------------------|

### *class* stream2py.source_reader.SourceReader(\*args, \*\*kwargs)

Bases: [`Source`](_autosummary/stream2py.protocols.html.md#stream2py.protocols.Source)

(deprecated) Abstract class interface to be used by StreamBuffer.

```pycon
>>> from stream2py import SourceReader
>>>
>>> class SimpleCounterString(SourceReader):
...     '''Count in range returning a string formatted as f"s{count}"'''
...     def __init__(self, start, stop):
...         assert start < stop
...         self.start = start
...         self.stop = stop
...         self.range_iterator = None
...         self.open_count = 0
...
...     def open(self):
...         self.open_count += 1
...         self.range_iterator = iter(range(self.start, self.stop))
...         self._closed = False
...
...     def read(self):
...         value = next(self.range_iterator, None)
...         if value is not None:
...             return f's{value}'
...         else:
...             return None
...
...     def close(self):
...         del self.range_iterator
...         self.range_iterator = None
...         self._closed = True
...
...     @property
...     def info(self):
...         return dict(start=self.start, stop=self.stop, open_count=self.open_count)
...
...     def key(self, data):
...         return int(data[1:])
...
>>> source_reader = SimpleCounterString(start=0, stop=10)
>>> source_reader.closed
True
>>> source_reader.open()
>>> source_reader.closed
False
>>> source_reader.info
{'start': 0, 'stop': 10, 'open_count': 1}
>>> source_reader.read()
's0'
>>> source_reader.close()
>>> source_reader.closed
True
>>> with source_reader:
...     source_reader.info
...     source_reader.read()
...     source_reader.read()
...     source_reader.read()
{'start': 0, 'stop': 10, 'open_count': 2}
's0'
's1'
's2'
```

#### *abstractmethod* close()

Close and clean up source reader.
Will be called when StreamBuffer stops or if an exception is raised during read and append
loop.

* **Return type:**
  [`None`](https://docs.python.org/3/builtins/constants.html#None)

#### *property* closed *: [bool](https://docs.python.org/3/builtins/functions.html#bool)*

True if the source reader is closed, False if it’s open.

Similar to io.IOBase.closed property.

* **Returns:**
  bool indicating whether the source is closed

#### *classmethod* get_timestamp()

Get timestamp of now as int microseconds

* **Return type:**
  [`int`](https://docs.python.org/3/builtins/functions.html#int)
* **Returns:**
  timestamp in UTC microseconds

#### *property* info *: [dict](https://docs.python.org/3/builtins/stdtypes.html#dict)*

A dict with important source info. Default can be init_kwargs and open timestamp.

This info will be available in BufferReader after SourceReader has opened but any changes
made between open and close are not guaranteed to be visible. Any continuous status updates
should be included with read data instead.

* **Returns:**
  dict

#### join_items(items)

(Optional) Join consecutive data items into one data item.

Implement together with `slice_item` to let `BufferReader.range(...,
join=True)` return the exact span asked for, as a single item (e.g. several
waveform chunks concatenated). Not abstract: sources whose items can’t be
joined just don’t override it.

* **Parameters:**
  **items** ([`list`](https://docs.python.org/3/builtins/stdtypes.html#list)) – non-empty list of data items, in key order
* **Return type:**
  [`Any`](https://docs.python.org/3/library/typing.html#typing.Any)
* **Returns:**
  one data item

#### *abstractmethod* key(data)

Converts data into a comparable value to sort by

* **Parameters:**
  **data** ([`Any`](https://docs.python.org/3/library/typing.html#typing.Any)) – the return value of the ‘read’ method
* **Return type:**
  [`ComparableType`](_autosummary/stream2py.utility.typing_hints.html.md#stream2py.utility.typing_hints.ComparableType)
* **Returns:**
  ComparableType

#### *abstractmethod* open()

Set up source to be read and set some source info affected by open time like the time of
open. Will be called in StreamBuffer immediately before first read.

* **Return type:**
  [`None`](https://docs.python.org/3/builtins/constants.html#None)

#### set_sleep_time_on_iter_none(sleep_time_s=0.001)

Set the sleep time of the iter yield loop when next data item is not yet available.

* **Parameters:**
  **sleep_time_s** ([`Union`](https://docs.python.org/3/library/typing.html#typing.Union)[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)]) – seconds to sleep

#### *property* sleep_time_on_read_none_s *: [Union](https://docs.python.org/3/library/typing.html#typing.Union)[[int](https://docs.python.org/3/builtins/functions.html#int), [float](https://docs.python.org/3/builtins/functions.html#float), [None](https://docs.python.org/3/builtins/constants.html#None)]*

Sets default sleep time for StreamBuffer when it reads None from SourceReader.
Useful when you know the period between data points.

For example if data only comes once every 10 seconds,
it would be inefficient to try and read again every 0.1 seconds and get None up to 100 times
but every 10 seconds or even 1 second would be much more reasonable depending on the use
case. Not necessary if read is a blocking function that always returns a data point.

Default is None which delegates the sleep time to StreamBuffer’s default.

* **Returns:**
  Optional[Union[int, float]] number of seconds to sleep

#### slice_item(item, start=None, stop=None)

(Optional) The part of `item` whose keys fall within `[start, stop]`.

`None` for a bound means unbounded on that side; an item that needs no
trimming can be returned as is. See `join_items`. Must be pure (items in the
buffer are shared, never mutate them) and keyed on absolute keys: it is applied
once per bound, so a lone item is sliced twice (`start` then `stop`).

* **Parameters:**
  **item** ([`Any`](https://docs.python.org/3/library/typing.html#typing.Any)) – a data item as returned by `read`
* **Return type:**
  [`Any`](https://docs.python.org/3/library/typing.html#typing.Any)
* **Returns:**
  a data item of the same kind, possibly shorter


# _autosummary/stream2py.sources.html.md

# stream2py.sources

A few light weight (builtin-only ) source readers you can use

### Modules

| [`http`](_autosummary/stream2py.sources.http.html.md#module-stream2py.sources.http)   | HTTP Response streaming reader with no external dependencies   |
|---------------------------------------------------------------------------------------|----------------------------------------------------------------|


# _autosummary/stream2py.sources.http.html.md

# stream2py.sources.http

## HTTP Response streaming reader with no external dependencies

### Classes

| [`HTTPResponseReader`](_autosummary/stream2py.sources.http.html.md#stream2py.sources.http.HTTPResponseReader)(url, \*[, method, ...])   |    |
|-----------------------------------------------------------------------------------------------|----|

### *class* stream2py.sources.http.HTTPResponseReader(url, , method='GET', headers=None, body='', response_format='bytes', response_read_args=(), encoding='')

Bases: [`SourceReader`](_autosummary/stream2py.source_reader.html.md#stream2py.source_reader.SourceReader)

#### close()

Close and clean up source reader.
Will be called when StreamBuffer stops or if an exception is raised during read and append
loop.

#### *property* info *: [dict](https://docs.python.org/3/builtins/stdtypes.html#dict)*

Provides a dict with init kwargs and read status

```pycon
>>> from pprint import pprint
>>> source = HTTPResponseReader('http://zombo.com')
>>> pprint(source.info)
{'url': 'http://zombo.com',
 'n_bytes_read': 0,
 'init_kwargs': {'method': 'GET,
                 'headers': {},
                 'body': None,
                 'response_format': 'bytes'}}
```

* **Returns:**
  dict

#### key(data)

Returns a slice of the data retrieved.

* **Parameters:**
  **data** – (start_index, end_index)
* **Return type:**
  [`ComparableType`](_autosummary/stream2py.utility.typing_hints.html.md#stream2py.utility.typing_hints.ComparableType)
* **Returns:**
  Union[str, bytes]

#### open()

Set up source to be read and set some source info affected by open time like the time of
open. Will be called in StreamBuffer immediately before first read.

#### read()

Returns a chunk of raw binary data.


# _autosummary/stream2py.stream_buffer.html.md

# stream2py.stream_buffer

A StreamBuffer has 2 jobs: First, it manages the open, read, and close of a SourceReader
and puts read data onto a thread-safe buffer.
Second, it is a factory of BufferReaders instances for multiple consumers.

### Classes

| [`StreamBuffer`](_autosummary/stream2py.stream_buffer.html.md#stream2py.stream_buffer.StreamBuffer)(source_reader, \*[, maxlen, ...])   | Handles starting and stopping SourceReader and making BufferReaders   |
|---------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------|

### *class* stream2py.stream_buffer.StreamBuffer(source_reader, , maxlen=100, sleep_time_on_read_none_s=None, auto_drop=True)

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)

Handles starting and stopping SourceReader and making BufferReaders

```pycon
>>> from stream2py import StreamBuffer
>>> from stream2py.examples.source_reader import SimpleCounterString
>>>
>>> source_reader = SimpleCounterString(start=0, stop=100)
>>> stream_buffer = StreamBuffer(source_reader=source_reader, maxlen=100)
>>> stream_buffer.is_running
False
>>> print(stream_buffer.source_reader_info)
None
>>> stream_buffer.start()
>>> assert stream_buffer.is_running
>>> stream_buffer.source_reader_info
{'start': 0, 'stop': 100, 'open_count': 1}
>>> open1_reader1 = stream_buffer.mk_reader()
>>> open1_reader1.next()
's0'
>>> open1_reader2 = stream_buffer.mk_reader()
>>> # readers from the same open instance are the same
>>> assert open1_reader1.is_same_buffer(open1_reader2) is True
>>> assert open1_reader1.next() != open1_reader2.next()  # item cursor position is different
>>> stream_buffer.stop()
>>>
>>> with stream_buffer:
...     stream_buffer.source_reader_info
...     open2_reader1  = stream_buffer.mk_reader()
...     open2_reader2  = stream_buffer.mk_reader()
...     # readers from the same open instance are the same
...     assert open2_reader1.is_same_buffer(open2_reader2) is True
...     # readers from the different open instances
...     assert open2_reader1.is_same_buffer(open1_reader1) is False
{'start': 0, 'stop': 100, 'open_count': 2}
```

#### attach_reader(reader)

Allows a StreamReader instance to read from this buffer.

* **Raises:**
  [**StreamNotStartedError**](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.StreamNotStartedError) – if the StreamBuffer hasn’t been started yet

#### drop(n=1)

Manually drop items from buffer when auto_drop is False

* **Parameters:**
  **n** – number of items to drop from the left side
* **Returns:**

#### *property* is_running *: [bool](https://docs.python.org/3/builtins/functions.html#bool)*

Checks if stop event has been set.

* **Returns:**
  bool

#### mk_reader(\*\*read_kwargs)

Makes a BufferReader instance for the currently running  StreamBuffer.
Reader must be made after start() to have data from said start.

* **Return type:**
  [`BufferReader`](_autosummary/stream2py.buffer_reader.html.md#stream2py.buffer_reader.BufferReader)
* **Returns:**
  BufferReader instance
* **Raises:**
  [**StreamNotStartedError**](_autosummary/stream2py.exceptions.html.md#stream2py.exceptions.StreamNotStartedError) – if the StreamBuffer hasn’t been started yet

#### *property* source_reader_info *: [Optional](https://docs.python.org/3/library/typing.html#typing.Optional)[[dict](https://docs.python.org/3/builtins/stdtypes.html#dict)]*

A dict with important source info set by SourceReader.

* **Returns:**
  dict or None

#### start()

Open and start reading from source_reader into buffer

#### stop()

Stop reading and close source_reader


# _autosummary/stream2py.stream_source.html.md

# stream2py.stream_source

A BufferReader gives data access to any number of consumers and provides methods to seek data such
as next(), range(), head(), tail().  Each BufferReader instance has it’s own cursor keeping track of
what data was last seen.

### Functions

| [`defaulted_values`](_autosummary/stream2py.stream_source.html.md#stream2py.stream_source.defaulted_values)(source_dict, defaults)   | Returns a the values of source_dict where None values are replaced with the corresponding `default_dict` values.   |
|--------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------|

### Classes

| [`StreamSource`](_autosummary/stream2py.stream_source.html.md#stream2py.stream_source.StreamSource)(\*[, reader_cls])   | Abstract parent class for reading from a source into a StreamBuffer.   |
|-----------------------------------------------------------------------------------|------------------------------------------------------------------------|

### *class* stream2py.stream_source.StreamSource(\*, reader_cls=<class 'stream2py.buffer_reader.BufferReader'>)

Bases: [`Source`](_autosummary/stream2py.protocols.html.md#stream2py.protocols.Source)

Abstract parent class for reading from a source into a StreamBuffer.

Provides interfaces for configuring the StreamBuffer and instantiating BufferReaders.

The open method will be called when creating a reader with open_reader().
The close method will be called when all open readers have been closed().
This will also stop the StreamBuffer. A closed source can be reopened by opening another reader.

```pycon
>>> from stream2py.examples.stream_source import SimpleCounterString
>>> source = SimpleCounterString(start=0, stop=10)
>>> with source.open_reader() as reader:
...     # the source will close when exiting the context
...     reader.read()
...     reader.read()
...     reader.read()
's0'
's1'
's2'
>>> reader2 = source.open_reader()
>>> reader2.read()
's0'
>>> reader2.close()  # closing the reader manually
```

#### configure_buffer(, maxlen=None, sleep_time_on_read_none_s=None, auto_drop=True)

Sets custom buffer parameters.

#### *classmethod* get_timestamp()

Get timestamp of now as int microseconds

* **Return type:**
  [`int`](https://docs.python.org/3/builtins/functions.html#int)
* **Returns:**
  timestamp in UTC microseconds

#### open_reader(read_size=1, strict_n=False, ignore_no_item_found=False)

Instantiates a reader.

* **Parameters:**
  * **read_size** – number of items to return by default when reading.
    Is also used as the “chunk size” when iterating.
  * **strict_n** – if True, by default a ValueError witll be raised if the
    exact number of requested items are not available when reading.
  * **ignore_no_item_found** – if True, by default return None when no next item
    instead of raising exception during a read

#### set_sleep_time_on_iter_none(sleep_time_s=0.1)

Set the sleep time of the iter yield loop when next data item is not yet available.

* **Parameters:**
  **sleep_time_s** ([`Union`](https://docs.python.org/3/library/typing.html#typing.Union)[[`int`](https://docs.python.org/3/builtins/functions.html#int), [`float`](https://docs.python.org/3/builtins/functions.html#float)]) – seconds to sleep

#### *property* sleep_time_on_read_none_s *: [Union](https://docs.python.org/3/library/typing.html#typing.Union)[[int](https://docs.python.org/3/builtins/functions.html#int), [float](https://docs.python.org/3/builtins/functions.html#float), [None](https://docs.python.org/3/builtins/constants.html#None)]*

Sets default sleep time for StreamBuffer when it reads None from SourceReader.
Useful when you know the period between data points.

For example if data only comes once every 10 seconds,
it would be inefficient to try and read again every 0.1 seconds and get None up to 100 times
but every 10 seconds or even 1 second would be much more reasonable depending on the use
case. Not necessary if read is a blocking function that always returns a data point.

Default is None which delegates the sleep time to StreamBuffer’s default.

* **Returns:**
  Optional[Union[int, float]] number of seconds to sleep

### stream2py.stream_source.defaulted_values(source_dict, defaults)

Returns a the values of source_dict where None values are replaced with the
corresponding `default_dict` values.

```pycon
>>> defaults = {'a': 1, 'b': 2, 'c': 3}
>>> tuple(defaulted_values({'a': None, 'b': 22, 'c': None}, defaults))
(1, 22, 3)
>>> tuple(defaulted_values({'a': False, 'b': True, 'c': None}, defaults))
(False, True, 3)
```


# _autosummary/stream2py.util.html.md

# stream2py.util

Util functions. Functions that are used by, or can be used with, stream2py objects
(but do not depend on stream2py objects themselves)

### Functions

| `forward_to_instance_enter`(obj)                             |                           |
|--------------------------------------------------------------|---------------------------|
| `forward_to_instance_exit`(obj, \*exception_info)            |                           |
| [`identity`](_autosummary/stream2py.util.html.md#stream2py.util.identity)(x) | Returns the input, as is. |
| `null_enter_func`(obj)                                       |                           |
| `null_exit_func`(obj, \*exception_info)                      |                           |

### Classes

| [`ContextualizeCall`](_autosummary/stream2py.util.html.md#stream2py.util.ContextualizeCall)(func[, enter_func, exit_func])   | Wraps a callable so that it also becomes a context manager.                            |
|-----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------|
| [`Timer`](_autosummary/stream2py.util.html.md#stream2py.util.Timer)([egress])                                    | A simple timer that can be used as a context manager, or started and stopped manually. |

### Exceptions

| [`EnterFunctionValidation`](_autosummary/stream2py.util.html.md#stream2py.util.EnterFunctionValidation)   | Raised if an function isn't a valid (context manager) \_\_enter_\_ function   |
|----------------------------------------------------------------------------|-------------------------------------------------------------------------------|
| [`ExitFunctionValidation`](_autosummary/stream2py.util.html.md#stream2py.util.ExitFunctionValidation)    | Raised if an function isn't a valid (context manager) \_\_exit_\_ function    |
| [`TypeValidationError`](_autosummary/stream2py.util.html.md#stream2py.util.TypeValidationError)       | Raised if an object is not valid                                              |

### *class* stream2py.util.ContextualizeCall(func, enter_func=<function null_enter_func>, exit_func=<function null_exit_func>)

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)

Wraps a callable so that it also becomes a context manager.

Both enter and exit functions can be specified.

This can be used, for instance, when the proper use of a function depends on some
objects being “in context” (i.e. calling the enter/exit of some objects).

This happens, for instance, when we want to use a method that depends on its
instance being in context. For convenience, this use case is implemented by the
`contextualize_with_instance` function of this module.

Below is a simple doctest, but check out more significant tests in the tests folder.
You can also check out:

This issue: [https://github.com/i2mint/stream2py/issues/18](https://github.com/i2mint/stream2py/issues/18)

This wiki: [https://github.com/i2mint/stream2py/wiki/Forwarding-context-management](https://github.com/i2mint/stream2py/wiki/Forwarding-context-management)

```pycon
>>> f = lambda x: x + 1
>>> hasattr(f, '__enter__') and hasattr(f, '__exit__')
False
>>> g = ContextualizeCall(f)
>>> hasattr(g, '__enter__') and hasattr(g, '__exit__')
True
>>> with g:
...     print(g(2))
3
```

In the example above we get the default null enter and exit functions.
They do nothing.
Let’s specify some that do something:

```pycon
>>> h = ContextualizeCall(f,
...     enter_func=lambda self: print('entering...'),
...     exit_func=lambda self, *errs: print('exiting...'),
... )
>>> with h:
...     print(h(2))
entering...
3
exiting...
```

#### NOTE
A `ContextualizeCall` instance will delegate all other attributes to the
wrapped `func` except for special methods. If a user needs to also delegate these
special methods, they’ll have to wrap `func` themselves.

### *exception* stream2py.util.EnterFunctionValidation

Bases: [`TypeValidationError`](_autosummary/stream2py.util.html.md#stream2py.util.TypeValidationError)

Raised if an function isn’t a valid (context manager) \_\_enter_\_ function

#### *static* is_valid(obj)

Returns True if and only if obj is considered valid

### *exception* stream2py.util.ExitFunctionValidation

Bases: [`TypeValidationError`](_autosummary/stream2py.util.html.md#stream2py.util.TypeValidationError)

Raised if an function isn’t a valid (context manager) \_\_exit_\_ function

#### *static* is_valid(obj)

Returns True if and only if obj is considered valid

### *class* stream2py.util.Timer(egress=<function identity>)

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)

A simple timer that can be used as a context manager, or started and stopped
manually.

```pycon
>>> from time import sleep
>>> timer = Timer()
>>> timer.start()
>>> sleep(0.01)
>>> timer.elapsed()
0.010056478977203369
>>> timer.stop()
```

You can also use a context block, and specify an `egress` function that will be
called on the elapsed seconds:

```pycon
>>> with Timer(lambda x: print(f"{int(x / (60 * 60 * 24))} days elapsed")) as timer:
...     sleep(0.01)
...     timer.elapsed()
0 days elapsed
```

The same timer instance can be reused across context blocks:

```pycon
>>> timer = Timer(lambda x: int(x / 3600))
>>> with timer:
...     sleep(0.01)
...     t = timer.elapsed()
>>> t
0
>>> with timer:
...     sleep(0.01)
```

Asking for the elapsed time of a timer that isn’t running tells you so, instead of
failing obscurely:

```pycon
>>> Timer().elapsed()
Traceback (most recent call last):
  ...
ValueError: The timer is not running....
```

Note that elapsed time is measured with `time.perf_counter`, a monotonic clock, so
it is unaffected by system clock adjustments. This also means `start_time` is not a
wall-clock timestamp; it is only meaningful as a reference point for a difference.

#### elapsed()

The (egress-transformed) seconds since the timer was started.

Raises `ValueError` if the timer is not running.

### *exception* stream2py.util.TypeValidationError

Bases: [`TypeError`](https://docs.python.org/3/builtins/exceptions.html#TypeError)

Raised if an object is not valid

#### *abstractmethod static* is_valid(obj)

Returns True if and only if obj is considered valid

* **Return type:**
  [`bool`](https://docs.python.org/3/builtins/functions.html#bool)

### stream2py.util.identity(x)

Returns the input, as is.


# _autosummary/stream2py.utility.html.md

# stream2py.utility

Low-level building blocks used by stream2py’s buffers and readers.

These are general-purpose data structures and synchronization primitives that do not
depend on stream2py objects themselves:

- [`stream2py.utility.sorted_collection`](_autosummary/stream2py.utility.sorted_collection.html.md#module-stream2py.utility.sorted_collection) – a sorted sequence with bisect-based lookup
- [`stream2py.utility.sorted_deque`](_autosummary/stream2py.utility.sorted_deque.html.md#module-stream2py.utility.sorted_deque) – a bounded sorted deque
- [`stream2py.utility.locked_sorted_deque`](_autosummary/stream2py.utility.locked_sorted_deque.html.md#module-stream2py.utility.locked_sorted_deque) – a sorted deque guarded by a lock
- [`stream2py.utility.reader_writer_lock`](_autosummary/stream2py.utility.reader_writer_lock.html.md#module-stream2py.utility.reader_writer_lock) – a multi-reader/single-writer lock
- [`stream2py.utility.typing_hints`](_autosummary/stream2py.utility.typing_hints.html.md#module-stream2py.utility.typing_hints) – shared type aliases

### Modules

| [`locked_sorted_deque`](_autosummary/stream2py.utility.locked_sorted_deque.html.md#module-stream2py.utility.locked_sorted_deque)   | Sorted Deque that handles Reader-Writer Priority   |
|---------------------------------------------------------------------------------------------------------------------|----------------------------------------------------|
| [`reader_writer_lock`](_autosummary/stream2py.utility.reader_writer_lock.html.md#module-stream2py.utility.reader_writer_lock)     | READER-WRITER LOCK WITH PRIORITY FOR WRITERS       |
| [`sorted_collection`](_autosummary/stream2py.utility.sorted_collection.html.md#module-stream2py.utility.sorted_collection)       | Sequence sorted by a key function.                 |
| [`sorted_deque`](_autosummary/stream2py.utility.sorted_deque.html.md#module-stream2py.utility.sorted_deque)                 | Deque sorted by a key function.                    |
| [`typing_hints`](_autosummary/stream2py.utility.typing_hints.html.md#module-stream2py.utility.typing_hints)                 | Typing hints                                       |


# _autosummary/stream2py.utility.locked_sorted_deque.html.md

# stream2py.utility.locked_sorted_deque

Sorted Deque that handles Reader-Writer Priority

### Classes

| [`RWLockSortedDeque`](_autosummary/stream2py.utility.locked_sorted_deque.html.md#stream2py.utility.locked_sorted_deque.RWLockSortedDeque)([iterable, key, maxlen])   |    |
|-----------------------------------------------------------------------------------------------|----|

### *class* stream2py.utility.locked_sorted_deque.RWLockSortedDeque(iterable=(), , key=None, maxlen=None)

Bases: [`RWLock`](_autosummary/stream2py.utility.reader_writer_lock.html.md#stream2py.utility.reader_writer_lock.RWLock)

```pycon
>>> import operator
>>> locked_deque = RWLockSortedDeque(
...     ((('plc', int(i // 10 * 10), i), f'data_{i}') for i in range(100)),
...     key=operator.itemgetter(0),
...     maxlen=100000)
>>> new_item = (('plc', 100000, 100000), 'new data')
>>> with locked_deque.writer_lock() as writer:
...     writer.append(new_item)
>>> with locked_deque.reader_lock() as reader:
...     print(reader.find_gt(('plc', 90, 91)))
...
(('plc', 90, 92), 'data_92')
>>> with locked_deque.reader_lock() as reader:
...     print(reader.find_gt(('plc', 90, 99)))
...
(('plc', 100000, 100000), 'new data')
```


# _autosummary/stream2py.utility.reader_writer_lock.html.md

# stream2py.utility.reader_writer_lock

READER-WRITER LOCK WITH PRIORITY FOR WRITERS

### Classes

| [`RWLock`](_autosummary/stream2py.utility.reader_writer_lock.html.md#stream2py.utility.reader_writer_lock.RWLock)()                                       | Synchronization object used in a solution of so-called second readers-writers problem.   |
|-------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------|
| [`RWLockTestCase`](_autosummary/stream2py.utility.reader_writer_lock.html.md#stream2py.utility.reader_writer_lock.RWLockTestCase)([methodName])                   |                                                                                          |
| [`Reader`](_autosummary/stream2py.utility.reader_writer_lock.html.md#stream2py.utility.reader_writer_lock.Reader)(buffer_, rw_lock, init_sleep_time, ...) |                                                                                          |
| [`Writer`](_autosummary/stream2py.utility.reader_writer_lock.html.md#stream2py.utility.reader_writer_lock.Writer)(buffer_, rw_lock, init_sleep_time, ...) |                                                                                          |

### *class* stream2py.utility.reader_writer_lock.RWLock

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)

Synchronization object used in a solution of so-called second
readers-writers problem. In this problem, many readers can simultaneously
access a share, and a writer has an exclusive access to this share.
Additionally, the following constraints should be met:

1. no reader should be kept waiting if the share is currently opened for
   : reading unless a writer is also waiting for the share,
2. no writer should be kept waiting for the share longer than absolutely
   : necessary.

The implementation is based on [1, secs. 4.2.2, 4.2.6, 4.2.7]
with a modification – adding an additional lock (C{self._\_readers_queue})
– in accordance with [2].

Sources:
[1] A.B. Downey: “The little book of semaphores”, Version 2.1.5, 2008
[2] P.J. Courtois, F. Heymans, D.L. Parnas:

```default
"Concurrent Control with 'Readers' and 'Writers'",
Communications of the ACM, 1971 (via [3])
```

[3] [http://en.wikipedia.org/wiki/Readers-writers_problem](http://en.wikipedia.org/wiki/Readers-writers_problem)

### *class* stream2py.utility.reader_writer_lock.RWLockTestCase(methodName='runTest')

Bases: [`TestCase`](https://docs.python.org/3/library/unittest.html#unittest.TestCase)

### *class* stream2py.utility.reader_writer_lock.Reader(buffer_, rw_lock, init_sleep_time, sleep_time)

Bases: [`Thread`](https://docs.python.org/3/library/threading.html#threading.Thread)

#### buffer_read

a copy of a the buffer read while in critical section

#### entry_time

Time of entry to the critical section

#### exit_time

Time of exit from the critical section

#### run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method
invokes the callable object passed to the object’s constructor as the
target argument, if any, with sequential and keyword arguments taken
from the args and kwargs arguments, respectively.

### *class* stream2py.utility.reader_writer_lock.Writer(buffer_, rw_lock, init_sleep_time, sleep_time, to_write)

Bases: [`Thread`](https://docs.python.org/3/library/threading.html#threading.Thread)

#### entry_time

Time of entry to the critical section

#### exit_time

Time of exit from the critical section

#### run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method
invokes the callable object passed to the object’s constructor as the
target argument, if any, with sequential and keyword arguments taken
from the args and kwargs arguments, respectively.


# _autosummary/stream2py.utility.sorted_collection.html.md

# stream2py.utility.sorted_collection

Sequence sorted by a key function.

### Classes

| [`SortedCollection`](_autosummary/stream2py.utility.sorted_collection.html.md#stream2py.utility.sorted_collection.SortedCollection)([iterable, key])   | Sequence sorted by a key function.   |
|--------------------------------------------------------------------------------------|--------------------------------------|

### *class* stream2py.utility.sorted_collection.SortedCollection(iterable=(), key=None)

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object)

Sequence sorted by a key function.

SortedCollection() is much easier to work with than using bisect() directly.
It supports key functions like those use in sorted(), min(), and max().
The result of the key function call is saved so that keys can be searched
efficiently.

Instead of returning an insertion-point which can be hard to interpret, the
five find-methods return a specific item in the sequence. They can scan for
exact matches, the last item less-than-or-equal to a key, or the first item
greater-than-or-equal to a key.

Once found, an item’s ordinal position can be located with the index() method.
New items can be added with the insert() and insert_right() methods.
Old items can be deleted with the remove() method.

The usual sequence methods are provided to support indexing, slicing,
length lookup, clearing, copying, forward and reverse iteration, contains
checking, item counts, item removal, and a nice looking repr.

Finding and indexing are O(log n) operations while iteration and insertion
are O(n).  The initial sort is O(n log n).

The key function is stored in the ‘key’ attibute for easy introspection or
so that you can assign a new key function (triggering an automatic re-sort).

In short, the class was designed to handle all of the common use cases for
bisect but with a simpler API and support for key functions.

```pycon
>>> from pprint import pprint
>>> from operator import itemgetter
```

```pycon
>>> s = SortedCollection(key=itemgetter(2))
>>> for record in [
...         ('roger', 'young', 30),
...         ('angela', 'jones', 28),
...         ('bill', 'smith', 22),
...         ('david', 'thomas', 32)]:
...     s.insert(record)
```

```pycon
>>> pprint(list(s))         # show records sorted by age
[('bill', 'smith', 22),
 ('angela', 'jones', 28),
 ('roger', 'young', 30),
 ('david', 'thomas', 32)]
```

```pycon
>>> s.find_le(29)           # find oldest person aged 29 or younger
('angela', 'jones', 28)
>>> s.find_lt(28)           # find oldest person under 28
('bill', 'smith', 22)
>>> s.find_gt(28)           # find youngest person over 28
('roger', 'young', 30)
```

```pycon
>>> r = s.find_ge(32)       # find youngest person aged 32 or older
>>> s.index(r)              # get the index of their record
3
>>> s[3]                    # fetch the record at that index
('david', 'thomas', 32)
```

```pycon
>>> s.key = itemgetter(0)   # now sort by first name
>>> pprint(list(s))
[('angela', 'jones', 28),
 ('bill', 'smith', 22),
 ('david', 'thomas', 32),
 ('roger', 'young', 30)]
```

#### count(item)

Return number of occurrences of item

#### find(k)

Return first item with a key == k.  Raise ValueError if not found.

#### find_ge(k)

Return first item with a key >= equal to k.  Raise ValueError if not found

#### find_gt(k)

Return first item with a key > k.  Raise ValueError if not found

#### find_le(k)

Return last item with a key <= k.  Raise ValueError if not found.

#### find_lt(k)

Return last item with a key < k.  Raise ValueError if not found.

#### index(item)

Find the position of an item.  Raise ValueError if not found.

#### insert(item)

Insert a new item.  If equal keys are found, add to the left

#### insert_right(item)

Insert a new item.  If equal keys are found, add to the right

#### *property* key

key function

#### remove(item)

Remove first occurence of item.  Raise ValueError if not found


# _autosummary/stream2py.utility.sorted_deque.html.md

# stream2py.utility.sorted_deque

Deque sorted by a key function.

### Classes

| [`SortedDeque`](_autosummary/stream2py.utility.sorted_deque.html.md#stream2py.utility.sorted_deque.SortedDeque)([iterable, key, maxlen])   | Deque sorted by a key function.   |
|-----------------------------------------------------------------------------------------|-----------------------------------|

### *class* stream2py.utility.sorted_deque.SortedDeque(iterable=(), key=None, maxlen=None)

Bases: [`SortedCollection`](_autosummary/stream2py.utility.sorted_collection.html.md#stream2py.utility.sorted_collection.SortedCollection)

Deque sorted by a key function.

Intent of SortedDeque is to efficiently search buffered data while efficiently maintaining a fixed length buffer
that is frequently popping the head and appending to the tail.

Overloaded SortedCollection methods \_\_contains_\_, index, and count due to the lack of
slicing in deques. Efficiency of said methods is still better than base deque.

Added an append method for cases where the new item to insert is known to be the last.

#### append(item)

Append item to the end and maintain key indexing.
Raise ValueError if item key is not greater than last item key.

* **Parameters:**
  **item** – item to append
* **Returns:**
  None

#### count(item)

Return number of occurrences of item

#### drop(n=1)

Remove n items from the left

* **Parameters:**
  **n**
* **Returns:**

#### find_last_gt(k)

Return last item with a key > k.  Raise ValueError if not found.
Basically a tail function with a greater than check

#### index(item)

Find the position of an item.  Raise ValueError if not found.

#### *property* maxlen

maxlen function

#### range(start, stop, step=None)

Return list of items within start and stop key range.

* **Parameters:**
  * **start** – starting key
  * **stop** – stopping key
  * **step**
* **Returns:**
  list with items in range

#### range_by_index(start_index, stop_index, step=None)

Return list of items within start and stop index range.

* **Parameters:**
  * **start_index** – starting index
  * **stop_index** – stopping index
  * **step**
* **Returns:**
  list with items in range

#### to_jdict()


# _autosummary/stream2py.utility.typing_hints.html.md

# stream2py.utility.typing_hints

Typing hints

### Classes

| [`ComparableType`](_autosummary/stream2py.utility.typing_hints.html.md#stream2py.utility.typing_hints.ComparableType)(\*args, \*\*kwargs)   |    |
|---------------------------------------------------------------------------------------|----|

### *class* stream2py.utility.typing_hints.ComparableType(\*args, \*\*kwargs)

Bases: [`Protocol`](https://docs.python.org/3/library/typing.html#typing.Protocol)


# about-this-build.html.md

<!-- generated by epythet -->

# About this build

This documentation was built on **2026-10-20 11:28 UTC** from commit <a href="https://github.com/i2mint/stream2py/commit/07dc7cfc7b24180f05ed284a67b4a1bf94a1bcc1"><code>07dc7cf</code></a> on branch <code>master</code>, for **stream2py 1.0.48** (from <code>pyproject.toml</code>).

#### WARNING
The documentation and the package may be misaligned:

- The documented version (1.0.48) is behind the latest release on PyPI (1.0.49): `pip install stream2py` gives newer code than these docs describe.

## Source

|                     |                                                                                                                                                         |
|---------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------|
| Commit              | <a href="https://github.com/i2mint/stream2py/commit/07dc7cfc7b24180f05ed284a67b4a1bf94a1bcc1"><code>07dc7cfc7b24180f05ed284a67b4a1bf94a1bcc1</code></a> |
| Branch              | <code>master</code>                                                                                                                                     |
| Tags at this commit | none                                                                                                                                                    |
| Working tree        | clean                                                                                                                                                   |
| Remote              | <code>https://github.com/i2mint/stream2py</code>                                                                                                        |

## Continuous integration

|              |                                                                                            |
|--------------|--------------------------------------------------------------------------------------------|
| Repository   | <code>i2mint/stream2py</code>                                                              |
| Run          | <a href="https://github.com/i2mint/stream2py/actions/runs/39414913420">39414913420</a>     |
| Ref          | <code>refs/heads/master</code>                                                             |
| Event commit | <code>07dc7cfc7b24180f05ed284a67b4a1bf94a1bcc1</code> (in the history of the built commit) |

## Tools

|          |         |
|----------|---------|
| epythet  | 0.2.19  |
| Sphinx   | 9.1.1   |
| docutils | 0.22.4  |
| Python   | 3.12.15 |

## Configuration as resolved

|               |                                                                  |
|---------------|------------------------------------------------------------------|
| theme         | <code>auto</code> (Sphinx theme <code>shibuya</code>)            |
| accent        | <code>#823774</code>                                             |
| api_generator | <code>autosummary</code>                                         |
| ignore        | <code>tests/</code>, <code>scrap/</code>, <code>examples/</code> |
| agent_outputs | <code>true</code>                                                |
| aggregates    | <code>md</code>                                                  |
| ai_artifacts  | <code>true</code>                                                |

## Package on PyPI

Latest release: <a href="https://pypi.org/project/stream2py/1.0.49/">1.0.49</a>, newer than the documented version (1.0.48).

## Reproduce

```bash
git clone https://github.com/i2mint/stream2py && cd stream2py
git checkout 07dc7cfc7b24180f05ed284a67b4a1bf94a1bcc1
pip install "epythet==0.2.19"
epythet quickstart . --ignore tests/ scrap/ examples/
```

The same data, for machines: <a href="build_info.json"><code>build_info.json</code></a> (schema version 1).


# api.html.md

# API reference

| [`stream2py`](_autosummary/stream2py.html.md#module-stream2py)   |    |
|-------------------------------------------------------------------------------|----|