WIP fix test_producer.py

Author: llucax

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

@@ -4,6 +4,7 @@
 """A post-fix formula evaluator that operates on `Sample` receivers."""
 
 import asyncio
+import logging
 from collections.abc import Callable
 from datetime import datetime
 from math import isinf, isnan
@@ -12,6 +13,8 @@
 from .._base_types import QuantityT, Sample
 from ._formula_steps import FormulaStep, MetricFetcher
 
+_logger = logging.getLogger(__name__)
+
 
 class FormulaEvaluator(Generic[QuantityT]):
     """A post-fix formula evaluator that operates on `Sample` receivers."""
@@ -97,10 +100,18 @@ async def apply(self) -> Sample[QuantityT]:
             RuntimeError: if some samples didn't arrive, or if formula application
                 failed.
         """
+
+        async def fetch_next(
+            name: str, fetcher: MetricFetcher[QuantityT]
+        ) -> Sample[QuantityT] | None:
+            val = await fetcher.fetch_next()
+            _logger.error(f">> FormulaEvaluator: {name=} {val=}")
+            return val
+
         eval_stack: list[float] = []
         ready_metrics, pending = await asyncio.wait(
             [
-                asyncio.create_task(fetcher.fetch_next(), name=name)
+                asyncio.create_task(fetch_next(name, fetcher), name=name)
                 for name, fetcher in self._metric_fetchers.items()
             ],
             return_when=asyncio.ALL_COMPLETED,

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

@@ -79,10 +79,16 @@ def generate(  # noqa: DOC502
 
         if self._config.allow_fallback:
             fallbacks = self._get_fallback_formulas(producer_components)
+            print(
+                f"ProducerPowerFormula: {dict({str(c): f for c, f in fallbacks.items()})}"
+            )
 
             for idx, (primary_component, fallback_formula) in enumerate(
                 fallbacks.items()
             ):
+                print(f"\t{idx}: {primary_component=}")
+                print(f"\t{is_not_meter(primary_component)=}")
+
                 if idx > 0:
                     builder.push_oper("+")
 
@@ -121,6 +127,9 @@ def _get_fallback_formulas(
             A dictionary mapping primary components to their FallbackFormulaMetricFetcher.
         """
         fallbacks = self._get_metric_fallback_components(components)
