Inprogress

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

Author: shsms

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

@@ -23,5 +23,5 @@ class EVDistributionConfig:
     initial_current: Current = field(default_factory=lambda: Current.from_amperes(10.0))
     """The initial current that can be allocated to an EV charger."""
 
-    increase_power_interval: timedelta = timedelta(minutes=1)
+    increase_power_interval: timedelta = timedelta(seconds=30)
     """The interval at which the power can be increased for an EV charger."""

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

@@ -1,14 +1,15 @@
 # License: MIT
 # Copyright © 2024 Frequenz Energy-as-a-Service GmbH
 
-"""Manage _ev chargers for the power distributor."""
+"""Manage EV chargers for the power distributor."""
 
+import asyncio
 import collections.abc
 import logging
 from datetime import datetime, timedelta, timezone
 
-from frequenz.channels import Sender
-from frequenz.channels.util import Merge
+from frequenz.channels import Broadcast, Sender
+from frequenz.channels.util import Merge, select, selected_from
 from typing_extensions import override
 
 from frequenz.sdk import microgrid
@@ -58,6 +59,9 @@ def __init__(
             component_status_tracker_type=EVChargerStatusTracker,
         )
         self._target_power = Power.zero()
+        self._target_power_channel = Broadcast[Power]("target_power")
+        self._target_power_tx = self._target_power_channel.new_sender()
+        self._task = None
 
     @override
     def component_ids(self) -> collections.abc.Set[int]:
@@ -67,6 +71,7 @@ def component_ids(self) -> collections.abc.Set[int]:
     @override
     async def start(self) -> None:
         """Start the ev charger data manager."""
+        self._task = asyncio.create_task(self._run_forever())
 
     @override
     async def distribute_power(self, request: Request) -> Result:
