### Complex Event Pipeline Example

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

A complex EventKit pipeline demonstrating map, star, take, and connect operations, culminating in plotting results.

```python
import eventkit as ev
import asyncio
import matplotlib.pyplot as plt

async def wave(a, b):
    c = 100
    return a + b / c, b - a / c


start = ev.Event()
pipe = start.map(wave).star().take(1000).connect(start.emit).list()
start.emit(0, 1)

await pipe;

plt.plot(pipe.value());
```

--------------------------------

### Event Processing Chain with Print Listener

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

An example of a complex event processing chain that includes using `print` as a listener within the chain.

```python
event = (
    ev.Sequence("abcde")
    .enumerate()
    .map(lambda i, c: (i, i + ord(c)))
    .connect(print)
    .star()
    .pluck(1)
    .map(chr)
)

await event.list()
```

--------------------------------

### Chainmap Example

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/api.html

Demonstrates using chainmap to combine nested events from a sequence.

```python
marbles = [
    'A    B    C    D           ',
    '\_       1    2    3    4',
    '\__                  K    L      M   N'
]

ev.Range(3).chainmap(lambda v: ev.Marble(marbles[v]))
# ->
# ['A', 'B', 'C', 'D', '1', '2', '3', '4', 'K', 'L', 'M', 'N']
```

--------------------------------

### Switchmap Example

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/api.html

Illustrates using switchmap to combine nested events, switching to the latest inner observable.

```python
marbles = [
    'A    B    C    D           ',
    '\_                 K    L      M   N',
    '\__      1    2      3    4'
]
ev.Range(3).switchmap(lambda v: Event.marble(marbles[v]))
# ->
# ['A', 'B', '1', '2', 'K', 'L', 'M', 'N']
```

--------------------------------

### Watcher Usage Example

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/asyncio/unix_events.html

Demonstrates how to use the child watcher to monitor subprocesses. It shows starting a process within a 'with' block and registering a callback for its termination.

```python
import asyncio
import signal
import os
import warnings

# Assuming AbstractChildWatcher, BaseChildWatcher, and SafeChildWatcher are defined elsewhere
# For demonstration, let's create dummy classes if they are not available in the context

class DummyAbstractEventLoop:
    def add_signal_handler(self, signum, handler):
        print(f"Added signal handler for {signum}")
    def remove_signal_handler(self, signum):
        print(f"Removed signal handler for {signum}")
    def call_exception_handler(self, exc_info):
        print(f"Exception handler called: {exc_info}")

class DummyAbstractChildWatcher:
    def add_child_handler(self, pid, callback, *args):
        raise NotImplementedError()
    def remove_child_handler(self, pid):
        raise NotImplementedError()
    def attach_loop(self, loop):
        raise NotImplementedError()
    def close(self):
        raise NotImplementedError()
    def __enter__(self):
        raise NotImplementedError()
    def __exit__(self, a, b, c):
        raise NotImplementedError()

class DummyBaseChildWatcher(DummyAbstractChildWatcher):
    def __init__(self):
        self._loop = None
        self._callbacks = {}

    def close(self):
        self.attach_loop(None)

    def _do_waitpid(self, expected_pid):
        raise NotImplementedError()

    def _do_waitpid_all(self):
        raise NotImplementedError()

    def attach_loop(self, loop):
        assert loop is None or isinstance(loop, DummyAbstractEventLoop)
        if self._loop is not None and loop is None and self._callbacks:
            warnings.warn(
                'A loop is being detached '
                'from a child watcher with pending handlers',
                RuntimeWarning)
        if self._loop is not None:
            self._loop.remove_signal_handler(signal.SIGCHLD)
        self._loop = loop
        if loop is not None:
            loop.add_signal_handler(signal.SIGCHLD, self._sig_chld)
            self._do_waitpid_all()

    def _sig_chld(self):
        try:
            self._do_waitpid_all()
        except Exception as exc:
            self._loop.call_exception_handler({
                'message': 'Unknown exception in SIGCHLD handler',
                'exception': exc,
            })

    def _compute_returncode(self, status):
        if os.WIFSIGNALED(status):
            return -os.WTERMSIG(status)
        elif os.WIFEXITED(status):
            return os.WEXITSTATUS(status)
        else:
            return status

class DummySafeChildWatcher(DummyBaseChildWatcher):
    def close(self):
        self._callbacks.clear()
        super().close()

    def __enter__(self):
        return self

    def __exit__(self, a, b, c):
        pass

    def add_child_handler(self, pid, callback, *args):
        if self._loop is None:
            raise RuntimeError(
                'Cannot add child handler without an attached event loop'
            )
        self._callbacks[pid] = (callback, args)

# --- Example Usage ---

def my_callback(pid, returncode, *args):
    print(f"Process {pid} terminated with return code {returncode}. Args: {args}")

# Create a dummy event loop and watcher
loop = DummyAbstractEventLoop()
watcher = DummySafeChildWatcher()

watcher.attach_loop(loop)

# Example of starting a process and monitoring it
# Note: subprocess.Popen is a placeholder here as we are in a dummy context
# In a real scenario, you would use subprocess.Popen

# Simulate a process PID
simulated_proc_pid = 12345

# Using the watcher in a 'with' block
with watcher:
    print("Inside watcher context...")
    # In a real scenario, you would start a process here:
    # proc = subprocess.Popen("sleep 1")
    # watcher.add_child_handler(proc.pid, my_callback, "extra_arg")
    
    # For demonstration, we add a handler directly
    watcher.add_child_handler(simulated_proc_pid, my_callback, "extra_arg")
    print(f"Handler added for PID {simulated_proc_pid}")

print("Exited watcher context.")

# Simulate SIGCHLD signal triggering the callback
# In a real scenario, this would be handled by the OS and the signal handler
# For demonstration, we manually call the internal handler if it were implemented
# watcher._sig_chld() # This would typically be called by the signal handler

watcher.close()
print("Watcher closed.")

```

