Un-Peekable-ification of the PowerDistributingActor (#800)

This PR replaces `Peekable`s with `LatestValueCache`.

And updates tests to not mock time, because that somehow makes a
difference when using actual receivers.

Author: shsms

src/frequenz/sdk/_internal/_channels.py

@@ -4,6 +4,7 @@
 """General purpose classes for use with channels."""
 
 import abc
+import asyncio
 import typing
 
 from frequenz.channels import Receiver
@@ -24,3 +25,26 @@ def new_receiver(self, maxsize: int = 50) -> Receiver[T]:
         Returns:
             A receiver instance.
         """
+
+
+class LatestValueCache(typing.Generic[T]):
+    """A cache that stores the latest value in a receiver."""
+
+    def __init__(self, receiver: Receiver[T]) -> None:
+        """Create a new cache.
+
+        Args:
+            receiver: The receiver to cache.
+        """
+        self._receiver = receiver
+        self._latest_value: T | None = None
+        self._task = asyncio.create_task(self._run())
+
+    @property
+    def latest_value(self) -> T | None:
+        """Get the latest value in the cache."""
+        return self._latest_value
+
+    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

@@ -11,9 +11,10 @@
 from datetime import timedelta
 
 import grpc
-from frequenz.channels import Peekable, Receiver, Sender
+from frequenz.channels import Receiver, Sender
 
 from .... import microgrid
+from ...._internal._channels import LatestValueCache
 from ...._internal._math import is_close_to_zero
 from ....microgrid import connection_manager
 from ....microgrid.component import BatteryData, ComponentCategory, InverterData
@@ -139,8 +140,8 @@ def __init__(
         self._bat_bats_map = maps["bat_bats"]
         self._inv_invs_map = maps["inv_invs"]
 
-        self._battery_receivers: dict[int, Peekable[BatteryData]] = {}
-        self._inverter_receivers: dict[int, Peekable[InverterData]] = {}
+        self._battery_caches: dict[int, LatestValueCache[BatteryData]] = {}
+        self._inverter_caches: dict[int, LatestValueCache[InverterData]] = {}
 
         self._component_pool_status_tracker = ComponentPoolStatusTracker(
             component_ids=set(self._battery_ids),
@@ -294,11 +295,11 @@ async def _create_channels(self) -> None:
         api = connection_manager.get().api_client
         for battery_id, inverter_ids in self._bat_invs_map.items():
             bat_recv: Receiver[BatteryData] = await api.battery_data(battery_id)
-            self._battery_receivers[battery_id] = bat_recv.into_peekable()
+            self._battery_caches[battery_id] = LatestValueCache(bat_recv)
 
             for inverter_id in inverter_ids:
                 inv_recv: Receiver[InverterData] = await api.inverter_data(inverter_id)
-                self._inverter_receivers[inverter_id] = inv_recv.into_peekable()
+                self._inverter_caches[inverter_id] = LatestValueCache(inv_recv)
 
     def _get_bounds(
         self,
@@ -370,10 +371,10 @@ def _check_request(
 
         for battery in request.component_ids:
             _logger.debug("Checking battery %d", battery)
-            if battery not in self._battery_receivers:
+            if battery not in self._battery_caches:
                 msg = (
                     f"No battery {battery}, available batteries: "
-                    f"{list(self._battery_receivers.keys())}"
+                    f"{list(self._battery_caches.keys())}"
                 )
                 return Error(request=request, msg=msg)
 
@@ -420,10 +421,11 @@ def _get_battery_inverter_data(
                 Return None if we could not replace NaN values.
         """
         battery_data_none = [
-            self._battery_receivers[battery_id].peek() for battery_id in battery_ids
+            self._battery_caches[battery_id].latest_value for battery_id in battery_ids
         ]
         inverter_data_none = [
-            self._inverter_receivers[inverter_id].peek() for inverter_id in inverter_ids
+            self._inverter_caches[inverter_id].latest_value
+            for inverter_id in inverter_ids
         ]
 
         # It means that nothing has been send on this channels, yet.
@@ -498,10 +500,10 @@ def _get_components_data(
         )
 
         for battery_id in working_batteries:
-            if battery_id not in self._battery_receivers:
+            if battery_id not in self._battery_caches:
                 raise KeyError(
                     f"No battery {battery_id}, "
-                    f"available batteries: {list(self._battery_receivers.keys())}"
+                    f"available batteries: {list(self._battery_caches.keys())}"
                 )
 
         connected_inverters = _get_all_from_map(self._bat_invs_map, batteries)

src/frequenz/sdk/microgrid/client/_client.py

@@ -98,10 +98,6 @@ async def meter_data(
     ) -> Receiver[MeterData]:
         """Return a channel receiver that provides a `MeterData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Args:
             component_id: id of the meter to get data for.
             maxsize: Size of the receiver's buffer.
@@ -118,10 +114,6 @@ async def battery_data(
     ) -> Receiver[BatteryData]:
         """Return a channel receiver that provides a `BatteryData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Args:
             component_id: id of the battery to get data for.
             maxsize: Size of the receiver's buffer.
@@ -138,10 +130,6 @@ async def inverter_data(
     ) -> Receiver[InverterData]:
         """Return a channel receiver that provides an `InverterData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Args:
             component_id: id of the inverter to get data for.
             maxsize: Size of the receiver's buffer.
@@ -158,10 +146,6 @@ async def ev_charger_data(
     ) -> Receiver[EVChargerData]:
         """Return a channel receiver that provides an `EvChargeData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Args:
             component_id: id of the ev charger to get data for.
             maxsize: Size of the receiver's buffer.
@@ -441,7 +425,7 @@ def _get_component_data_channel(
         if component_id in self._component_streams:
             return self._component_streams[component_id]
         task_name = f"raw-component-data-{component_id}"
-        chan = Broadcast[_GenericComponentData](task_name)
+        chan = Broadcast[_GenericComponentData](task_name, resend_latest=True)
         self._component_streams[component_id] = chan
 
         self._streaming_tasks[component_id] = asyncio.create_task(
@@ -493,10 +477,6 @@ async def meter_data(  # noqa: DOC502 (ValueError is raised indirectly by _expec
     ) -> Receiver[MeterData]:
         """Return a channel receiver that provides a `MeterData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Raises:
             ValueError: if the given id is unknown or has a different type.
 
@@ -523,10 +503,6 @@ async def battery_data(  # noqa: DOC502 (ValueError is raised indirectly by _exp
     ) -> Receiver[BatteryData]:
         """Return a channel receiver that provides a `BatteryData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Raises:
             ValueError: if the given id is unknown or has a different type.
 
@@ -553,10 +529,6 @@ async def inverter_data(  # noqa: DOC502 (ValueError is raised indirectly by _ex
     ) -> Receiver[InverterData]:
         """Return a channel receiver that provides an `InverterData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Raises:
             ValueError: if the given id is unknown or has a different type.
 
@@ -583,10 +555,6 @@ async def ev_charger_data(  # noqa: DOC502 (ValueError is raised indirectly by _
     ) -> Receiver[EVChargerData]:
         """Return a channel receiver that provides an `EvChargeData` stream.
 
-        If only the latest value is required, the `Receiver` returned by this
-        method can be converted into a `Peekable` with the `into_peekable`
-        method on the `Receiver.`
-
         Raises:
             ValueError: if the given id is unknown or has a different type.
 

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

@@ -12,6 +12,7 @@
 from frequenz.channels.util import Timer, select, selected_from
 
 from ..._internal._asyncio import cancel_and_await
+from ..._internal._channels import LatestValueCache
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory
 
@@ -93,16 +94,16 @@ async def _run(self) -> None:
             _logger.error(err)
             raise RuntimeError(err)
 
-        meter_data = (
+        meter_data = LatestValueCache(
             await api_client.meter_data(next(iter(meters)).component_id)
-        ).into_peekable()
+        )
         latest_bound: dict[int, ComponentCurrentLimit] = {}
 
         bound_chan = self._bounds_rx
         timer = Timer.timeout(timedelta(self._repeat_interval.total_seconds()))
 
         async for selected in select(bound_chan, timer):
-            meter = meter_data.peek()
+            meter = meter_data.latest_value
             if meter is None:
                 raise ValueError("Meter channel closed.")