Use a sentinel in LatestValueCache to denote if the cache is empty (#846)

This is necessary because `None` is a valid value that could have been
sent through a channel, so the having a `None` value might not always
mean that the cache is empty.

Closes #820

Author: shsms

src/frequenz/sdk/_internal/_channels.py

@@ -27,6 +27,10 @@ def new_receiver(self, *, maxsize: int = 50) -> Receiver[T]:
         """
 
 
+class _Sentinel:
+    """A sentinel to denote that no value has been received yet."""
+
+
 class LatestValueCache(typing.Generic[T]):
     """A cache that stores the latest value in a receiver."""
 
@@ -37,14 +41,34 @@ def __init__(self, receiver: Receiver[T]) -> None:
             receiver: The receiver to cache.
         """
         self._receiver = receiver
-        self._latest_value: T | None = None
+        self._latest_value: T | _Sentinel = _Sentinel()
         self._task = asyncio.create_task(self._run())
 
-    @property
-    def latest_value(self) -> T | None:
-        """Get the latest value in the cache."""
+    def get(self) -> T:
+        """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

src/frequenz/sdk/actor/power_distributing/_component_managers/_battery_manager.py

@@ -419,27 +419,27 @@ def _get_battery_inverter_data(
             Data for the battery and adjacent inverter without NaN values.
                 Return None if we could not replace NaN values.
         """
-        battery_data_none = [
-            self._battery_caches[battery_id].latest_value for battery_id in battery_ids
-        ]
-        inverter_data_none = [
-            self._inverter_caches[inverter_id].latest_value
-            for inverter_id in inverter_ids
-        ]
-
-        # It means that nothing has been send on this channels, yet.
+        # It means that nothing has been send on these channels, yet.
         # This should be handled by BatteryStatus. BatteryStatus should not return
         # this batteries as working.
-        if not all(battery_data_none) or not all(inverter_data_none):
+        if not all(
+            self._battery_caches[bat_id].has_value for bat_id in battery_ids
+        ) or not all(
+            self._inverter_caches[inv_id].has_value for inv_id in inverter_ids
+        ):
             _logger.error(
                 "Battery %s or inverter %s send no data, yet. They should be not used.",
                 battery_ids,
                 inverter_ids,
             )
             return None
 
-        battery_data = typing.cast(list[BatteryData], battery_data_none)
-        inverter_data = typing.cast(list[InverterData], inverter_data_none)
+        battery_data = [
+            self._battery_caches[battery_id].get() for battery_id in battery_ids
+        ]
+        inverter_data = [
+            self._inverter_caches[inverter_id].get() for inverter_id in inverter_ids
+        ]
 
         DataType = typing.TypeVar("DataType", BatteryData, InverterData)
 

src/frequenz/sdk/timeseries/ev_charger_pool/_set_current_bounds.py

@@ -103,7 +103,7 @@ async def _run(self) -> None:
         timer = Timer.timeout(timedelta(self._repeat_interval.total_seconds()))
 
         async for selected in select(bound_chan, timer):
-            meter = meter_data.latest_value
+            meter = meter_data.get()
             if meter is None:
                 raise ValueError("Meter channel closed.")