Change PowerDistributor to ignore components with NaN metrics (#247)

Each float metric returned from api can be float("NaN"). If that happen
PowerDistributor would fail because it is impossible to make math
calcuation on NaN value.
This PR is just a quick fix of this problem.
* if power bounds are NaN, then it tries to replace them with
corresponding power bounds from adjacent component. If these components
are also NaN, then it ignores battery.
* if other metrics metrics are NaN then it ignores battery.

More sophisticated solution should be done in the future, maybe use
resampler.

Author: ela-kotulska-frequenz

RELEASE_NOTES.md

@@ -7,6 +7,9 @@
 ## Upgrading
 
 * Remove `_soc` formula from the LogicalMeter. This feature has been moved to the BatteryPool.
+* Upgrade PowerDistributingActor to handle components with the NaN metrics (#247):
+    * if power bounds are NaN, then it tries to replace them with corresponding power bounds from adjacent component. If these components are also NaN, then it ignores battery.
+    * if other metrics metrics are NaN then it ignores battery.
 
 ## New Features
 

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

@@ -16,8 +16,8 @@
 import asyncio
 import logging
 from asyncio.tasks import ALL_COMPLETED
-from dataclasses import dataclass
-from math import ceil, floor
+from dataclasses import dataclass, replace
+from math import ceil, floor, isnan
 from typing import (  # pylint: disable=unused-import
     Any,
     Dict,
@@ -266,7 +266,7 @@ async def run(self) -> None:
 
             try:
                 pairs_data: List[InvBatPair] = self._get_components_data(
-                    self._all_battery_status.get_working_batteries(request.batteries)
+                    request.batteries
                 )
             except KeyError as err:
                 await user.channel.send(Error(request, str(err)))
@@ -548,8 +548,8 @@ def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
             Pairs of battery and adjacent inverter data.
         """
         pairs_data: List[InvBatPair] = []
-
-        for battery_id in batteries:
+        working_batteries = self._all_battery_status.get_working_batteries(batteries)
+        for battery_id in working_batteries:
             if battery_id not in self._battery_receivers:
                 raise KeyError(
                     f"No battery {battery_id}, "
@@ -558,24 +558,100 @@ def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
 
             inverter_id: int = self._bat_inv_map[battery_id]
 
-            battery_data: Optional[BatteryData] = self._battery_receivers[
-                battery_id
-            ].peek()
-
-            if battery_data is None:
-                _logger.warning("None returned from battery receiver %d.", battery_id)
+            data = self._get_battery_inverter_data(battery_id, inverter_id)
+            if data is None:
+                _logger.warning(
+                    "Skipping battery %d because its message isn't correct.",
+                    battery_id,
+                )
                 continue
 
-            inverter_data: Optional[InverterData] = self._inverter_receivers[
-                inverter_id
-            ].peek()
+            pairs_data.append(data)
+        return pairs_data
 
-            if inverter_data is None:
-                _logger.warning("None returned from inverter receiver %d.", inverter_id)
-                continue
+    def _get_battery_inverter_data(
+        self, battery_id: int, inverter_id: int
+    ) -> Optional[InvBatPair]:
+        """Get battery and inverter data if they are correct.
 
-            pairs_data.append(InvBatPair(battery_data, inverter_data))
-        return pairs_data
+        Each float data from the microgrid can be "NaN".
+        We can't do math operations on "NaN".
+        So check all the metrics and:
+        * if power bounds are NaN, then try to replace it with the corresponding
+          power bounds from the adjacent component. If metric in the adjacent component
+          is also NaN, then return None.
+        * if other metrics are NaN then return None. We can't assume anything for other
+          metrics.
+
+        Args:
+            battery_id: battery id
+            inverter_id: inverter id
+
+        Returns:
+            Data for the battery and adjacent inverter without NaN values.
+            Return None if we could not replace NaN values.
+        """
+        battery_data = self._battery_receivers[battery_id].peek()
+        inverter_data = self._inverter_receivers[inverter_id].peek()
+
+        # It means that nothing has been send on this channels, yet.
+        # This should be handled by BatteryStatus. BatteryStatus should not return
+        # this batteries as working.
+        if battery_data is None or inverter_data is None:
+            _logger.error(
+                "Battery %d or inverter %d send no data, yet. They should be not used.",
+                battery_id,
+                inverter_id,
+            )
+            return None
+
+        not_replaceable_metrics = [
+            battery_data.soc,
+            battery_data.soc_lower_bound,
+            battery_data.soc_upper_bound,
+            # We could replace capacity with 0, but it won't change distribution.
+            # This battery will be ignored in distribution anyway.
+            battery_data.capacity,
+        ]
+        if any(map(isnan, not_replaceable_metrics)):
+            _logger.debug("Some metrics for battery %d are NaN", battery_id)
+            return None
+
+        replaceable_metrics = [
+            battery_data.power_lower_bound,
+            battery_data.power_upper_bound,
+            inverter_data.active_power_lower_bound,
+            inverter_data.active_power_upper_bound,
+        ]
+
+        # If all values are ok then return them.
+        if not any(map(isnan, replaceable_metrics)):
+            return InvBatPair(battery_data, inverter_data)
+
+        # Replace NaN with the corresponding value in the adjacent component.
+        # If both metrics are None, return None to ignore this battery.
+        replaceable_pairs = [
+            ("power_lower_bound", "active_power_lower_bound"),
+            ("power_upper_bound", "active_power_upper_bound"),
+        ]
+
+        battery_new_metrics = {}
+        inverter_new_metrics = {}
+        for bat_attr, inv_attr in replaceable_pairs:
+            bat_bound = getattr(battery_data, bat_attr)
+            inv_bound = getattr(inverter_data, inv_attr)
+            if isnan(bat_bound) and isnan(inv_bound):
+                _logger.debug("Some metrics for battery %d are NaN", battery_id)
+                return None
+            if isnan(bat_bound):
+                battery_new_metrics[bat_attr] = inv_bound
+            elif isnan(inv_bound):
+                inverter_new_metrics[inv_attr] = bat_bound
+
+        return InvBatPair(
+            replace(battery_data, **battery_new_metrics),
+            replace(inverter_data, **inverter_new_metrics),
+        )
 
     async def _create_channels(self) -> None:
         """Create channels to get data of components in microgrid."""

tests/actor/test_power_distributing.py

@@ -183,6 +183,188 @@ async def test_power_distributor_one_user(self, mocker: MockerFixture) -> None:
         assert result.excess_power == 200
         assert result.request == request
 
+    async def test_battery_soc_nan(self, mocker: MockerFixture) -> None:
+        # pylint: disable=too-many-locals
+        """Test if battery with SoC==NaN is not used."""
+        mock_microgrid = await self.init_mock_microgrid(mocker)
+
+        await mock_microgrid.send(
+            battery_msg(
+                106,
+                soc=Metric(float("NaN"), Bound(20, 80)),
+                capacity=Metric(98000),
+                power=Bound(-1000, 1000),
+            )
+        )
+
+        channel = Bidirectional[Request, Result]("user1", "power_distributor")
+
+        request = Request(
+            power=1200,
+            batteries={106, 206},
+            request_timeout_sec=SAFETY_TIMEOUT,
+        )
+
+        attrs = {"get_working_batteries.return_value": request.batteries}
+        mocker.patch(
+            "frequenz.sdk.actor.power_distributing.power_distributing.BatteryPoolStatus",
+            return_value=MagicMock(spec=BatteryPoolStatus, **attrs),
+        )
+
+        mocker.patch("asyncio.sleep", new_callable=AsyncMock)
+        battery_status_channel = Broadcast[BatteryStatus]("battery_status")
+        distributor = PowerDistributingActor(
+            {"user1": channel.service_handle},
+            battery_status_sender=battery_status_channel.new_sender(),
+        )
+
+        attrs = {"get_working_batteries.return_value": request.batteries}
+        mocker.patch(
+            "frequenz.sdk.actor.power_distributing.power_distributing.BatteryPoolStatus",
+            return_value=MagicMock(spec=BatteryPoolStatus, **attrs),
+        )
+
+        client_handle = channel.client_handle
+        await client_handle.send(request)
+
+        done, pending = await asyncio.wait(
+            [asyncio.create_task(client_handle.receive())],
+            timeout=SAFETY_TIMEOUT,
+        )
+        await distributor._stop_actor()
+
+        assert len(pending) == 0
+        assert len(done) == 1
+
+        result: Result = done.pop().result()
+        assert isinstance(result, Success)
+        assert result.used_batteries == {206}
+        assert result.succeed_power == 500
+        assert result.excess_power == 700
+        assert result.request == request
+
+    async def test_battery_capacity_nan(self, mocker: MockerFixture) -> None:
+        # pylint: disable=too-many-locals
+        """Test if power distribution works with single user works."""
+        mock_microgrid = await self.init_mock_microgrid(mocker)
+
+        await mock_microgrid.send(
+            battery_msg(
+                106,
+                soc=Metric(40, Bound(20, 80)),
+                capacity=Metric(float("NaN")),
+                power=Bound(-1000, 1000),
+            )
+        )
+
+        channel = Bidirectional[Request, Result]("user1", "power_distributor")
+
+        request = Request(
+            power=1200,
+            batteries={106, 206},
+            request_timeout_sec=SAFETY_TIMEOUT,
+        )
+        attrs = {"get_working_batteries.return_value": request.batteries}
+        mocker.patch(
+            "frequenz.sdk.actor.power_distributing.power_distributing.BatteryPoolStatus",
+            return_value=MagicMock(spec=BatteryPoolStatus, **attrs),
+        )
+
+        mocker.patch("asyncio.sleep", new_callable=AsyncMock)
+        battery_status_channel = Broadcast[BatteryStatus]("battery_status")
+        distributor = PowerDistributingActor(
+            {"user1": channel.service_handle},
+            battery_status_sender=battery_status_channel.new_sender(),
+        )
+
+        client_handle = channel.client_handle
+        await client_handle.send(request)
+
+        done, pending = await asyncio.wait(
+            [asyncio.create_task(client_handle.receive())],
+            timeout=SAFETY_TIMEOUT,
+        )
+        await distributor._stop_actor()
+
+        assert len(pending) == 0
+        assert len(done) == 1
+
+        result: Result = done.pop().result()
+        assert isinstance(result, Success)
+        assert result.used_batteries == {206}
+        assert result.succeed_power == 500
+        assert result.excess_power == 700
+        assert result.request == request
+
+    async def test_battery_power_bounds_nan(self, mocker: MockerFixture) -> None:
+        # pylint: disable=too-many-locals
+        """Check if missign"""
+        mock_microgrid = await self.init_mock_microgrid(mocker)
+
+        # Battery 206 should work even if his inverter sends NaN
+        await mock_microgrid.send(
+            inverter_msg(
+                205,
+                power=Bound(float("NaN"), float("NaN")),
+            )
+        )
+
+        # Battery 106 should not work because both battery and inverter sends NaN
+        await mock_microgrid.send(
+            inverter_msg(
+                105,
+                power=Bound(-1000, float("NaN")),
+            )
+        )
+
+        await mock_microgrid.send(
+            battery_msg(
+                106,
+                soc=Metric(40, Bound(20, 80)),
+                capacity=Metric(float(98000)),
+                power=Bound(float("NaN"), float("NaN")),
+            )
+        )
+
+        channel = Bidirectional[Request, Result]("user1", "power_distributor")
+
+        request = Request(
+            power=1200,
+            batteries={106, 206},
+            request_timeout_sec=SAFETY_TIMEOUT,
+        )
+        attrs = {"get_working_batteries.return_value": request.batteries}
+        mocker.patch(
+            "frequenz.sdk.actor.power_distributing.power_distributing.BatteryPoolStatus",
+            return_value=MagicMock(spec=BatteryPoolStatus, **attrs),
+        )
+
+        mocker.patch("asyncio.sleep", new_callable=AsyncMock)
+        battery_status_channel = Broadcast[BatteryStatus]("battery_status")
+        distributor = PowerDistributingActor(
+            {"user1": channel.service_handle},
+            battery_status_sender=battery_status_channel.new_sender(),
+        )
+
+        client_handle = channel.client_handle
+        await client_handle.send(request)
+
+        done, pending = await asyncio.wait(
+            [asyncio.create_task(client_handle.receive())],
+            timeout=SAFETY_TIMEOUT,
+        )
+        await distributor._stop_actor()
+
+        assert len(pending) == 0
+        assert len(done) == 1
+
+        result: Result = done.pop().result()
+        assert isinstance(result, Success)
+        assert result.used_batteries == {206}
+        assert result.succeed_power == 1000
+        assert result.excess_power == 200
+        assert result.request == request
+
     async def test_power_distributor_two_users(self, mocker: MockerFixture) -> None:
         # pylint: disable=too-many-locals
         """Test if power distribution works with two users."""