diff --git a/src/frequenz/sdk/microgrid/_graph.py b/src/frequenz/sdk/microgrid/_graph.py index 15e1ea6ff..cbe04df26 100644 --- a/src/frequenz/sdk/microgrid/_graph.py +++ b/src/frequenz/sdk/microgrid/_graph.py @@ -30,7 +30,7 @@ import networkx as nx from .client import Connection, MicrogridApiClient -from .component import Component, ComponentCategory +from .component import Component, ComponentCategory, InverterType _logger = logging.getLogger(__name__) @@ -113,6 +113,159 @@ def successors(self, component_id: int) -> Set[Component]: KeyError: if the specified `component_id` is not in the graph """ + @abstractmethod + def is_pv_inverter(self, component: Component) -> bool: + """Check if the specified component is a PV inverter. + + Args: + component: component to check. + + Returns: + Whether the specified component is a PV inverter. + """ + + @abstractmethod + def is_pv_meter(self, component: Component) -> bool: + """Check if the specified component is a PV meter. + + This is done by checking if the component has only PV inverters as its + successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is a PV meter. + """ + + @abstractmethod + def is_pv_chain(self, component: Component) -> bool: + """Check if the specified component is part of a PV chain. + + A component is part of a PV chain if it is a PV meter or a PV inverter. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of a PV chain. + """ + + @abstractmethod + def is_battery_inverter(self, component: Component) -> bool: + """Check if the specified component is a battery inverter. + + Args: + component: component to check. + + Returns: + Whether the specified component is a battery inverter. + """ + + @abstractmethod + def is_battery_meter(self, component: Component) -> bool: + """Check if the specified component is a battery meter. + + This is done by checking if the component has only battery inverters as its + predecessors. + + Args: + component: component to check. + + Returns: + Whether the specified component is a battery meter. + """ + + @abstractmethod + def is_battery_chain(self, component: Component) -> bool: + """Check if the specified component is part of a battery chain. + + A component is part of a battery chain if it is a battery meter or a battery + inverter. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of a battery chain. + """ + + @abstractmethod + def is_ev_charger(self, component: Component) -> bool: + """Check if the specified component is an EV charger. + + Args: + component: component to check. + + Returns: + Whether the specified component is an EV charger. + """ + + @abstractmethod + def is_ev_charger_meter(self, component: Component) -> bool: + """Check if the specified component is an EV charger meter. + + This is done by checking if the component has only EV chargers as its + successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is an EV charger meter. + """ + + @abstractmethod + def is_ev_charger_chain(self, component: Component) -> bool: + """Check if the specified component is part of an EV charger chain. + + A component is part of an EV charger chain if it is an EV charger meter or an + EV charger. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of an EV charger chain. + """ + + @abstractmethod + def is_chp(self, component: Component) -> bool: + """Check if the specified component is a CHP. + + Args: + component: component to check. + + Returns: + Whether the specified component is a CHP. + """ + + @abstractmethod + def is_chp_meter(self, component: Component) -> bool: + """Check if the specified component is a CHP meter. + + This is done by checking if the component has only CHPs as its successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is a CHP meter. + """ + + @abstractmethod + def is_chp_chain(self, component: Component) -> bool: + """Check if the specified component is part of a CHP chain. + + A component is part of a CHP chain if it is a CHP meter or a CHP. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of a CHP chain. + """ + class _MicrogridComponentGraph(ComponentGraph): """ComponentGraph implementation designed to work with the microgrid API. @@ -352,6 +505,190 @@ def validate(self) -> None: self._validate_junctions() self._validate_leaf_components() + def is_pv_inverter(self, component: Component) -> bool: + """Check if the specified component is a PV inverter. + + Args: + component: component to check. + + Returns: + Whether the specified component is a PV inverter. + """ + return ( + component.category == ComponentCategory.INVERTER + and component.type == InverterType.SOLAR + ) + + def is_pv_meter(self, component: Component) -> bool: + """Check if the specified component is a PV meter. + + This is done by checking if the component has only PV inverters as its + successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is a PV meter. + """ + successors = self.successors(component.component_id) + return ( + component.category == ComponentCategory.METER + and len(successors) > 0 + and all( + self.is_pv_inverter(successor) + for successor in self.successors(component.component_id) + ) + ) + + def is_pv_chain(self, component: Component) -> bool: + """Check if the specified component is part of a PV chain. + + A component is part of a PV chain if it is either a PV inverter or a PV + meter. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of a PV chain. + """ + return self.is_pv_inverter(component) or self.is_pv_meter(component) + + def is_ev_charger(self, component: Component) -> bool: + """Check if the specified component is an EV charger. + + Args: + component: component to check. + + Returns: + Whether the specified component is an EV charger. + """ + return component.category == ComponentCategory.EV_CHARGER + + def is_ev_charger_meter(self, component: Component) -> bool: + """Check if the specified component is an EV charger meter. + + This is done by checking if the component has only EV chargers as its + successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is an EV charger meter. + """ + successors = self.successors(component.component_id) + return ( + component.category == ComponentCategory.METER + and len(successors) > 0 + and all(self.is_ev_charger(successor) for successor in successors) + ) + + def is_ev_charger_chain(self, component: Component) -> bool: + """Check if the specified component is part of an EV charger chain. + + A component is part of an EV charger chain if it is either an EV charger or an + EV charger meter. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of an EV charger chain. + """ + return self.is_ev_charger(component) or self.is_ev_charger_meter(component) + + def is_battery_inverter(self, component: Component) -> bool: + """Check if the specified component is a battery inverter. + + Args: + component: component to check. + + Returns: + Whether the specified component is a battery inverter. + """ + return ( + component.category == ComponentCategory.INVERTER + and component.type == InverterType.BATTERY + ) + + def is_battery_meter(self, component: Component) -> bool: + """Check if the specified component is a battery meter. + + This is done by checking if the component has only battery inverters as + its successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is a battery meter. + """ + successors = self.successors(component.component_id) + return ( + component.category == ComponentCategory.METER + and len(successors) > 0 + and all(self.is_battery_inverter(successor) for successor in successors) + ) + + def is_battery_chain(self, component: Component) -> bool: + """Check if the specified component is part of a battery chain. + + A component is part of a battery chain if it is either a battery inverter or a + battery meter. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of a battery chain. + """ + return self.is_battery_inverter(component) or self.is_battery_meter(component) + + def is_chp(self, component: Component) -> bool: + """Check if the specified component is a CHP. + + Args: + component: component to check. + + Returns: + Whether the specified component is a CHP. + """ + return component.category == ComponentCategory.CHP + + def is_chp_meter(self, component: Component) -> bool: + """Check if the specified component is a CHP meter. + + This is done by checking if the component has only CHPs as its + successors. + + Args: + component: component to check. + + Returns: + Whether the specified component is a CHP meter. + """ + successors = self.successors(component.component_id) + return ( + component.category == ComponentCategory.METER + and len(successors) > 0 + and all(self.is_chp(successor) for successor in successors) + ) + + def is_chp_chain(self, component: Component) -> bool: + """Check if the specified component is part of a CHP chain. + + A component is part of a CHP chain if it is either a CHP or a CHP meter. + + Args: + component: component to check. + + Returns: + Whether the specified component is part of a CHP chain. + """ + return self.is_chp(component) or self.is_chp_meter(component) + def _validate_graph(self) -> None: """Check that the underlying graph data is valid. diff --git a/src/frequenz/sdk/microgrid/component/_component.py b/src/frequenz/sdk/microgrid/component/_component.py index 59cfba505..3836d8d11 100644 --- a/src/frequenz/sdk/microgrid/component/_component.py +++ b/src/frequenz/sdk/microgrid/component/_component.py @@ -64,11 +64,11 @@ class ComponentCategory(Enum): BATTERY = microgrid_pb.ComponentCategory.COMPONENT_CATEGORY_BATTERY EV_CHARGER = microgrid_pb.ComponentCategory.COMPONENT_CATEGORY_EV_CHARGER LOAD = microgrid_pb.ComponentCategory.COMPONENT_CATEGORY_LOAD + CHP = microgrid_pb.ComponentCategory.COMPONENT_CATEGORY_CHP # types not yet supported by the API but which can be inferred # from available graph info PV_ARRAY = 1000001 - CHP = 1000002 # combined heat and power plant def _component_category_from_protobuf( diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py index fd3383b8a..b389c97eb 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py @@ -31,6 +31,8 @@ from ._formula_steps import ( Adder, Averager, + Clipper, + ConstantValue, Divider, FormulaStep, MetricFetcher, @@ -481,6 +483,49 @@ def push_metric( ) self._steps.append(fetcher) + def push_constant(self, value: float) -> None: + """Push a constant value into the engine. + + Args: + value: The constant value to push. + """ + self._steps.append(ConstantValue(value)) + + def push_clipper(self, min_value: float | None, max_value: float | None) -> None: + """Push a clipper step into the engine. + + The clip will be applied on the last value available on the evaluation stack, + before the clip step is called. + + So if an entire expression needs to be clipped, the expression should be + enclosed in parentheses, before the clip step is added. + + For example, this clips the output of the entire expression: + + ```python + builder.push_oper("(") + builder.push_metric("metric_1", receiver_1) + builder.push_oper("+") + builder.push_metric("metric_2", receiver_2) + builder.push_oper(")") + builder.push_clipper(min_value=0.0, max_value=None) + ``` + + And this clips the output of metric_2 only, and not the final result: + + ```python + builder.push_metric("metric_1", receiver_1) + builder.push_oper("+") + builder.push_metric("metric_2", receiver_2) + builder.push_clipper(min_value=0.0, max_value=None) + ``` + + Args: + min_value: The minimum value to clip to. + max_value: The maximum value to clip to. + """ + self._steps.append(Clipper(min_value, max_value)) + def push_average(self, metrics: List[Tuple[str, Receiver[Sample], bool]]) -> None: """Push an average calculator into the engine. diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py index 8bb0c550c..de20d877e 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py @@ -4,6 +4,8 @@ """Generators for formulas from component graphs.""" from ._battery_power_formula import BatteryPowerFormula +from ._chp_power_formula import CHPPowerFormula +from ._consumer_power_formula import ConsumerPowerFormula from ._ev_charger_current_formula import EVChargerCurrentFormula from ._ev_charger_power_formula import EVChargerPowerFormula from ._formula_generator import ( @@ -11,6 +13,7 @@ FormulaGenerationError, FormulaGenerator, FormulaGeneratorConfig, + FormulaType, ) from ._grid_current_formula import GridCurrentFormula from ._grid_power_formula import GridPowerFormula @@ -22,9 +25,12 @@ # "FormulaGenerator", "FormulaGeneratorConfig", + "FormulaType", # # Power Formula generators # + "CHPPowerFormula", + "ConsumerPowerFormula", "GridPowerFormula", "BatteryPowerFormula", "EVChargerPowerFormula", diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py index 741f42f62..8e9dba008 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py @@ -12,6 +12,7 @@ NON_EXISTING_COMPONENT_ID, ComponentNotFound, FormulaGenerator, + FormulaType, ) _logger = logging.getLogger(__name__) @@ -67,9 +68,19 @@ def generate( "Can't find inverters for all batteries from the component graph." ) + builder.push_oper("(") + builder.push_oper("(") for idx, comp in enumerate(battery_inverters): if idx > 0: builder.push_oper("+") builder.push_component_metric(comp.component_id, nones_are_zeros=True) + builder.push_oper(")") + if self._config.formula_type == FormulaType.PRODUCTION: + builder.push_oper("*") + builder.push_constant(-1) + builder.push_oper(")") + + if self._config.formula_type != FormulaType.PASSIVE_SIGN_CONVENTION: + builder.push_clipper(0.0, None) return builder.build() diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_chp_power_formula.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_chp_power_formula.py new file mode 100644 index 000000000..7613036fa --- /dev/null +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_chp_power_formula.py @@ -0,0 +1,105 @@ +# License: MIT +# Copyright © 2023 Frequenz Energy-as-a-Service GmbH + +"""Formula generator from component graph for CHP Power.""" + +from __future__ import annotations + +import logging +from collections import abc + +from ....microgrid import connection_manager +from ....microgrid.component import ComponentCategory, ComponentMetricId +from ..._formula_engine import FormulaEngine +from ._formula_generator import ( + NON_EXISTING_COMPONENT_ID, + FormulaGenerationError, + FormulaGenerator, + FormulaType, +) + +_logger = logging.getLogger(__name__) + + +class CHPPowerFormula(FormulaGenerator): + """Formula generator for CHP Power.""" + + def generate(self) -> FormulaEngine: + """Make a formula for the cumulative CHP power of a microgrid. + + The calculation is performed by adding the active power measurements from + dedicated meters attached to CHPs. + + Returns: + A formula engine that will calculate cumulative CHP power values. + + Raises: + FormulaGenerationError: If there's no dedicated meter attached to every CHP. + + """ + builder = self._get_builder("chp-power", ComponentMetricId.ACTIVE_POWER) + + chp_meter_ids = self._get_chp_meters() + if not chp_meter_ids: + _logger.warning("No CHPs found in the component graph.") + builder.push_component_metric( + NON_EXISTING_COMPONENT_ID, nones_are_zeros=True + ) + return builder.build() + + builder.push_oper("(") + builder.push_oper("(") + for idx, chp_meter_id in enumerate(chp_meter_ids): + if idx > 0: + builder.push_oper("+") + builder.push_component_metric(chp_meter_id, nones_are_zeros=False) + builder.push_oper(")") + if self._config.formula_type == FormulaType.PRODUCTION: + builder.push_oper("*") + builder.push_constant(-1) + builder.push_oper(")") + + if self._config.formula_type != FormulaType.PASSIVE_SIGN_CONVENTION: + builder.push_clipper(0.0, None) + + return builder.build() + + def _get_chp_meters(self) -> abc.Set[int]: + """Get the meter IDs of the CHPs from the component graph. + + Returns: + A set of meter IDs of the CHPs in the component graph. If no CHPs are + found, None is returned. + + Raises: + FormulaGenerationError: If there's no dedicated meter attached to every CHP. + """ + component_graph = connection_manager.get().component_graph + chps = list( + comp + for comp in component_graph.components() + if comp.category == ComponentCategory.CHP + ) + + chp_meters: set[int] = set() + for chp in chps: + predecessors = component_graph.predecessors(chp.component_id) + if len(predecessors) != 1: + raise FormulaGenerationError( + f"CHP {chp.component_id} has {len(predecessors)} predecessors. " + " Expected exactly one." + ) + meter = next(iter(predecessors)) + if meter.category != ComponentCategory.METER: + raise FormulaGenerationError( + f"CHP {chp.component_id} has a predecessor of category " + f"{meter.category}. Expected ComponentCategory.METER." + ) + meter_successors = component_graph.successors(meter.component_id) + if not all(successor in chps for successor in meter_successors): + raise FormulaGenerationError( + f"Meter {meter.component_id} connected to CHP {chp.component_id}" + "has non-chp successors." + ) + chp_meters.add(meter.component_id) + return chp_meters diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_consumer_power_formula.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_consumer_power_formula.py new file mode 100644 index 000000000..1308c1cb2 --- /dev/null +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_consumer_power_formula.py @@ -0,0 +1,132 @@ +# License: MIT +# Copyright © 2022 Frequenz Energy-as-a-Service GmbH + +"""Formula generator from component graph for Consumer Power.""" + +from __future__ import annotations + +from collections import abc + +from ....microgrid import connection_manager +from ....microgrid.component import Component, ComponentCategory, ComponentMetricId +from .._formula_engine import FormulaEngine +from .._resampled_formula_builder import ResampledFormulaBuilder +from ._formula_generator import ComponentNotFound, FormulaGenerator + + +class ConsumerPowerFormula(FormulaGenerator): + """Formula generator from component graph for calculating the Consumer Power. + + The consumer power is calculated by summing up the power of all components that + are not part of a battery, CHP, PV or EV charger chain. + """ + + def generate(self) -> FormulaEngine: + """Generate formula for calculating consumer power from the component graph. + + Returns: + A formula engine that will calculate the consumer power. + + Raises: + ComponentNotFound: If the component graph does not contain a consumer power + component. + RuntimeError: If the grid component has a single successor that is not a + meter. + """ + builder = self._get_builder("consumer-power", ComponentMetricId.ACTIVE_POWER) + component_graph = connection_manager.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("Grid component not found in the component graph.") + + grid_successors = component_graph.successors(grid_component.component_id) + if not grid_successors: + raise ComponentNotFound("No components found in the component graph.") + + if len(grid_successors) == 1: + grid_meter = next(iter(grid_successors)) + if grid_meter.category != ComponentCategory.METER: + raise RuntimeError( + "Only grid successor in the component graph is not a meter." + ) + return self._gen_with_grid_meter(builder, grid_meter) + return self._gen_without_grid_meter(builder, grid_successors) + + def _gen_with_grid_meter( + self, + builder: ResampledFormulaBuilder, + grid_meter: Component, + ) -> FormulaEngine: + """Generate formula for calculating consumer power with grid meter. + + Args: + builder: The formula engine builder. + grid_meter: The grid meter component. + + Returns: + A formula engine that will calculate the consumer power. + """ + component_graph = connection_manager.get().component_graph + successors = component_graph.successors(grid_meter.component_id) + + builder.push_component_metric(grid_meter.component_id, nones_are_zeros=False) + + for successor in successors: + # If the component graph supports additional types of grid successors in the + # future, additional checks need to be added here. + if ( + component_graph.is_battery_chain(successor) + or component_graph.is_chp_chain(successor) + or component_graph.is_pv_chain(successor) + or component_graph.is_ev_charger_chain(successor) + ): + builder.push_oper("-") + nones_are_zeros = True + if successor.category == ComponentCategory.METER: + nones_are_zeros = False + builder.push_component_metric( + successor.component_id, nones_are_zeros=nones_are_zeros + ) + + return builder.build() + + def _gen_without_grid_meter( + self, + builder: ResampledFormulaBuilder, + grid_successors: abc.Iterable[Component], + ) -> FormulaEngine: + """Generate formula for calculating consumer power without a grid meter. + + Args: + builder: The formula engine builder. + grid_successors: The grid successors. + + Returns: + A formula engine that will calculate the consumer power. + """ + component_graph = connection_manager.get().component_graph + is_first = True + for successor in grid_successors: + # If the component graph supports additional types of grid successors in the + # future, additional checks need to be added here. + if ( + component_graph.is_battery_chain(successor) + or component_graph.is_chp_chain(successor) + or component_graph.is_pv_chain(successor) + or component_graph.is_ev_charger_chain(successor) + ): + continue + if not is_first: + builder.push_oper("+") + is_first = False + builder.push_component_metric(successor.component_id, nones_are_zeros=False) + + return builder.build() diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_power_formula.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_power_formula.py index 79ec78799..fe85df57f 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_power_formula.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_power_formula.py @@ -7,7 +7,7 @@ from ....microgrid.component import ComponentMetricId from .._formula_engine import FormulaEngine -from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator +from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator, FormulaType _logger = logging.getLogger(__name__) @@ -38,10 +38,19 @@ def generate(self) -> FormulaEngine: ) return builder.build() + builder.push_oper("(") + builder.push_oper("(") for idx, component_id in enumerate(component_ids): if idx > 0: builder.push_oper("+") - builder.push_component_metric(component_id, nones_are_zeros=True) + builder.push_oper(")") + if self._config.formula_type == FormulaType.PRODUCTION: + builder.push_oper("*") + builder.push_constant(-1) + builder.push_oper(")") + + if self._config.formula_type != FormulaType.PASSIVE_SIGN_CONVENTION: + builder.push_clipper(0.0, None) return builder.build() diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py index 26b6af7f9..5d53c6fbe 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py @@ -9,6 +9,7 @@ from abc import ABC, abstractmethod from collections import abc from dataclasses import dataclass +from enum import Enum from frequenz.channels import Sender @@ -29,11 +30,31 @@ class ComponentNotFound(FormulaGenerationError): NON_EXISTING_COMPONENT_ID = sys.maxsize +class FormulaType(Enum): + """Enum representing type of formula outputs.""" + + PASSIVE_SIGN_CONVENTION = 1 + """Formula output will be signed values, following the passive sign convention, with + consumption from the grid being positive and production to the grid being negative. + """ + + PRODUCTION = 2 + """Formula output will be unsigned values representing production to the grid. When + power is being consumed from the grid instead, this formula will output zero. + """ + + CONSUMPTION = 3 + """Formula output will be unsigned values representing consumption from the grid. + When power is being produced to the grid instead, this formula will output zero. + """ + + @dataclass(frozen=True) class FormulaGeneratorConfig: """Config for formula generators.""" component_ids: abc.Set[int] | None = None + formula_type: FormulaType = FormulaType.PASSIVE_SIGN_CONVENTION class FormulaGenerator(ABC): diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_grid_power_formula.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_grid_power_formula.py index 7ca6ea398..f4657fef9 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_grid_power_formula.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_grid_power_formula.py @@ -6,7 +6,7 @@ from ....microgrid import connection_manager from ....microgrid.component import ComponentCategory, ComponentMetricId from .._formula_engine import FormulaEngine -from ._formula_generator import ComponentNotFound, FormulaGenerator +from ._formula_generator import ComponentNotFound, FormulaGenerator, FormulaType class GridPowerFormula(FormulaGenerator): @@ -42,7 +42,18 @@ def generate( 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. + # directly connected to the grid. If the requested formula type is + # `PASSIVE_SIGN_CONVENTION`, there is nothing more to do. If the requested + # formula type is `PRODUCTION`, the formula output is negated, then clipped to + # 0. If the requested formula type is `CONSUMPTION`, the formula output is + # already positive, so it is just clipped to 0. + # + # So the formulas would look like: + # - `PASSIVE_SIGN_CONVENTION`: `(grid-successor-1 + grid-successor-2 + ...)` + # - `PRODUCTION`: `max(0, -(grid-successor-1 + grid-successor-2 + ...))` + # - `CONSUMPTION`: `max(0, (grid-successor-1 + grid-successor-2 + ...))` + builder.push_oper("(") + builder.push_oper("(") for idx, comp in enumerate(grid_successors): if idx > 0: builder.push_oper("+") @@ -66,5 +77,14 @@ def generate( builder.push_component_metric( comp.component_id, nones_are_zeros=nones_are_zeros ) + builder.push_oper(")") + + if self._config.formula_type == FormulaType.PRODUCTION: + builder.push_oper("*") + builder.push_constant(-1) + builder.push_oper(")") + + if self._config.formula_type != FormulaType.PASSIVE_SIGN_CONVENTION: + builder.push_clipper(0.0, None) return builder.build() diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_pv_power_formula.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_pv_power_formula.py index 981b24c12..5bf03ba0d 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_pv_power_formula.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_pv_power_formula.py @@ -8,7 +8,7 @@ from ....microgrid import connection_manager from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType from .._formula_engine import FormulaEngine -from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator +from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator, FormulaType _logger = logging.getLogger(__name__) @@ -49,10 +49,20 @@ def generate(self) -> FormulaEngine: ) return builder.build() + builder.push_oper("(") + builder.push_oper("(") for idx, comp in enumerate(pv_inverters): if idx > 0: builder.push_oper("+") builder.push_component_metric(comp.component_id, nones_are_zeros=True) + builder.push_oper(")") + if self._config.formula_type == FormulaType.PRODUCTION: + builder.push_oper("*") + builder.push_constant(-1) + builder.push_oper(")") + + if self._config.formula_type != FormulaType.PASSIVE_SIGN_CONVENTION: + builder.push_clipper(0.0, None) return builder.build() diff --git a/src/frequenz/sdk/timeseries/_formula_engine/_formula_steps.py b/src/frequenz/sdk/timeseries/_formula_engine/_formula_steps.py index 68ec9f028..d220a02de 100644 --- a/src/frequenz/sdk/timeseries/_formula_engine/_formula_steps.py +++ b/src/frequenz/sdk/timeseries/_formula_engine/_formula_steps.py @@ -197,6 +197,69 @@ def apply(self, eval_stack: List[float]) -> None: eval_stack.append(avg) +class ConstantValue(FormulaStep): + """A formula step for inserting a constant value.""" + + def __init__(self, value: float) -> None: + """Create a `ConstantValue` instance. + + Args: + value: The constant value. + """ + self._value = value + + def __repr__(self) -> str: + """Return a string representation of the step. + + Returns: + A string representation of the step. + """ + return str(self._value) + + def apply(self, eval_stack: List[float]) -> None: + """Push the constant value to the eval_stack. + + Args: + eval_stack: An evaluation stack, to append the constant value to. + """ + eval_stack.append(self._value) + + +class Clipper(FormulaStep): + """A formula step for clipping a value between a minimum and maximum.""" + + def __init__(self, min_val: float | None, max_val: float | None) -> None: + """Create a `Clipper` instance. + + Args: + min_val: The minimum value. + max_val: The maximum value. + """ + self._min_val = min_val + self._max_val = max_val + + def __repr__(self) -> str: + """Return a string representation of the step. + + Returns: + A string representation of the step. + """ + return f"clip({self._min_val}, {self._max_val})" + + def apply(self, eval_stack: List[float]) -> None: + """Clip the value at the top of the eval_stack. + + Args: + eval_stack: An evaluation stack, to apply the formula step on. + """ + val = eval_stack.pop() + if self._min_val is not None: + val = max(val, self._min_val) + if self._max_val is not None: + val = min(val, self._max_val) + eval_stack.append(val) + + class MetricFetcher(FormulaStep): """A formula step for fetching a value from a metric Receiver.""" diff --git a/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py b/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py index e7d1f5e55..b4b83ff20 100644 --- a/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py +++ b/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py @@ -23,6 +23,7 @@ from .._formula_engine._formula_generators import ( BatteryPowerFormula, FormulaGeneratorConfig, + FormulaType, ) from ._methods import AggregateMethod, SendOnUpdate from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator @@ -105,6 +106,8 @@ def battery_ids(self) -> Set[int]: def power(self) -> FormulaEngine: """Fetch the total power of the batteries in the pool. + This formula produces values that are in the Passive Sign Convention (PSC). + If a formula engine to calculate this metric is not already running, it will be started. @@ -115,11 +118,70 @@ def power(self) -> FormulaEngine: A FormulaEngine that will calculate and stream the total power of all batteries in the pool. """ - return self._formula_pool.from_generator( + engine = self._formula_pool.from_generator( "battery_pool_power", BatteryPowerFormula, - FormulaGeneratorConfig(component_ids=self._batteries), - ) # type: ignore[return-value] + FormulaGeneratorConfig( + component_ids=self._batteries, + formula_type=FormulaType.PASSIVE_SIGN_CONVENTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def production_power(self) -> FormulaEngine: + """Fetch the total production power of the batteries in the pool. + + This formula produces positive values when producing power and 0 otherwise. + + If a formula engine to calculate this metric is not already running, it will be + started. + + A receiver from the formula engine can be obtained by calling the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream the total production power of + all batteries in the pool. + """ + engine = self._formula_pool.from_generator( + "battery_pool_production_power", + BatteryPowerFormula, + FormulaGeneratorConfig( + component_ids=self._batteries, + formula_type=FormulaType.PRODUCTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def consumption_power(self) -> FormulaEngine: + """Fetch the total consumption power of the batteries in the pool. + + This formula produces positive values when consuming power and 0 otherwise. + + If a formula engine to calculate this metric is not already running, it will be + started. + + A receiver from the formula engine can be obtained by calling the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream the total consumption power of + all batteries in the pool. + """ + engine = self._formula_pool.from_generator( + "battery_pool_consumption_power", + BatteryPowerFormula, + FormulaGeneratorConfig( + component_ids=self._batteries, + formula_type=FormulaType.CONSUMPTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine async def soc( self, maxsize: int | None = RECEIVER_MAX_SIZE diff --git a/src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py b/src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py index 35fdbac8e..50f29eaa2 100644 --- a/src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py +++ b/src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py @@ -24,6 +24,7 @@ EVChargerCurrentFormula, EVChargerPowerFormula, FormulaGeneratorConfig, + FormulaType, ) from ._set_current_bounds import BoundsSetter, ComponentCurrentLimit from ._state_tracker import EVChargerState, StateTracker @@ -108,6 +109,8 @@ def component_ids(self) -> abc.Set[int]: def current(self) -> FormulaEngine3Phase: """Fetch the total current for the EV Chargers in the pool. + This formula produces values that are in the Passive Sign Convention (PSC). + If a formula engine to calculate EV Charger current is not already running, it will be started. @@ -118,16 +121,20 @@ def current(self) -> FormulaEngine3Phase: A FormulaEngine that will calculate and stream the total current of all EV Chargers. """ - return self._formula_pool.from_generator( + engine = self._formula_pool.from_generator( "ev_charger_total_current", EVChargerCurrentFormula, FormulaGeneratorConfig(component_ids=self._component_ids), - ) # type: ignore[return-value] + ) + assert isinstance(engine, FormulaEngine3Phase) + return engine @property def power(self) -> FormulaEngine: """Fetch the total power for the EV Chargers in the pool. + This formula produces values that are in the Passive Sign Convention (PSC). + If a formula engine to calculate EV Charger power is not already running, it will be started. @@ -138,11 +145,70 @@ def power(self) -> FormulaEngine: A FormulaEngine that will calculate and stream the total power of all EV Chargers. """ - return self._formula_pool.from_generator( - "ev_charger_total_power", + engine = self._formula_pool.from_generator( + "ev_charger_power", EVChargerPowerFormula, - FormulaGeneratorConfig(component_ids=self._component_ids), - ) # type: ignore[return-value] + FormulaGeneratorConfig( + component_ids=self._component_ids, + formula_type=FormulaType.PASSIVE_SIGN_CONVENTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def production_power(self) -> FormulaEngine: + """Fetch the total power produced by the EV Chargers in the pool. + + This formula produces positive values when producing power and 0 otherwise. + + If a formula engine to calculate EV Charger power is not already running, it + will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream the production power of all + EV Chargers. + """ + engine = self._formula_pool.from_generator( + "ev_charger_production_power", + EVChargerPowerFormula, + FormulaGeneratorConfig( + component_ids=self._component_ids, + formula_type=FormulaType.PRODUCTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def consumption_power(self) -> FormulaEngine: + """Fetch the total power consumed by the EV Chargers in the pool. + + This formula produces positive values when consuming power and 0 otherwise. + + If a formula engine to calculate EV Charger power is not already running, it + will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream the consumption power of all + EV Chargers. + """ + engine = self._formula_pool.from_generator( + "ev_charger_consumption_power", + EVChargerPowerFormula, + FormulaGeneratorConfig( + component_ids=self._component_ids, + formula_type=FormulaType.CONSUMPTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine def component_data(self, component_id: int) -> Receiver[EVChargerData]: """Stream 3-phase current values and state of an EV Charger. diff --git a/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py b/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py index cecb88087..be36c76ee 100644 --- a/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py +++ b/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py @@ -5,7 +5,6 @@ from __future__ import annotations -import logging import uuid from frequenz.channels import Sender @@ -14,13 +13,15 @@ from ...microgrid.component import ComponentMetricId from .._formula_engine import FormulaEngine, FormulaEngine3Phase, FormulaEnginePool from .._formula_engine._formula_generators import ( + CHPPowerFormula, + ConsumerPowerFormula, + FormulaGeneratorConfig, + FormulaType, GridCurrentFormula, GridPowerFormula, PVPowerFormula, ) -_logger = logging.getLogger(__name__) - class LogicalMeter: """A logical meter for calculating high level metrics in a microgrid. @@ -138,6 +139,8 @@ def start_formula( def grid_power(self) -> FormulaEngine: """Fetch the grid power for the microgrid. + This formula produces values that are in the Passive Sign Convention (PSC). + If a formula engine to calculate grid power is not already running, it will be started. @@ -147,15 +150,65 @@ def grid_power(self) -> FormulaEngine: Returns: A FormulaEngine that will calculate and stream grid power. """ - return self._formula_pool.from_generator( + engine = self._formula_pool.from_generator( "grid_power", GridPowerFormula, - ) # type: ignore[return-value] + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def grid_consumption_power(self) -> FormulaEngine: + """Fetch the grid consumption power for the microgrid. + + This formula produces positive values when consuming power and 0 otherwise. + + If a formula engine to calculate grid consumption power is not already running, + it will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream grid consumption power. + """ + engine = self._formula_pool.from_generator( + "grid_consumption_power", + GridPowerFormula, + FormulaGeneratorConfig(formula_type=FormulaType.CONSUMPTION), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def grid_production_power(self) -> FormulaEngine: + """Fetch the grid production power for the microgrid. + + This formula produces positive values when producing power and 0 otherwise. + + If a formula engine to calculate grid production power is not already running, + it will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream grid production power. + """ + engine = self._formula_pool.from_generator( + "grid_production_power", + GridPowerFormula, + FormulaGeneratorConfig(formula_type=FormulaType.PRODUCTION), + ) + assert isinstance(engine, FormulaEngine) + return engine @property def grid_current(self) -> FormulaEngine3Phase: """Fetch the grid power for the microgrid. + This formula produces values that are in the Passive Sign Convention (PSC). + If a formula engine to calculate grid current is not already running, it will be started. @@ -165,15 +218,67 @@ def grid_current(self) -> FormulaEngine3Phase: Returns: A FormulaEngine that will calculate and stream grid current. """ - return self._formula_pool.from_generator( + engine = self._formula_pool.from_generator( "grid_current", GridCurrentFormula, - ) # type: ignore[return-value] + ) + assert isinstance(engine, FormulaEngine3Phase) + return engine + + @property + def consumer_power(self) -> FormulaEngine: + """Fetch the consumer power for the microgrid. + + Under normal circumstances this is expected to correspond to the gross + consumption of the site excluding active parts and battery. + + This formula produces values that are in the Passive Sign Convention (PSC). + + If a formula engine to calculate consumer power is not already running, it will + be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream consumer power. + """ + engine = self._formula_pool.from_generator( + "consumer_power", + ConsumerPowerFormula, + ) + assert isinstance(engine, FormulaEngine) + return engine @property def pv_power(self) -> FormulaEngine: + """Fetch the PV power in the microgrid. + + This formula produces values that are in the Passive Sign Convention (PSC). + + If a formula engine to calculate PV power is not already running, it will be + started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream PV total power. + """ + engine = self._formula_pool.from_generator( + "pv_power", + PVPowerFormula, + FormulaGeneratorConfig(formula_type=FormulaType.PASSIVE_SIGN_CONVENTION), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def pv_production_power(self) -> FormulaEngine: """Fetch the PV power production in the microgrid. + This formula produces positive values when producing power and 0 otherwise. + If a formula engine to calculate PV power production is not already running, it will be started. @@ -183,7 +288,106 @@ def pv_power(self) -> FormulaEngine: Returns: A FormulaEngine that will calculate and stream PV power production. """ - return self._formula_pool.from_generator( - "pv_power", + engine = self._formula_pool.from_generator( + "pv_production_power", PVPowerFormula, - ) # type: ignore[return-value] + FormulaGeneratorConfig(formula_type=FormulaType.PRODUCTION), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def pv_consumption_power(self) -> FormulaEngine: + """Fetch the PV power consumption in the microgrid. + + This formula produces positive values when consuming power and 0 otherwise. + + If a formula engine to calculate PV power consumption is not already running, it + will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream PV power consumption. + """ + engine = self._formula_pool.from_generator( + "pv_consumption_power", + PVPowerFormula, + FormulaGeneratorConfig(formula_type=FormulaType.CONSUMPTION), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def chp_power(self) -> FormulaEngine: + """Fetch the CHP power production in the microgrid. + + This formula produces values that are in the Passive Sign Convention (PSC). + + If a formula engine to calculate CHP power production is not already running, it + will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream CHP power production. + """ + engine = self._formula_pool.from_generator( + "chp_power", + CHPPowerFormula, + FormulaGeneratorConfig(formula_type=FormulaType.PASSIVE_SIGN_CONVENTION), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def chp_production_power(self) -> FormulaEngine: + """Fetch the CHP power production in the microgrid. + + This formula produces positive values when producing power and 0 otherwise. + + If a formula engine to calculate CHP power production is not already running, it + will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream CHP power production. + """ + engine = self._formula_pool.from_generator( + "chp_production_power", + CHPPowerFormula, + FormulaGeneratorConfig( + formula_type=FormulaType.PRODUCTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine + + @property + def chp_consumption_power(self) -> FormulaEngine: + """Fetch the CHP power consumption in the microgrid. + + This formula produces positive values when consuming power and 0 otherwise. + + If a formula engine to calculate CHP power consumption is not already running, + it will be started. + + A receiver from the formula engine can be created using the `new_receiver` + method. + + Returns: + A FormulaEngine that will calculate and stream CHP power consumption. + """ + engine = self._formula_pool.from_generator( + "chp_consumption_power", + CHPPowerFormula, + FormulaGeneratorConfig( + formula_type=FormulaType.CONSUMPTION, + ), + ) + assert isinstance(engine, FormulaEngine) + return engine diff --git a/tests/timeseries/_battery_pool/test_battery_pool.py b/tests/timeseries/_battery_pool/test_battery_pool.py index d7d138152..c3ace4119 100644 --- a/tests/timeseries/_battery_pool/test_battery_pool.py +++ b/tests/timeseries/_battery_pool/test_battery_pool.py @@ -13,7 +13,7 @@ import async_solipsism import pytest -from frequenz.channels import Receiver, Sender +from frequenz.channels import Broadcast, Receiver, Sender from pytest_mock import MockerFixture from frequenz.sdk import microgrid @@ -23,7 +23,8 @@ ) from frequenz.sdk.actor import ResamplerConfig from frequenz.sdk.actor.power_distributing import BatteryStatus -from frequenz.sdk.microgrid.component import ComponentCategory +from frequenz.sdk.microgrid.component import ComponentCategory, ComponentMetricId +from frequenz.sdk.timeseries import Sample from frequenz.sdk.timeseries.battery_pool import ( BatteryPool, Bound, @@ -35,6 +36,7 @@ battery_inverter_mapping, ) +from ...timeseries.mock_microgrid import MockMicrogrid from ...utils.component_data_streamer import MockComponentDataStreamer from ...utils.component_data_wrapper import BatteryDataWrapper, InverterDataWrapper from ...utils.component_graph_utils import ( @@ -439,6 +441,64 @@ async def run_test_battery_status_channel( # pylint: disable=too-many-arguments compare_messages(msg, all_pool_result, waiting_time_sec) +async def test_battery_pool_power(mocker: MockerFixture) -> None: + """Test `BatteryPool.{,production,consumption}_power` methods.""" + mockgrid = MockMicrogrid(grid_side_meter=True) + mockgrid.add_batteries(2) + await mockgrid.start(mocker) + + channels: dict[int, Broadcast[Sample]] = { + meter_id: Broadcast(f"#{meter_id}") + for meter_id in [*mockgrid.meter_ids, *mockgrid.battery_inverter_ids] + } + senders: list[Sender[Sample]] = [ + channels[component_id].new_sender() + for component_id in mockgrid.battery_inverter_ids + ] + + async def send_resampled_data( + now: datetime, + meter_data: list[float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) + + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() + + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, + ) + + battery_pool = microgrid.battery_pool() + power_receiver = battery_pool.power.new_receiver() + consumption_receiver = battery_pool.consumption_power.new_receiver() + production_receiver = battery_pool.production_power.new_receiver() + + now = datetime.now(tz=timezone.utc) + await send_resampled_data(now, [2.0, 3.0]) + assert (await power_receiver.receive()).value == 5.0 + assert (await consumption_receiver.receive()).value == 5.0 + assert (await production_receiver.receive()).value == 0.0 + + await send_resampled_data(now + timedelta(seconds=1), [-2.0, -5.0]) + assert (await power_receiver.receive()).value == -7.0 + assert (await consumption_receiver.receive()).value == 0.0 + assert (await production_receiver.receive()).value == 7.0 + + await send_resampled_data(now + timedelta(seconds=3), [2.0, -5.0]) + assert (await power_receiver.receive()).value == -3.0 + assert (await consumption_receiver.receive()).value == 0.0 + assert (await production_receiver.receive()).value == 3.0 + + await mockgrid.cleanup() + + async def run_capacity_test(setup_args: SetupArgs) -> None: """Test if capacity metric is working as expected. diff --git a/tests/timeseries/mock_microgrid.py b/tests/timeseries/mock_microgrid.py index a3d4c3ff2..4aae39d77 100644 --- a/tests/timeseries/mock_microgrid.py +++ b/tests/timeseries/mock_microgrid.py @@ -40,6 +40,8 @@ class MockMicrogrid: # pylint: disable=too-many-instance-attributes grid_id = 1 main_meter_id = 4 + + chp_id_suffix = 5 evc_id_suffix = 6 meter_id_suffix = 7 inverter_id_suffix = 8 @@ -73,6 +75,7 @@ def __init__( if self._grid_side_meter: self._connect_to = self.main_meter_id + self.chp_ids: list[int] = [] self.battery_inverter_ids: list[int] = [] self.pv_inverter_ids: list[int] = [] self.battery_ids: list[int] = [] @@ -194,6 +197,37 @@ def _start_ev_charger_streaming(self, evc_id: int) -> None: ), ) + def add_chps(self, count: int) -> None: + """Add CHPs with connected meters to the mock microgrid. + + Args: + count: number of CHPs to add. + """ + for _ in range(count): + meter_id = self._id_increment * 10 + self.meter_id_suffix + chp_id = self._id_increment * 10 + self.chp_id_suffix + self._id_increment += 1 + + self.meter_ids.append(meter_id) + self.chp_ids.append(chp_id) + + self._components.add( + Component( + meter_id, + ComponentCategory.METER, + ) + ) + self._components.add( + Component( + chp_id, + ComponentCategory.CHP, + ) + ) + + self._start_meter_streaming(meter_id) + self._connections.add(Connection(self._connect_to, meter_id)) + self._connections.add(Connection(meter_id, chp_id)) + def add_batteries(self, count: int) -> None: """Add batteries with connected inverters and meters to the microgrid. diff --git a/tests/timeseries/test_ev_charger_pool.py b/tests/timeseries/test_ev_charger_pool.py index 4a539a3b9..f2bd744ec 100644 --- a/tests/timeseries/test_ev_charger_pool.py +++ b/tests/timeseries/test_ev_charger_pool.py @@ -6,11 +6,10 @@ from __future__ import annotations import asyncio -from datetime import datetime -from math import isclose +from datetime import datetime, timezone from typing import Any -from frequenz.channels import Broadcast, Receiver +from frequenz.channels import Broadcast, Receiver, Sender from pytest_mock import MockerFixture from frequenz.sdk import microgrid @@ -24,10 +23,6 @@ EVChargerState, StateTracker, ) -from tests.timeseries._formula_engine.utils import ( - get_resampled_stream, - synchronize_receivers, -) from tests.timeseries.mock_microgrid import MockMicrogrid @@ -83,40 +78,55 @@ async def test_ev_power( # pylint: disable=too-many-locals self, mocker: MockerFixture, ) -> None: - """Test the battery power and pv power formulas.""" + """Test the ev power formula.""" mockgrid = MockMicrogrid(grid_side_meter=False) - mockgrid.add_ev_chargers(5) + mockgrid.add_ev_chargers(3) await mockgrid.start(mocker) - logical_meter = microgrid.logical_meter() + channels: dict[int, Broadcast[Sample]] = { + meter_id: Broadcast(f"#{meter_id}") + for meter_id in [*mockgrid.meter_ids, *mockgrid.evc_ids] + } + senders: list[Sender[Sample]] = [ + channels[component_id].new_sender() for component_id in mockgrid.evc_ids + ] + + async def send_resampled_data( + now: datetime, + meter_data: list[float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) - ev_pool = microgrid.ev_charger_pool() + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() - main_meter_recv = get_resampled_stream( - logical_meter._namespace, # pylint: disable=protected-access - mockgrid.main_meter_id, - ComponentMetricId.ACTIVE_POWER, + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, ) - grid_power_recv = logical_meter.grid_power.new_receiver() - ev_power_recv = ev_pool.power.new_receiver() - - await synchronize_receivers([grid_power_recv, main_meter_recv, ev_power_recv]) - ev_results = [] - for _ in range(10): - grid_pow = await grid_power_recv.receive() - ev_pow = await ev_power_recv.receive() - main_pow = await main_meter_recv.receive() + ev_pool = microgrid.ev_charger_pool() + power_receiver = ev_pool.power.new_receiver() + production_receiver = ev_pool.production_power.new_receiver() + consumption_receiver = ev_pool.consumption_power.new_receiver() - assert grid_pow is not None and grid_pow.value is not None - assert ev_pow is not None and ev_pow.value is not None - assert main_pow is not None and main_pow.value is not None - assert isclose(grid_pow.value, ev_pow.value + main_pow.value) + now = datetime.now(tz=timezone.utc) + await send_resampled_data(now, [2.0, 4.0, 10.0]) + assert (await power_receiver.receive()).value == 16.0 + assert (await production_receiver.receive()).value == 0.0 + assert (await consumption_receiver.receive()).value == 16.0 - ev_results.append(ev_pow.value) + await send_resampled_data(now, [2.0, 4.0, -10.0]) + assert (await power_receiver.receive()).value == -4.0 + assert (await production_receiver.receive()).value == 4.0 + assert (await consumption_receiver.receive()).value == 0.0 await mockgrid.cleanup() - assert len(ev_results) == 10 async def test_ev_component_data(self, mocker: MockerFixture) -> None: """Test the component_data method of EVChargerPool.""" diff --git a/tests/timeseries/test_formula_engine.py b/tests/timeseries/test_formula_engine.py index 07cea7be5..0a134def6 100644 --- a/tests/timeseries/test_formula_engine.py +++ b/tests/timeseries/test_formula_engine.py @@ -603,3 +603,116 @@ async def test_nones_are_skipped(self) -> None: ([None, None, None], 0.0), ], ) + + +class TestConstantValue: + """Tests for the constant value step.""" + + async def test_constant_value(self) -> None: + """Test using constant values in formulas.""" + + channel_1 = Broadcast[Sample]("channel_1") + channel_2 = Broadcast[Sample]("channel_2") + + sender_1 = channel_1.new_sender() + sender_2 = channel_2.new_sender() + + builder = FormulaBuilder("test_constant_value") + builder.push_metric("channel_1", channel_1.new_receiver(), False) + builder.push_oper("+") + builder.push_constant(2.0) + builder.push_oper("*") + builder.push_metric("channel_2", channel_2.new_receiver(), False) + + engine = builder.build() + + results_rx = engine.new_receiver() + + now = datetime.now() + await sender_1.send(Sample(now, 10.0)) + await sender_2.send(Sample(now, 15.0)) + assert (await results_rx.receive()).value == 40.0 + + await sender_1.send(Sample(now, -10.0)) + await sender_2.send(Sample(now, 15.0)) + assert (await results_rx.receive()).value == 20.0 + + builder = FormulaBuilder("test_constant_value") + builder.push_oper("(") + builder.push_metric("channel_1", channel_1.new_receiver(), False) + builder.push_oper("+") + builder.push_constant(2.0) + builder.push_oper(")") + builder.push_oper("*") + builder.push_metric("channel_2", channel_2.new_receiver(), False) + + engine = builder.build() + + results_rx = engine.new_receiver() + + now = datetime.now() + await sender_1.send(Sample(now, 10.0)) + await sender_2.send(Sample(now, 15.0)) + assert (await results_rx.receive()).value == 180.0 + + await sender_1.send(Sample(now, -10.0)) + await sender_2.send(Sample(now, 15.0)) + assert (await results_rx.receive()).value == -120.0 + + +class TestClipper: + """Tests for the clipper step.""" + + async def test_clipper(self) -> None: + """Test the usage of clipper in formulas.""" + channel_1 = Broadcast[Sample]("channel_1") + channel_2 = Broadcast[Sample]("channel_2") + + sender_1 = channel_1.new_sender() + sender_2 = channel_2.new_sender() + + builder = FormulaBuilder("test_clipper") + builder.push_metric("channel_1", channel_1.new_receiver(), False) + builder.push_oper("+") + builder.push_metric("channel_2", channel_2.new_receiver(), False) + builder.push_clipper(0.0, 100.0) + engine = builder.build() + + results_rx = engine.new_receiver() + + now = datetime.now() + await sender_1.send(Sample(now, 10.0)) + await sender_2.send(Sample(now, 150.0)) + assert (await results_rx.receive()).value == 110.0 + + await sender_1.send(Sample(now, 200.0)) + await sender_2.send(Sample(now, -10.0)) + assert (await results_rx.receive()).value == 200.0 + + await sender_1.send(Sample(now, 200.0)) + await sender_2.send(Sample(now, 10.0)) + assert (await results_rx.receive()).value == 210.0 + + builder = FormulaBuilder("test_clipper") + builder.push_oper("(") + builder.push_metric("channel_1", channel_1.new_receiver(), False) + builder.push_oper("+") + builder.push_metric("channel_2", channel_2.new_receiver(), False) + builder.push_oper(")") + builder.push_clipper(0.0, 100.0) + engine = builder.build() + + results_rx = engine.new_receiver() + + now = datetime.now() + await sender_1.send(Sample(now, 10.0)) + await sender_2.send(Sample(now, 150.0)) + assert (await results_rx.receive()).value == 100.0 + + await sender_1.send(Sample(now, 200.0)) + await sender_2.send(Sample(now, -10.0)) + assert (await results_rx.receive()).value == 100.0 + + await sender_1.send(Sample(now, 25.0)) + await sender_2.send(Sample(now, -10.0)) + assert (await results_rx.receive()).value == 15.0 diff --git a/tests/timeseries/test_logical_meter.py b/tests/timeseries/test_logical_meter.py index 02ef5d99c..00b32ac14 100644 --- a/tests/timeseries/test_logical_meter.py +++ b/tests/timeseries/test_logical_meter.py @@ -5,10 +5,15 @@ from __future__ import annotations +from datetime import datetime, timezone +from typing import Any + +from frequenz.channels import Broadcast, Receiver, Sender from pytest_mock import MockerFixture from frequenz.sdk import microgrid from frequenz.sdk.microgrid.component import ComponentMetricId +from frequenz.sdk.timeseries import Sample from ._formula_engine.utils import ( equal_float_lists, @@ -96,73 +101,241 @@ async def test_grid_power_2( assert len(results) == 10 assert equal_float_lists(results, meter_sums) - async def test_battery_and_pv_power( # pylint: disable=too-many-locals + async def test_grid_production_consumption_power( self, mocker: MockerFixture, ) -> None: - """Test the battery power and pv power formulas.""" + """Test the grid production and consumption power formulas.""" mockgrid = MockMicrogrid(grid_side_meter=False) - mockgrid.add_batteries(3) - mockgrid.add_solar_inverters(2) + mockgrid.add_batteries(2) + mockgrid.add_solar_inverters(1) await mockgrid.start(mocker) - battery_pool = microgrid.battery_pool() + + channels: dict[int, Broadcast[Sample]] = { + meter_id: Broadcast(f"#{meter_id}") for meter_id in mockgrid.meter_ids + } + senders: list[Sender[Sample]] = [ + channels[meter_id].new_sender() for meter_id in mockgrid.meter_ids + ] + + async def send_resampled_data( + now: datetime, + meter_data: tuple[float | None, float | None, float | None, float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) + + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() + + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, + ) + logical_meter = microgrid.logical_meter() + grid_recv = logical_meter.grid_power.new_receiver() + grid_production_recv = logical_meter.grid_production_power.new_receiver() + grid_consumption_recv = logical_meter.grid_consumption_power.new_receiver() + + now = datetime.now() + await send_resampled_data(now, (1.0, 2.0, 3.0, 4.0)) + assert (await grid_recv.receive()).value == 10.0 + assert (await grid_production_recv.receive()).value == 0.0 + assert (await grid_consumption_recv.receive()).value == 10.0 + + await send_resampled_data(now, (1.0, 2.0, -3.0, -4.0)) + assert (await grid_recv.receive()).value == -4.0 + assert (await grid_production_recv.receive()).value == 4.0 + assert (await grid_consumption_recv.receive()).value == 0.0 + + async def test_chp_power(self, mocker: MockerFixture) -> None: + """Test the chp power formula.""" + mockgrid = MockMicrogrid(grid_side_meter=False) + mockgrid.add_chps(1) + mockgrid.add_batteries(2) + await mockgrid.start(mocker) - battery_power_recv = battery_pool.power.new_receiver() - pv_power_recv = logical_meter.pv_power.new_receiver() + assert len(mockgrid.meter_ids) == 4 - bat_inv_receivers = [ - get_resampled_stream( - battery_pool._namespace, # pylint: disable=protected-access - meter_id, - ComponentMetricId.ACTIVE_POWER, - ) - for meter_id in mockgrid.battery_inverter_ids + channels: dict[int, Broadcast[Sample]] = { + meter_id: Broadcast(f"#{meter_id}") for meter_id in [*mockgrid.meter_ids] + } + senders: list[Sender[Sample]] = [ + channels[component_id].new_sender() for component_id in mockgrid.meter_ids ] - pv_inv_receivers = [ - get_resampled_stream( - logical_meter._namespace, # pylint: disable=protected-access - meter_id, - ComponentMetricId.ACTIVE_POWER, - ) - for meter_id in mockgrid.pv_inverter_ids + async def send_resampled_data( + now: datetime, + meter_data: list[float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) + + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() + + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, + ) + + logical_meter = microgrid.logical_meter() + chp_power_receiver = logical_meter.chp_power.new_receiver() + chp_production_power_receiver = ( + logical_meter.chp_production_power.new_receiver() + ) + chp_consumption_power_receiver = ( + logical_meter.chp_consumption_power.new_receiver() + ) + + now = datetime.now(tz=timezone.utc) + await send_resampled_data(now, [1.0, 2.0, 3.0, 4.0]) + assert (await chp_power_receiver.receive()).value == 2.0 + assert (await chp_production_power_receiver.receive()).value == 0.0 + assert (await chp_consumption_power_receiver.receive()).value == 2.0 + + await send_resampled_data(now, [-4.0, -12.0, None, 10.2]) + assert (await chp_power_receiver.receive()).value == -12.0 + assert (await chp_production_power_receiver.receive()).value == 12.0 + assert (await chp_consumption_power_receiver.receive()).value == 0.0 + + async def test_pv_power(self, mocker: MockerFixture) -> None: + """Test the pv power formula.""" + mockgrid = MockMicrogrid(grid_side_meter=False) + mockgrid.add_solar_inverters(2) + await mockgrid.start(mocker) + + assert len(mockgrid.pv_inverter_ids) == 2 + + channels: dict[int, Broadcast[Sample]] = { + inv_id: Broadcast(f"#{inv_id}") for inv_id in [*mockgrid.pv_inverter_ids] + } + senders: list[Sender[Sample]] = [ + channels[component_id].new_sender() + for component_id in mockgrid.pv_inverter_ids ] - await synchronize_receivers( - [battery_power_recv, pv_power_recv, *bat_inv_receivers, *pv_inv_receivers] + async def send_resampled_data( + now: datetime, + meter_data: list[float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) + + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() + + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, ) - battery_results = [] - pv_results = [] - battery_inv_sums = [] - pv_inv_sums = [] - for _ in range(10): - bat_inv_sum = 0.0 - pv_inv_sum = 0.0 - for recv in bat_inv_receivers: - val = await recv.receive() - assert val is not None and val.value is not None and val.value > 0.0 - bat_inv_sum += val.value - battery_inv_sums.append(bat_inv_sum) + logical_meter = microgrid.logical_meter() + pv_power_receiver = logical_meter.pv_power.new_receiver() + pv_production_power_receiver = logical_meter.pv_production_power.new_receiver() + pv_consumption_power_receiver = ( + logical_meter.pv_consumption_power.new_receiver() + ) - for recv in pv_inv_receivers: - val = await recv.receive() - assert val is not None and val.value is not None and val.value > 0.0 - pv_inv_sum += val.value - pv_inv_sums.append(pv_inv_sum) + now = datetime.now(tz=timezone.utc) + await send_resampled_data(now, [-1.0, -2.0]) + assert (await pv_power_receiver.receive()).value == -3.0 + assert (await pv_production_power_receiver.receive()).value == 3.0 + assert (await pv_consumption_power_receiver.receive()).value == 0.0 - val = await battery_power_recv.receive() - assert val is not None and val.value is not None - battery_results.append(val.value) + async def test_consumer_power_grid_meter(self, mocker: MockerFixture) -> None: + """Test the consumer power formula with a grid meter.""" + mockgrid = MockMicrogrid(grid_side_meter=True) + mockgrid.add_batteries(2) + mockgrid.add_solar_inverters(2) + await mockgrid.start(mocker) - val = await pv_power_recv.receive() - assert val is not None and val.value is not None - pv_results.append(val.value) + assert len(mockgrid.meter_ids) == 5 - await mockgrid.cleanup() + channels: dict[int, Broadcast[Sample]] = { + meter_id: Broadcast(f"#{meter_id}") for meter_id in [*mockgrid.meter_ids] + } + senders: list[Sender[Sample]] = [ + channels[component_id].new_sender() for component_id in mockgrid.meter_ids + ] + + async def send_resampled_data( + now: datetime, + meter_data: list[float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) + + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() + + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, + ) + + logical_meter = microgrid.logical_meter() + consumer_power_receiver = logical_meter.consumer_power.new_receiver() + + now = datetime.now(tz=timezone.utc) + await send_resampled_data(now, [20.0, 2.0, 3.0, 4.0, 5.0]) + assert (await consumer_power_receiver.receive()).value == 6.0 + + async def test_consumer_power_no_grid_meter(self, mocker: MockerFixture) -> None: + """Test the consumer power formula without a grid meter.""" + mockgrid = MockMicrogrid(grid_side_meter=False) + mockgrid.add_batteries(2) + mockgrid.add_solar_inverters(2) + await mockgrid.start(mocker) + + assert len(mockgrid.meter_ids) == 5 + + channels: dict[int, Broadcast[Sample]] = { + meter_id: Broadcast(f"#{meter_id}") for meter_id in [*mockgrid.meter_ids] + } + senders: list[Sender[Sample]] = [ + channels[component_id].new_sender() for component_id in mockgrid.meter_ids + ] + + async def send_resampled_data( + now: datetime, + meter_data: list[float | None], + ) -> None: + """Send resampled data to the channels.""" + for sender, value in zip(senders, meter_data): + await sender.send(Sample(now, value)) + + def mock_resampled_receiver( + _1: Any, component_id: int, _2: ComponentMetricId + ) -> Receiver[Sample]: + return channels[component_id].new_receiver() + + mocker.patch( + "frequenz.sdk.timeseries._formula_engine._resampled_formula_builder" + ".ResampledFormulaBuilder._get_resampled_receiver", + mock_resampled_receiver, + ) + + logical_meter = microgrid.logical_meter() + consumer_power_receiver = logical_meter.consumer_power.new_receiver() - assert len(battery_results) == 10 - assert equal_float_lists(battery_results, battery_inv_sums) - assert len(pv_results) == 10 - assert equal_float_lists(pv_results, pv_inv_sums) + now = datetime.now(tz=timezone.utc) + await send_resampled_data(now, [20.0, 2.0, 3.0, 4.0, 5.0]) + assert (await consumer_power_receiver.receive()).value == 20.0