Skip to content

Commit cd8d39e

Browse files
ela-kotulska-frequenzela-kotulska-frequenz
committed
Refactor GridPowerFormulaBase to handle multiple quantities
Use dedicated quanitity ReactivePower in GridReactivePowerFormula Signed-off-by: Elzbieta Kotulska <[email protected]>
1 parent 3e0e1ef commit cd8d39e

File tree

3 files changed

+110
-47
lines changed

3 files changed

+110
-47
lines changed

‎src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_grid_power_formula.py‎

Lines changed: 49 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,14 +4,17 @@
44
"""Formula generator from component graph for Grid Power."""
55

66

7-
from frequenz.client.microgrid import ComponentMetricId
7+
from frequenz.client.microgrid import Component, ComponentMetricId
88
from frequenz.quantities import Power
99

1010
from .._formula_engine import FormulaEngine
11+
from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
12+
from ._formula_generator import FormulaGeneratorConfig
1113
from ._grid_power_formula_base import GridPowerFormulaBase
14+
from ._simple_power_formula import SimplePowerFormula
1215

1316

14-
class GridPowerFormula(GridPowerFormulaBase):
17+
class GridPowerFormula(GridPowerFormulaBase[Power]):
1518
"""Creates a formula engine from the component graph for calculating grid power."""
1619

