diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 96a0240b..fbd4aa0a 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -15,3 +15,5 @@ ## Bug Fixes + +- Docstrings now use absolute cross-references to public paths instead of relative ones. Relative cross-references are expanded from the module where an object is defined since mkdocstrings-python 2.0.9, which broke some links, and projects rendering docstrings inherited from this library needed `relative_crossrefs: true` for them to work at all. See [frequenz-repo-config-python#655](https://github.com/frequenz-floss/frequenz-repo-config-python/issues/655) for details. diff --git a/mkdocs.yml b/mkdocs.yml index 3f6c8f24..da4ec9e7 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -121,7 +121,6 @@ plugins: show_source: true show_symbol_type_toc: true signature_crossrefs: true - relative_crossrefs: true inventories: # See https://mkdocstrings.github.io/python/usage/#import for details - https://docs.python.org/3/objects.inv diff --git a/pyproject.toml b/pyproject.toml index 262e204e..309fbcca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,7 +57,7 @@ dev-mkdocs = [ "mkdocs-literate-nav == 0.6.3", "mkdocs-macros-plugin == 1.5.0", "mkdocs-material == 9.7.7", - "mkdocstrings[python] == 1.0.6", + "mkdocstrings[python] == 1.0.7", "mkdocstrings-python == 2.0.7", "pymdownx-superfence-filter-lines == 0.1.0", "griffe==2.3.0", diff --git a/src/frequenz/channels/__init__.py b/src/frequenz/channels/__init__.py index 627a890f..b0657de5 100644 --- a/src/frequenz/channels/__init__.py +++ b/src/frequenz/channels/__init__.py @@ -11,71 +11,71 @@ Base classes: -* [`Receiver`][.Receiver]: An object that can wait for and +* [`Receiver`][frequenz.channels.Receiver]: An object that can wait for and consume messages from a channel. -* [`Sender`][.Sender]: An object that can send messages to +* [`Sender`][frequenz.channels.Sender]: An object that can send messages to a channel. Channels: -* [`Anycast`][.Anycast]: A channel that supports multiple +* [`Anycast`][frequenz.channels.Anycast]: A channel that supports multiple senders and multiple receivers. A message sent through a sender will be received by exactly one receiver. -* [`Broadcast`][.Broadcast]: A channel to broadcast messages +* [`Broadcast`][frequenz.channels.Broadcast]: A channel to broadcast messages from multiple senders to multiple receivers. Each message sent through any of the senders is received by all of the receivers. Utilities to work with channels: -* [`merge`][.merge]: Merge messages coming from multiple receivers into +* [`merge`][frequenz.channels.merge]: Merge messages coming from multiple receivers into a single stream. -* [`select`][.select]: Iterate over the messages of all - [receivers][.Receiver] as new messages become available. +* [`select`][frequenz.channels.select]: Iterate over the messages of all + [receivers][frequenz.channels.Receiver] as new messages become available. -* [`LatestValueCache`][.LatestValueCache]: A cache that stores +* [`LatestValueCache`][frequenz.channels.LatestValueCache]: A cache that stores the latest value in a receiver, providing a way to look up the latest value in a stream, without having to wait, as long as there has been one value received. Exception classes: -* [`Error`][.Error]: Base class for all errors in this +* [`Error`][frequenz.channels.Error]: Base class for all errors in this library. -* [`ChannelError`][.ChannelError]: Base class for all errors +* [`ChannelError`][frequenz.channels.ChannelError]: Base class for all errors related to channels. -* [`ChannelClosedError`][.ChannelClosedError]: Error raised when +* [`ChannelClosedError`][frequenz.channels.ChannelClosedError]: Error raised when trying to operate (send, receive, etc.) through a closed channel. -* [`SenderError`][.SenderError]: Base class for all errors +* [`SenderError`][frequenz.channels.SenderError]: Base class for all errors related to senders. -* [`ReceiverError`][.ReceiverError]: Base class for all errors +* [`ReceiverError`][frequenz.channels.ReceiverError]: Base class for all errors related to receivers. -* [`ReceiverStoppedError`][.ReceiverStoppedError]: A receiver +* [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]: A receiver stopped producing messages. -* [`SelectError`][.SelectError]: Base class for all errors - related to [`select`][.select]. +* [`SelectError`][frequenz.channels.SelectError]: Base class for all errors + related to [`select`][frequenz.channels.select]. -* [`UnhandledSelectedError`][.UnhandledSelectedError]: An error - raised by [`select`][.select] that was not handled by the +* [`UnhandledSelectedError`][frequenz.channels.UnhandledSelectedError]: An error + raised by [`select`][frequenz.channels.select] that was not handled by the user. Extra utility receivers: -* [`Event`][.event.Event]: A receiver that generates a message when +* [`Event`][frequenz.channels.event.Event]: A receiver that generates a message when an event is set. -* [`FileWatcher`][.file_watcher.FileWatcher]: A receiver that +* [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher]: A receiver that generates a message when a file is added, modified or deleted. -* [`Timer`][.timer.Timer]: A receiver that generates a message after a +* [`Timer`][frequenz.channels.timer.Timer]: A receiver that generates a message after a given amount of time. """ diff --git a/src/frequenz/channels/_anycast.py b/src/frequenz/channels/_anycast.py index 7ccb809c..75cfa72d 100644 --- a/src/frequenz/channels/_anycast.py +++ b/src/frequenz/channels/_anycast.py @@ -79,12 +79,12 @@ class Anycast(Generic[ChannelMessageT]): To create a new [senders][frequenz.channels.Sender] and [receivers][frequenz.channels.Receiver] you can use the - [`.new_sender()`][.new_sender] and - [`.new_receiver()`][.new_receiver] methods + [`.new_sender()`][frequenz.channels.Anycast.new_sender] and + [`.new_receiver()`][frequenz.channels.Anycast.new_receiver] methods respectively. When the channel is not needed anymore, it should be closed with the - [`.aclose()`][.aclose] method. This will prevent further + [`.aclose()`][frequenz.channels.Anycast.aclose] method. This will prevent further attempts to [`send()`][frequenz.channels.Sender.send] data. Receivers will still be able to drain the pending messages on the channel, but after that, subsequent [`receive()`][frequenz.channels.Receiver.receive] calls will raise a @@ -265,7 +265,7 @@ def limit(self) -> int: """The maximum number of messages that can be stored in the channel's buffer. If the length of channel's buffer reaches the limit, then the sender - blocks at the [`send()`][...Sender.send] method until + blocks at the [`send()`][frequenz.channels.Sender.send] method until a message is consumed. """ maxlen = self._deque.maxlen @@ -275,12 +275,12 @@ def limit(self) -> int: async def aclose(self) -> None: """Close the channel. - Any further attempts to [`send()`][...Sender.send] data - will raise a [`SenderError`][...SenderError]. + Any further attempts to [`send()`][frequenz.channels.Sender.send] data + will raise a [`SenderError`][frequenz.channels.SenderError]. Receivers will still be able to drain the pending messages on the channel, but after that, subsequent - [`receive()`][...Receiver.receive] calls will return `None` + [`receive()`][frequenz.channels.Receiver.receive] calls will return `None` immediately. """ self._closed = True @@ -290,8 +290,11 @@ async def aclose(self) -> None: self._recv_cv.notify_all() @deprecated("The close() method is deprecated, use aclose() instead") - async def close(self) -> None: # noqa: D402 - """Close the channel, deprecated alias for [`.aclose()`][..aclose].""" # noqa: D402 + async def close(self) -> None: + """Close the channel. + + Deprecated alias for [`.aclose()`][frequenz.channels.Anycast.aclose]. + """ return await self.aclose() def new_sender(self) -> Sender[ChannelMessageT]: @@ -351,7 +354,7 @@ async def send(self, message: _T, /) -> None: Raises: SenderError: If the underlying channel was closed. - A [`ChannelClosedError`][...ChannelClosedError] is + A [`ChannelClosedError`][frequenz.channels.ChannelClosedError] is set as the cause. SenderClosedError: If this sender was closed. """ @@ -387,7 +390,7 @@ async def aclose(self) -> None: After closing, the sender will not be able to send any more messages. Any attempt to send a message through a closed sender will raise a - [`SenderError`][...SenderError]. + [`SenderError`][frequenz.channels.SenderError]. """ self._closed = True @@ -430,10 +433,11 @@ def __init__(self, channel: Anycast[_T], /) -> None: async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`.ready()`][..ready] has finished, the message should be read with - a call to [`.consume()`][..consume] ([`receive()`][...Receiver.receive] - or iterated over). The receiver will - remain ready (this method will return immediately) until it is + Once a call to [`.ready()`][frequenz.channels.Receiver.ready] has finished, the + message should be read with a call to + [`.consume()`][frequenz.channels.Receiver.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -460,7 +464,7 @@ async def ready(self) -> bool: @override def consume(self) -> _T: - """Return the latest message once [`ready()`][..ready] is complete. + """Return the latest message once [`ready()`][frequenz.channels.Receiver.ready] is complete. Returns: The next message that was received. diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py index 0f010ea0..a5f80621 100644 --- a/src/frequenz/channels/_broadcast.py +++ b/src/frequenz/channels/_broadcast.py @@ -63,12 +63,12 @@ class Broadcast( # pylint: disable=too-many-instance-attributes To create a new [senders][frequenz.channels.Sender] and [receivers][frequenz.channels.Receiver] you can use the - [`.new_sender()`][.new_sender] and - [`.new_receiver()`][.new_receiver] methods + [`.new_sender()`][frequenz.channels.Broadcast.new_sender] and + [`.new_receiver()`][frequenz.channels.Broadcast.new_receiver] methods respectively. When a channel is not needed anymore, it should be closed with - [`.aclose()`][.aclose]. This will prevent further + [`.aclose()`][frequenz.channels.Broadcast.aclose]. This will prevent further attempts to [`send()`][frequenz.channels.Sender.send] data, and will allow receivers to drain the pending items on their queues, but after that, subsequent [`receive()`][frequenz.channels.Receiver.receive] calls will @@ -200,13 +200,13 @@ def __init__( name: The name of the channel. This is for logging purposes, and it will be shown in the string representation of the channel. resend_latest: When True, every time a new receiver is created with - [`.new_receiver()`][..new_receiver], the last message seen by the channel - will be sent to the - new receiver automatically. This allows new receivers on slow streams to - get the latest message as soon as they are created, without having to - wait for the next message on the channel to arrive. It is safe to be - set in data/reporting channels, but is not recommended for use in - channels that stream control instructions. + [`.new_receiver()`][frequenz.channels.Broadcast.new_receiver], the last + message seen by the channel will be sent to the new receiver + automatically. This allows new receivers on slow streams to get the + latest message as soon as they are created, without having to wait for + the next message on the channel to arrive. It is safe to be set in + data/reporting channels, but is not recommended for use in channels that + stream control instructions. auto_close: If True, the channel will be closed when all senders or all receivers are closed. """ @@ -239,8 +239,8 @@ def __init__( self.resend_latest: bool = resend_latest """Whether to resend the latest message to new receivers. - When `True`, every time a new receiver is created with [`.new_receiver()`][..new_receiver], - it will + When `True`, every time a new receiver is created with + [`.new_receiver()`][frequenz.channels.Broadcast.new_receiver], it will automatically get sent the latest message on the channel. This allows new receivers on slow streams to get the latest message as soon as they are created, without having to wait for the next message on the channel to arrive. @@ -270,12 +270,12 @@ def is_closed(self) -> bool: async def aclose(self) -> None: """Close this channel. - Any further attempts to [`send()`][...Sender.send] data - will raise a [`SenderError`][...SenderError]. + Any further attempts to [`send()`][frequenz.channels.Sender.send] data + will raise a [`SenderError`][frequenz.channels.SenderError]. Receivers will still be able to drain the pending items on their queues, but after that, subsequent - [`receive()`][...Receiver.receive] calls will return `None` + [`receive()`][frequenz.channels.Receiver.receive] calls will return `None` immediately. """ self._latest = None @@ -284,8 +284,11 @@ async def aclose(self) -> None: self._recv_cv.notify_all() @deprecated("The close() method is deprecated, use aclose() instead") - async def close(self) -> None: # noqa: D402 - """Close the channel, deprecated alias for [`.aclose()`][..aclose].""" # noqa: D402 + async def close(self) -> None: + """Close the channel. + + Deprecated alias for [`.aclose()`][frequenz.channels.Broadcast.aclose]. + """ return await self.aclose() def new_sender(self) -> BroadcastSender[ChannelMessageT]: @@ -340,7 +343,7 @@ class BroadcastSender(CloneableSubscribableSender[_T]): """A sender to send messages to the broadcast channel. Should not be created directly, but through the - [`.clone()`][.clone] + [`.clone()`][frequenz.channels.BroadcastSender.clone] method of an existing sender. """ @@ -372,7 +375,7 @@ async def send(self, message: _T, /) -> None: Raises: SenderError: If the underlying channel was closed. - A [`ChannelClosedError`][...ChannelClosedError] is + A [`ChannelClosedError`][frequenz.channels.ChannelClosedError] is set as the cause. SenderClosedError: If this sender was closed. """ @@ -408,7 +411,7 @@ async def aclose(self) -> None: After a sender is closed, it can no longer be used to send messages. Any attempt to send a message through a closed sender will raise a - [`SenderClosedError`][...SenderClosedError]. + [`SenderClosedError`][frequenz.channels.SenderClosedError]. """ if self._closed: return @@ -491,7 +494,7 @@ def __init__( get dropped just in this receiver. Args: - channel: A reference to the [`Broadcast`][...Broadcast] + channel: A reference to the [`Broadcast`][frequenz.channels.Broadcast] channel that this receiver belongs to. name: A name to identify the receiver in the logs. If `None` an `id(self)`-based name will be used. This is only for debugging @@ -549,10 +552,11 @@ def __len__(self) -> int: async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`.ready()`][..ready] has finished, the message should be read with - a call to [`.consume()`][..consume] ([`receive()`][...Receiver.receive] - or iterated over). The receiver will - remain ready (this method will return immediately) until it is + Once a call to [`.ready()`][frequenz.channels.BroadcastReceiver.ready] has + finished, the message should be read with a call to + [`.consume()`][frequenz.channels.BroadcastReceiver.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -580,7 +584,10 @@ async def ready(self) -> bool: @override def consume(self) -> _T: - """Return the latest message once [`ready()`][..ready] is complete. + """Return the latest message once the receiver is ready. + + [`ready()`][frequenz.channels.BroadcastReceiver.ready] must be called before each call to + [`consume()`][frequenz.channels.BroadcastReceiver.consume]. Returns: The next message that was received. @@ -603,7 +610,7 @@ def close(self) -> None: After calling this method, new messages will not be received. Once the receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. """ self._closed = True self._channel._receivers.pop( # pylint: disable=protected-access @@ -805,7 +812,7 @@ def __new__( name: The name of the channel. This is for logging purposes, and it will be shown in the string representation of the channel. resend_latest: When True, every time a new receiver is created with - [`Broadcast.new_receiver()`][...Broadcast.new_receiver], + [`Broadcast.new_receiver()`][frequenz.channels.Broadcast.new_receiver], the last message seen by the channel will be sent to the new receiver automatically. This allows new receivers on slow streams to get the latest message as soon as they are created, without having to diff --git a/src/frequenz/channels/_exceptions.py b/src/frequenz/channels/_exceptions.py index b9e5f95d..191d83cc 100644 --- a/src/frequenz/channels/_exceptions.py +++ b/src/frequenz/channels/_exceptions.py @@ -6,11 +6,11 @@ # Exceptions All exceptions generated by this library inherit from the -[`Error`][..Error] exception. +[`Error`][frequenz.channels.Error] exception. Exceptions generated by channels inherit from the -[`ChannelError`][..ChannelError] exception. When there is an attempt to -use a closed channel, a [`ChannelClosedError`][..ChannelClosedError] +[`ChannelError`][frequenz.channels.ChannelError] exception. When there is an attempt to +use a closed channel, a [`ChannelClosedError`][frequenz.channels.ChannelClosedError] exception is raised. # Causes @@ -40,8 +40,8 @@ Tip: If you are using the async iteration interface for receivers, then you can access the cause of the - [`ReceiverStoppedError`][..ReceiverStoppedError] exception by - explicitly calling [`receive()`][..Receiver.receive] on the + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError] exception by + explicitly calling [`receive()`][frequenz.channels.Receiver.receive] on the receiver after the iteration is done: ```python show_lines="6:" diff --git a/src/frequenz/channels/_latest_value_cache.py b/src/frequenz/channels/_latest_value_cache.py index 2530dbe4..1e85786f 100644 --- a/src/frequenz/channels/_latest_value_cache.py +++ b/src/frequenz/channels/_latest_value_cache.py @@ -1,19 +1,18 @@ # License: MIT # Copyright © 2024 Frequenz Energy-as-a-Service GmbH -"""Caches the latest value received from a [`Receiver`][..Receiver]. +"""Caches the latest value received from a [`Receiver`][frequenz.channels.Receiver]. It provides a way to look up the latest value in a stream whenever required, as long as there has been one value received. -[`LatestValueCache`][..LatestValueCache] takes a -[`Receiver`][..Receiver] as an argument and stores the latest -value received by that receiver. As soon as a value is received, its -[`has_value`][..LatestValueCache.has_value] method returns -`True`, and its [`get`][..LatestValueCache.get] method returns -the latest value received. The [`get`][..LatestValueCache.get] method will -raise an exception if called -before any messages have been received from the receiver. +[`LatestValueCache`][frequenz.channels.LatestValueCache] takes a +[`Receiver`][frequenz.channels.Receiver] as an argument and stores the latest value +received by that receiver. As soon as a value is received, its +[`has_value`][frequenz.channels.LatestValueCache.has_value] method returns `True`, and +its [`get`][frequenz.channels.LatestValueCache.get] method returns the latest value +received. The [`get`][frequenz.channels.LatestValueCache.get] method will raise an +exception if called before any messages have been received from the receiver. Example: ```python @@ -55,7 +54,7 @@ class LatestValueCache(typing.Generic[T_co]): It provides a way to look up the latest value in a stream without any delay, as long as there has been one value received. - Takes ownership of the [`Receiver`][..Receiver]. + Takes ownership of the [`Receiver`][frequenz.channels.Receiver]. When the cache is stopped, the receiver will be closed. """ @@ -88,7 +87,7 @@ def get(self) -> T_co: """Return the latest value that has been received. This raises a `ValueError` if no value has been received yet. - Use [`has_value`][..has_value] to + Use [`has_value`][frequenz.channels.LatestValueCache.has_value] to check whether a value has been received yet, before trying to access the value, to avoid the exception. @@ -117,7 +116,7 @@ async def _run(self) -> None: self._latest_value = value async def stop(self) -> None: - """Stop the cache and close the owned [`Receiver`][...Receiver].""" + """Stop the cache and close the owned [`Receiver`][frequenz.channels.Receiver].""" self._receiver.close() self._stopped = True if not self._task.done(): diff --git a/src/frequenz/channels/_merge.py b/src/frequenz/channels/_merge.py index b412d4ca..68c1d134 100644 --- a/src/frequenz/channels/_merge.py +++ b/src/frequenz/channels/_merge.py @@ -6,7 +6,7 @@ # Usage If you just need to receive the same type of messages but from multiple sources in one -stream, you can use [`merge()`][.merge] to create a new receiver that +stream, you can use [`merge()`][frequenz.channels.merge] to create a new receiver that will receive messages from all the given receivers: ```python show_lines="8:" @@ -27,21 +27,21 @@ This can be helpful when you just need to receive messages and don't care about where are they coming from specifically. If you need to know where the message came -from, you can use [`select()`][..select] instead. +from, you can use [`select()`][frequenz.channels.select] instead. # Stopping A merge receiver will be stopped automatically when all the receivers that it merges are stopped. When using the async iterator interface, this means that the iterator will stop as soon as all the receivers are stopped. When using -[`receive()`][..Receiver.receive], this means that the method will raise -a [`ReceiverStoppedError`][..ReceiverStoppedError] exception as soon as +[`receive()`][frequenz.channels.Receiver.receive], this means that the method will raise +a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError] exception as soon as all the receivers are stopped. If you want to stop a merge receiver manually, you can use the -[`stop()`][.Merger.stop] method. +[`stop()`][frequenz.channels.Merger.stop] method. -When using [`receive()`][..Receiver.receive], you should make sure to +When using [`receive()`][frequenz.channels.Receiver.receive], you should make sure to either stop all the receivers that you are merging, or to stop the merge receiver manually. This is to make sure that all the tasks created by the merge receiver are cleaned up properly. @@ -84,8 +84,8 @@ def merge(*receivers: Receiver[ReceiverMessageT_co]) -> Merger[ReceiverMessageT_ *receivers: The receivers to merge. Returns: - A [`Merger`][..Merger] that merges the messages coming from multiple - receivers into a single stream. + A [`Merger`][frequenz.channels.Merger] that merges the messages coming from + multiple receivers into a single stream. Raises: ValueError: If no receivers are provided. @@ -100,8 +100,9 @@ class Merger(Receiver[ReceiverMessageT_co]): """A receiver that merges messages coming from multiple receivers into a single stream. Tip: - Please consider using the more idiomatic [`merge()`][..merge] - function instead of creating a [`Merger`][..Merger] instance directly. + Please consider using the more idiomatic [`merge()`][frequenz.channels.merge] + function instead of creating a [`Merger`][frequenz.channels.Merger] instance + directly. """ def __init__( @@ -141,10 +142,11 @@ async def stop(self) -> None: async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`ready()`][..ready] has finished, the message should be read with - a call to [`consume()`][..consume] ([`receive()`][...Receiver.receive] - or iterated over). The receiver will - remain ready (this method will return immediately) until it is + Once a call to [`ready()`][frequenz.channels.Merger.ready] has finished, the + message should be read with a call to + [`consume()`][frequenz.channels.Merger.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -177,7 +179,7 @@ async def ready(self) -> bool: @override def consume(self) -> ReceiverMessageT_co: - """Return the latest message once [`ready()`][..ready] is complete. + """Return the latest message once [`ready()`][frequenz.channels.Merger.ready] is complete. Returns: The next message that was received. @@ -198,7 +200,7 @@ def close(self) -> None: After calling this method, new messages will not be received. Once the receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. """ for task in self._pending: if not task.done() and task.get_loop().is_running(): diff --git a/src/frequenz/channels/_oneshot.py b/src/frequenz/channels/_oneshot.py index 3147d4ac..4d2bb9c4 100644 --- a/src/frequenz/channels/_oneshot.py +++ b/src/frequenz/channels/_oneshot.py @@ -25,7 +25,7 @@ class _Oneshot(typing.Generic[ChannelMessageT]): A one-shot channel is a channel that can only send one message. After the first message is sent, the sender is closed and any further attempts to send a message - will raise a [`SenderClosedError`][..SenderClosedError]. + will raise a [`SenderClosedError`][frequenz.channels.SenderClosedError]. """ def __init__(self) -> None: @@ -116,7 +116,7 @@ class OneshotChannel( A one-shot channel is a channel that can only send one message. After the first message is sent, the sender is closed and any further attempts to send a message - will raise a [`SenderClosedError`][..SenderClosedError]. + will raise a [`SenderClosedError`][frequenz.channels.SenderClosedError]. Example: Sending a message from one task to another ```python diff --git a/src/frequenz/channels/_receiver.py b/src/frequenz/channels/_receiver.py index ef1ea594..f1ce77ac 100644 --- a/src/frequenz/channels/_receiver.py +++ b/src/frequenz/channels/_receiver.py @@ -6,10 +6,10 @@ # Receivers Messages are received from [channels](/user-guide/channels/index.md) through -[`Receiver`][..Receiver] objects. [`Receiver`][..Receiver]s -are usually created by calling `channel.new_receiver()` and are [async -iterators][typing.AsyncIterator], so the easiest way to receive messages from them as -a stream is to use `async for`: +[`Receiver`][frequenz.channels.Receiver] objects. +[`Receiver`][frequenz.channels.Receiver]s are usually created by calling +`channel.new_receiver()` and are [async iterators][typing.AsyncIterator], so the easiest +way to receive messages from them as a stream is to use `async for`: ```python show_lines="6:" from frequenz.channels import Anycast @@ -22,7 +22,7 @@ ``` If you need to receive messages in different places or expecting a particular -sequence, you can use the [`receive()`][..Receiver.receive] method: +sequence, you can use the [`receive()`][frequenz.channels.Receiver.receive] method: ```python show_lines="6:" from frequenz.channels import Anycast @@ -40,7 +40,7 @@ # Message Transformation If you need to transform the received messages, receivers provide a -[`map()`][..Receiver.map] method to easily do so: +[`map()`][frequenz.channels.Receiver.map] method to easily do so: ```python show_lines="6:" from frequenz.channels import Anycast @@ -52,13 +52,13 @@ print(message) ``` -[`map()`][..Receiver.map] returns a new full receiver, so you can +[`map()`][frequenz.channels.Receiver.map] returns a new full receiver, so you can use it in any of the ways described above. # Message Filtering If you need to filter the received messages, receivers provide a -[`filter()`][..Receiver.filter] method to easily do so: +[`filter()`][frequenz.channels.Receiver.filter] method to easily do so: ```python show_lines="6:" from frequenz.channels import Anycast @@ -70,8 +70,8 @@ print(message) # Only even numbers will be printed ``` -As with [`map()`][..Receiver.map], -[`filter()`][..Receiver.filter] returns a new full receiver, so you can +As with [`map()`][frequenz.channels.Receiver.map], +[`filter()`][frequenz.channels.Receiver.filter] returns a new full receiver, so you can use it in any of the ways described above. # Error Handling @@ -82,13 +82,13 @@ [Error Handling](/user-guide/error-handling/) section of the user guide. If there is an error while receiving a message, -a [`ReceiverError`][..ReceiverError] exception is raised for both -[`receive()`][..Receiver.receive] method and async iteration +a [`ReceiverError`][frequenz.channels.ReceiverError] exception is raised for both +[`receive()`][frequenz.channels.Receiver.receive] method and async iteration interface. If the receiver has completely stopped (for example the underlying channel was -closed), a [`ReceiverStoppedError`][..ReceiverStoppedError] exception -is raised by [`receive()`][..Receiver.receive] method. +closed), a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError] exception +is raised by [`receive()`][frequenz.channels.Receiver.receive] method. ```python show_lines="6:" from frequenz.channels import Anycast @@ -128,14 +128,13 @@ When a particular stream of data is no-longer required, it is a good practice to close the receiver to free up resources. This is especially important in applications that are repeatedly creating new receivers. The -[`Receiver.close()`][..Receiver.close] method can be used for this +[`Receiver.close()`][frequenz.channels.Receiver.close] method can be used for this purpose. -After [`Receiver.close()`][..Receiver.close] is called, we can still receive -messages that are in the receiver's -buffer, but as soon as the receiver's buffer has been drained, trying to receive further -messages from a *closed* receiver will raise a -[`ReceiverStoppedError`][..ReceiverStoppedError]. +After [`Receiver.close()`][frequenz.channels.Receiver.close] is called, we can still +receive messages that are in the receiver's buffer, but as soon as the receiver's buffer +has been drained, trying to receive further messages from a *closed* receiver will raise +a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. # Advanced Usage @@ -146,22 +145,22 @@ ignore this section. Receivers extend on the [async iterator protocol][typing.AsyncIterator] by providing -a [`ready()`][..Receiver.ready] and -a [`consume()`][..Receiver.consume] method. +a [`ready()`][frequenz.channels.Receiver.ready] and +a [`consume()`][frequenz.channels.Receiver.consume] method. -The [`ready()`][..Receiver.ready] method is used to await until the +The [`ready()`][frequenz.channels.Receiver.ready] method is used to await until the receiver has a new message available, but without actually consuming it. The -[`consume()`][..Receiver.consume] method consumes the next available +[`consume()`][frequenz.channels.Receiver.consume] method consumes the next available message and returns it. -[`ready()`][..Receiver.ready] can be called multiple times, and it +[`ready()`][frequenz.channels.Receiver.ready] can be called multiple times, and it will return immediately if the receiver is already ready. -[`consume()`][..Receiver.consume] must be called only after -[`ready()`][..Receiver.ready] is done and only once, until the next -call to [`ready()`][..Receiver.ready]. +[`consume()`][frequenz.channels.Receiver.consume] must be called only after +[`ready()`][frequenz.channels.Receiver.ready] is done and only once, until the next +call to [`ready()`][frequenz.channels.Receiver.ready]. -Exceptions are never raised by [`ready()`][..Receiver.ready], they -are always delayed until [`consume()`][..Receiver.consume] is +Exceptions are never raised by [`ready()`][frequenz.channels.Receiver.ready], they +are always delayed until [`consume()`][frequenz.channels.Receiver.consume] is called. """ @@ -208,9 +207,11 @@ async def __anext__(self) -> ReceiverMessageT_co: # noqa: DOC503 async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`ready()`][..ready] has finished, the message should be read with - a call to [`consume()`][..consume] ([`receive()`][..receive] or iterated over). The receiver - will remain ready (this method will return immediately) until it is + Once a call to [`ready()`][frequenz.channels.Receiver.ready] has finished, the + message should be read with a call to + [`consume()`][frequenz.channels.Receiver.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -219,9 +220,10 @@ async def ready(self) -> bool: @abstractmethod def consume(self) -> ReceiverMessageT_co: - """Return the latest message once [`ready()`][..ready] is complete. + """Return the latest message once [`ready()`][frequenz.channels.Receiver.ready] is complete. - [`ready()`][..ready] must be called before each call to [`consume()`][..consume]. + [`ready()`][frequenz.channels.Receiver.ready] must be called before each call to + [`consume()`][frequenz.channels.Receiver.consume]. Returns: The next message received. @@ -236,7 +238,7 @@ def close(self) -> None: After calling this method, new messages will not be available from the receiver. Once the receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. """ raise NotImplementedError("close() must be implemented by subclasses") @@ -406,7 +408,7 @@ def __init__(self, message: str, receiver: Receiver[ReceiverMessageT_co]): Args: message: The error message. - receiver: The [`Receiver`][...Receiver] where the + receiver: The [`Receiver`][frequenz.channels.Receiver] where the error happened. """ super().__init__(message) @@ -421,7 +423,7 @@ def __init__(self, receiver: Receiver[ReceiverMessageT_co]): """Initialize this error. Args: - receiver: The [`Receiver`][...Receiver] where the + receiver: The [`Receiver`][frequenz.channels.Receiver] where the error happened. """ super().__init__(f"Receiver {receiver} was stopped", receiver) @@ -462,9 +464,11 @@ def __init__( async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`ready()`][..ready] has finished, the message should be read with - a call to [`consume()`][..consume] ([`receive()`][..receive] or iterated over). The receiver - will remain ready (this method will return immediately) until it is + Once a call to [`ready()`][frequenz.channels.Receiver.ready] has finished, the + message should be read with a call to + [`consume()`][frequenz.channels.Receiver.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -476,7 +480,10 @@ async def ready(self) -> bool: # explicitly raise anything. @override def consume(self) -> MappedMessageT_co: # noqa: DOC502 - """Return a transformed message once [`ready()`][..ready] is complete. + """Return a transformed message once the receiver is ready. + + [`ready()`][frequenz.channels.Receiver.ready] must be called before each call to + [`consume()`][frequenz.channels.Receiver.consume]. Returns: The next message that was received. @@ -493,7 +500,7 @@ def close(self) -> None: After calling this method, new messages will not be received. Once the receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. """ self._receiver.close() @@ -550,9 +557,11 @@ def __init__( async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`ready()`][..ready] has finished, the message should be read with - a call to [`consume()`][..consume] ([`receive()`][..receive] or iterated over). The receiver - will remain ready (this method will return immediately) until it is + Once a call to [`ready()`][frequenz.channels.Receiver.ready] has finished, the + message should be read with a call to + [`consume()`][frequenz.channels.Receiver.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -568,7 +577,10 @@ async def ready(self) -> bool: @override def consume(self) -> ReceiverMessageT_co: - """Return a transformed message once [`ready()`][..ready] is complete. + """Return a transformed message once the receiver is ready. + + [`ready()`][frequenz.channels.Receiver.ready] must be called before each call to + [`consume()`][frequenz.channels.Receiver.consume]. Returns: The next message that was received. @@ -592,7 +604,7 @@ def close(self) -> None: After calling this method, new messages will not be received. Once the receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. """ self._receiver.close() diff --git a/src/frequenz/channels/_select.py b/src/frequenz/channels/_select.py index d6130b5e..079d54b0 100644 --- a/src/frequenz/channels/_select.py +++ b/src/frequenz/channels/_select.py @@ -8,15 +8,15 @@ If you need to receive different types of messages from different receivers, you need to know the source of a particular received message to know the type of the message. -[`select()`][..select] allows you to do that. It is an +[`select()`][frequenz.channels.select] allows you to do that. It is an [async iterator][typing.AsyncIterator] that will iterate over the messages of all receivers as they receive new messages. -It yields a [`Selected`][..Selected] object that will tell you the +It yields a [`Selected`][frequenz.channels.Selected] object that will tell you the source of the received message. To make sure the received message is *cast* to the -correct type, you need to use the [`selected_from()`][..selected_from] +correct type, you need to use the [`selected_from()`][frequenz.channels.selected_from] function to check the source of the message, and the -[`message`][..Selected.message] attribute to access the message: +[`message`][frequenz.channels.Selected.message] attribute to access the message: ```python show_lines="8:" from frequenz.channels import Anycast, ReceiverStoppedError, select, selected_from @@ -37,11 +37,11 @@ Tip: To prevent common bugs, like when a new receiver is added to the select loop but - the handling code is forgotten, [`select()`][..select] will check + the handling code is forgotten, [`select()`][frequenz.channels.select] will check that all the selected receivers are handled in the if-chain. If this happens, it will raise an - [`UnhandledSelectedError`][..UnhandledSelectedError] exception. + [`UnhandledSelectedError`][frequenz.channels.UnhandledSelectedError] exception. If for some reason you want to ignore a received message, just add the receiver to the if-chain and do nothing with the message: @@ -63,13 +63,14 @@ # Stopping -The [`select()`][..select] async iterator will stop as soon as all the receivers -are stopped. You can also end the iteration early by breaking out of the loop as normal. +The [`select()`][frequenz.channels.select] async iterator will stop as soon as all the +receivers are stopped. You can also end the iteration early by breaking out of the loop +as normal. -When a single [`Receiver`][..Receiver] is stopped, it will be reported -via the [`Selected`][..Selected] object. You can use the -[`was_stopped`][..Selected.was_stopped] method to check if the -selected [`Receiver`][..Receiver] was stopped: +When a single [`Receiver`][frequenz.channels.Receiver] is stopped, it will be reported +via the [`Selected`][frequenz.channels.Selected] object. You can use the +[`was_stopped`][frequenz.channels.Selected.was_stopped] method to check if the +selected [`Receiver`][frequenz.channels.Receiver] was stopped: ```python show_lines="8:" from frequenz.channels import Anycast, select, selected_from @@ -89,10 +90,10 @@ ``` Tip: - The [`was_stopped`][..Selected.was_stopped] method is a + The [`was_stopped`][frequenz.channels.Selected.was_stopped] method is a convenience method that is equivalent to checking if the - [`exception`][..Selected.exception] attribute is an instance of - [`ReceiverStoppedError`][..ReceiverStoppedError]. + [`exception`][frequenz.channels.Selected.exception] attribute is an instance of + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. # Error Handling @@ -101,8 +102,8 @@ [Error Handling](/user-guide/error-handling/) section of the user guide. If a receiver raises an exception while receiving a message, the exception will be -raised by the [`message`][..Selected.message] attribute of the -[`Selected`][..Selected] object. +raised by the [`message`][frequenz.channels.Selected.message] attribute of the +[`Selected`][frequenz.channels.Selected] object. You can use a try-except block to handle exceptions as usual: @@ -126,8 +127,8 @@ # ... ``` -The [`Selected`][..Selected] object also has a -[`exception`][..Selected.exception] attribute that will contain the +The [`Selected`][frequenz.channels.Selected] object also has a +[`exception`][frequenz.channels.Selected.exception] attribute that will contain the exception that was raised by the receiver. """ @@ -152,17 +153,17 @@ def __repr__(self) -> str: class Selected(Generic[ReceiverMessageT_co]): - """A result of a [`select()`][..select] iteration. + """A result of a [`select()`][frequenz.channels.select] iteration. The selected receiver is consumed immediately and the received message is stored in the instance, unless there was an exception while receiving the message, in which case the exception is stored instead. - [`Selected`][..Selected] instances should be used in conjunction with the - [`selected_from()`][..selected_from] function to determine + [`Selected`][frequenz.channels.Selected] instances should be used in conjunction + with the [`selected_from()`][frequenz.channels.selected_from] function to determine which receiver was selected. - Please see [`select()`][..select] for an example. + Please see [`select()`][frequenz.channels.select] for an example. """ def __init__(self, receiver: Receiver[ReceiverMessageT_co], /) -> None: @@ -170,9 +171,9 @@ def __init__(self, receiver: Receiver[ReceiverMessageT_co], /) -> None: The receiver is consumed immediately when creating the instance and the received message is stored in the instance for later use as - [`message`][..message]. If there was an exception + [`message`][frequenz.channels.Selected.message]. If there was an exception while receiving the message, then the exception is stored in the instance - instead (as [`exception`][..exception]). + instead (as [`exception`][frequenz.channels.Selected.exception]). Args: receiver: The receiver that was selected. @@ -207,7 +208,7 @@ def message(self) -> ReceiverMessageT_co: # noqa: DOC503 Raises: Exception: If there was an exception while receiving the message. Normally - this should be an [`Error`][...Error] + this should be an [`Error`][frequenz.channels.Error] instance, but catches all exceptions in case some receivers can raise anything else. """ @@ -251,19 +252,19 @@ def __repr__(self) -> str: def selected_from( selected: Selected[Any], receiver: Receiver[ReceiverMessageT_co] ) -> TypeGuard[Selected[ReceiverMessageT_co]]: - """Return whether the given receiver was selected by [`select()`][..select]. + """Return whether the given receiver was selected by [`select()`][frequenz.channels.select]. This function is used in conjunction with the - [`Selected`][..Selected] class to determine which receiver was - selected in [`select()`][..select] iteration. + [`Selected`][frequenz.channels.Selected] class to determine which receiver was + selected in [`select()`][frequenz.channels.select] iteration. It also works as a [`TypeGuard`][typing.TypeGuard] to narrow the type of the - [`Selected`][..Selected] instance to the type of the receiver. + [`Selected`][frequenz.channels.Selected] instance to the type of the receiver. - Please see [`select()`][..select] for an example. + Please see [`select()`][frequenz.channels.select] for an example. Args: - selected: The result of a [`select()`][..select] iteration. + selected: The result of a [`select()`][frequenz.channels.select] iteration. receiver: The receiver to check if it was the source of a select operation. Returns: @@ -273,21 +274,21 @@ def selected_from( class SelectError(Error): - """An error that happened during a [`select()`][..select] operation. + """An error that happened during a [`select()`][frequenz.channels.select] operation. - This exception is raised when a [`select()`][..select] iteration fails. - It is raised as a single exception when one receiver fails during normal operation - (while calling [`ready()`][..Receiver.ready] for example). + This exception is raised when a [`select()`][frequenz.channels.select] iteration + fails. It is raised as a single exception when one receiver fails during normal + operation (while calling [`ready()`][frequenz.channels.Receiver.ready] for example). It is raised as a group exception ([`BaseExceptionGroup`][]) when a - [`select()`][..select] loop is cleaning up after it's done. + [`select()`][frequenz.channels.select] loop is cleaning up after it's done. """ class UnhandledSelectedError(SelectError, Generic[ReceiverMessageT_co]): - """A receiver was not handled in a [`select()`][..select] iteration. + """A receiver was not handled in a [`select()`][frequenz.channels.select] iteration. - This exception is raised when a [`select()`][..select] iteration - finishes without a call to [`selected_from()`][..selected_from] for + This exception is raised when a [`select()`][frequenz.channels.select] iteration + finishes without a call to [`selected_from()`][frequenz.channels.selected_from] for the selected receiver. """ @@ -349,31 +350,31 @@ async def select( # noqa: DOC503 This function is used to iterate over the messages of all receivers as they receive new messages. It is used in conjunction with the - [`Selected`][..Selected] class and the - [`selected_from()`][..selected_from] function to determine + [`Selected`][frequenz.channels.Selected] class and the + [`selected_from()`][frequenz.channels.selected_from] function to determine to determine which receiver was selected in a select operation. An exhaustiveness check is performed at runtime to make sure all selected receivers are handled in the if-chain, so you should call - [`selected_from()`][..selected_from] with all the receivers passed to - [`select()`][..select] inside the select loop, even if you plan to ignore - a message, to signal [`select()`][..select] that you are purposefully - ignoring the message. + [`selected_from()`][frequenz.channels.selected_from] with all the receivers passed + to [`select()`][frequenz.channels.select] inside the select loop, even if you plan + to ignore a message, to signal [`select()`][frequenz.channels.select] that you are + purposefully ignoring the message. Note: - The [`select()`][..select] function is intended to be used in cases where - the set of receivers is static and known beforehand. If you need to dynamically add/remove - receivers from a select loop, there are a few alternatives. Depending on your - use case, one or the other could work better for you: + The [`select()`][frequenz.channels.select] function is intended to be used in + cases where the set of receivers is static and known beforehand. If you need to + dynamically add/remove receivers from a select loop, there are a few + alternatives. Depending on your use case, one or the other could work better + for you: - * Use [`merge()`][..merge]: this is useful when you have an + * Use [`merge()`][frequenz.channels.merge]: this is useful when you have an unknown number of receivers of the same type that can be handled as a group. * Use tasks to manage each receiver individually: this is better if there are no relationships between the receivers. - * Break the [`select()`][..select] loop and start a new one with the - new set of receivers - (this should be the last resort, as it has some performance implications - because the loop needs to be restarted). + * Break the [`select()`][frequenz.channels.select] loop and start a new one with + the new set of receivers (this should be the last resort, as it has some + performance implications because the loop needs to be restarted). Example: ```python @@ -428,9 +429,9 @@ async def select( # noqa: DOC503 and receivers fail while cleaning up. SelectError: If there is an error while selecting receivers during normal operation. For example if a receiver raises an exception in the - [`ready()`][..Receiver.ready] + [`ready()`][frequenz.channels.Receiver.ready] method. Normal errors while receiving messages are not raised, but reported - via the [`Selected`][..Selected] instance. + via the [`Selected`][frequenz.channels.Selected] instance. """ receivers_map: dict[str, Receiver[Any]] = {str(hash(r)): r for r in receivers} pending: set[asyncio.Task[bool]] = set() diff --git a/src/frequenz/channels/_sender.py b/src/frequenz/channels/_sender.py index 7149880c..8a5a4445 100644 --- a/src/frequenz/channels/_sender.py +++ b/src/frequenz/channels/_sender.py @@ -6,9 +6,9 @@ # Senders Messages are sent to a [channel](/user-guide/channels) through -[`Sender`][..Sender] objects. [`Sender`][..Sender]s are +[`Sender`][frequenz.channels.Sender] objects. [`Sender`][frequenz.channels.Sender]s are usually created by calling `channel.new_sender()`, and are a very simple abstraction -that only provides a single [`send()`][..Sender.send] method: +that only provides a single [`send()`][frequenz.channels.Sender.send] method: ```python show_lines="6:" from frequenz.channels import Anycast @@ -19,12 +19,12 @@ await sender.send("Hello, world!") ``` -Although [`send()`][..Sender.send] is an asynchronous method, some +Although [`send()`][frequenz.channels.Sender.send] is an asynchronous method, some channels may implement it in a synchronous, non-blocking way. For example, buffered channels that drop messages when the buffer is full could guarantee that -[`send()`][..Sender.send] never blocks. However, please keep in mind +[`send()`][frequenz.channels.Sender.send] never blocks. However, please keep in mind that the [asyncio][] event loop could give control to another task at any time, -effectively making the [`send()`][..Sender.send] method blocking. +effectively making the [`send()`][frequenz.channels.Sender.send] method blocking. # Error Handling @@ -34,7 +34,7 @@ [Error Handling](/user-guide/error-handling/) section of the user guide. If there is any failure sending a message, -a [`SenderError`][..SenderError] exception is raised. +a [`SenderError`][frequenz.channels.SenderError] exception is raised. ```python show_lines="6:" from frequenz.channels import Anycast @@ -79,12 +79,12 @@ async def aclose(self) -> None: After a sender is closed, it can no longer be used to send messages. Any attempt to send a message through a closed sender will raise a - [`SenderClosedError`][...SenderClosedError]. + [`SenderClosedError`][frequenz.channels.SenderClosedError]. """ class SenderError(Error, Generic[SenderMessageT_co]): - """An error that originated in a [`Sender`][..Sender]. + """An error that originated in a [`Sender`][frequenz.channels.Sender]. All exceptions generated by senders inherit from this exception. """ @@ -94,12 +94,12 @@ def __init__(self, message: str, sender: Sender[SenderMessageT_co]): Args: message: The error message. - sender: The [`Sender`][...Sender] where the error + sender: The [`Sender`][frequenz.channels.Sender] where the error happened. """ super().__init__(message) self.sender: Sender[SenderMessageT_co] = sender - """The [`Sender`][...Sender] where the error happened.""" + """The [`Sender`][frequenz.channels.Sender] where the error happened.""" class SenderClosedError(SenderError[SenderMessageT_co]): @@ -109,32 +109,33 @@ def __init__(self, sender: Sender[SenderMessageT_co]): """Initialize this error. Args: - sender: The [`Sender`][...Sender] that was closed. + sender: The [`Sender`][frequenz.channels.Sender] that was closed. """ super().__init__("Sender is closed", sender) class SubscribableSender(Sender[SenderMessageT_contra], ABC): - """A [`Sender`][..Sender] that can be subscribed to.""" + """A [`Sender`][frequenz.channels.Sender] that can be subscribed to.""" @abstractmethod def subscribe(self) -> Receiver[SenderMessageT_contra]: """Subscribe to this sender. Returns: - A new [`Receiver`][...Receiver] attached to this sender's channel. + A new [`Receiver`][frequenz.channels.Receiver] attached to this sender's + channel. """ class CloneableSender(Sender[SenderMessageT_contra], ABC): - """A [`Sender`][..Sender] that can be cloned.""" + """A [`Sender`][frequenz.channels.Sender] that can be cloned.""" @abstractmethod def clone(self) -> CloneableSender[SenderMessageT_contra]: """Clone this sender. Returns: - A new [`CloneableSender`][...CloneableSender] + A new [`CloneableSender`][frequenz.channels.CloneableSender] that sends messages to the same channel as this sender. """ @@ -144,4 +145,4 @@ class CloneableSubscribableSender( CloneableSender[SenderMessageT_contra], ABC, ): - """A [`Sender`][..Sender] that can be both cloned and subscribed to.""" + """A [`Sender`][frequenz.channels.Sender] that can be both cloned and subscribed to.""" diff --git a/src/frequenz/channels/event.py b/src/frequenz/channels/event.py index c40b4f6d..87c58af5 100644 --- a/src/frequenz/channels/event.py +++ b/src/frequenz/channels/event.py @@ -5,12 +5,12 @@ !!! Tip inline end - Read the [`Event`][.Event] documentation for more + Read the [`Event`][frequenz.channels.event.Event] documentation for more information. This module contains the following: -* [`Event`][.Event]: +* [`Event`][frequenz.channels.event.Event]: {{docstring_summary("frequenz.channels.event.Event")}} """ @@ -27,18 +27,18 @@ class Event(Receiver[None]): # Usage There are cases where it is useful to be able to send a signal to - a [`select()`][...select] loop, for example, to stop a loop from + 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`][..Event] receiver and - call [`set()`][.set] on it when you want to make it + 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()`][.stop]. + [`stop()`][frequenz.channels.event.Event.stop]. # Example diff --git a/src/frequenz/channels/experimental/_grouping_latest_value_cache.py b/src/frequenz/channels/experimental/_grouping_latest_value_cache.py index 96f59029..0e31ab45 100644 --- a/src/frequenz/channels/experimental/_grouping_latest_value_cache.py +++ b/src/frequenz/channels/experimental/_grouping_latest_value_cache.py @@ -21,17 +21,17 @@ ValueT_co = TypeVar("ValueT_co", covariant=True) """Covariant type variable for the values cached by the -[`GroupingLatestValueCache`][..GroupingLatestValueCache]. +[`GroupingLatestValueCache`][frequenz.channels.experimental.GroupingLatestValueCache]. """ DefaultT = TypeVar("DefaultT") """Type variable for the default value returned by -[`GroupingLatestValueCache.get`][..GroupingLatestValueCache.get]. +[`GroupingLatestValueCache.get`][frequenz.channels.experimental.GroupingLatestValueCache.get]. """ HashableT = TypeVar("HashableT", bound=Hashable) """Type variable for the keys used to group values in the -[`GroupingLatestValueCache`][..GroupingLatestValueCache]. +[`GroupingLatestValueCache`][frequenz.channels.experimental.GroupingLatestValueCache]. """ @@ -49,21 +49,22 @@ class GroupingLatestValueCache(Mapping[HashableT, ValueT_co]): It provides a way to look up on demand, the latest value in a stream for any key, as long as there has been at least one value received for that key. - [`GroupingLatestValueCache`][..GroupingLatestValueCache] - takes a [`Receiver`][...Receiver] and a `key` function as arguments and - stores the latest value received by that receiver for each key separately. - - The [`GroupingLatestValueCache`][..GroupingLatestValueCache] - implements the [`Mapping`][collections.abc.Mapping] - interface, so it can be used like a dictionary. Additionally other methods from - [`MutableMapping`][collections.abc.MutableMapping] are implemented, but only - methods removing items from the cache are allowed, such as - [`pop()`][.pop], - [`popitem()`][.popitem], - [`clear()`][.clear], and - [`__delitem__()`][.__delitem__]. - Other update methods are not provided because the user should not update the - cache values directly. + [`GroupingLatestValueCache`][frequenz.channels.experimental.GroupingLatestValueCache] + takes a [`Receiver`][frequenz.channels.Receiver] and a `key` function as arguments + and stores the latest value received by that receiver for each key separately. + + The + [`GroupingLatestValueCache`][frequenz.channels.experimental.GroupingLatestValueCache] + implements the [`Mapping`][collections.abc.Mapping] interface, so it can be used + like a dictionary. Additionally other methods from + [`MutableMapping`][collections.abc.MutableMapping] are implemented, but only methods + removing items from the cache are allowed, such as + [`pop()`][frequenz.channels.experimental.GroupingLatestValueCache.pop], + [`popitem()`][frequenz.channels.experimental.GroupingLatestValueCache.popitem], + [`clear()`][frequenz.channels.experimental.GroupingLatestValueCache.clear], and + [`__delitem__()`][frequenz.channels.experimental.GroupingLatestValueCache.__delitem__]. + Other update methods are not provided because the user should not update the cache + values directly. Example: ```python diff --git a/src/frequenz/channels/experimental/_nop_receiver.py b/src/frequenz/channels/experimental/_nop_receiver.py index 83916339..a5878474 100644 --- a/src/frequenz/channels/experimental/_nop_receiver.py +++ b/src/frequenz/channels/experimental/_nop_receiver.py @@ -36,12 +36,12 @@ async def ready(self) -> bool: @override def consume(self) -> ReceiverMessageT_co: # noqa: DOC503 (raised indirectly) - """Raise [`ReceiverError`][....ReceiverError] or stop. + """Raise [`ReceiverError`][frequenz.channels.ReceiverError] or stop. - If the [`NopReceiver`][...NopReceiver] is not closed, - raise [`ReceiverError`][....ReceiverError]. - If the receiver is closed, then raise - [`ReceiverStoppedError`][....ReceiverStoppedError]. + If the [`NopReceiver`][frequenz.channels.experimental.NopReceiver] is not + closed, raise [`ReceiverError`][frequenz.channels.ReceiverError]. If the + receiver is closed, then raise + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. Returns: The next message received. diff --git a/src/frequenz/channels/experimental/_optional_receiver.py b/src/frequenz/channels/experimental/_optional_receiver.py index 1cabc8f1..0039fea4 100644 --- a/src/frequenz/channels/experimental/_optional_receiver.py +++ b/src/frequenz/channels/experimental/_optional_receiver.py @@ -3,10 +3,10 @@ """A receiver that will wait indefinitely if there is no underlying receiver. -The [`OptionalReceiver`][.OptionalReceiver] is useful when the -underlying receiver is not set initially. -Instead of making `if-else` branches to check if the receiver is set, you can use -this receiver to wait indefinitely if it is not set. +The [`OptionalReceiver`][frequenz.channels.experimental.OptionalReceiver] is useful when +the underlying receiver is not set initially. Instead of making `if-else` branches to +check if the receiver is set, you can use this receiver to wait indefinitely if it is +not set. """ import asyncio @@ -37,10 +37,12 @@ def __init__(self, receiver: Receiver[ReceiverMessageT_co] | None): async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`ready()`][..ready] has finished, the message should be read with - a call to [`consume()`][..consume] - ([`receive()`][....Receiver.receive] or iterated over). The receiver will - remain ready (this method will return immediately) until it is + Once a call to + [`ready()`][frequenz.channels.experimental.OptionalReceiver.ready] has finished, + the message should be read with a call to + [`consume()`][frequenz.channels.experimental.OptionalReceiver.consume] + ([`receive()`][frequenz.channels.Receiver.receive] or iterated over). The + receiver will remain ready (this method will return immediately) until it is consumed. Returns: @@ -55,9 +57,11 @@ async def ready(self) -> bool: @override def consume(self) -> ReceiverMessageT_co: # noqa: DOC503 (raised indirectly) - """Return the latest message once [`ready()`][..ready] is complete. + """Return the latest message once the receiver is ready. - [`ready()`][..ready] must be called before each call to [`consume()`][..consume]. + [`ready()`][frequenz.channels.experimental.OptionalReceiver.ready] must be + called before each call to + [`consume()`][frequenz.channels.experimental.OptionalReceiver.consume]. Returns: The next message received. diff --git a/src/frequenz/channels/experimental/_pipe.py b/src/frequenz/channels/experimental/_pipe.py index b54f1917..3c4ceea1 100644 --- a/src/frequenz/channels/experimental/_pipe.py +++ b/src/frequenz/channels/experimental/_pipe.py @@ -3,9 +3,9 @@ """Pipe between a receiver and a sender. -The [`Pipe`][.Pipe] class takes a receiver and a sender and -creates a pipe between them by -forwarding all the messages received by the receiver to the sender. +The [`Pipe`][frequenz.channels.experimental.Pipe] class takes a receiver and a sender +and creates a pipe between them by forwarding all the messages received by the receiver +to the sender. """ from __future__ import annotations @@ -21,9 +21,9 @@ class Pipe(typing.Generic[ChannelMessageT]): """A pipe between two channels. - The [`Pipe`][..Pipe] class takes a receiver and a sender and - creates a pipe between them - by forwarding all the messages received by the receiver to the sender. + The [`Pipe`][frequenz.channels.experimental.Pipe] class takes a receiver and a + sender and creates a pipe between them by forwarding all the messages received by + the receiver to the sender. Example: ```python diff --git a/src/frequenz/channels/experimental/_relay_sender.py b/src/frequenz/channels/experimental/_relay_sender.py index 5960229d..b1d223c0 100644 --- a/src/frequenz/channels/experimental/_relay_sender.py +++ b/src/frequenz/channels/experimental/_relay_sender.py @@ -3,9 +3,8 @@ """A Sender for sending messages to multiple senders. -The [`RelaySender`][.RelaySender] class takes multiple senders and -forwards all the messages sent to it, -to the senders it was created with. +The [`RelaySender`][frequenz.channels.experimental.RelaySender] class takes multiple +senders and forwards all the messages sent to it, to the senders it was created with. """ import asyncio @@ -17,10 +16,11 @@ class RelaySender(Sender[SenderMessageT_contra]): - """A [`Sender`][...Sender] for sending messages to multiple senders. + """A [`Sender`][frequenz.channels.Sender] for sending messages to multiple senders. - The [`RelaySender`][..RelaySender] class takes multiple senders and - forwards all the messages sent to it, to the senders it was created with. + The [`RelaySender`][frequenz.channels.experimental.RelaySender] class takes multiple + senders and forwards all the messages sent to it, to the senders it was created + with. Example: ```python @@ -42,7 +42,7 @@ class RelaySender(Sender[SenderMessageT_contra]): """ def __init__(self, *senders: Sender[SenderMessageT_contra]) -> None: - """Create a new [`RelaySender`][...RelaySender]. + """Create a new [`RelaySender`][frequenz.channels.experimental.RelaySender]. Args: *senders: The senders to send messages to. diff --git a/src/frequenz/channels/file_watcher.py b/src/frequenz/channels/file_watcher.py index 5e9dd875..c728e779 100644 --- a/src/frequenz/channels/file_watcher.py +++ b/src/frequenz/channels/file_watcher.py @@ -5,16 +5,16 @@ !!! Tip inline end - Read the [`FileWatcher`][.FileWatcher] + Read the [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] documentation for more information. This module contains the following: -* [`FileWatcher`][.FileWatcher]: +* [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher]: {{docstring_summary("frequenz.channels.file_watcher.FileWatcher")}} -* [`Event`][.Event]: +* [`Event`][frequenz.channels.file_watcher.Event]: {{docstring_summary("frequenz.channels.file_watcher.Event")}} -* [`EventType`][.EventType]: +* [`EventType`][frequenz.channels.file_watcher.EventType]: {{docstring_summary("frequenz.channels.file_watcher.EventType")}} """ @@ -61,38 +61,39 @@ class FileWatcher(Receiver[Event]): # Usage - A [`FileWatcher`][..FileWatcher] receiver can be used + A [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] receiver can be used to watch for changes in a set of files. It will generate an - [`Event`][..Event] message every time a file is + [`Event`][frequenz.channels.file_watcher.Event] message every time a file is created, modified or deleted, depending on the type of events that it is configured to watch for. - The [`EventType`][..EventType] of the event contains the - [`type`][..Event.type] of change that was observed and - the [`path`][..Event.path] where the change was + The [`EventType`][frequenz.channels.file_watcher.EventType] of the event contains + the [`type`][frequenz.channels.file_watcher.Event.type] of change that was observed + and the [`path`][frequenz.channels.file_watcher.Event.path] where the change was observed. Note: - The owner of the [`FileWatcher`][..FileWatcher] + The owner of the [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] receiver is responsible for recreating the - [`FileWatcher`][..FileWatcher] after it has been - cancelled or stopped. - For example, if a [`Task`][asyncio.Task] uses an asynchronous iterator to consume - events from the [`FileWatcher`][..FileWatcher] and the - task is cancelled, the [`FileWatcher`][..FileWatcher] will + [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] after it has been + cancelled or stopped. For example, if a [`Task`][asyncio.Task] uses an + asynchronous iterator to consume events from the + [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] and the task is + cancelled, the [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] will also stop. Therefore, the same `FileWatcher` instance cannot be reused for a new task to consume events. In this case, a new - [`FileWatcher`][..FileWatcher] instance must be created. + [`FileWatcher`][frequenz.channels.file_watcher.FileWatcher] instance must be + created. # Event Types The following event types are available: - * [`CREATE`][..EventType.CREATE]: + * [`CREATE`][frequenz.channels.file_watcher.EventType.CREATE]: {{docstring_summary("frequenz.channels.file_watcher.EventType.CREATE")}} - * [`MODIFY`][..EventType.MODIFY]: + * [`MODIFY`][frequenz.channels.file_watcher.EventType.MODIFY]: {{docstring_summary("frequenz.channels.file_watcher.EventType.MODIFY")}} - * [`DELETE`][..EventType.DELETE]: + * [`DELETE`][frequenz.channels.file_watcher.EventType.DELETE]: {{docstring_summary("frequenz.channels.file_watcher.EventType.DELETE")}} # Example @@ -193,11 +194,12 @@ def __del__(self) -> None: async def ready(self) -> bool: """Wait until the receiver is ready with a message or an error. - Once a call to [`ready()`][..ready] has finished, the message should be read with - a call to [`consume()`][..consume] - ([`receive()`][..receive] or iterated over). The receiver will - remain ready (this method will return immediately) until it is - consumed. + Once a call to [`ready()`][frequenz.channels.file_watcher.FileWatcher.ready] has + finished, the message should be read with a call to + [`consume()`][frequenz.channels.file_watcher.FileWatcher.consume] + ([`receive()`][frequenz.channels.file_watcher.FileWatcher.receive] or iterated + over). The receiver will remain ready (this method will return immediately) + until it is consumed. Returns: Whether the receiver is still active. @@ -220,7 +222,11 @@ async def ready(self) -> bool: @override def consume(self) -> Event: - """Return the latest event once [`ready()`][..ready] is complete. + """Return the latest event once the receiver is ready. + + [`ready()`][frequenz.channels.file_watcher.FileWatcher.ready] must be called + before each call to + [`consume()`][frequenz.channels.file_watcher.FileWatcher.consume]. Returns: The next event that was received. diff --git a/src/frequenz/channels/timer.py b/src/frequenz/channels/timer.py index 26c41958..3b4cf7a4 100644 --- a/src/frequenz/channels/timer.py +++ b/src/frequenz/channels/timer.py @@ -14,7 +14,7 @@ If you need to do something as periodically as possible (avoiding [drifts](#missed-ticks-and-drifting)), you can use -a [`Timer`][.Timer] like this: +a [`Timer`][frequenz.channels.timer.Timer] like this: Example: Periodic Timer Example ```python @@ -38,7 +38,7 @@ async def main() -> None: for every missed tick. If, instead, you need a timeout, for example to abort waiting for other receivers after -a certain amount of time, you can use a [`Timer`][.Timer] like +a certain amount of time, you can use a [`Timer`][frequenz.channels.timer.Timer] like this: Example: Timeout Example @@ -76,7 +76,7 @@ async def main() -> None: # Missed Ticks And Drifting -A [`Timer`][.Timer] can be used to send messages at regular +A [`Timer`][frequenz.channels.timer.Timer] can be used to send messages at regular time intervals, but there is one fundamental issue with timers in the [asyncio][] world: the event loop could give control to another task at any time, and that task can take a long time to finish, making the time it takes the next timer message to be received @@ -88,11 +88,11 @@ async def main() -> None: The following built-in policies are available: -* [`SkipMissedAndDrift`][.SkipMissedAndDrift]: +* [`SkipMissedAndDrift`][frequenz.channels.timer.SkipMissedAndDrift]: {{docstring_summary("frequenz.channels.timer.SkipMissedAndDrift")}} -* [`SkipMissedAndResync`][.SkipMissedAndResync]: +* [`SkipMissedAndResync`][frequenz.channels.timer.SkipMissedAndResync]: {{docstring_summary("frequenz.channels.timer.SkipMissedAndResync")}} -* [`TriggerAllMissed`][.TriggerAllMissed]: +* [`TriggerAllMissed`][frequenz.channels.timer.TriggerAllMissed]: {{docstring_summary("frequenz.channels.timer.TriggerAllMissed")}} """ @@ -126,8 +126,8 @@ class MissedTickPolicy(abc.ABC): """A policy to handle timer missed ticks. To implement a custom policy you need to subclass - [`MissedTickPolicy`][..MissedTickPolicy] and implement the - [`calculate_next_tick_time`][.calculate_next_tick_time] + [`MissedTickPolicy`][frequenz.channels.timer.MissedTickPolicy] and implement the + [`calculate_next_tick_time`][frequenz.channels.timer.MissedTickPolicy.calculate_next_tick_time] method. Example: @@ -161,10 +161,10 @@ def calculate_next_tick_time( ) -> int: """Calculate the next tick time according to the missed tick policy. - This method is called by [`ready()`][...Timer.ready] after it - has determined that the timer has triggered. It will check if the timer has missed - any ticks and handle them according to - [`missed_tick_policy`][...Timer.missed_tick_policy]. + This method is called by [`ready()`][frequenz.channels.timer.Timer.ready] after + it has determined that the timer has triggered. It will check if the timer has + missed any ticks and handle them according to + [`missed_tick_policy`][frequenz.channels.timer.Timer.missed_tick_policy]. Args: interval: The interval between ticks (in microseconds). @@ -174,7 +174,7 @@ def calculate_next_tick_time( Returns: The next tick time (in microseconds) according to - [`missed_tick_policy`][...Timer.missed_tick_policy]. + [`missed_tick_policy`][frequenz.channels.timer.Timer.missed_tick_policy]. """ def __repr__(self) -> str: @@ -185,7 +185,7 @@ def __repr__(self) -> str: class TriggerAllMissed(MissedTickPolicy): """A policy that triggers all the missed ticks immediately until it catches up. - The [`TriggerAllMissed`][..TriggerAllMissed] policy will + The [`TriggerAllMissed`][frequenz.channels.timer.TriggerAllMissed] policy will trigger all missed ticks immediately until it catches up with the current time. This means that if the timer is delayed for any reason, when it finally gets some time to run, it will trigger all the missed ticks in a burst. The drift is not @@ -252,11 +252,11 @@ class SkipMissedAndResync(MissedTickPolicy): """A policy that drops all the missed ticks, triggers immediately and resyncs. If ticks are missed, the - [`SkipMissedAndResync`][..SkipMissedAndResync] policy will - make the [`Timer`][..Timer] trigger immediately and it will + [`SkipMissedAndResync`][frequenz.channels.timer.SkipMissedAndResync] policy will + make the [`Timer`][frequenz.channels.timer.Timer] trigger immediately and it will schedule to trigger again on the next multiple of the - [`interval`][..Timer.interval], effectively skipping any missed - ticks, but re-syncing with the original start time. + [`interval`][frequenz.channels.timer.Timer.interval], effectively skipping any + missed ticks, but re-syncing with the original start time. Example: This example represents a timer with interval 1 second. @@ -318,14 +318,14 @@ def calculate_next_tick_time( class SkipMissedAndDrift(MissedTickPolicy): """A policy that drops all the missed ticks, triggers immediately and resets. - The [`SkipMissedAndDrift`][..SkipMissedAndDrift] policy will + The [`SkipMissedAndDrift`][frequenz.channels.timer.SkipMissedAndDrift] policy will behave effectively as if the timer was - [`reset()`][..Timer.reset] every time it is triggered. This means - the start time will change and the drift will be accumulated each time a tick is - delayed. Only the relative drift will be returned on each tick. + [`reset()`][frequenz.channels.timer.Timer.reset] every time it is triggered. This + means the start time will change and the drift will be accumulated each time a tick + is delayed. Only the relative drift will be returned on each tick. The reset happens only if the delay is larger than the - [delay tolerance][.delay_tolerance], so it + [delay tolerance][frequenz.channels.timer.SkipMissedAndDrift.delay_tolerance], so it is possible to ignore small delays and not drift in those cases. Example: @@ -340,7 +340,9 @@ class SkipMissedAndDrift(MissedTickPolicy): 3. The third tick, `T2.2`, happens at 2.3 seconds (0.1 seconds late), so it also triggers immediately but **it doesn't drift** because the delay is **under - the [`delay_tolerance`][.delay_tolerance]**. The next tick is expected at 3.2 seconds. + the + [`delay_tolerance`][frequenz.channels.timer.SkipMissedAndDrift.delay_tolerance]**. + The next tick is expected at 3.2 seconds. 4. The fourth tick, `T4.2`, triggers at 4.3 seconds (1.1 seconds late), so it also triggers immediately but the timer has drifted by 1.1 seconds, so a @@ -392,10 +394,11 @@ def calculate_next_tick_time( ) -> int: """Calculate the next tick time. - If the drift is larger than [`delay_tolerance`][..delay_tolerance], then it returns `now + - interval` (so the timer drifts), otherwise it returns - `scheduled_tick_time + interval` (we consider the delay too small and - avoid small drifts). + If the drift is larger than + [`delay_tolerance`][frequenz.channels.timer.SkipMissedAndDrift.delay_tolerance], + then it returns `now + interval` (so the timer drifts), otherwise it returns + `scheduled_tick_time + interval` (we consider the delay too small and avoid + small drifts). Args: now: The current loop time (in microseconds). @@ -423,25 +426,25 @@ def __repr__(self) -> str: class Timer(Receiver[timedelta]): """A receiver that sends a message regularly. - [`Timer`][..Timer]s are started by default after they are + [`Timer`][frequenz.channels.timer.Timer]s are started by default after they are created. This can be disabled by using `auto_start=False` option when creating them. In this case, the timer will not be started until - [`reset()`][.reset] is called. Receiving from the timer - (either using [`receive()`][.receive] or using the + [`reset()`][frequenz.channels.timer.Timer.reset] is called. Receiving from the timer + (either using [`receive()`][frequenz.channels.timer.Timer.receive] or using the async iterator interface) will also start the timer at that point. Timers need to be created in a context where an [`asyncio`][] loop is already running. If no - [`loop`][.loop] is specified, the current running loop + [`loop`][frequenz.channels.timer.Timer.loop] is specified, the current running loop is used. - Timers can be stopped by calling [`stop()`][.stop]. + Timers can be stopped by calling [`stop()`][frequenz.channels.timer.Timer.stop]. A stopped timer will raise - a [`ReceiverStoppedError`][...ReceiverStoppedError] or stop the async + a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError] or stop the async iteration as usual. Once a timer is explicitly stopped, it can only be started again by explicitly - calling [`reset()`][.reset] (trying to receive from it + calling [`reset()`][frequenz.channels.timer.Timer.reset] (trying to receive from it or using the async iterator interface will keep failing). Timer messages are [`timedelta`][datetime.timedelta]s containing the drift of the @@ -457,14 +460,14 @@ class Timer(Receiver[timedelta]): Even when the [`asyncio`][] loop's monotonic clock is a [`float`][], timers use `int`s to represent time internally. The internal time is tracked in microseconds, so the timer resolution is 1 microsecond - ([`interval`][.interval] must be at least + ([`interval`][frequenz.channels.timer.Timer.interval] must be at least 1 microsecond). This is to avoid floating point errors when performing calculations with time, which can lead to issues that are very hard to reproduce and debug. If the timer is delayed too much, then it will behave according to the - [`missed_tick_policy`][.missed_tick_policy]. Missing + [`missed_tick_policy`][frequenz.channels.timer.Timer.missed_tick_policy]. Missing ticks might or might not trigger a message and the drift could be accumulated or not depending on the chosen policy. """ @@ -483,17 +486,17 @@ def __init__( # noqa: DOC503 pylint: disable=too-many-arguments ) -> None: """Initialize this timer. - See the [class documentation][..] for details. + See the [class documentation][frequenz.channels.timer.Timer] for details. Args: interval: The time between timer ticks. Must be at least 1 microsecond. - missed_tick_policy: The policy of the timer when it misses a tick. - Commonly one of [`TriggerAllMissed`][...TriggerAllMissed], - [`SkipMissedAndResync`][...SkipMissedAndResync], - [`SkipMissedAndDrift`][...SkipMissedAndDrift] - or a custom class deriving from - [`MissedTickPolicy`][...MissedTickPolicy]. See the + missed_tick_policy: The policy of the timer when it misses a tick. Commonly + one of [`TriggerAllMissed`][frequenz.channels.timer.TriggerAllMissed], + [`SkipMissedAndResync`][frequenz.channels.timer.SkipMissedAndResync], + [`SkipMissedAndDrift`][frequenz.channels.timer.SkipMissedAndDrift] or a + custom class deriving from + [`MissedTickPolicy`][frequenz.channels.timer.MissedTickPolicy]. See the documentation of each class for more details. auto_start: Whether the timer should be started when the instance is created. This can only be `True` if there is @@ -538,8 +541,8 @@ def __init__( # noqa: DOC503 pylint: disable=too-many-arguments self._missed_tick_policy: MissedTickPolicy = missed_tick_policy """The policy of the timer when it misses a tick. - See the documentation of [`MissedTickPolicy`][..MissedTickPolicy] - for details. + See the documentation of + [`MissedTickPolicy`][frequenz.channels.timer.MissedTickPolicy] for details. """ self._reset_event = asyncio.Event() @@ -559,11 +562,11 @@ def __init__( # noqa: DOC503 pylint: disable=too-many-arguments * If `_next_msg_time` is `None`, it means it wasn't started yet (it was created with `auto_start=False`). Any receiving method will start - it by calling [`reset()`][.reset] in this case. + it by calling [`reset()`][frequenz.channels.timer.Timer.reset] in this case. * If `_next_msg_time` is not `None`, it means there was a request to stop it. In this case receiving methods will raise - a [`ReceiverStoppedError`][...ReceiverStoppedError]. + a [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. """ self._next_tick_time: int | None = None @@ -576,10 +579,10 @@ def __init__( # noqa: DOC503 pylint: disable=too-many-arguments self._current_drift: timedelta | None = None """The difference between `_next_msg_time` and the triggered time. - This is calculated by [`ready()`][.ready] but is returned by [`consume()`][.consume]. If - `None` it means `ready()` wasn't called and `consume()` will assert. - `consume()` will set it back to `None` to tell `ready()` that it needs - to wait again. + This is calculated by [`ready()`][frequenz.channels.timer.Timer.ready] but is + returned by [`consume()`][frequenz.channels.timer.Timer.consume]. If `None` it + means `ready()` wasn't called and `consume()` will assert. `consume()` will set + it back to `None` to tell `ready()` that it needs to wait again. """ if auto_start: @@ -658,13 +661,14 @@ def reset( # noqa: DOC503 def stop(self) -> None: """Stop the timer. - Once `stop` has been called, all subsequent calls to [`ready()`][..ready] will - immediately return False and calls to [`consume()`][..consume] / - [`receive()`][..receive] or any - use of the async iterator interface will raise - a [`ReceiverStoppedError`][....ReceiverStoppedError]. + Once `stop` has been called, all subsequent calls to + [`ready()`][frequenz.channels.timer.Timer.ready] will immediately return False + and calls to [`consume()`][frequenz.channels.timer.Timer.consume] / + [`receive()`][frequenz.channels.timer.Timer.receive] or any use of the async + iterator interface will raise a + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. - You can restart the timer with [`reset()`][..reset]. + You can restart the timer with [`reset()`][frequenz.channels.timer.Timer.reset]. """ self._stopped = True # We need to make sure it's not None, otherwise `ready()` will start it @@ -675,12 +679,13 @@ def stop(self) -> None: # exceptions are raised indirectly. @override async def ready(self) -> bool: # noqa: DOC502 - """Wait until the timer [`interval`][..interval] elapses. + """Wait until the timer [`interval`][frequenz.channels.timer.Timer.interval] elapses. - Once a call to [`ready()`][..ready] has finished, the resulting tick information - must be read with a call to [`consume()`][..consume] - ([`receive()`][..receive] or iterated over) - to tell the timer it should wait for the next interval. + Once a call to [`ready()`][frequenz.channels.timer.Timer.ready] has finished, + the resulting tick information must be read with a call to + [`consume()`][frequenz.channels.timer.Timer.consume] + ([`receive()`][frequenz.channels.timer.Timer.receive] or iterated over) to tell + the timer it should wait for the next interval. The timer will remain ready (this method will return immediately) until it is consumed. @@ -748,18 +753,20 @@ async def ready(self) -> bool: # noqa: DOC502 @override def consume(self) -> timedelta: - """Return the latest drift once [`ready()`][..ready] is complete. + """Return the latest drift once the timer has triggered. - Once the timer has triggered ([`ready()`][..ready] is done), this method returns the - difference between when the timer should have triggered and the time when - it actually triggered. See the class documentation for more details. + Once the timer has triggered ([`ready()`][frequenz.channels.timer.Timer.ready] + is done), this method returns the difference between when the timer should have + triggered and the time when it actually triggered. See the class documentation + for more details. Returns: The difference between when the timer should have triggered and the time when it actually did. Raises: - ReceiverStoppedError: If the timer was stopped via [`stop()`][..stop]. + ReceiverStoppedError: If the timer was stopped via + [`stop()`][frequenz.channels.timer.Timer.stop]. """ # If it was stopped and there it no pending result, we raise # (if there is a pending result, then we still want to return it first)