--------------------------------

### Timerange Event Example

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Demonstrates creating a Timerange event that produces absolute datetimes at specified intervals. It uses `await event.last()` to capture the output.

```python
event = ev.Timerange(0, 2, 0.25)
event += print

# await event.last() # This would be used in an async context
```

--------------------------------

### Timer Event Example

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Illustrates the use of a Timer event to emit a value a specified number of times with a delay. It uses `await timer.last()` to keep the output visible.

```python
timer = ev.Timer(0.25, count=10)
timer += print

# await timer.last() # This would be used in an async context
```

--------------------------------

### Logger Initialization and Configuration

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/logging.html

This section covers the initialization of the root logger and manager, and the `basicConfig` function for setting up basic logging configurations. `basicConfig` allows customization of handlers, formatters, and log levels for simple script setups.

```python
root = RootLogger(WARNING)
Logger.root = root
Logger.manager = Manager(Logger.root)

def basicConfig(**kwargs):
    """
    Do basic configuration for the logging system.

    This function does nothing if the root logger already has handlers
    configured. It is a convenience method intended for use by simple scripts
    to do one-shot configuration of the logging package.

    The default behaviour is to create a StreamHandler which writes to
    sys.stderr, set a formatter using the BASIC_FORMAT format string, and
    add the handler to the root logger.

    A number of optional keyword arguments may be specified, which can alter
    the default behaviour.

    filename  Specifies that a FileHandler be created, using the specified
              filename, rather than a StreamHandler.
    filemode  Specifies the mode to open the file, if filename is specified
              (if filemode is unspecified, it defaults to 'a').
    format    Use the specified format string for the handler.
    datefmt   Use the specified date/time format.
    style     If a format string is specified, use this to specify the
              type of format string (possible values '%', '{', '$', for
              %-formatting, :meth:`str.format` and :class:`string.Template`
              - defaults to '%').
    level     Set the root logger level to the specified level.
    stream    Use the specified stream to initialize the StreamHandler. Note
              that this argument is incompatible with 'filename' - if both
              are present, 'stream' is ignored.
    handlers  If specified, this should be an iterable of already created
              handlers, which will be added to the root handler. Any handler
              in the list which does not have a formatter assigned will be
              assigned the formatter created in this function.

    Note that you could specify a stream created using open(filename, mode)
    rather than passing the filename and mode in. However, it should be
    remembered that StreamHandler does not close its stream (since it may be
    using sys.stdout or sys.stderr), whereas FileHandler closes its stream
    when the handler is closed.

    .. versionchanged:: 3.2
       Added the `style` parameter.

    .. versionchanged:: 3.3
       Added the `handlers` parameter. A `ValueError` is now thrown for
       incompatible arguments (e.g. `handlers` specified together with

```

--------------------------------

### Event Class Initialization

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/event.html

Demonstrates the initialization of the Event class, including the creation of optional error and done sub-events, and the setup of internal slots for listeners.