1720
def generate( # noqa: DOC502
@@ -32,3 +35,47 @@ def generate( # noqa: DOC502
3235
Power.from_watts,
3336
)
3437
return self._generate(builder)
38+
39+
def _get_fallback_formulas(
40+
self, components: set[Component]
41+
) -> dict[Component, FallbackFormulaMetricFetcher[Power] | None]:
42+
"""Find primary and fallback components and create fallback formulas.
43+
44+
The primary component is the one that will be used to calculate the grid power.
45+
If it is not available, the fallback formula will be used instead.
46+
Fallback formulas calculate the grid power using the fallback components.
47+
Fallback formulas are wrapped in `FallbackFormulaMetricFetcher`.
48+
49+
Args:
50+
components: The producer components.
51+
52+
Returns:
53+
A dictionary mapping primary components to their FallbackFormulaMetricFetcher.
54+
"""
55+
fallbacks = self._get_metric_fallback_components(components)
56+
57+
fallback_formulas: dict[
58+
Component, FallbackFormulaMetricFetcher[Power] | None
59+
] = {}
60+
61+
for primary_component, fallback_components in fallbacks.items():
62+
if len(fallback_components) == 0:
63+
fallback_formulas[primary_component] = None
64+
continue
65+
66+
fallback_ids = [c.component_id for c in fallback_components]
67+
generator = SimplePowerFormula(
68+
f"{self._namespace}_fallback_{fallback_ids}",
69+
self._channel_registry,
70+
self._resampler_subscription_sender,
71+
FormulaGeneratorConfig(
72+
component_ids=set(fallback_ids),
73+
allow_fallback=False,
74+
),
75+
)
76+
77+
fallback_formulas[primary_component] = FallbackFormulaMetricFetcher(
78+
generator
79+
)
80+
81+
return fallback_formulas

‎src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_grid_power_formula_base.py‎

Lines changed: 8 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -3,28 +3,23 @@
33

44
"""Base formula generator from component graph for Grid Power."""
55

6-
from abc import ABC
6+
from abc import ABC, abstractmethod
77

88
from frequenz.client.microgrid import Component, ComponentCategory
9-
from frequenz.quantities import Power
109

10+
from ..._base_types import QuantityT
1111
from .._formula_engine import FormulaEngine
1212
from .._resampled_formula_builder import ResampledFormulaBuilder
1313
from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
14-
from ._formula_generator import (
15-
ComponentNotFound,
16-
FormulaGenerator,
17-
FormulaGeneratorConfig,
18-
)
19-
from ._simple_power_formula import SimplePowerFormula
14+
from ._formula_generator import ComponentNotFound, FormulaGenerator
2015

2116

22-
class GridPowerFormulaBase(FormulaGenerator[Power], ABC):
17+
class GridPowerFormulaBase(FormulaGenerator[QuantityT], ABC):
2318
"""Base class for grid power formula generators."""
2419

2520
def _generate(
26-
self, builder: ResampledFormulaBuilder[Power]
27-
) -> FormulaEngine[Power]:
21+
self, builder: ResampledFormulaBuilder[QuantityT]
22+
) -> FormulaEngine[QuantityT]:
2823
"""Generate a formula for calculating grid power from the component graph.
2924
3025
Args:
@@ -92,9 +87,10 @@ def _generate(
9287

9388
return builder.build()
9489

90+
@abstractmethod
9591
def _get_fallback_formulas(
9692
self, components: set[Component]
97-
) -> dict[Component, FallbackFormulaMetricFetcher[Power] | None]:
93+
) -> dict[Component, FallbackFormulaMetricFetcher[QuantityT] | None]:
9894
"""Find primary and fallback components and create fallback formulas.
9995
10096
The primary component is the one that will be used to calculate the producer power.
@@ -108,30 +104,3 @@ def _get_fallback_formulas(
108104
Returns:
109105
A dictionary mapping primary components to their FallbackFormulaMetricFetcher.
110106
"""
111-
fallbacks = self._get_metric_fallback_components(components)
112-
113-
fallback_formulas: dict[
114-
Component, FallbackFormulaMetricFetcher[Power] | None
115-
] = {}
116-
117-
for primary_component, fallback_components in fallbacks.items():
118-
if len(fallback_components) == 0:
119-
fallback_formulas[primary_component] = None
120-
continue
121-
122-
fallback_ids = [c.component_id for c in fallback_components]
123-
generator = SimplePowerFormula(
124-
f"{self._namespace}_fallback_{fallback_ids}",
125-
self._channel_registry,
126-
self._resampler_subscription_sender,
127-
FormulaGeneratorConfig(
128-
component_ids=set(fallback_ids),
129-
allow_fallback=False,
130-
),
131-
)
132-
133-
fallback_formulas[primary_component] = FallbackFormulaMetricFetcher(
134-
generator
135-
)
136-
137-
return fallback_formulas

‎src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_grid_reactive_power_formula.py‎

Lines changed: 53 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,20 +4,23 @@
44
"""Formula generator from component graph for Grid Reactive Power."""
55

66

7-
from frequenz.client.microgrid import ComponentMetricId
8-
from frequenz.quantities import Power
7+
from frequenz.client.microgrid import Component, ComponentMetricId
8+
from frequenz.quantities import ReactivePower
99

1010
from .._formula_engine import FormulaEngine
11+
from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
12+
from ._formula_generator import FormulaGeneratorConfig
1113
from ._grid_power_formula_base import GridPowerFormulaBase
14+
from ._simple_power_formula import SimpleReactivePowerFormula
1215

1316

14-
class GridReactivePowerFormula(GridPowerFormulaBase):
17+
class GridReactivePowerFormula(GridPowerFormulaBase[ReactivePower]):
1518
"""Creates a formula engine from the component graph for calculating grid reactive power."""
1619

1720
def generate( # noqa: DOC502
1821
# * ComponentNotFound is raised indirectly by _get_grid_component_successors
1922
self,
20-
) -> FormulaEngine[Power]:
23+
) -> FormulaEngine[ReactivePower]:
2124
"""Generate a formula for calculating grid reactive power from the component graph.
2225
2326
Returns:
@@ -27,8 +30,52 @@ def generate( # noqa: DOC502
2730
ComponentNotFound: when the component graph doesn't have a `GRID` component.
2831
"""
2932
builder = self._get_builder(
30-
f"grid-{ComponentMetricId.REACTIVE_POWER.value}",
33+
"grid_reactive_power_formula",
3134
ComponentMetricId.REACTIVE_POWER,
32-
Power.from_watts,
35+
ReactivePower.from_volt_amperes_reactive,
3336
)
3437
return self._generate(builder)
38+
39+
def _get_fallback_formulas(
40+
self, components: set[Component]
41+
) -> dict[Component, FallbackFormulaMetricFetcher[ReactivePower] | None]:
42+
"""Find primary and fallback components and create fallback formulas.
43+
44+
The primary component is the one that will be used to calculate the grid reactive power.
45+
If it is not available, the fallback formula will be used instead.
46+
Fallback formulas calculate the grid power using the fallback components.
47+
Fallback formulas are wrapped in `FallbackFormulaMetricFetcher`.
48+
49+
Args:
50+
components: The producer components.
51+
52+
Returns:
53+
A dictionary mapping primary components to their FallbackFormulaMetricFetcher.
54+
"""
55+
fallbacks = self._get_metric_fallback_components(components)
56+
57+
fallback_formulas: dict[
58+
Component, FallbackFormulaMetricFetcher[ReactivePower] | None
59+
] = {}
60+
61+
for primary_component, fallback_components in fallbacks.items():
62+
if len(fallback_components) == 0:
63+
fallback_formulas[primary_component] = None
64+
continue
65+
66+
fallback_ids = [c.component_id for c in fallback_components]
67+
generator = SimpleReactivePowerFormula(
68+
f"{self._namespace}_fallback_{fallback_ids}",
69+
self._channel_registry,
70+
self._resampler_subscription_sender,
71+
FormulaGeneratorConfig(
72+
component_ids=set(fallback_ids),
73+
allow_fallback=False,
74+
),
75+
)
76+
77+
fallback_formulas[primary_component] = FallbackFormulaMetricFetcher(
78+
generator
79+
)
80+
81+
return fallback_formulas

0 commit comments

Comments
 (0)