Skip to content

event

frequenz.channels.event ¤

A receiver that can be made ready directly.

Tip

Read the Event documentation for more information.

This module contains the following:

  • Event: A receiver that can be made ready directly.

Classes¤

frequenz.channels.event.Event ¤

Bases: Receiver[None]

A receiver that can be made ready directly.

Usage¤

There are cases where it is useful to be able to send a signal to a select() loop, for example, to stop a loop from outside the loop itself.

To do that, you can use an Event receiver and call set() on it when you want to make it ready.

Stopping¤

The receiver will be re-activated (will keep blocking) after the current set event is received. To stop the receiver completely, you can call stop().

Example¤
Exit after printing the first 5 numbers 
import asyncio

from frequenz.channels import Anycast, select, selected_from
from frequenz.channels.event import Event

channel: Anycast[int] = Anycast(name="channel")
receiver = channel.new_receiver()
sender = channel.new_sender()
stop_event = Event(name="stop")


async def do_work() -> None:
    async for selected in select(receiver, stop_event):
        if selected_from(selected, receiver):
            print(selected.message)
        elif selected_from(selected, stop_event):
            print("Stop event triggered")
            stop_event.stop()
            break


async def send_stuff() -> None:
    for i in range(10):
        if stop_event.is_stopped:
            break
        await asyncio.sleep(1)
        await sender.send(i)


async def main() -> None:
    async with asyncio.TaskGroup() as task_group:
        task_group.create_task(do_work(), name="do_work")
        task_group.create_task(send_stuff(), name="send_stuff")
        await asyncio.sleep(5.5)
        stop_event.set()


asyncio.run(main())
Source code in frequenz/channels/event.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class Event(_receiver.Receiver[None]):
    """A receiver that can be made ready directly.

    # Usage

    There are cases where it is useful to be able to send a signal to
    a [`select()`][frequenz.channels.select] loop, for example, to stop a loop from
    outside the loop itself.

    To do that, you can use an [`Event`][frequenz.channels.event.Event] receiver and
    call [`set()`][frequenz.channels.event.Event.set] on it when you want to make it
    ready.

    # Stopping

    The receiver will be re-activated (will keep blocking) after the current set
    event is received. To stop the receiver completely, you can call
    [`stop()`][frequenz.channels.event.Event.stop].

    # Example

    Example: Exit after printing the first 5 numbers
        ```python
        import asyncio

        from frequenz.channels import Anycast, select, selected_from
        from frequenz.channels.event import Event

        channel: Anycast[int] = Anycast(name="channel")
        receiver = channel.new_receiver()
        sender = channel.new_sender()
        stop_event = Event(name="stop")


        async def do_work() -> None:
            async for selected in select(receiver, stop_event):
                if selected_from(selected, receiver):
                    print(selected.message)
                elif selected_from(selected, stop_event):
                    print("Stop event triggered")
                    stop_event.stop()
                    break


        async def send_stuff() -> None:
            for i in range(10):
                if stop_event.is_stopped:
                    break
                await asyncio.sleep(1)
                await sender.send(i)


        async def main() -> None:
            async with asyncio.TaskGroup() as task_group:
                task_group.create_task(do_work(), name="do_work")
                task_group.create_task(send_stuff(), name="send_stuff")
                await asyncio.sleep(5.5)
                stop_event.set()


        asyncio.run(main())
        ```
    """

    def __init__(self, *, name: str | None = None) -> None:
        """Initialize this event.

        Args:
            name: The name of the receiver.  If `None` an `id(self)`-based name will be
                used. This is only for debugging purposes, it will be shown in the
                string representation of the receiver.
        """
        self._event: _asyncio.Event = _asyncio.Event()
        """The event that is set when the receiver is ready."""

        self._name: str = f"{id(self):_}" if name is None else name
        """The name of the receiver.

        This is for debugging purposes, it will be shown in the string representation
        of the receiver.
        """

        self._is_set: bool = False
        """Whether the receiver is ready to be consumed.

        This is used to differentiate between when the receiver was stopped (the event
        is triggered too) but still there is an event to be consumed and when it was
        stopped but was not explicitly set().
        """

        self._is_stopped: bool = False
        """Whether the receiver is stopped."""

    @property
    def name(self) -> str:
        """The name of this receiver.

        This is for debugging purposes, it will be shown in the string representation
        of this receiver.
        """
        return self._name

    @property
    def is_set(self) -> bool:
        """Whether this receiver is set (ready)."""
        return self._is_set

    @property
    def is_stopped(self) -> bool:
        """Whether this receiver is stopped."""
        return self._is_stopped

    def stop(self) -> None:
        """Stop this receiver."""
        self._is_stopped = True
        self._event.set()

    def set(self) -> None:
        """Trigger the event (make the receiver ready)."""
        self._is_set = True
        self._event.set()

    async def ready(self) -> bool:
        """Wait until this receiver is ready.

        Returns:
            Whether this receiver is still running.
        """
        if self._is_stopped:
            return False
        await self._event.wait()
        return not self._is_stopped

    def consume(self) -> None:
        """Consume the event.

        This makes this receiver wait again until the event is set again.

        Raises:
            ReceiverStoppedError: If this receiver is stopped.
        """
        if not self._is_set and self._is_stopped:
            raise _receiver.ReceiverStoppedError(self)

        assert self._is_set, "calls to `consume()` must be follow a call to `ready()`"

        self._is_set = False
        self._event.clear()

    def __str__(self) -> str:
        """Return a string representation of this event."""
        return f"{type(self).__name__}({self._name!r})"

    def __repr__(self) -> str:
        """Return a string representation of this event."""
        return (
            f"<{type(self).__name__} name={self._name!r} is_set={self.is_set!r} "
            f"is_stopped={self.is_stopped!r}>"
        )