```python
class Event:
    """
    Enable event passing between loosely coupled components.
    The event emits values to connected listeners and has
    a selection of operators to create general data flow pipelines.

    Args:
        name: Name to use for this event.
    """

    __slots__ = (
        'error_event', 'done_event', '_name', '_value',
        '_slots', '_done', '_source', '__weakref__')

    NO_VALUE = NO_VALUE
    logger = logging.getLogger(__name__)

    error_event: Optional["Event"]
    done_event: Optional["Event"]
    _name: str
    _value: AnyType
    _slots: List[List]
    _done: bool
    _source: Optional["Event"]

    def __init__(self, name: str = '', _with_error_done_events: bool = True):
        self.error_event = None
        """
        Sub event that emits errors from this event as
        `emit(source, exception)`.
        """
        self.done_event = None
        """
        Sub event that emits when this event is done as
        `emit(source)`.
        """
        if _with_error_done_events:
            self.error_event = Event('error', False)
            self.done_event = Event('done', False)
        self._slots = []  # list of [obj, weakref, func] sublists
        self._name = name or self.__class__.__qualname__
        self._value = NO_VALUE
        self._done = False
        self._source = None
```

--------------------------------

### Basic Logging Configuration

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/logging.html

Configures the root logger with handlers, formatters, and levels. It handles thread safety and ensures proper setup for logging output.

```python
def basicConfig(**kwargs):
    """
    DobasicConfig() does nothing if the root logger already has handlers configured.
    Otherwise, it adds a handler to the root logger. The handler is a
    StreamHandler that writes to sys.stderr. The logging level is set to
    WARNING. The message format is:
    '%(levelname)s:%(name)s:%(message)s'

    The handler is created from the following keyword arguments, if supplied:

    filename=None
        Name of the file to log to. If this is not specified, the output
        will be sent to sys.stderr.
    filemode='a'
        The mode to open the file specified by filename. Defaults to 'a' (append).
    format=None
        The format for the output. If this is not specified, a default
        format will be used.
    datefmt=None
        The format for date/time, which is used by the Formatter. If not
        specified, the format is '%Y-%m-%dT%H:%M:%S'
    style='%' 
        Indicates the type of format string to be used. The default is '%'. 
        Other supported styles are '{' and '$'.
    level=0
        The lowest logging level that will be processed by the root logger.
        The default is WARNING.
    stream=None
        Use the given stream to output logging data. If this is not specified,
        and a filename is not specified, the output will be sent to sys.stderr.
    handlers=None
        A list of handlers to add to the root logger. If this is specified,
        then filename, filemode, format, datefmt, style, level, and stream
        are ignored.
    """
    # Add thread safety in case someone mistakenly calls
    # basicConfig() from multiple threads
    _acquireLock()
    try:
        if len(root.handlers) == 0:
            handlers = kwargs.pop("handlers", None)
            if handlers is None:
                if "stream" in kwargs and "filename" in kwargs:
                    raise ValueError("'stream' and 'filename' should not be "
                                     "specified together")
            else:
                if "stream" in kwargs or "filename" in kwargs:
                    raise ValueError("'stream' or 'filename' should not be "
                                     "specified together with 'handlers'")
            if handlers is None:
                filename = kwargs.pop("filename", None)
                mode = kwargs.pop("filemode", 'a')
                if filename:
                    h = FileHandler(filename, mode)
                else:
                    stream = kwargs.pop("stream", None)
                    h = StreamHandler(stream)
                handlers = [h]
            dfs = kwargs.pop("datefmt", None)
            style = kwargs.pop("style", '%')
            if style not in _STYLES:
                raise ValueError('Style must be one of: %s' % ','.join(
                                 _STYLES.keys()))
            fs = kwargs.pop("format", _STYLES[style][1])
            fmt = Formatter(fs, dfs, style)
            for h in handlers:
                if h.formatter is None:
                    h.setFormatter(fmt)
                root.addHandler(h)
            level = kwargs.pop("level", None)
            if level is not None:
                root.setLevel(level)
            if kwargs:
                keys = ', '.join(kwargs.keys())
                raise ValueError('Unrecognised argument(s): %s' % keys)
    finally:
        _releaseLock()
```

--------------------------------

### UNIX Domain Socket Server Setup

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/asyncio/unix_events.html

Configures and binds a UNIX domain stream socket for server operations. Handles potential stale sockets and address-in-use errors. Supports abstract namespace sockets.