+        print(
+            f"fallback components for {ProducerPowerFormula}: {dict({str(c): [str(cc) for cc in f] for c, f in fallbacks.items()})}"
+        )
 
         fallback_formulas: dict[
             Component, FallbackFormulaMetricFetcher[Power] | None

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

@@ -477,6 +477,7 @@ async def _synchronize_and_fetch_fallback(
         # We need to save value, because
         # primary_fetcher_value.timestamp < self._latest_fallback_sample.timestamp
         # In that case we should wait for our time window.
+        print(f"MetricFetcher<{self._name}>: {self._latest_fallback_sample=}")
         if self._latest_fallback_sample is None:
             self._latest_fallback_sample = await self._fetch_from_fallback(
                 fallback_fetcher
@@ -516,9 +517,11 @@ async def fetch_next(self) -> Sample[QuantityT] | None:
 
     async def _fetch_next(self) -> Sample[QuantityT] | None:
         # First fetch from primary stream
+        print(f"MetricFetcher<{self._name}>: Fetching from primary stream")
         primary_value: Sample[QuantityT] | None = None
         try:
             primary_value = await self._stream.receive()
+            print(f"MetricFetcher<{self._name}>: RECEIVED {primary_value=}")
         except ReceiverStoppedError:
             if self._is_stopped:
                 _logger.debug("Stream for metric fetcher %s closed.", self._name)
@@ -532,13 +535,15 @@ async def _fetch_next(self) -> Sample[QuantityT] | None:
 
         # We have no fallback, so we just return primary value even if it is not correct.
         if self._fallback is None:
+            print(f"MetricFetcher<{self._name}>: NO FALLBACK, returning primary")
             return primary_value
 
         is_primary_value_valid = primary_value is not None and self._is_value_valid(
             primary_value.value
         )
 
         if is_primary_value_valid:
+            print(f"MetricFetcher<{self._name}>: Primary value is valid")
             # Primary stream is good again, so we can stop fallback and return primary_value.
             if self._fallback.is_running:
                 _logger.info(
@@ -560,6 +565,7 @@ async def _fetch_next(self) -> Sample[QuantityT] | None:
             self._fallback.start()
             return primary_value
 
+        print(f"MetricFetcher<{self._name}>: Fetching from fallback stream")
         return await self._synchronize_and_fetch_fallback(primary_value, self._fallback)
 
     @property

tests/timeseries/test_producer.py

@@ -3,6 +3,7 @@
 
 """Test the logical component for calculating high level producer metrics."""
 
+import asyncio
 from contextlib import AsyncExitStack
 
 from frequenz.quantities import Power
@@ -114,59 +115,66 @@ async def test_producer_fallback_formula(self, mocker: MockerFixture) -> None:
 
             # fmt: off
             expected_input_output: list[
-                tuple[list[float | None], list[float | None], list[float | None], Power | None]
+                tuple[float, list[float | None], list[float | None], list[float | None], Power | None]
             ] = [
-                # ([pv_meter_power], [pv_inverter_power], [chp_power], expected_power)
+                # (test number, [pv_meter_power], [pv_inverter_power], [chp_power], expected_power)
+                # Case 1: All components are available
                 # Add power from meters and chp
-                ([-1.0, -2.0], [None, -200.0], [300], Power.from_watts(297.0)),
-                ([-1.0, -10], [-100.0, -200.0], [400], Power.from_watts(389.0)),
+                (1.1, [-1.0, -2.0], [None, -200.0], [300], Power.from_watts(297.0)),
+                (1.2, [-1.0, -10], [-100.0, -200.0], [400], Power.from_watts(389.0)),
                 # Case 2: The first meter is unavailable (None).
                 # Subscribe to the fallback inverter, but return None as the result,
                 # according to the "nones-are-zero" rule
-                ([None, -2.0], [-100, -200.0], [400], None),
+                (2.1, [None, -2.0], [-100, -200.0], [400], None),
                 # Case 3: First meter is unavailable (None). Fallback inverter provides
                 # a value.
                 # Add second meter, first inverter and chp power
-                ([None, -2.0], [-100, -200.0], [400], Power.from_watts(298.0)),
-                ([None, -2.0], [-50, -200.0], [300], Power.from_watts(248.0)),
+                (3.1, [None, -2.0], [-100, -200.0], [400], Power.from_watts(298.0)),
+                (3.2, [None, -2.0], [-50, -200.0], [300], Power.from_watts(248.0)),
                 # Case 4: Both first meter and its fallback inverter are unavailable
                 # (None). Return 0 from failing component according to the
                 # "nones-are-zero" rule.
-                ([None, -2.0], [None, -200.0], [300], Power.from_watts(298.0)),
-                ([None, -10.0], [-20.0, -200.0], [300], Power.from_watts(270.0)),
+                (4.1, [None, -2.0], [None, -200.0], [300], Power.from_watts(298.0)),
+                (4.2, [None, -10.0], [-20.0, -200.0], [300], Power.from_watts(270.0)),
                 # Case 5: CHP is unavailable. Return 0 from failing component
                 # according to the "nones-are-zero" rule.
-                ([None, -10.0], [-20.0, -200.0], [None], Power.from_watts(-30.0)),
+                (5.1, [None, -10.0], [-20.0, -200.0], [None], Power.from_watts(-30.0)),
                 # Case 6: Both meters are unavailable (None). Subscribe for fallback inverter
-                ([None, None], [-20.0, -200.0], [None], None),
-                ([None, None], [-20.0, -200.0], [None], Power.from_watts(-220.0)),
-                ([None, None], [None, -200.0], [None], Power.from_watts(-200.0)),
+                (6.1, [None, None], [-20.0, -200.0], [None], None),
+                (6.2, [None, None], [-20.0, -200.0], [None], Power.from_watts(-220.0)),
+                (6.3, [None, None], [None, -200.0], [None], Power.from_watts(-200.0)),
                 # Case 7: All components are unavailable (None). Return 0 according to the
                 # "nones-are-zero" rule.
-                ([None, None], [None, None], [None], Power.from_watts(0)),
-                ([None, None], [None, None], [None], Power.from_watts(0)),
-                ([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)),
+                (7.1, [None, None], [None, None], [None], Power.from_watts(0)),
+                (7.2, [None, None], [None, None], [None], Power.from_watts(0)),
+                (7.3, [None, None], [None, None], [300.0], Power.from_watts(300.0)),
+                (7.4, [-200.0, None], [None, -100.0], [50.0], Power.from_watts(-250.0)),
+                (7.5, [-200.0, -200.0], [-10.0, -20.0], [50.0], Power.from_watts(-350.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)),
+                (8.1, [None, -200.0], [-10.0, -100.0], [50.0], None),
+                (8.2, [None, -200.0], [-10.0, -100.0], [50.0], Power.from_watts(-160)),
 
             ]
             # fmt: on
 
-            for idx, (
+            for (
+                idx,
                 meter_power,
                 pv_inverter_power,
                 chp_power,
                 expected_power,
-            ) in enumerate(expected_input_output):
+            ) in expected_input_output:
+                print(
+                    f"\n----------------------------------------------------\nTEST CASE {idx}"
+                )
                 await mockgrid.mock_resampler.send_chp_power(chp_power)
                 await mockgrid.mock_resampler.send_meter_power(meter_power)
                 await mockgrid.mock_resampler.send_pv_inverter_power(pv_inverter_power)
                 mockgrid.mock_resampler.next_ts()
+                await asyncio.sleep(0.1)
 
                 result = await producer_power_receiver.receive()
+                print(f">>>>> result: {result}")
                 assert result.value == expected_power, (
                     f"Test case {idx} failed:"
                     + f" meter_power: {meter_power}"