Generate formulas automatically from the component graph (#103)

Author: sahas-subramanian-frequenz

src/frequenz/sdk/microgrid/client/_client.py

@@ -23,7 +23,10 @@
     InverterData,
     MeterData,
 )
-from ..component._component import _component_category_from_protobuf
+from ..component._component import (
+    _component_category_from_protobuf,
+    _component_type_from_protobuf,
+)
 from ._connection import Connection
 from ._retry import LinearBackoff, RetryStrategy
 
@@ -229,7 +232,11 @@ async def components(self) -> Iterable[Component]:
             component_list.components,
         )
         result: Iterable[Component] = map(
-            lambda c: Component(c.id, _component_category_from_protobuf(c.category)),
+            lambda c: Component(
+                c.id,
+                _component_category_from_protobuf(c.category),
+                _component_type_from_protobuf(c.category, c.inverter),
+            ),
             components_only,
         )
 

src/frequenz/sdk/microgrid/component/__init__.py

@@ -6,7 +6,7 @@
 This package provides classes to operate con microgrid components.
 """
 
-from ._component import Component, ComponentCategory, ComponentMetricId
+from ._component import Component, ComponentCategory, ComponentMetricId, InverterType
 from ._component_data import (
     BatteryData,
     ComponentData,
@@ -25,5 +25,6 @@
     "EVChargerCableState",
     "EVChargerData",
     "InverterData",
+    "InverterType",
     "MeterData",
 ]

src/frequenz/sdk/microgrid/component/_component.py

@@ -7,10 +7,52 @@
 
 from dataclasses import dataclass
 from enum import Enum
+from typing import Optional
 
+import frequenz.api.microgrid.inverter_pb2 as inverter_pb
 import frequenz.api.microgrid.microgrid_pb2 as microgrid_pb
 
 
+class ComponentType(Enum):
+    """A base class from which individual component types are derived."""
+
+
+class InverterType(ComponentType):
+    """Enum representing inverter types."""
+
+    NONE = inverter_pb.Type.TYPE_UNSPECIFIED
+    BATTERY = inverter_pb.Type.TYPE_BATTERY
+    SOLAR = inverter_pb.Type.TYPE_SOLAR
+    HYBRID = inverter_pb.Type.TYPE_HYBRID
+
+
+def _component_type_from_protobuf(
+    component_category: microgrid_pb.ComponentCategory.ValueType,
+    component_type: inverter_pb.Type.ValueType,
+) -> Optional[ComponentType]:
+    """Convert a protobuf InverterType message to Component enum.
+
+    For internal-only use by the `microgrid` package.
+
+    Args:
+        component_category: category the type belongs to.
+        component_type: protobuf enum to convert.
+
+    Returns:
+        Enum value corresponding to the protobuf message.
+    """
+    # ComponentType values in the protobuf definition are not unique across categories
+    # as of v0.11.0, so we need to check the component category first, before doing any
+    # component type checks.
+    if component_category == microgrid_pb.ComponentCategory.COMPONENT_CATEGORY_INVERTER:
+        if not any(t.value == component_type for t in InverterType):
+            return None
+
+        return InverterType(component_type)
+
+    return None
+
+
 class ComponentCategory(Enum):
     """Possible types of microgrid component."""
 
@@ -62,6 +104,7 @@ class Component:
 
     component_id: int
     category: ComponentCategory
+    type: Optional[ComponentType] = None
 
     def is_valid(self) -> bool:
         """Check if this instance contains valid data.

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