```python
            path = os.fspath(path)
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

            # Check for abstract socket. `str` and `bytes` paths are supported.
            if path[0] not in (0, '\x00'):
                try:
                    if stat.S_ISSOCK(os.stat(path).st_mode):
                        os.remove(path)
                except FileNotFoundError:
                    pass
                except OSError as err:
                    # Directory may have permissions only to create socket.
                    logger.error('Unable to check or remove stale UNIX socket ' \
                                 '%r: %r', path, err)

            try:
                sock.bind(path)
            except OSError as exc:
                sock.close()
                if exc.errno == errno.EADDRINUSE:
                    # Let's improve the error message by adding
                    # with what exact address it occurs.
                    msg = f'Address {path!r} is already in use'
                    raise OSError(errno.EADDRINUSE, msg) from None
                else:
                    raise
            except:
                sock.close()
                raise
        else:
            if sock is None:
                raise ValueError(
                    'path was not specified, and no sock specified')

            if (
                sock.family != socket.AF_UNIX or
                sock.type != socket.SOCK_STREAM
            ):
                raise ValueError(
                    f'A UNIX Domain Stream Socket was expected, got {sock!r}')

        sock.setblocking(False)
        server = base_events.Server(self, [sock], protocol_factory,
                                    ssl, backlog, ssl_handshake_timeout)
        if start_serving:
            server._start_serving()
            # Skip one loop iteration so that all 'loop.add_reader'
            # go through.
            await tasks.sleep(0, loop=self)

        return server
```

--------------------------------

### Unix Subprocess Transport Initialization

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/asyncio/unix_events.html

Handles the setup and initialization of a Unix subprocess transport, including managing standard input redirection using socket pairs for cross-platform compatibility.

```python
class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):

    def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
        stdin_w = None
        if stdin == subprocess.PIPE:
            # Use a socket pair for stdin, since not all platforms
            # support selecting read events on the write end of a
            # socket (which we use in order to detect closing of the
            # other end).  Notably this is needed on AIX, and works
            # just fine on other platforms.
            stdin, stdin_w = socket.socketpair()
        try:
            self._proc = subprocess.Popen(
                args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
                universal_newlines=False, bufsize=bufsize, **kwargs)
            if stdin_w is not None:
                stdin.close()
                self._proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
                stdin_w = None
        finally:
            if stdin_w is not None:
                stdin.close()
                stdin_w.close()
```

--------------------------------

### Applying Event Operators to Futures

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Demonstrates using EventKit operators like Zip with a list of asyncio Futures.

```python
import eventkit as ev
import asyncio

async def getFut(result):
    fut = asyncio.Future()
    loop = asyncio.get_event_loop()
    loop.call_later(1, fut.set_result, result)
    return fut

futs = [getFut(i) for i in range(10)]

await ev.Zip(*futs)
```

--------------------------------

### Python Logger Class Methods

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/logging.html

Details methods of the Python Logger class, including getting effective logging levels, checking if a level is enabled, getting child loggers, and representation methods.

```python
def getEffectiveLevel(self):
    """
    Get the effective level for this logger.

    Loop through this logger and its parents in the logger hierarchy,
    looking for a non-zero logging level. Return the first one found.
    """
    logger = self
    while logger:
        if logger.level:
            return logger.level
        logger = logger.parent
    return NOTSET

def isEnabledFor(self, level):
    """
    Is this logger enabled for level 'level'?
    """
    try:
        return self._cache[level]
    except KeyError:
        _acquireLock()
        if self.manager.disable >= level:
            is_enabled = self._cache[level] = False
        else:
            is_enabled = self._cache[level] = level >= self.getEffectiveLevel()
        _releaseLock()

        return is_enabled

def getChild(self, suffix):
    """
    Get a logger which is a descendant to this one.

    This is a convenience method, such that

    logging.getLogger('abc').getChild('def.ghi')

is the same as

    logging.getLogger('abc.def.ghi')

    It's useful, for example, when the parent logger is named using
    __name__ rather than a literal string.
    """
    if self.root is not self:
        suffix = '.'.join((self.name, suffix))
    return self.manager.getLogger(suffix)

def __repr__(self):
    level = getLevelName(self.getEffectiveLevel())
    return '<%s %s (%s)>' % (self.__class__.__name__, self.name, level)

def __reduce__(self):
    # In general, only the root logger will not be accessible via its name.
    # However, the root logger's class has its own __reduce__ method.
    if getLogger(self.name) is not self:
        import pickle
        raise pickle.PicklingError('logger cannot be pickled')
    return getLogger, (self.name,)
```

