Introduction

The primary use cases of eventkit are

• to send events between loosely coupled components;

• to compose all kinds of event-driven data pipelines.

The interface is kept as Pythonic as possible, with familiar names from Python and its libraries where possible. For scheduling asyncio is used and there is seamless integration with it.

See the examples and the introduction notebook to get a true feel for the possibilities.

Note: the package is published as aeventkit and imported as eventkit.

Installation

pip install aeventkit

pip install aeventkit[numpy]  # with the Array operators

Python 3.11 or later is required (tested on CPython 3.11 to 3.15 and PyPy 3.11).

Upgrading from 2.x? See Upgrading to 3.0 and the changelog.

Examples

Create an event and connect two listeners

import eventkit as ev

def f(a, b):
    print(a * b)

def g(a, b):
    print(a / b)

event = ev.Event()
event += f
event += g
event.emit(10, 5)

Create a simple pipeline

import eventkit as ev

event = (
    ev.Sequence('abcde')
    .map(str.upper)
    .enumerate()
)

print(event.run())  # in Jupyter: await event.list()

Output:

[(0, 'A'), (1, 'B'), (2, 'C'), (3, 'D'), (4, 'E')]

Create a pipeline to get a running average and standard deviation

(with the numpy extra: pip install aeventkit[numpy])

import random
import eventkit as ev

source = ev.Range(1000).map(lambda i: random.gauss(0, 1))

event = source.array(500)[ev.ArrayMean, ev.ArrayStd].zip()

print(event.last().run())  # in Jupyter: await event.last()

Output:

[(0.00790957852672618, 1.0345673260655333)]

Combine async iterators together

import asyncio
import eventkit as ev

async def ait(r):
    for i in r:
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for t in ev.Zip(ait('XYZ'), ait('123')):
        print(t)

asyncio.run(main())  # in Jupyter: await main()

Output:

('X', '1')
('Y', '2')
('Z', '3')

Real-time video analysis pipeline

self.video = VideoStream(conf.CAM_ID)
scene = self.video | FaceTracker | SceneAnalyzer
lastScene = scene.aiter(skip_to_last=True)
async for frame, persons in lastScene:
    ...

Full source code

Parallel processing

poolmap runs a function on the values of an event in a concurrent.futures executor: a process pool for CPU-bound work on all cores, an interpreter pool (Python 3.14+), or a thread pool for blocking I/O. Results stay in order unless ordered=False, and chunksize batches small tasks:

import bz2
from concurrent.futures import ProcessPoolExecutor

import eventkit as ev

if __name__ == "__main__":
    data = [b"A" * 1_000_000] * 1000
    with ProcessPoolExecutor() as pool:
        sizes = ev.Sequence(data).poolmap(pool, bz2.compress, chunksize=10).map(len)
        print(sizes.mean().last().run())

Inspired by:

• Qt Signals & Slots

• itertools

• aiostream

• Bacon

• aioreactive

• Reactive extensions

• underscore.js

• .NET Events

Documentation

The complete API documentation.