@@ -0,0 +1,31 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Generators for formulas from component graphs."""
+
+from ._battery_power_formula import BatteryPowerFormula
+from ._formula_generator import (
+    ComponentNotFound,
+    FormulaGenerationError,
+    FormulaGenerator,
+)
+from ._grid_power_formula import GridPowerFormula
+from ._pv_power_formula import PVPowerFormula
+
+__all__ = [
+    #
+    # Base class
+    #
+    "FormulaGenerator",
+    #
+    # Formula generators
+    #
+    "GridPowerFormula",
+    "BatteryPowerFormula",
+    "PVPowerFormula",
+    #
+    # Exceptions
+    #
+    "ComponentNotFound",
+    "FormulaGenerationError",
+]

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

@@ -0,0 +1,54 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for Grid Power."""
+
+from .....sdk import microgrid
+from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType
+from .._formula_engine import FormulaEngine
+from ._formula_generator import ComponentNotFound, FormulaGenerator
+
+
+class BatteryPowerFormula(FormulaGenerator):
+    """Creates a formula engine from the component graph for calculating grid power."""
+
+    async def generate(
+        self,
+    ) -> FormulaEngine:
+        """Make a formula for the cumulative AC battery power of a microgrid.
+
+        The calculation is performed by adding the Active Powers of all the inverters
+        that are attached to batteries.
+
+        If there's no data coming from an inverter, that inverter's power will be
+        treated as 0.
+
+        Returns:
+            A formula engine that will calculate cumulative battery power values.
+
+        Raises:
+            ComponentNotFound: if there are no batteries in the component graph, or if
+                they don't have an inverter as a predecessor.
+            FormulaGenerationError: If a battery has a non-inverter predecessor
+                in the component graph.
+        """
+        builder = self._get_builder(ComponentMetricId.ACTIVE_POWER)
+        component_graph = microgrid.get().component_graph
+        battery_inverters = list(
+            comp
+            for comp in component_graph.components()
+            if comp.category == ComponentCategory.INVERTER
+            and comp.type == InverterType.BATTERY
+        )
+
+        if not battery_inverters:
+            raise ComponentNotFound(
+                "Unable to find any battery inverters in the component graph."
+            )
+
+        for idx, comp in enumerate(battery_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/_formula_generators/_formula_generator.py

@@ -0,0 +1,59 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Base class for formula generators that use the component graphs."""
+
+from abc import ABC, abstractmethod
+
+from frequenz.channels import Sender
+
+from ....actor import ChannelRegistry, ComponentMetricRequest
+from ....microgrid.component import ComponentMetricId
+from .._formula_engine import FormulaEngine
+from .._resampled_formula_builder import ResampledFormulaBuilder
+
+
+class FormulaGenerationError(Exception):
+    """An error encountered during formula generation from the component graph."""
+
+
+class ComponentNotFound(FormulaGenerationError):
+    """Indicates that a component required for generating a formula is not found."""
+
+
+class FormulaGenerator(ABC):
+    """A class for generating formulas from the component graph."""
+
+    def __init__(
+        self,
+        namespace: str,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
+    ) -> None:
+        """Create a `FormulaGenerator` instance.
+
+        Args:
+            namespace: A namespace to use with the data-pipeline.
+            channel_registry: A channel registry instance shared with the resampling
+                actor.
+            resampler_subscription_sender: A sender for sending metric requests to the
+                resampling actor.
+        """
+        self._channel_registry = channel_registry
+        self._resampler_subscription_sender = resampler_subscription_sender
+        self._namespace = namespace
+
+    def _get_builder(
+        self, component_metric_id: ComponentMetricId
+    ) -> ResampledFormulaBuilder:
+        builder = ResampledFormulaBuilder(
+            self._namespace,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            component_metric_id,
+        )
+        return builder
+
+    @abstractmethod
+    async def generate(self) -> FormulaEngine:
+        """Generate a formula engine, based on the component graph."""

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

@@ -0,0 +1,67 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for Grid Power."""
+
+from .....sdk import microgrid
+from ....microgrid.component import ComponentCategory, ComponentMetricId
+from .._formula_engine import FormulaEngine
+from ._formula_generator import ComponentNotFound, FormulaGenerator
+
+
+class GridPowerFormula(FormulaGenerator):
+    """Creates a formula engine from the component graph for calculating grid power."""
+
+    async def generate(
+        self,
+    ) -> FormulaEngine:
+        """Generate a formula for calculating grid power from the component graph.
+
+        Returns:
+            A formula engine that will calculate grid power values.
+
+        Raises:
+            ComponentNotFound: when the component graph doesn't have a `GRID` component.
+        """
+        builder = self._get_builder(ComponentMetricId.ACTIVE_POWER)
+        component_graph = microgrid.get().component_graph
+        grid_component = next(
+            (
+                comp
+                for comp in component_graph.components()
+                if comp.category == ComponentCategory.GRID
+            ),
+            None,
+        )
+
+        if grid_component is None:
+            raise ComponentNotFound(
+                "Unable to find a GRID component from the component graph."
+            )
+
+        grid_successors = component_graph.successors(grid_component.component_id)
+
+        # generate a formula that just adds values from all commponents that are
+        # directly connected to the grid.
+        for idx, comp in enumerate(grid_successors):
+            if idx > 0:
+                builder.push_oper("+")
+
+            # Ensure the device has an `ACTIVE_POWER` metric.  When inverters
+            # produce `None` samples, those inverters are excluded from the
+            # calculation by treating their `None` values as `0`s.
+            #
+            # This is not possible for Meters, so when they produce `None`
+            # values, those values get propagated as the output.
+            if comp.category == ComponentCategory.INVERTER:
+                nones_are_zeros = True
+            elif comp.category == ComponentCategory.METER:
+                nones_are_zeros = False
+            else:
+                continue
+
+            await builder.push_component_metric(
+                comp.component_id, nones_are_zeros=nones_are_zeros
+            )
+
+        return builder.build()

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()