--------------------------------

### Piping and Forking: Sampling Min/Max

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Illustrates forking an event stream to apply different operators (Min, Max) and then combining the results using zip.

```python
from math import sin

event = ev.Range(6).map(sin)[ev.Min, ev.Max].zip()

await event.list()
```

--------------------------------

### Connecting and Emitting Events

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Demonstrates how to create an Event, connect listener functions (f, g, print), emit values, and disconnect listeners using '+=' and '-=' operators.

```python
import eventkit as ev

def f(i):
    print("f got", i)

def g(i):
    print("g got", i)

event = ev.Event()
event += f
event += g
event.emit(42)

event += print
event.emit(43)

event -= g
event -= print
event.emit(44)

event.clear()
event.emit(45)
```

--------------------------------

### Event Enumeration

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/api.html

Augments each source event emission with a sequential count. Supports custom start and step values for the count.

```python
enumerate(_start=0_, _step=1_)
```

--------------------------------

### Sample Class Initialization and Methods

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/timing.html

Details the Sample class, which is used for sampling operations based on a timer. It includes initialization, handling source data, and managing timer connections and disconnections.

```python
class Sample(Op):
    __slots__ = ('_timer',)

    def __init__(self, timer, source=None):
        Op.__init__(self, source)
        self._timer = timer
        timer.connect(
            self._on_timer,
            self.on_source_error,
            self.on_source_done)

    def on_source(self, *args):
        self._value = args

    def _on_timer(self, *args):
        if self._value is not NO_VALUE:
            self.emit(*self._value)

    def on_source_done(self, source):
        Op.on_source_done(self, self)
        self._timer.disconnect(
            self._on_timer,
            self.on_source_error,
            self.on_source_done)
        self._timer = None
```

--------------------------------

### Emitting Multiple Arguments

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Shows how to emit multiple arguments to connected listeners and how to connect class methods as listeners.

```python
def f(a, b):
    print("f got", a, b)

event = ev.Event()
event += f
event.emit(7, "A")

class A:
    def on_event(self, s, t):
        print("on_event got", s, t)

a = A()
event = ev.Event()
event += a.on_event
event.emit("Whats", "up?")

del a
event.emit("Still", "there?")
```

--------------------------------

### Combine Events with Chain

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Illustrates chaining event streams, processing them sequentially and queuing emissions from subsequent sources.

```python
e1 = ev.Sequence("ABC", 0.01)
e2 = ev.Sequence("123", 0.02)
e3 = ev.Sequence("XYZ", 0.03)

event = e1.chain(e2, e3)

await event.list()
```

--------------------------------

### Event Counting

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/api.html

Emits a count corresponding to the number of emissions received from the source event. Supports custom start and step values for the count.

```python
count(_start=0_, _step=1_)
```

--------------------------------

### EventKit Pipeline Construction Syntaxes

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/ARCHITECTURE.md

Demonstrates multiple equivalent syntaxes for building event pipelines in EventKit, including method chaining, the pipe operator, explicit piping, and constructor chaining.

```python
import eventkit

# Method chaining (most common)
event = source.map(func).filter(pred).take(10)

# Pipe operator
event = source | eventkit.Map(func) | eventkit.Filter(pred) | eventkit.Take(10)

# Explicit piping
event = source.pipe(eventkit.Map(func), eventkit.Filter(pred), eventkit.Take(10))

# Constructor chaining
event = eventkit.Take(10, eventkit.Filter(pred, eventkit.Map(func, source)))
```

--------------------------------

### Pattern Matching Example (Python 3.10+)

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/TODO.md

Utilize Python 3.10+ structural pattern matching for cleaner and more readable complex conditional logic.

```python
def process_command(command):
    match command:
        case {"action": "create", "data": data_payload}:
            print(f"Creating with data: {data_payload}")
        case {"action": "delete", "id": item_id}:
            print(f"Deleting item with ID: {item_id}")
        case {"action": "update", "id": item_id, "data": update_data}:
            print(f"Updating item {item_id} with data: {update_data}")
        case _:
            print("Unknown command")
```

--------------------------------

### Ziplatest Class Initialization and Source Handling

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/combine.html

