Keep battery pool metric cache and resend 0 W after failed commands

RELEASE_NOTES.md

@@ -2,15 +2,20 @@
 
 ## Summary
 
-This release updates the microgrid component graph library to v0.6.0, which stops clamping the consumer and producer formulas.  Consumer power can now be negative, and a producer that draws power now makes the producer total less negative instead of counting as zero.
+<!-- Here goes a general summary of what this release is about -->
 
 ## Upgrading
 
-- The minimum supported version of [`frequenz-microgrid-component-graph`](https://github.com/frequenz-floss/frequenz-microgrid-component-graph-python) is now [v0.6.0](https://github.com/frequenz-floss/frequenz-microgrid-component-graph-python/releases/tag/v0.6.0), which stops clamping the consumer and producer formulas. This changes the values streamed by `microgrid.consumer().power` and `microgrid.producer().power`.
-  - Consumer power used to be clamped at zero. It can now be negative, for example when unmodeled production or a measurement mismatch is larger than the consumption. To get the old result, use `consumer.power.max(Power.zero()).build("consumer_power")`.
-  - Producer power used to clamp each producer at zero. A producer that draws power now adds a positive value instead of zero, so the total can be positive. For example, PV producing 10 kW and a CHP drawing 2 kW used to give -10 kW and now give -8 kW. There is one exception: when the two share a meter below the grid meter, that meter sends data, and `disable_fallback_components` is off, the meter measures them together, so they gave -8 kW before too. `producer.power.min(Power.zero()).build("producer_power")` clamps the total at zero, but it still differs from the old result when one producer draws power while another produces.
-  - With `ComponentGraphConfig(include_phantom_loads_in_consumer_formula=True)`, consumer power is unchanged: it still clamps each of its terms at zero.
+<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+
+## New Features
+
+<!-- Here goes the main new features and examples or instructions on how to use them -->
 
 ## Bug Fixes
 
+<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+
 * Make the PowerManager fall back to zero instead of changing the requested power direction when a target is snapped outside asymmetric exclusion bounds.
+* The battery pool no longer drops the cached data of batteries that stop working. A battery that started working again was left out of the pool's metrics until its next data sample arrived. When every working battery had just started working again, `system_power_bounds` briefly reported no bounds, and if an actor had requested power, the power manager set the target power to zero.
+* After a power request to a battery's inverter fails or times out, the next request that includes that inverter is now always sent, even if it is 0 W. Before, a 0 W request was skipped when the last successful request to that inverter was also 0 W. The failed request may have been applied anyway, so the battery could keep running at that request's power. As before, a battery whose request failed is left out of new requests for a while: 1 second at first, and up to 30 seconds if its requests keep failing. This only happens while another battery in the new request is working. If none is, the failed battery is used again right away.

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_battery_manager.py

@@ -184,7 +184,12 @@ def __init__(
         """The distribution algorithm used to distribute power between batteries."""
 
         self._last_set_powers: dict[ComponentId, Power] = {}
-        """The last power value set to each battery's inverter."""
+        """The last power value set to each battery's inverter.
+
+        An inverter is missing if it was never sent a request, or if its last
+        request failed. A failed request may still have been applied, so the
+        next request to that inverter is always sent.
+        """
 
     @override
     def component_ids(self) -> collections.abc.Set[ComponentId]:
@@ -772,6 +777,9 @@ def _parse_result(
             if failed:
                 failed_power += distribution[inverter_id]
                 failed_batteries.update(battery_ids)
+                # The command may still have been applied, so the next one must
+                # be sent even if it is 0 W.
+                self._last_set_powers.pop(inverter_id, None)
             else:
                 self._last_set_powers[inverter_id] = distribution[inverter_id]
 

src/frequenz/sdk/timeseries/battery_pool/_methods.py

@@ -14,9 +14,6 @@
 
 from ..._internal._asyncio import cancel_and_await, run_forever
 from ..._internal._constants import RECEIVER_MAX_SIZE, WAIT_FOR_COMPONENT_DATA_SEC
-from ...microgrid._power_distributing._component_managers._battery_manager import (
-    _get_battery_inverter_mappings,
-)
 from ._component_metric_fetcher import (
     ComponentMetricFetcher,
     LatestBatteryMetricsFetcher,
@@ -86,13 +83,6 @@ def __init__(
             min_update_interval: Minimum frequency for sending update about the change.
         """
         self._metric_calculator: MetricCalculator[T] = metric_calculator
-        self._bat_inv_map = _get_battery_inverter_mappings(
-            self._metric_calculator.batteries,
-            inv_bats=False,
-            bat_bats=False,
-            inv_invs=False,
-        )["bat_invs"]
-
         self._working_batteries: set[ComponentId] = working_batteries.intersection(
             metric_calculator.batteries
         )
@@ -149,13 +139,6 @@ def update_working_batteries(self, new_working_batteries: set[ComponentId]) -> N
         # int he PowerBounds metrics.
         new_set = new_working_batteries.intersection(self._metric_calculator.batteries)
 
-        stopped_working = self._working_batteries - new_set
-        for battery_id in stopped_working:
-            # Removed cached metrics for components that stopped working.
-            self._cached_metrics.pop(battery_id, None)
-            for inv_id in self._bat_inv_map[battery_id]:
-                self._cached_metrics.pop(inv_id, None)
-
         if new_set != self._working_batteries:
             self._working_batteries = new_set
             self._update_event.set()

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -10,7 +10,7 @@
 import dataclasses
 import logging
 import math
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping
 from contextlib import AsyncExitStack
 from dataclasses import dataclass, is_dataclass, replace
 from datetime import datetime, timedelta, timezone
@@ -382,6 +382,135 @@ async def test_battery_pool_power_bounds(setup_batteries_pool: SetupArgs) -> Non
     await run_power_bounds_test(setup_batteries_pool)
 
 
+async def start_streaming_power_bounds(
+    setup_args: SetupArgs,
+    inverter_bounds: Mapping[ComponentId, tuple[float, float]] | None = None,
+) -> tuple[set[ComponentId], dict[ComponentId, frozenset[ComponentId]]]:
+    """Mark all batteries as working and stream power bounds for them.
+
+    All batteries are working and sending data, not just the ones in the battery
+    pool.
+
+    Args:
+        setup_args: Needed sdk tools and tools for mocking microgrid.
+        inverter_bounds: The inclusion bounds to stream for the inverters of the
+            given batteries. The inverters of other batteries get (-900, 6000).
+
+    Returns:
+        All the batteries in the microgrid, and the inverters of each battery.
+    """
+    all_batteries = get_components(setup_args.mock_microgrid, Battery)
+    await setup_args.battery_status_sender.send(
+        ComponentPoolStatus(working=all_batteries, uncertain=set())
+    )
+    bat_invs_map = _get_battery_inverter_mappings(
+        all_batteries,
+        inv_bats=False,
+        bat_bats=False,
+        inv_invs=False,
+    )["bat_invs"]
+
+    for battery_id, inverter_ids in bat_invs_map.items():
+        # Sampling rate choose to reflect real application.
+        setup_args.streamer.start_streaming(
+            BatteryDataWrapper(
+                component_id=battery_id,
+                timestamp=datetime.now(tz=timezone.utc),
+                power_inclusion_lower_bound=-1000,
+                power_inclusion_upper_bound=5000,
+                power_exclusion_lower_bound=-300,
+                power_exclusion_upper_bound=300,
+            ),
+            sampling_rate=0.05,
+        )
+        lower, upper = (inverter_bounds or {}).get(battery_id, (-900, 6000))
+        for inverter_id in inverter_ids:
+            setup_args.streamer.start_streaming(
+                InverterDataWrapper(
+                    component_id=inverter_id,
+                    timestamp=datetime.now(tz=timezone.utc),
+                    active_power_inclusion_lower_bound=lower,
+                    active_power_inclusion_upper_bound=upper,
+                    active_power_exclusion_lower_bound=-200,
+                    active_power_exclusion_upper_bound=200,
+                ),
+                sampling_rate=0.1,
+            )
+    return all_batteries, bat_invs_map
+
+
+async def test_power_bounds_when_working_batteries_swap(
+    setup_batteries_pool: SetupArgs,
+) -> None:
+    """Test that batteries rejoining the working set count towards the bounds at once.
+
+    The metric fetchers clear a component's data when none arrives within
+    MAX_BATTERY_DATA_AGE_SEC, and the status trackers stop counting batteries
+    with stale data as working. So a battery that leaves the working set and
+    comes back must not have to wait for its next data sample before its bounds
+    are used again.
+
+    Args:
+        setup_batteries_pool: Fixture that creates needed microgrid tools.
+    """
+    battery_pool = setup_batteries_pool.battery_pool
+    battery_status_sender = setup_batteries_pool.battery_status_sender
+    first, second = sorted(battery_pool.component_ids)
+
+    # Give the second battery different bounds, so that swapping which battery
+    # is working changes the result.
+    all_batteries, _ = await start_streaming_power_bounds(
+        setup_batteries_pool, {second: (-500, 3000)}
+    )
+
+    receiver = battery_pool.system_power_bounds.new_receiver(limit=50)
+    waiting_time_sec = 1.0
+
+    msg = await asyncio.wait_for(
+        receiver.receive(), timeout=WAIT_FOR_COMPONENT_DATA_SEC + waiting_time_sec
+    )
+    now = datetime.now(tz=timezone.utc)
+    compare_messages(
+        msg,
+        SystemBounds(
+            timestamp=now,
+            inclusion_bounds=Bounds(Power.from_watts(-1400), Power.from_watts(8000)),
+            exclusion_bounds=Bounds(Power.from_watts(-600), Power.from_watts(600)),
+        ),
+    )
+
+    # No new data arrives from here on, so the status changes below land between
+    # two samples. They must happen within MAX_BATTERY_DATA_AGE_SEC, after which
+    # the cached data expires.
+    await setup_batteries_pool.streamer.stop()
+
+    await battery_status_sender.send(
+        ComponentPoolStatus(working=all_batteries - {second}, uncertain={second})
+    )
+    msg = await asyncio.wait_for(receiver.receive(), timeout=waiting_time_sec)
+    compare_messages(
+        msg,
+        SystemBounds(
+            timestamp=now,
+            inclusion_bounds=Bounds(Power.from_watts(-900), Power.from_watts(5000)),
+            exclusion_bounds=Bounds(Power.from_watts(-300), Power.from_watts(300)),
+        ),
+    )
+
+    await battery_status_sender.send(
+        ComponentPoolStatus(working=all_batteries - {first}, uncertain={first})
+    )
+    msg = await asyncio.wait_for(receiver.receive(), timeout=waiting_time_sec)
+    compare_messages(
+        msg,
+        SystemBounds(
+            timestamp=now,
+            inclusion_bounds=Bounds(Power.from_watts(-500), Power.from_watts(3000)),
+            exclusion_bounds=Bounds(Power.from_watts(-300), Power.from_watts(300)),
+        ),
+    )
+
+
 async def test_all_batteries_temperature(setup_all_batteries: SetupArgs) -> None:
     """Test temperature for battery pool with all components in the microgrid.
 
@@ -860,7 +989,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
     compare_messages(msg, Sample(now, Percentage.from_percent(50.0)))
 
 
-async def run_power_bounds_test(  # pylint: disable=too-many-locals
+async def run_power_bounds_test(
     setup_args: SetupArgs,
 ) -> None:
     """Test if power bounds metric is working as expected.
@@ -869,48 +998,10 @@ async def run_power_bounds_test(  # pylint: disable=too-many-locals
         setup_args: Needed sdk tools and tools for mocking microgrid.
     """
     battery_pool = setup_args.battery_pool
-    mock_microgrid = setup_args.mock_microgrid
     streamer = setup_args.streamer
     battery_status_sender = setup_args.battery_status_sender
 
-    # All batteries are working and sending data. Not just the ones in the
-    # battery pool.
-    all_batteries = get_components(mock_microgrid, Battery)
-    await battery_status_sender.send(
-        ComponentPoolStatus(working=all_batteries, uncertain=set())
-    )
-    bat_invs_map = _get_battery_inverter_mappings(
-        all_batteries,
-        inv_bats=False,
-        bat_bats=False,
-        inv_invs=False,
-    )["bat_invs"]
-
-    for battery_id, inverter_ids in bat_invs_map.items():
-        # Sampling rate choose to reflect real application.
-        streamer.start_streaming(
-            BatteryDataWrapper(
-                component_id=battery_id,
-                timestamp=datetime.now(tz=timezone.utc),
-                power_inclusion_lower_bound=-1000,
-                power_inclusion_upper_bound=5000,
-                power_exclusion_lower_bound=-300,
-                power_exclusion_upper_bound=300,
-            ),
-            sampling_rate=0.05,
-        )
-        for inverter_id in inverter_ids:
-            streamer.start_streaming(
-                InverterDataWrapper(
-                    component_id=inverter_id,
-                    timestamp=datetime.now(tz=timezone.utc),
-                    active_power_inclusion_lower_bound=-900,
-                    active_power_inclusion_upper_bound=6000,
-                    active_power_exclusion_lower_bound=-200,
-                    active_power_exclusion_upper_bound=200,
-                ),
-                sampling_rate=0.1,
-            )
+    all_batteries, bat_invs_map = await start_streaming_power_bounds(setup_args)
 
     receiver = battery_pool.system_power_bounds.new_receiver(limit=50)
 

tests/timeseries/_battery_pool/test_battery_pool_control_methods.py

@@ -618,3 +618,48 @@ async def test_no_resend_0w(self, mocks: Mocks, mocker: MockerFixture) -> None:
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 25.0) for inv_id in mocks.microgrid.battery_inverter_ids
         ]
+
+    async def test_resend_0w_after_failure(
+        self, mocks: Mocks, mocker: MockerFixture
+    ) -> None:
+        """Test that 0W is sent again to an inverter whose last command failed.
+
+        A command that timed out may still have been applied. The battery status
+        is mocked, so the failed battery is not blocked before the next request.
+        """
+        set_power = cast(
+            AsyncMock,
+            microgrid.connection_manager.get().api_client.set_component_power_active,
+        )
+        await self._patch_battery_pool_status(mocks, mocker)
+        await self._init_data_for_batteries(mocks)
+        await self._init_data_for_inverters(mocks)
+
+        battery_pool = microgrid.new_battery_pool(priority=5)
+        bounds_rx = battery_pool.power_status.new_receiver()
+        self._assert_report(
+            await bounds_rx.receive(), power=None, lower=-4000.0, upper=4000.0
+        )
+
+        await battery_pool.propose_power(Power.from_watts(0.0))
+        await asyncio.sleep(1.0)  # Wait for the power to be distributed.
+        assert set_power.call_count == 4
+
+        # The command to the first inverter times out.
+        failing_inverter = mocks.microgrid.battery_inverter_ids[0]
+
+        async def side_effect(inv_id: int, _: float) -> None:
+            if inv_id == failing_inverter:
+                await asyncio.sleep(1000.0)
+
+        set_power.side_effect = side_effect
+        await battery_pool.propose_power(Power.from_watts(100.0))
+        await asyncio.sleep(20.0)  # Wait for the requests to time out.
+
+        set_power.side_effect = None
+        set_power.reset_mock()
+        await battery_pool.propose_power(Power.from_watts(0.0))
+        await asyncio.sleep(1.0)
+        assert sorted(set_power.call_args_list) == [
+            mocker.call(inv_id, 0.0) for inv_id in mocks.microgrid.battery_inverter_ids
+        ]