Change PowerDistributor to ignore components with NaN metrics

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.

Signed-off-by: ela-kotulska-frequenz <[email protected]>

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,
@@ -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."""