Attributes¤
is_set property ¤
is_set: bool

Whether this receiver is set (ready).

is_stopped property ¤
is_stopped: bool

Whether this receiver is stopped.

name property ¤
name: str

The name of this receiver.

This is for debugging purposes, it will be shown in the string representation of this receiver.

Functions¤
__aiter__ ¤
__aiter__() -> Self

Get an async iterator over the received messages.

RETURNS DESCRIPTION
Self

This receiver, as it is already an async iterator.
Source code in frequenz/channels/_receiver.py 
191
192
193
194
195
196
197
def __aiter__(self) -> Self:
    """Get an async iterator over the received messages.

    Returns:
        This receiver, as it is already an async iterator.
    """
    return self
__anext__ async ¤
__anext__() -> ReceiverMessageT_co

Await the next message in the async iteration over received messages.

RETURNS DESCRIPTION
ReceiverMessageT_co

The next received message.
RAISES DESCRIPTION
StopAsyncIteration

If the receiver stopped producing messages.
ReceiverError

If there is some problem with the receiver.
Source code in frequenz/channels/_receiver.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def __anext__(self) -> ReceiverMessageT_co:
    """Await the next message in the async iteration over received messages.

    Returns:
        The next received message.

    Raises:
        StopAsyncIteration: If the receiver stopped producing messages.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        await self.ready()
        return self.consume()
    except ReceiverStoppedError as exc:
        raise StopAsyncIteration() from exc
__init__ ¤
__init__(*, name: str | None = None) -> None

Initialize this event.

PARAMETER DESCRIPTION
name

The name of the receiver. If None an id(self)-based name will be used. This is only for debugging purposes, it will be shown in the string representation of the receiver.

TYPE: str | None DEFAULT: None

Source code in frequenz/channels/event.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __init__(self, *, name: str | None = None) -> None:
    """Initialize this event.

    Args:
        name: The name of the receiver.  If `None` an `id(self)`-based name will be
            used. This is only for debugging purposes, it will be shown in the
            string representation of the receiver.
    """
    self._event: _asyncio.Event = _asyncio.Event()
    """The event that is set when the receiver is ready."""

    self._name: str = f"{id(self):_}" if name is None else name
    """The name of the receiver.

    This is for debugging purposes, it will be shown in the string representation
    of the receiver.
    """

    self._is_set: bool = False
    """Whether the receiver is ready to be consumed.

    This is used to differentiate between when the receiver was stopped (the event
    is triggered too) but still there is an event to be consumed and when it was
    stopped but was not explicitly set().
    """

    self._is_stopped: bool = False
    """Whether the receiver is stopped."""
__repr__ ¤
__repr__() -> str

Return a string representation of this event.

Source code in frequenz/channels/event.py 
175
176
177
178
179
180
def __repr__(self) -> str:
    """Return a string representation of this event."""
    return (
        f"<{type(self).__name__} name={self._name!r} is_set={self.is_set!r} "
        f"is_stopped={self.is_stopped!r}>"
    )
__str__ ¤
__str__() -> str

Return a string representation of this event.

Source code in frequenz/channels/event.py 
171
172
173
def __str__(self) -> str:
    """Return a string representation of this event."""
    return f"{type(self).__name__}({self._name!r})"
consume ¤
consume() -> None

Consume the event.

This makes this receiver wait again until the event is set again.

RAISES DESCRIPTION
ReceiverStoppedError

If this receiver is stopped.
Source code in frequenz/channels/event.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def consume(self) -> None:
    """Consume the event.

    This makes this receiver wait again until the event is set again.

    Raises:
        ReceiverStoppedError: If this receiver is stopped.
    """
    if not self._is_set and self._is_stopped:
        raise _receiver.ReceiverStoppedError(self)

    assert self._is_set, "calls to `consume()` must be follow a call to `ready()`"

    self._is_set = False
    self._event.clear()
map ¤
map(
    mapping_function: Callable[
        [ReceiverMessageT_co], MappedMessageT_co
    ]
) -> Receiver[MappedMessageT_co]

Apply a mapping function on the received message.

Tip

The returned receiver type won't have all the methods of the original receiver. If you need to access methods of the original receiver that are not part of the Receiver interface you should save a reference to the original receiver and use that instead.

PARAMETER DESCRIPTION
mapping_function

The function to be applied on incoming messages.

TYPE: Callable[[ReceiverMessageT_co], MappedMessageT_co]

RETURNS DESCRIPTION
Receiver[MappedMessageT_co]

A new receiver that applies the function on the received messages.
Source code in frequenz/channels/_receiver.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def map(
    self, mapping_function: Callable[[ReceiverMessageT_co], MappedMessageT_co], /
) -> Receiver[MappedMessageT_co]:
    """Apply a mapping function on the received message.

    Tip:
        The returned receiver type won't have all the methods of the original
        receiver. If you need to access methods of the original receiver that are
        not part of the `Receiver` interface you should save a reference to the
        original receiver and use that instead.

    Args:
        mapping_function: The function to be applied on incoming messages.

    Returns:
        A new receiver that applies the function on the received messages.
    """
    return _Mapper(receiver=self, mapping_function=mapping_function)
ready async ¤
ready() -> bool

Wait until this receiver is ready.

RETURNS DESCRIPTION
bool

Whether this receiver is still running.
Source code in frequenz/channels/event.py 
144
145
146
147
148
149
150
151
152
153
async def ready(self) -> bool:
    """Wait until this receiver is ready.

    Returns:
        Whether this receiver is still running.
    """
    if self._is_stopped:
        return False
    await self._event.wait()
    return not self._is_stopped
receive async ¤
receive() -> ReceiverMessageT_co

Receive a message.

RETURNS DESCRIPTION
ReceiverMessageT_co

The received message.
RAISES DESCRIPTION
ReceiverStoppedError

If there is some problem with the receiver.
ReceiverError

If there is some problem with the receiver.
Source code in frequenz/channels/_receiver.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def receive(self) -> ReceiverMessageT_co:
    """Receive a message.

    Returns:
        The received message.

    Raises:
        ReceiverStoppedError: If there is some problem with the receiver.
        ReceiverError: If there is some problem with the receiver.
    """
    try:
        received = await self.__anext__()  # pylint: disable=unnecessary-dunder-call
    except StopAsyncIteration as exc:
        # If we already had a cause and it was the receiver was stopped,
        # then reuse that error, as StopAsyncIteration is just an artifact
        # introduced by __anext__.
        if (
            isinstance(exc.__cause__, ReceiverStoppedError)
            # pylint is not smart enough to figure out we checked above
            # this is a ReceiverStoppedError and thus it does have
            # a receiver member
            and exc.__cause__.receiver is self  # pylint: disable=no-member
        ):
            raise exc.__cause__
        raise ReceiverStoppedError(self) from exc
    return received
set ¤
set() -> None

Trigger the event (make the receiver ready).

Source code in frequenz/channels/event.py 
139
140
141
142
def set(self) -> None:
    """Trigger the event (make the receiver ready)."""
    self._is_set = True
    self._event.set()
stop ¤
stop() -> None

Stop this receiver.

Source code in frequenz/channels/event.py 
134
135
136
137
def stop(self) -> None:
    """Stop this receiver."""
    self._is_stopped = True
    self._event.set()