Initializes the Ziplatest operator, which combines values from multiple event sources. It manages the state of primed sources and callbacks for each source. The `_set_sources` method prepares the initial sources, extracts their values, and connects callbacks to handle updates and completion.

```python
class Ziplatest(JoinOp):
    __slots__ = ('_values', '_is_primed', '_source2cbs')

    def __init__(self, *sources, partial=True):
        JoinOp.__init__(self)
        self._is_primed = partial
        self._source2cbs = defaultdict(list)  # map from source to callbacks
        if sources:
            self._set_sources(*sources)

    def _set_sources(self, *sources):
        sources = [Event.create(s) for s in sources]
        self._sources = deque(s for s in sources if not s.done())
        if not self._sources:
            self.set_done()
            return
        self._values = [s.value() for s in sources]
        for i, source in enumerate(self._sources):
            cb = functools.partial(self._on_source_i, i)
            source.connect(cb, self.on_source_error, self.on_source_done)
            self._source2cbs[source].append(cb)
```

--------------------------------

### Pluck Operator for Suffix

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Illustrates the `pluck` operator to extract a specific property (file suffix) from emitted objects (Path objects) and `unique` to get distinct suffixes.

```python
from pathlib import Path

# Assuming there are files in the current directory
# files = Path(".").glob("*")

# event = ev.Sequence(files).pluck(".suffix").unique()

# await event.list() # This would be used in an async context
```

--------------------------------

### Piping and Forking: Multiple Delays

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Demonstrates forking an event stream to apply multiple delay operators and then merging the results.

```python
event = ev.Range(5, interval=0.1)[
    ev.Op, ev.Delay(0.1), ev.Delay(0.2), ev.Delay(0.5)
].merge()

await event.list()
```

--------------------------------

### Create a Simple EventKit Pipeline

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/README.rst

Illustrates building a basic data processing pipeline using EventKit's sequence, map, and enumerate operators. The pipeline transforms a sequence of characters.

```python
import eventkit as ev

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

print(event.run())
```

--------------------------------

### Create Event and Connect Listeners

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/README.rst

Demonstrates how to create an event, attach listener functions to it, and emit events with arguments. This is fundamental for inter-component communication.

```python
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)
```

--------------------------------

### TaskGroup Usage Example (Python 3.11+)

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/TODO.md

Employ `asyncio.TaskGroup` for improved concurrent task management, resource handling, and error propagation in Python 3.11+.

```python
import asyncio

async def task1():
    await asyncio.sleep(1)
    return "Result 1"

async def task2():
    await asyncio.sleep(2)
    return "Result 2"

async def main():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(task1())
        task_b = tg.create_task(task2())

    print(f"Task A finished with: {task_a.result()}")
    print(f"Task B finished with: {task_b.result()}")

asyncio.run(main())
```

--------------------------------

### Count Operation

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/aggregate.html

The Count class implements an operation that counts items in an event stream. It takes an optional start value and step for the count.

```python
import operator
import itertools
from collections import deque

from ..util import NO_VALUE
from .op import Op
from .transform import Iterate

class Count(Iterate):
    __slots__ = ()

    def __init__(self, start=0, step=1, source=None):
        it = itertools.count(start, step)
        Iterate.__init__(self, it, source)
```

--------------------------------

### Logger Class Management

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/logging.html

Functions to set and get the logger class used by the logging system. Ensures the provided class is a subclass of the base Logger.

```python
def setLoggerClass(klass):
    """
    Set the class to be used when instantiating a logger. The class should
    define __init__() such that only a name argument is required, and the
    __init__() should call Logger.__init__()
    """
    if klass != Logger:
        if not issubclass(klass, Logger):
            raise TypeError("logger not derived from logging.Logger: "
                            + klass.__name__)
    global _loggerClass
    _loggerClass = klass

def getLoggerClass():
    """
    Return the class to be used when instantiating a logger.
    """
    return _loggerClass
```

--------------------------------

### EventKit Event Creation and Initialization

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/api.html

Provides methods for initializing and creating various types of events within the EventKit framework. This includes setting up events on objects, creating events from different sources like iterables and awaitables, and managing event sequences and repetitions.

