Add a formula generator for PV power

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/sdk/timeseries/logical_meter/_formula_generators/__init__.py

@@ -10,11 +10,22 @@
     FormulaGenerator,
 )
 from ._grid_power_formula import GridPowerFormula
+from ._pv_power_formula import PVPowerFormula
 
 __all__ = [
+    #
+    # Base class
+    #
+    "FormulaGenerator",
+    #
+    # Formula generators
+    #
     "GridPowerFormula",
     "BatteryPowerFormula",
+    "PVPowerFormula",
+    #
+    # Exceptions
+    #
     "ComponentNotFound",
-    "FormulaGenerator",
     "FormulaGenerationError",
 ]

src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_pv_power_formula.py

@@ -0,0 +1,45 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator for PV Power, from the component graph."""
+
+from .....sdk import microgrid
+from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType
+from .._formula_engine import FormulaEngine
+from ._formula_generator import ComponentNotFound, FormulaGenerator
+
+
+class PVPowerFormula(FormulaGenerator):
+    """Creates a formula engine for calculating the PV power production."""
+
+    async def generate(self) -> FormulaEngine:
+        """Make a formula for the PV power production of a microgrid.
+
+        Returns:
+            A formula engine that will calculate PV power production values.
+
+        Raises:
+            ComponentNotFound: if there are no PV inverters in the component graph.
+        """
+        builder = self._get_builder(ComponentMetricId.ACTIVE_POWER)
+
+        component_graph = microgrid.get().component_graph
+        pv_inverters = list(
+            comp
+            for comp in component_graph.components()
+            if comp.category == ComponentCategory.INVERTER
+            and comp.type == InverterType.SOLAR
+        )
+
+        if not pv_inverters:
+            raise ComponentNotFound(
+                "Unable to find any PV inverters in the component graph."
+            )
+
+        for idx, comp in enumerate(pv_inverters):
+            if idx > 0:
+                builder.push_oper("+")
+
+            await builder.push_component_metric(comp.component_id, nones_are_zeros=True)
+
+        return builder.build()

src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py

@@ -17,7 +17,12 @@
 from ...microgrid.component import ComponentMetricId
 from .. import Sample
 from ._formula_engine import FormulaEngine
-from ._formula_generators import BatteryPowerFormula, FormulaGenerator, GridPowerFormula
+from ._formula_generators import (
+    BatteryPowerFormula,
+    FormulaGenerator,
+    GridPowerFormula,
+    PVPowerFormula,
+)
 from ._resampled_formula_builder import ResampledFormulaBuilder
 
 logger = logging.Logger(__name__)
@@ -170,3 +175,15 @@ async def battery_power(self) -> Receiver[Sample]:
 
         """
         return await self._get_formula_stream("battery_power", BatteryPowerFormula)
+
+    async def pv_power(self) -> Receiver[Sample]:
+        """Fetch the PV power production in the microgrid.
+
+        If a formula engine to calculate PV power production is not
+        already running, it will be started.  Else, we'll just get a new
+        receiver to the already existing data stream.
+
+        Returns:
+            A *new* receiver that will stream PV power production values.
+        """
+        return await self._get_formula_stream("pv_power", PVPowerFormula)