Add a stop method for LatestValueCache

And update users of `LatestValueCache` to call `stop` from their
`stop` methods.

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

Author: shsms

src/frequenz/sdk/_internal/_channels.py

@@ -9,6 +9,8 @@
 
 from frequenz.channels import Receiver
 
+from ._asyncio import cancel_and_await
+
 T_co = typing.TypeVar("T_co", covariant=True)
 
 
@@ -72,3 +74,7 @@ def has_value(self) -> bool:
     async def _run(self) -> None:
         async for value in self._receiver:
             self._latest_value = value
+
+    async def stop(self) -> None:
+        """Stop the cache."""
+        await cancel_and_await(self._task)

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

@@ -188,6 +188,10 @@ async def start(self) -> None:
     @override
     async def stop(self) -> None:
         """Stop the battery data manager."""
+        for bat_cache in self._battery_caches.values():
+            await bat_cache.stop()
+        for inv_cache in self._inverter_caches.values():
+            await inv_cache.stop()
         await self._component_pool_status_tracker.stop()
 
     @override

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

@@ -93,6 +93,7 @@ async def distribute_power(self, request: Request) -> None:
     @override
     async def stop(self) -> None:
         """Stop the ev charger manager."""
+        await self._voltage_cache.stop()
         await self._component_pool_status_tracker.stop()
 
     def _get_ev_charger_ids(self) -> collections.abc.Set[int]:

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

@@ -10,7 +10,7 @@
 
 from frequenz.channels import Broadcast, Sender, select, selected_from
 from frequenz.channels.timer import SkipMissedAndDrift, Timer
-from frequenz.client.microgrid import ComponentCategory
+from frequenz.client.microgrid import ComponentCategory, MeterData
 
 from ..._internal._asyncio import cancel_and_await
 from ..._internal._channels import LatestValueCache
@@ -55,6 +55,7 @@ def __init__(self, repeat_interval: timedelta) -> None:
         )
         self._bounds_rx = self._bounds_chan.new_receiver()
         self._bounds_tx = self._bounds_chan.new_sender()
+        self._meter_data_cache: LatestValueCache[MeterData] | None = None
 
     async def set(self, component_id: int, max_amps: float) -> None:
         """Send the given current limit to the microgrid for the given component id.
@@ -75,6 +76,8 @@ def new_bounds_sender(self) -> Sender[ComponentCurrentLimit]:
 
     async def stop(self) -> None:
         """Stop the BoundsSetter."""
+        if self._meter_data_cache is not None:
+            await self._meter_data_cache.stop()
         await self._bounds_chan.close()
         await cancel_and_await(self._task)
 
@@ -96,7 +99,7 @@ async def _run(self) -> None:
             _logger.error(err)
             raise RuntimeError(err)
 
-        meter_data = LatestValueCache(
+        self._meter_data_cache = LatestValueCache(
             await api_client.meter_data(next(iter(meters)).component_id)
         )
         latest_bound: dict[int, ComponentCurrentLimit] = {}
@@ -107,7 +110,7 @@ async def _run(self) -> None:
         )
 
         async for selected in select(bound_chan, timer):
-            meter = meter_data.get()
+            meter = self._meter_data_cache.get()
             if meter is None:
                 raise ValueError("Meter channel closed.")