@@ -78,10 +83,9 @@ async def distribute_power(self, request: Request) -> Result:
         Returns:
             Result of the distribution.
         """
-        self._target_power = request.power
-        used_power = self._evc_states.get_ev_total_used_power()
-        if self._target_power < used_power:
-            self._throttle_ev_chargers(used_power - self._target_power)
+        await self._target_power_tx.send(request.power)
+
+        # TODO: check max_power check based on sum of bounds
         return Success(
             request,
             Power.zero(),
@@ -143,10 +147,11 @@ def _act_on_new_data(self, ev_data: EVChargerData) -> list[tuple[int, Power]]:
             if ev_previously_connected:
                 _logger.info("EV disconnected from EV charger %s", component_id)
             self._evc_states.get(component_id).update_state(ev_data)
-            return [(component_id, Power.zero())]
+            if self._evc_states.get(component_id).last_allocation > Power.zero():
+                return [(component_id, Power.zero())]
 
         # else if last throttling was less than 'increase_power_interval', do nothing.
-        now = datetime.utcnow()
+        now = datetime.now(tz=timezone.utc)
         last_throttling_time = self._evc_states.get(component_id).last_reallocation_time
         if last_throttling_time is not None:
             dur = now - last_throttling_time
@@ -181,7 +186,6 @@ def _act_on_new_data(self, ev_data: EVChargerData) -> list[tuple[int, Power]]:
             evc.last_allocation + allottable_power,
             Power.from_watts(evc.last_data.active_power_inclusion_upper_bound),
         )
-        evc.update_last_allocation(target_power, now)
         _logger.info(
             "Increasing power to EV charger %s from %s to %s",
             component_id,
@@ -190,38 +194,74 @@ def _act_on_new_data(self, ev_data: EVChargerData) -> list[tuple[int, Power]]:
         )
         return [(component_id, target_power)]
 
+    async def _run_forever(self) -> None:
+        while True:
+            try:
+                await self._run()
+            except:  # pylint: disable=bare-except
+                _logger.exception("Recovering from an error in EV charger manager.")
+                await asyncio.sleep(1.0)
+
     async def _run(self) -> None:
         api = microgrid.connection_manager.get().api_client
         ev_charger_data_rx = Merge(
             *[await api.ev_charger_data(evc_id) for evc_id in self._ev_charger_ids]
         )
-        async for evc_data in ev_charger_data_rx:
+        target_power_rx = self._target_power_channel.new_receiver()
+        async for selected in select(ev_charger_data_rx, target_power_rx):
             bounds_changes = []
-            # If a new ev charger is added, add it to the state tracker, with
-            # now as the last reallocation time and last charging time.
-            #
-            # This means it won't be assigned any power until the reallocation duration
-            # has passed.
-            if evc_data.component_id not in self._evc_states:
-                now = datetime.now(tz=timezone.utc)
-                self._evc_states.add_evc(
-                    EvcState(
-                        component_id=evc_data.component_id,
-                        last_data=evc_data,
-                        power=Power.from_watts(evc_data.active_power),
-                        last_allocation=Power.zero(),
-                        last_reallocation_time=now,
-                        last_charging_time=now,
+            if selected_from(selected, ev_charger_data_rx):
+                evc_data = selected.value
+                # If a new ev charger is added, add it to the state tracker, with
+                # now as the last reallocation time and last charging time.
+                #
+                # This means it won't be assigned any power until the reallocation
+                # duration has passed.
+                if evc_data.component_id not in self._evc_states:
+                    now = datetime.now(tz=timezone.utc)
+                    self._evc_states.add_evc(
+                        EvcState(
+                            component_id=evc_data.component_id,
+                            last_data=evc_data,
+                            power=Power.zero(),
+                            last_allocation=Power.zero(),
+                            last_reallocation_time=now,
+                            last_charging_time=now,
+                        )
+                    )
+                    bounds_changes = [(evc_data.component_id, Power.zero())]
+
+                # See if the ev charger has room for more power, and if the last
+                # allocation was not in the last reallocation duration.
+                else:
+                    bounds_changes = self._act_on_new_data(evc_data)
+
+            elif selected_from(selected, target_power_rx):
+                self._target_power = selected.value
+                _logger.debug("New target power: %s", self._target_power)
+                used_power = self._evc_states.get_ev_total_used_power()
+                if self._target_power < used_power:
+                    bounds_changes = self._throttle_ev_chargers(
+                        used_power - self._target_power
                     )
-                )
 
-            # See if the ev charger has room for more power, and if the last allocation
-            # was not in the last reallocation duration.
+            if bounds_changes:
+                _logger.debug("Setting power to EV chargers: %s", bounds_changes)
             else:
-                bounds_changes = self._act_on_new_data(evc_data)
-
+                continue
+            now = datetime.now(tz=timezone.utc)
             for component_id, power in bounds_changes:
-                await api.set_power(component_id, power.as_watts())
+                try:
+                    self._evc_states.get(component_id).update_last_allocation(
+                        power, now
+                    )
+                    await api.set_power(component_id, power.as_watts())
+                except Exception:  # pylint: disable=bare-except
+                    _logger.error(
+                        "Failed to set power to EV charger %s to %s",
+                        component_id,
+                        power,
+                    )
 
     def _throttle_ev_chargers(self, throttle_by: Power) -> list[tuple[int, Power]]:
         """Reduce EV charging power to meet the target power.

src/frequenz/sdk/actor/power_distributing/_component_status/_ev_charger_status_tracker.py

