Import the LatestValueCache implementation from the Frequenz SDK

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/channels/__init__.py

@@ -35,6 +35,11 @@
 * [select][frequenz.channels.select]: Iterate over the messages of all
   [receivers][frequenz.channels.Receiver] as new messages become available.
 
+* [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][frequenz.channels.Error]: Base class for all errors in this
@@ -85,6 +90,7 @@
     SenderMessageT_co,
     SenderMessageT_contra,
 )
+from ._latest_value_cache import LatestValueCache
 from ._merge import Merger, merge
 from ._receiver import Receiver, ReceiverError, ReceiverStoppedError
 from ._select import (
@@ -104,6 +110,7 @@
     "ChannelMessageT",
     "Error",
     "ErroredChannelT_co",
+    "LatestValueCache",
     "MappedMessageT_co",
     "Merger",
     "Receiver",

src/frequenz/channels/_latest_value_cache.py

@@ -0,0 +1,103 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""A cache that stores the latest value in a receiver.
+
+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.
+"""
+
+import asyncio
+import typing
+
+from ._receiver import Receiver
+
+T_co = typing.TypeVar("T_co", covariant=True)
+
+
+class _Sentinel:
+    """A sentinel to denote that no value has been received yet."""
+
+    def __str__(self) -> str:
+        """Return a string representation of this sentinel."""
+        return "<no value received yet>"
+
+
+class LatestValueCache(typing.Generic[T_co]):
+    """A cache that stores the latest value in a receiver.
+
+    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.
+    """
+
+    def __init__(
+        self, receiver: Receiver[T_co], *, unique_id: str | None = None
+    ) -> None:
+        """Create a new cache.
+
+        Args:
+            receiver: The receiver to cache.
+            unique_id: A string to help uniquely identify this instance. If not
+                provided, a unique identifier will be generated from the object's
+                [`id()`][id]. It is used mostly for debugging purposes.
+        """
+        self._receiver = receiver
+        self._unique_id: str = hex(id(self)) if unique_id is None else unique_id
+        self._latest_value: T_co | _Sentinel = _Sentinel()
+        self._task = asyncio.create_task(
+            self._run(), name=f"LatestValueCache«{self._unique_id}»"
+        )
+
+    @property
+    def unique_id(self) -> str:
+        """The unique identifier of this instance."""
+        return self._unique_id
+
+    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` to
+        check whether a value has been received yet, before trying to access the value,
+        to avoid the exception.
+
+        Returns:
+            The latest value that has been received.
+
+        Raises:
+            ValueError: If no value has been received yet.
+        """
+        if isinstance(self._latest_value, _Sentinel):
+            raise ValueError("No value has been received yet.")
+        return self._latest_value
+
+    def has_value(self) -> bool:
+        """Check whether a value has been received yet.
+
+        Returns:
+            `True` if a value has been received, `False` otherwise.
+        """
+        return not isinstance(self._latest_value, _Sentinel)
+
+    async def _run(self) -> None:
+        async for value in self._receiver:
+            self._latest_value = value
+
+    async def stop(self) -> None:
+        """Stop the cache."""
+        if not self._task.done():
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+
+    def __repr__(self) -> str:
+        """Return a string representation of this cache."""
+        return (
+            f"<LatestValueCache latest_value={self._latest_value!r}, "
+            f"receiver={self._receiver!r}, unique_id={self._unique_id!r}>"
+        )
+
+    def __str__(self) -> str:
+        """Return the last value seen by this cache."""
+        return str(self._latest_value)