```APIDOC
Event.init(_obj_, _event_names_)
  - Convenience function for initializing multiple events as members of the given object.
  - Parameters:
    - _obj_: The object to attach events to.
    - _event_names_ (Iterable): Names to use for the created events.

Event.create(_obj_)
  - Create an event from an async iterator, awaitable, or event constructor without arguments.
  - Parameters:
    - _obj_: The source object. If it’s already an event then it is passed as-is.

Event.wait(_future_)
  - Create a new event that emits the value of the awaitable when it becomes available and then set this event done.
  - 'wait()' and '__await__()' are each other’s inverse.
  - Parameters:
    - _future_ (Awaitable): Future to wait on.
  - Return type: Wait

Event.aiterate(_ait_)
  - Create a new event that emits the yielded values from the asynchronous iterator.
  - The asynchronous iterator serves as a source for both the time and value of emits.
  - 'aiterate()' and '__aiter__()' are each other’s inverse.
  - Parameters:
    - _ait_ (AsyncIterable): The asynchronous source iterator. It must `await` at least once; If necessary use: `await asyncio.sleep(0)`.
  - Return type: Aiterate

Event.sequence(_values_, _interval_=0, _times_=None)
  - Create a new event that emits the given values. Supply at most one `interval` or `times`.
  - Parameters:
    - _values_ (Iterable): The source values.
    - _interval_ (float): Time interval in seconds between values.
    - _times_ (Optional[Iterable[float]]): Relative times for individual values, in seconds since start of event. The sequence should match `values`.
  - Return type: Sequence

Event.repeat(_value=<NoValue>_, _count_=1, _interval_=0, _times_=None)
  - Create a new event that repeats `value` a number of `count` times.
  - Parameters:
    - _value_: The value to emit.
    - _count_: Number of times to emit.
    - _interval_ (float): Time interval in seconds between values.
    - _times_ (Optional[Iterable[float]]): Relative times for individual values, in seconds since start of event. The sequence should match `values`.
  - Return type: Repeat

Event.range(*args, _interval_=0, _times_=None)
  - Create a new event that emits the values from a range.
  - Parameters:
    - args: Same as for built-in `range`.
    - _interval_ (float): Time interval in seconds between values.
    - _times_ (Optional[Iterable[float]]): Relative times for individual values, in seconds since start of event. The sequence should match the range.
  - Return type: Range

Event.timerange(_start_=0, _end_=None, _step_=1)
  - Create a new event that emits the datetime value, at that datetime, from a range of datetimes.
  - Parameters:
    - _start_:
      Start time, can be specified as:
        - `datetime.datetime`.
        - `datetime.time`: Today is used as date.
        - `int` or `float`: Number of seconds relative to now. Values will be quantized to the given step.
    - _end_:
      End time, can be specified as:
        - `datetime.datetime`.
        - `datetime.time`: Today is used as date.
        - `None`: No end limit.
    - _step_: Number of seconds, or `datetime.timedelta`, to space between values.
  - Return type: Timerange

Event.timer(_interval_, _count_=None)
  - Create a new timer event that emits at regularly paced intervals the number of seconds since starting it.
  - Parameters:
    - _interval_ (float): Time interval in seconds between emits.
    - _count_ (Optional[int]): Number of times to emit, or `None` for no limit.
  - Return type: Timer
```

--------------------------------

### Real-time Video Analysis Pipeline

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/README.rst

An example of a real-time video analysis pipeline using EventKit, chaining operations like VideoStream, FaceTracker, and SceneAnalyzer. It processes frames asynchronously.

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

--------------------------------

### Distributed Computing with EventKit and Distex

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/README.rst

Shows how to use EventKit with the distex library for distributed computing. It demonstrates compressing large data chunks in parallel using poolmap.

```python
from distex import Pool
import eventkit as ev
import bz2

pool = Pool()
data = [b'A' * 1000000] * 1000

pipe = ev.Sequence(data).poolmap(pool, bz2.compress).map(len).mean().last()

print(pipe.run())
pool.shutdown()
```

--------------------------------

### DropWhile Operator

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/select.html

The DropWhile operator skips items from the source as long as a given predicate is true. It starts emitting items only after the predicate returns False.

```python
class DropWhile(Op):
    __slots__ = ('_predicate', '_drop')

    def __init__(self, predicate=lambda x: not(x), source=None):
        Op.__init__(self, source)
        self._predicate = predicate
        self._drop = True

    def on_source(self, *args):
        if self._drop:
            self._drop = self._predicate(*args)
        if not self._drop:
            self.emit(*args)
```

--------------------------------

### Combine Events with Zip

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/notebooks/eventkit_introduction.ipynb

Shows how to zip event streams, creating tuples of emitted values once all sources have emitted a value.