@@ -72,7 +72,7 @@ def __init__(  # pylint: disable=too-many-arguments
     @override
     def start(self) -> None:
         """Start the status tracker."""
-        self._tasks.add(asyncio.create_task(self._run()))
+        self._tasks.add(asyncio.create_task(self._run_forever()))
 
     def _is_working(self, ev_data: EVChargerData) -> bool:
         """Return whether the given EV charger can be assigned power.
@@ -99,9 +99,9 @@ async def _run_forever(self) -> None:
         while True:
             try:
                 await self._run()
-            except Exception as ex:  # pylint: disable=broad-except
+            except:  # pylint: disable=broad-except
                 _logger.exception(
-                    "Restarting after exception in EVChargerStatusTracker: %s", ex
+                    "Restarting after exception in EVChargerStatusTracker"
                 )
                 await asyncio.sleep(1.0)
 
@@ -155,12 +155,15 @@ async def _run(self) -> None:
         api_client = connection_manager.get().api_client
         ev_data_rx = await api_client.ev_charger_data(self._component_id)
         set_power_result_rx = self._set_power_result_receiver
-        missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift())
+        # TODO: Add missing data timer once resets are fixed in channels
+        # missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift())
+        missing_data_timer = Timer(timedelta(seconds=100.0), SkipMissedAndDrift())
         async for selected in select(
             ev_data_rx, set_power_result_rx, missing_data_timer
         ):
             new_status = ComponentStatusEnum.NOT_WORKING
             if selected_from(selected, ev_data_rx):
+                missing_data_timer.reset()
                 new_status = self._handle_ev_data(selected.value)
             elif selected_from(selected, set_power_result_rx):
                 new_status = self._handle_set_power_result(selected.value)
@@ -172,7 +175,10 @@ async def _run(self) -> None:
                 )
 
             # Send status update if status changed
-            if self._blocking_status.is_blocked():
+            if (
+                self._blocking_status.is_blocked()
+                and new_status != ComponentStatusEnum.NOT_WORKING
+            ):
                 new_status = ComponentStatusEnum.UNCERTAIN
 
             if new_status != self._last_status:

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

@@ -5,7 +5,9 @@
 
 
 import asyncio
+import logging
 from collections import abc
+from datetime import datetime
 
 from frequenz.channels import Receiver, Sender
 from frequenz.channels.util import Merge, select, selected_from
@@ -16,6 +18,8 @@
 from .. import Power
 from .._base_types import Bounds, SystemBounds
 
+_logger = logging.getLogger(__name__)
+
 
 class EVCSystemBoundsTracker(BackgroundService):
     """Track the system bounds for the EV chargers."""
@@ -41,12 +45,15 @@ def __init__(
         self._bounds_sender = bounds_sender
         self._latest_component_data: dict[int, microgrid.component.EVChargerData] = {}
         self._last_sent_bounds: SystemBounds | None = None
+        self._component_pool_status = ComponentPoolStatus(set(), set())
 
     def start(self) -> None:
         """Start the EV charger system bounds tracker."""
         self._tasks.add(asyncio.create_task(self._run()))
 
     async def _send_bounds(self) -> None:
+        if not self._latest_component_data:
+            return
         inclusion_bounds = Bounds(
             lower=Power.from_watts(
                 sum(
@@ -102,25 +109,29 @@ async def _run(self) -> None:
             )
         )
 
-        component_pool_status = ComponentPoolStatus(set(), set())
-
-        async for selected in select(status_rx, ev_data_rx):
-            if selected_from(selected, status_rx):
-                status = selected.value
-                component_pool_status = status
-                for comp_id in self._latest_component_data:
+        try:
+            async for selected in select(status_rx, ev_data_rx):
+                if selected_from(selected, status_rx):
+                    self._component_pool_status = selected.value
+                    to_pop = []
+                    for comp_id in self._latest_component_data:
+                        if (
+                            comp_id not in self._component_pool_status.working
+                            and comp_id not in self._component_pool_status.uncertain
+                        ):
+                            to_pop.append(comp_id)
+                    for comp_id in to_pop:
+                        self._latest_component_data.pop(comp_id, None)
+                elif selected_from(selected, ev_data_rx):
+                    data = selected.value
+                    comp_id = data.component_id
                     if (
-                        comp_id not in component_pool_status.working
-                        and comp_id not in component_pool_status.uncertain
+                        comp_id not in self._component_pool_status.working
+                        and comp_id not in self._component_pool_status.uncertain
                     ):
-                        self._latest_component_data.pop(comp_id, None)
-            elif selected_from(selected, ev_data_rx):
-                if (
-                    comp_id not in component_pool_status.working
-                    and comp_id not in component_pool_status.uncertain
-                ):
-                    continue
-                data = selected.value
-                self._latest_component_data[data.component_id] = data
-
-            await self._send_bounds()
+                        continue
+                    self._latest_component_data[data.component_id] = data
+
+                await self._send_bounds()
+        except:
+            _logger.exception("bounds tracker failed")