Implement a stop method on MetricFetcher

And use it to stop the fallback fetcher when it is no longer needed.

Signed-off-by: Elzbieta Kotulska <[email protected]>

Author: ela-kotulska-frequenz

src/frequenz/sdk/timeseries/formula_engine/_formula_engine.py

@@ -117,9 +117,10 @@ async def stop(self) -> None:
         if self._task is None:
             return
         await cancel_and_await(self._task)
+
         _, fetchers = self._builder.finalize()
         for fetcher in fetchers.values():
-            fetcher.stream.close()
+            await fetcher.stop()
 
     @classmethod
     def from_receiver(

src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_fallback_formula_metric_fetcher.py

@@ -77,3 +77,13 @@ def consume(self) -> Sample[QuantityT]:
         ), f"Fallback metric fetcher: {self.name} was not started"
 
         return self._receiver.consume()
+
+    async def stop(self) -> None:
+        """Stop fallback metric fetcher, by closing the connection."""
+        if self._formula_engine is not None:
+            await self._formula_engine.stop()
+            self._formula_engine = None
+
+        if self._receiver is not None:
+            self._receiver.close()
+            self._receiver = None

src/frequenz/sdk/timeseries/formula_engine/_formula_steps.py

@@ -368,6 +368,10 @@ def is_running(self) -> bool:
     def start(self) -> None:
         """Initialize the metric fetcher and start fetching samples."""
 
+    @abstractmethod
+    async def stop(self) -> None:
+        """Stope the fetcher if is running."""
+
 
 class MetricFetcher(Generic[QuantityT], FormulaStep):
     """A formula step for fetching a value from a metric Receiver."""
@@ -406,6 +410,16 @@ def stream(self) -> Receiver[Sample[QuantityT]]:
         """
         return self._stream
 
+    async def stop(self) -> None:
+        """Stop metric fetcher.
+
+        If metric fetcher is stopped, it can't be started again.
+        There is no use-case now to start it again.
+        """
+        self.stream.close()
+        if self._fallback:
+            await self._fallback.stop()
+
     def stream_name(self) -> str:
         """Return the name of the stream.
 
@@ -497,6 +511,9 @@ async def _fetch_next(self) -> Sample[QuantityT] | None:
         )
 
         if is_primary_value_valid:
+            # Primary stream is good again, so we can stop fallback and return primary_value.
+            if self._fallback.is_running:
+                await self._fallback.stop()
             return primary_value
 
         if not self._fallback.is_running:

tests/microgrid/test_grid.py

@@ -480,6 +480,8 @@ async def test_grid_fallback_formula_without_grid_meter(mocker: MockerFixture) -
             ([2000, 1000], [-200, 1000, None], Power.from_watts(3000)),
             # battery start working
             ([2000, 10], [-200, 1000, 100], Power.from_watts(2110)),
+            # No primary value, start fallback formula
+            ([2000, None], [-200, 1000, 100], None),
             ([2000, None], [-200, 1000, 100], Power.from_watts(2900)),
         ]
         # fmt: on

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -642,8 +642,9 @@ async def test_battery_power_fallback_formula(
             ([-1.0, None], [100.0, 100.0, None], Power.from_watts(200.0)),
             # Case 4: bat_inv_meter is available, ignore fallback inverters
             ([-1.0, 10], [100.0, 100.0, None], Power.from_watts(10.0)),
-            # Case 4: all components are unavailable (None). Return 0 according to the
-            # "nones-are-zero" rule.
+            # Case 4: all components are unavailable (None). Start fallback formula.
+            # Returned power = 0 according to the "nones-are-zero" rule.
+            ([-1.0, None], [None, None, None], None),
             ([-1.0, None], [None, None, None], Power.from_watts(0.0)),
             # Case 5: Components becomes available
             ([-1.0, None], [None, None, 100.0], Power.from_watts(100.0)),

tests/timeseries/test_consumer.py

@@ -113,7 +113,8 @@ async def test_consumer_power_fallback_formula_with_grid_meter(
                 ([None, 100, -50], [100], [-200, -300], None),
                 ([None, 200, -50], [100], [-200, -300], None),
                 ([100, 100, -50], [100], [-200, -300], Power.from_watts(350)),
-                # Case 5: Only grid meter is working
+                # Case 5: Only grid meter is working, subscribe for fallback formula
+                ([100, None, None], [None], [None, None], None),
                 ([100, None, None], [None], [None, None], Power.from_watts(100)),
                 ([-500, None, None], [None], [None, None], Power.from_watts(-500)),
                 # Case 6: Nothing is working

tests/timeseries/test_producer.py

@@ -148,9 +148,9 @@ async def test_producer_fallback_formula(self, mocker: MockerFixture) -> None:
                 ([None, None], [None, None], [300.0], Power.from_watts(300.0)),
                 ([-200.0, None], [None, -100.0], [50.0], Power.from_watts(-250.0)),
                 ([-200.0, -200.0], [-10.0, -20.0], [50.0], Power.from_watts(-350.0)),
-                # Case 8: Meter is unavailable but we already subscribed for inverter
-                # So don't return None in this case. Just proper formula result.
-                ([None, -200.0], [-10.0, -100.0], [50.0], Power.from_watts(-160.0)),
+                # Case 8: Meter is unavailable, start fallback formula.
+                ([None, -200.0], [-10.0, -100.0], [50.0], None),
+                ([None, -200.0], [-10.0, -100.0], [50.0], Power.from_watts(-160)),
 
             ]
             # fmt: on