```python
e1 = ev.Sequence("ABC", 0.01)
e2 = ev.Sequence("123", 0.02)
e3 = ev.Sequence("XYZ", 0.03)

event = e1.zip(e2, e3)

await event.list()
```

--------------------------------

### Skip Operator

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/select.html

The Skip operator skips a specified number of items from the source before starting to emit. It takes a count argument to determine how many items to skip.

```python
class Skip(Op):
    __slots__ = ('_count', '_n')

    def __init__(self, count=1, source=None):
        Op.__init__(self, source)
        self._count = count
        self._n = 0

    def on_source(self, *args):
        self._n += 1
        if self._n == self._count:
            self._source -= self.on_source
            self._source += self.emit
```

--------------------------------

### Product Operation

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/aggregate.html

The Product class inherits from Reduce and calculates the product of values in an event stream. It uses the operator.mul function and allows specifying a starting value.

```python
class Product(Reduce):
    __slots__ = ()

    def __init__(self, start=1, source=None):
        Reduce.__init__(self, operator.mul, start, source)
```

--------------------------------

### EventKit API Reference

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/event.html

Comprehensive API documentation for the EventKit library, detailing static methods for event creation and management. This includes initializing events on objects, creating events from awaitables or async iterators, and defining sequences or repeated emissions.

```APIDOC
EventKit:
  __aiter__(): Alias for aiter.
  __contains__(c): Checks if a callable is already connected.
  __reduce__(): Custom reduction for pickling, excluding slots.

  @staticmethod init(obj, event_names: Iterable):
    Convenience function for initializing multiple events as members of the given object.
    Args:
      event_names: Names to use for the created events.

  @staticmethod create(obj):
    Create an event from an async iterator, awaitable, or event constructor without arguments.
    Args:
      obj: The source object. If it's already an event then it is passed as-is.
    Returns:
      An Event object.
    Raises:
      ValueError: If the object type is invalid.

  @staticmethod wait(future: Awaitable) -> "Wait":
    Create a new event that emits the value of the awaitable when it becomes available and then set this event done.
    Args:
      future: Future to wait on.
    Returns:
      A Wait event object.

  @staticmethod aiterate(ait: AsyncIterable) -> "Aiterate":
    Create a new event that emits the yielded values from the asynchronous iterator.
    The asynchronous iterator serves as a source for both the time and value of emits.
    Args:
      ait: The asynchronous source iterator. It must `await` at least once; If necessary use:
        await asyncio.sleep(0)
    Returns:
      An Aiterate event object.

  @staticmethod sequence(values: Iterable, interval: float = 0, times: Union[Iterable[float], None] = None) -> "Sequence":
    Create a new event that emits the given values.
    Supply at most one `interval` or `times`.
    Args:
      values: The source values.
      interval: Time interval in seconds between values.
      times: Relative times for individual values, in seconds since start of event. The sequence should match `values`.
    Returns:
      A Sequence event object.

  @staticmethod repeat(value=NO_VALUE, count=1, interval: float = 0, times: Union[Iterable[float], None] = None) -> "Repeat":
    Create a new event that repeats `value` a number of `count` times.
    Args:
      value: The value to emit.
      count: Number of times to emit.
      interval: Time interval in seconds between values.
      times: Relative times for individual values, in seconds since start of event. The sequence should match `values`.
    Returns:
      A Repeat event object.
```

--------------------------------

### Sum Operation

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/_modules/eventkit/ops/aggregate.html

The Sum class inherits from Reduce and calculates the sum of values in an event stream. It uses the operator.add function and allows specifying a starting value.

```python
class Sum(Reduce):
    __slots__ = ()

    def __init__(self, start=0, source=None):
        Reduce.__init__(self, operator.add, start, source)
```

--------------------------------

### Distributed Computing with eventkit and distex

Source: https://github.com/ib-api-reloaded/eventkit/blob/main/docs/html/index.html

Illustrates using eventkit with the distex library for distributed computation. It shows how to create a Pool, apply a function (bz2.compress) to data in parallel using poolmap, and then compute the mean of the results. The pool.shutdown() method is used to clean up resources.

```python
from distex import Pool
import eventkit as ev
import bz2

pool = Pool()
# await pool  # un-comment in Jupyter
data = [b'A' * 1000000] * 1000

pipe = ev.Sequence(data).poolmap(pool, bz2.compress).map(len).mean().last()

print(pipe.run())  # in Jupyter: print(await pipe)
pool.shutdown()
```