Skip to content

formula_engine

frequenz.sdk.timeseries.formula_engine ¤

Provides a way for the SDK to apply formulas on resampled data streams.

Formula Engine¤

FormulaEngines are used in the SDK to calculate and stream metrics like grid_power, consumer_power, etc., which are building blocks of the Frequenz SDK Microgrid Model.

The SDK creates the formulas by analysing the configuration of components in the Component Graph.

Streaming Interface¤

The FormulaEngine.new_receiver() method can be used to create a Receiver that streams the Samples calculated by the formula engine.

from frequenz.sdk import microgrid

battery_pool = microgrid.new_battery_pool(priority=5)

async for power in battery_pool.power.new_receiver():
    print(f"{power=}")

Composition¤

Composite FormulaEngines can be built using arithmetic operations on FormulaEngines streaming the same type of data.

For example, if you're interested in a particular composite metric that can be calculated by subtracting new_battery_pool().power and new_ev_charger_pool().power from the grid().power, we can build a FormulaEngine that provides a stream of this calculated metric as follows:

from frequenz.sdk import microgrid

battery_pool = microgrid.new_battery_pool(priority=5)
ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)
grid = microgrid.grid()

# apply operations on formula engines to create a formula engine that would
# apply these operations on the corresponding data streams.
net_power = (
    grid.power - (battery_pool.power + ev_charger_pool.power)
).build("net_power")

async for power in net_power.new_receiver():
    print(f"{power=}")

Formula Engine 3-Phase¤

A FormulaEngine3Phase is similar to a FormulaEngine, except that they stream 3-phase samples. All the current formulas (like Grid.current_per_phase, EVChargerPool.current_per_phase, etc.) are implemented as per-phase formulas.

Streaming Interface¤

The FormulaEngine3Phase.new_receiver() method can be used to create a Receiver that streams the Sample3Phase values calculated by the formula engine.

from frequenz.sdk import microgrid

ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)

async for sample in ev_charger_pool.current_per_phase.new_receiver():
    print(f"Current: {sample}")

Composition¤

FormulaEngine3Phase instances can be composed together, just like FormulaEngine instances.

from frequenz.sdk import microgrid

ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)
grid = microgrid.grid()

# Calculate grid consumption current that's not used by the EV chargers
other_current = (grid.current_per_phase - ev_charger_pool.current_per_phase).build(
    "other_current"
)

async for sample in other_current.new_receiver():
    print(f"Other current: {sample}")

Classes¤

frequenz.sdk.timeseries.formula_engine.FormulaEngine ¤

Bases: Generic[QuantityT]

An engine to apply formulas on resampled data streams.

Please refer to the module documentation for more information on how formula engines are used throughout the SDK.

Streaming the power of a battery pool. 
from frequenz.sdk import microgrid

battery_pool = microgrid.new_battery_pool(priority=5)

async for power in battery_pool.power.new_receiver():
    print(f"{power=}")
Composition of formula engines. 
from frequenz.sdk import microgrid

battery_pool = microgrid.new_battery_pool(priority=5)
ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)
grid = microgrid.grid()

# apply operations on formula engines to create a formula engine that would
# apply these operations on the corresponding data streams.
net_power = (
    grid.power - (battery_pool.power + ev_charger_pool.power)
).build("net_power")

async for power in net_power.new_receiver():
    print(f"{power=}")
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
class FormulaEngine(Generic[QuantityT]):
    """An engine to apply formulas on resampled data streams.

    Please refer to the [module documentation][frequenz.sdk.timeseries.formula_engine]
    for more information on how formula engines are used throughout the SDK.

    Example: Streaming the power of a battery pool.
        ```python
        from frequenz.sdk import microgrid

        battery_pool = microgrid.new_battery_pool(priority=5)

        async for power in battery_pool.power.new_receiver():
            print(f"{power=}")
        ```

    Example: Composition of formula engines.
        ```python
        from frequenz.sdk import microgrid

        battery_pool = microgrid.new_battery_pool(priority=5)
        ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)
        grid = microgrid.grid()

        # apply operations on formula engines to create a formula engine that would
        # apply these operations on the corresponding data streams.
        net_power = (
            grid.power - (battery_pool.power + ev_charger_pool.power)
        ).build("net_power")

        async for power in net_power.new_receiver():
            print(f"{power=}")
        ```
    """

    def __init__(
        self,
        builder: FormulaBuilder[QuantityT],
        create_method: Callable[[float], QuantityT],
    ) -> None:
        """Create a `FormulaEngine` instance.

        Args:
            builder: A `FormulaBuilder` instance to get the formula steps and metric
                fetchers from.
            create_method: A method to generate the output `Sample` value with.  If the
                formula is for generating power values, this would be
                `Power.from_watts`, for example.
        """
        self._higher_order_builder = HigherOrderFormulaBuilder
        self._name: str = builder.name
        self._builder: FormulaBuilder[QuantityT] = builder
        self._create_method: Callable[[float], QuantityT] = create_method
        self._channel: Broadcast[Sample[QuantityT]] = Broadcast(name=self._name)
        self._task: asyncio.Task[None] | None = None

    async def _stop(self) -> None:
        """Stop a running formula engine."""
        if self._task is None:
            return
        await cancel_and_await(self._task)

    @classmethod
    def from_receiver(
        cls,
        name: str,
        receiver: Receiver[Sample[QuantityT]],
        create_method: Callable[[float], QuantityT],
        *,
        nones_are_zeros: bool = False,
    ) -> FormulaEngine[QuantityT]:
        """
        Create a formula engine from a receiver.

        Can be used to compose a formula engine with a receiver. When composing
        the new engine with other engines, make sure that receiver gets data
        from the same resampler and that the `create_method`s match.

        Example:
            ```python
            from frequenz.sdk import microgrid
            from frequenz.quantities import Power

            async def run() -> None:
                producer_power_engine = microgrid.producer().power
                consumer_power_recv = microgrid.consumer().power.new_receiver()

                excess_power_recv = (
                    (
                        producer_power_engine
                        + FormulaEngine.from_receiver(
                            "consumer power",
                            consumer_power_recv,
                            Power.from_watts,
                        )
                    )
                    .build("excess power")
                    .new_receiver()
                )

            asyncio.run(run())
            ```

        Args:
            name: A name for the formula engine.
            receiver: A receiver that streams `Sample`s.
            create_method: A method to generate the output `Sample` value with,
                e.g. `Power.from_watts`.
            nones_are_zeros: If `True`, `None` values in the receiver are treated as 0.

        Returns:
            A formula engine that streams the `Sample`s from the receiver.
        """
        builder = FormulaBuilder(name, create_method)
        builder.push_metric(name, receiver, nones_are_zeros=nones_are_zeros)
        return cls(builder, create_method)

    def __add__(
        self,
        other: (
            FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
        ),
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """Return a formula builder that adds (data in) `other` to `self`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder(self, self._create_method) + other

    def __sub__(
        self,
        other: (
            FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
        ),
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """Return a formula builder that subtracts (data in) `other` from `self`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder(self, self._create_method) - other

    def __mul__(
        self,
        other: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | float,
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """Return a formula builder that multiplies (data in) `self` with `other`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder(self, self._create_method) * other

    def __truediv__(
        self,
        other: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | float,
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """Return a formula builder that divides (data in) `self` by `other`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder(self, self._create_method) / other

    def max(
        self,
        other: (
            FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
        ),
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """Return a formula engine that outputs the maximum of `self` and `other`.

        Args:
            other: A formula receiver, a formula builder or a QuantityT instance
                corresponding to a sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder(self, self._create_method).max(other)

    def min(
        self,
        other: (
            FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
        ),
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """Return a formula engine that outputs the minimum of `self` and `other`.

        Args:
            other: A formula receiver, a formula builder or a QuantityT instance
                corresponding to a sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder(self, self._create_method).min(other)

    def consumption(
        self,
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """
        Return a formula builder that applies the consumption operator on `self`.

        The consumption operator returns either the identity if the power value is
        positive or 0.
        """
        return HigherOrderFormulaBuilder(self, self._create_method).consumption()

    def production(
        self,
    ) -> HigherOrderFormulaBuilder[QuantityT]:
        """
        Return a formula builder that applies the production operator on `self`.

        The production operator returns either the absolute value if the power value is
        negative or 0.
        """
        return HigherOrderFormulaBuilder(self, self._create_method).production()

    async def _run(self) -> None:
        await self._builder.subscribe()
        steps, metric_fetchers = self._builder.finalize()
        evaluator = FormulaEvaluator[QuantityT](
            self._name, steps, metric_fetchers, self._create_method
        )
        sender = self._channel.new_sender()
        while True:
            try:
                msg = await evaluator.apply()
            except asyncio.CancelledError:
                _logger.exception("FormulaEngine task cancelled: %s", self._name)
                raise
            except Exception as err:  # pylint: disable=broad-except
                _logger.warning(
                    "Formula application failed: %s. Error: %s", self._name, err
                )
            else:
                await sender.send(msg)

    def __str__(self) -> str:
        """Return a string representation of the formula.

        Returns:
            A string representation of the formula.
        """
        steps = (
            self._builder._build_stack
            if len(self._builder._build_stack) > 0
            else self._builder._steps
        )
        return format_formula(steps)

    def new_receiver(
        self, name: str | None = None, max_size: int = 50
    ) -> Receiver[Sample[QuantityT]]:
        """Create a new receiver that streams the output of the formula engine.

        Args:
            name: An optional name for the receiver.
            max_size: The size of the receiver's buffer.

        Returns:
            A receiver that streams output `Sample`s from the formula engine.
        """
        if self._task is None:
            self._task = asyncio.create_task(self._run())

        recv = self._channel.new_receiver(name=name, limit=max_size)

        # This is a hack to ensure that the lifetime of the engine is tied to the
        # lifetime of the receiver.  This is necessary because the engine is a task that
        # runs forever, and in cases where higher order built for example with the below
        # idiom, the user would hold no references to the engine and it could get
        # garbage collected before the receiver.  This behaviour is explained in the
        # `asyncio.create_task` docs here:
        # https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
        #
        #     formula = (grid_power_engine + bat_power_engine).build().new_receiver()
        recv._engine_reference = self  # type: ignore # pylint: disable=protected-access
        return recv
Functions¤
__add__ ¤
__add__(
    other: (
        FormulaEngine[QuantityT]
        | HigherOrderFormulaBuilder[QuantityT]
        | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]

Return a formula builder that adds (data in) other to self.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT

RETURNS DESCRIPTION
HigherOrderFormulaBuilder[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def __add__(
    self,
    other: (
        FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]:
    """Return a formula builder that adds (data in) `other` to `self`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder(self, self._create_method) + other
__init__ ¤
__init__(
    builder: FormulaBuilder[QuantityT],
    create_method: Callable[[float], QuantityT],
) -> None

Create a FormulaEngine instance.

PARAMETER DESCRIPTION
builder

A FormulaBuilder instance to get the formula steps and metric fetchers from.

TYPE: FormulaBuilder[QuantityT]

create_method

A method to generate the output Sample value with. If the formula is for generating power values, this would be Power.from_watts, for example.

TYPE: Callable[[float], QuantityT]

Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __init__(
    self,
    builder: FormulaBuilder[QuantityT],
    create_method: Callable[[float], QuantityT],
) -> None:
    """Create a `FormulaEngine` instance.

    Args:
        builder: A `FormulaBuilder` instance to get the formula steps and metric
            fetchers from.
        create_method: A method to generate the output `Sample` value with.  If the
            formula is for generating power values, this would be
            `Power.from_watts`, for example.
    """
    self._higher_order_builder = HigherOrderFormulaBuilder
    self._name: str = builder.name
    self._builder: FormulaBuilder[QuantityT] = builder
    self._create_method: Callable[[float], QuantityT] = create_method
    self._channel: Broadcast[Sample[QuantityT]] = Broadcast(name=self._name)
    self._task: asyncio.Task[None] | None = None
__mul__ ¤
__mul__(
    other: (
        FormulaEngine[QuantityT]
        | HigherOrderFormulaBuilder[QuantityT]
        | float
    ),
) -> HigherOrderFormulaBuilder[QuantityT]

Return a formula builder that multiplies (data in) self with other.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | float

RETURNS DESCRIPTION
HigherOrderFormulaBuilder[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def __mul__(
    self,
    other: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | float,
) -> HigherOrderFormulaBuilder[QuantityT]:
    """Return a formula builder that multiplies (data in) `self` with `other`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder(self, self._create_method) * other
__str__ ¤
__str__() -> str

Return a string representation of the formula.

RETURNS DESCRIPTION
str

A string representation of the formula.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
322
323
324
325
326
327
328
329
330
331
332
333
def __str__(self) -> str:
    """Return a string representation of the formula.

    Returns:
        A string representation of the formula.
    """
    steps = (
        self._builder._build_stack
        if len(self._builder._build_stack) > 0
        else self._builder._steps
    )
    return format_formula(steps)
__sub__ ¤
__sub__(
    other: (
        FormulaEngine[QuantityT]
        | HigherOrderFormulaBuilder[QuantityT]
        | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]

Return a formula builder that subtracts (data in) other from self.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT

RETURNS DESCRIPTION
HigherOrderFormulaBuilder[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def __sub__(
    self,
    other: (
        FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]:
    """Return a formula builder that subtracts (data in) `other` from `self`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder(self, self._create_method) - other
__truediv__ ¤
__truediv__(
    other: (
        FormulaEngine[QuantityT]
        | HigherOrderFormulaBuilder[QuantityT]
        | float
    ),
) -> HigherOrderFormulaBuilder[QuantityT]

Return a formula builder that divides (data in) self by other.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | float

RETURNS DESCRIPTION
HigherOrderFormulaBuilder[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def __truediv__(
    self,
    other: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | float,
) -> HigherOrderFormulaBuilder[QuantityT]:
    """Return a formula builder that divides (data in) `self` by `other`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder(self, self._create_method) / other
consumption ¤
consumption() -> HigherOrderFormulaBuilder[QuantityT]

Return a formula builder that applies the consumption operator on self.

The consumption operator returns either the identity if the power value is positive or 0.

Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
280
281
282
283
284
285
286
287
288
289
def consumption(
    self,
) -> HigherOrderFormulaBuilder[QuantityT]:
    """
    Return a formula builder that applies the consumption operator on `self`.

    The consumption operator returns either the identity if the power value is
    positive or 0.
    """
    return HigherOrderFormulaBuilder(self, self._create_method).consumption()
from_receiver classmethod ¤
from_receiver(
    name: str,
    receiver: Receiver[Sample[QuantityT]],
    create_method: Callable[[float], QuantityT],
    *,
    nones_are_zeros: bool = False
) -> FormulaEngine[QuantityT]

Create a formula engine from a receiver.

Can be used to compose a formula engine with a receiver. When composing the new engine with other engines, make sure that receiver gets data from the same resampler and that the create_methods match.

Example 
from frequenz.sdk import microgrid
from frequenz.quantities import Power

async def run() -> None:
    producer_power_engine = microgrid.producer().power
    consumer_power_recv = microgrid.consumer().power.new_receiver()

    excess_power_recv = (
        (
            producer_power_engine
            + FormulaEngine.from_receiver(
                "consumer power",
                consumer_power_recv,
                Power.from_watts,
            )
        )
        .build("excess power")
        .new_receiver()
    )

asyncio.run(run())
PARAMETER DESCRIPTION
name

A name for the formula engine.

TYPE: str

receiver

A receiver that streams Samples.

TYPE: Receiver[Sample[QuantityT]]

create_method

A method to generate the output Sample value with, e.g. Power.from_watts.

TYPE: Callable[[float], QuantityT]

nones_are_zeros

If True, None values in the receiver are treated as 0.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
FormulaEngine[QuantityT]

A formula engine that streams the Samples from the receiver.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@classmethod
def from_receiver(
    cls,
    name: str,
    receiver: Receiver[Sample[QuantityT]],
    create_method: Callable[[float], QuantityT],
    *,
    nones_are_zeros: bool = False,
) -> FormulaEngine[QuantityT]:
    """
    Create a formula engine from a receiver.

    Can be used to compose a formula engine with a receiver. When composing
    the new engine with other engines, make sure that receiver gets data
    from the same resampler and that the `create_method`s match.

    Example:
        ```python
        from frequenz.sdk import microgrid
        from frequenz.quantities import Power

        async def run() -> None:
            producer_power_engine = microgrid.producer().power
            consumer_power_recv = microgrid.consumer().power.new_receiver()

            excess_power_recv = (
                (
                    producer_power_engine
                    + FormulaEngine.from_receiver(
                        "consumer power",
                        consumer_power_recv,
                        Power.from_watts,
                    )
                )
                .build("excess power")
                .new_receiver()
            )

        asyncio.run(run())
        ```

    Args:
        name: A name for the formula engine.
        receiver: A receiver that streams `Sample`s.
        create_method: A method to generate the output `Sample` value with,
            e.g. `Power.from_watts`.
        nones_are_zeros: If `True`, `None` values in the receiver are treated as 0.

    Returns:
        A formula engine that streams the `Sample`s from the receiver.
    """
    builder = FormulaBuilder(name, create_method)
    builder.push_metric(name, receiver, nones_are_zeros=nones_are_zeros)
    return cls(builder, create_method)
max ¤
max(
    other: (
        FormulaEngine[QuantityT]
        | HigherOrderFormulaBuilder[QuantityT]
        | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]

Return a formula engine that outputs the maximum of self and other.

PARAMETER DESCRIPTION
other

A formula receiver, a formula builder or a QuantityT instance corresponding to a sub-expression.

TYPE: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT

RETURNS DESCRIPTION
HigherOrderFormulaBuilder[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def max(
    self,
    other: (
        FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]:
    """Return a formula engine that outputs the maximum of `self` and `other`.

    Args:
        other: A formula receiver, a formula builder or a QuantityT instance
            corresponding to a sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder(self, self._create_method).max(other)
min ¤
min(
    other: (
        FormulaEngine[QuantityT]
        | HigherOrderFormulaBuilder[QuantityT]
        | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]

Return a formula engine that outputs the minimum of self and other.

PARAMETER DESCRIPTION
other

A formula receiver, a formula builder or a QuantityT instance corresponding to a sub-expression.

TYPE: FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT

RETURNS DESCRIPTION
HigherOrderFormulaBuilder[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def min(
    self,
    other: (
        FormulaEngine[QuantityT] | HigherOrderFormulaBuilder[QuantityT] | QuantityT
    ),
) -> HigherOrderFormulaBuilder[QuantityT]:
    """Return a formula engine that outputs the minimum of `self` and `other`.

    Args:
        other: A formula receiver, a formula builder or a QuantityT instance
            corresponding to a sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder(self, self._create_method).min(other)
new_receiver ¤
new_receiver(
    name: str | None = None, max_size: int = 50
) -> Receiver[Sample[QuantityT]]

Create a new receiver that streams the output of the formula engine.

PARAMETER DESCRIPTION
name

An optional name for the receiver.

TYPE: str | None DEFAULT: None

max_size

The size of the receiver's buffer.

TYPE: int DEFAULT: 50

RETURNS DESCRIPTION
Receiver[Sample[QuantityT]]

A receiver that streams output Samples from the formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def new_receiver(
    self, name: str | None = None, max_size: int = 50
) -> Receiver[Sample[QuantityT]]:
    """Create a new receiver that streams the output of the formula engine.

    Args:
        name: An optional name for the receiver.
        max_size: The size of the receiver's buffer.

    Returns:
        A receiver that streams output `Sample`s from the formula engine.
    """
    if self._task is None:
        self._task = asyncio.create_task(self._run())

    recv = self._channel.new_receiver(name=name, limit=max_size)

    # This is a hack to ensure that the lifetime of the engine is tied to the
    # lifetime of the receiver.  This is necessary because the engine is a task that
    # runs forever, and in cases where higher order built for example with the below
    # idiom, the user would hold no references to the engine and it could get
    # garbage collected before the receiver.  This behaviour is explained in the
    # `asyncio.create_task` docs here:
    # https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
    #
    #     formula = (grid_power_engine + bat_power_engine).build().new_receiver()
    recv._engine_reference = self  # type: ignore # pylint: disable=protected-access
    return recv
production ¤
production() -> HigherOrderFormulaBuilder[QuantityT]

Return a formula builder that applies the production operator on self.

The production operator returns either the absolute value if the power value is negative or 0.

Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
291
292
293
294
295
296
297
298
299
300
def production(
    self,
) -> HigherOrderFormulaBuilder[QuantityT]:
    """
    Return a formula builder that applies the production operator on `self`.

    The production operator returns either the absolute value if the power value is
    negative or 0.
    """
    return HigherOrderFormulaBuilder(self, self._create_method).production()

frequenz.sdk.timeseries.formula_engine.FormulaEngine3Phase ¤

Bases: Generic[QuantityT]

An engine to apply formulas on 3-phase resampled data streams.

Please refer to the module documentation for more information on how formula engines are used throughout the SDK.

Streaming the current of an EV charger pool. 
from frequenz.sdk import microgrid

ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)

async for sample in ev_charger_pool.current_per_phase.new_receiver():
    print(f"Current: {sample}")
Composition of formula engines. 
from frequenz.sdk import microgrid

ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)
grid = microgrid.grid()

# Calculate grid consumption current that's not used by the EV chargers
other_current = (grid.current_per_phase - ev_charger_pool.current_per_phase).build(
    "other_current"
)

async for sample in other_current.new_receiver():
    print(f"Other current: {sample}")
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
class FormulaEngine3Phase(Generic[QuantityT]):
    """An engine to apply formulas on 3-phase resampled data streams.

    Please refer to the [module documentation][frequenz.sdk.timeseries.formula_engine]
    for more information on how formula engines are used throughout the SDK.

    Example: Streaming the current of an EV charger pool.
        ```python
        from frequenz.sdk import microgrid

        ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)

        async for sample in ev_charger_pool.current_per_phase.new_receiver():
            print(f"Current: {sample}")
        ```

    Example: Composition of formula engines.
        ```python
        from frequenz.sdk import microgrid

        ev_charger_pool = microgrid.new_ev_charger_pool(priority=5)
        grid = microgrid.grid()

        # Calculate grid consumption current that's not used by the EV chargers
        other_current = (grid.current_per_phase - ev_charger_pool.current_per_phase).build(
            "other_current"
        )

        async for sample in other_current.new_receiver():
            print(f"Other current: {sample}")
        ```
    """

    def __init__(
        self,
        name: str,
        create_method: Callable[[float], QuantityT],
        phase_streams: tuple[
            FormulaEngine[QuantityT],
            FormulaEngine[QuantityT],
            FormulaEngine[QuantityT],
        ],
    ) -> None:
        """Create a `FormulaEngine3Phase` instance.

        Args:
            name: A name for the formula.
            create_method: A method to generate the output `Sample` value with.  If the
                formula is for generating power values, this would be
                `Power.from_watts`, for example.
            phase_streams: output streams of formula engines running per-phase formulas.
        """
        self._higher_order_builder = HigherOrderFormulaBuilder3Phase
        self._name: str = name
        self._create_method: Callable[[float], QuantityT] = create_method
        self._channel: Broadcast[Sample3Phase[QuantityT]] = Broadcast(name=self._name)
        self._task: asyncio.Task[None] | None = None
        self._streams: tuple[
            FormulaEngine[QuantityT],
            FormulaEngine[QuantityT],
            FormulaEngine[QuantityT],
        ] = phase_streams

    async def _stop(self) -> None:
        """Stop a running formula engine."""
        if self._task is None:
            return
        await cancel_and_await(self._task)

    def __add__(
        self,
        other: (
            FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
        ),
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """Return a formula builder that adds (data in) `other` to `self`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method) + other

    def __sub__(
        self: FormulaEngine3Phase[QuantityT],
        other: (
            FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
        ),
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """Return a formula builder that subtracts (data in) `other` from `self`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method) - other

    def __mul__(
        self: FormulaEngine3Phase[QuantityT],
        other: (
            FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
        ),
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """Return a formula builder that multiplies (data in) `self` with `other`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method) * other

    def __truediv__(
        self: FormulaEngine3Phase[QuantityT],
        other: (
            FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
        ),
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """Return a formula builder that divides (data in) `self` by `other`.

        Args:
            other: A formula receiver, or a formula builder instance corresponding to a
                sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method) / other

    def max(
        self: FormulaEngine3Phase[QuantityT],
        other: (
            FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
        ),
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """Return a formula engine that outputs the maximum of `self` and `other`.

        Args:
            other: A formula receiver, a formula builder or a QuantityT instance
                corresponding to a sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method).max(other)

    def min(
        self: FormulaEngine3Phase[QuantityT],
        other: (
            FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
        ),
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """Return a formula engine that outputs the minimum of `self` and `other`.

        Args:
            other: A formula receiver, a formula builder or a QuantityT instance
                corresponding to a sub-expression.

        Returns:
            A formula builder that can take further expressions, or can be built
                into a formula engine.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method).min(other)

    def consumption(
        self: FormulaEngine3Phase[QuantityT],
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """
        Return a formula builder that applies the consumption operator on `self`.

        The consumption operator returns either the identity if the power value is
        positive or 0.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method).consumption()

    def production(
        self: FormulaEngine3Phase[QuantityT],
    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
        """
        Return a formula builder that applies the production operator on `self`.

        The production operator returns either the absolute value if the power value is
        negative or 0.
        """
        return HigherOrderFormulaBuilder3Phase(self, self._create_method).production()

    async def _run(self) -> None:
        sender = self._channel.new_sender()
        phase_1_rx = self._streams[0].new_receiver()
        phase_2_rx = self._streams[1].new_receiver()
        phase_3_rx = self._streams[2].new_receiver()

        while True:
            try:
                phase_1 = await phase_1_rx.receive()
                phase_2 = await phase_2_rx.receive()
                phase_3 = await phase_3_rx.receive()
                msg = Sample3Phase(
                    phase_1.timestamp,
                    phase_1.value,
                    phase_2.value,
                    phase_3.value,
                )
            except asyncio.CancelledError:
                _logger.exception("FormulaEngine task cancelled: %s", self._name)
                break
            else:
                await sender.send(msg)

    def new_receiver(
        self, name: str | None = None, max_size: int = 50
    ) -> Receiver[Sample3Phase[QuantityT]]:
        """Create a new receiver that streams the output of the formula engine.

        Args:
            name: An optional name for the receiver.
            max_size: The size of the receiver's buffer.

        Returns:
            A receiver that streams output `Sample`s from the formula engine.
        """
        if self._task is None:
            self._task = asyncio.create_task(self._run())

        recv = self._channel.new_receiver(name=name, limit=max_size)

        # This is a hack to ensure that the lifetime of the engine is tied to the
        # lifetime of the receiver.  This is necessary because the engine is a task that
        # runs forever, and in cases where higher order built for example with the below
        # idiom, the user would hold no references to the engine and it could get
        # garbage collected before the receiver.  This behaviour is explained in the
        # `asyncio.create_task` docs here:
        # https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
        #
        #     formula = (grid_power_engine + bat_power_engine).build().new_receiver()
        recv._engine_reference = self  # type: ignore # pylint: disable=protected-access
        return recv
Functions¤
__add__ ¤
__add__(
    other: (
        FormulaEngine3Phase[QuantityT]
        | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula builder that adds (data in) other to self.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]

RETURNS DESCRIPTION
HigherOrderFormulaBuilder3Phase[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def __add__(
    self,
    other: (
        FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """Return a formula builder that adds (data in) `other` to `self`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method) + other
__init__ ¤
__init__(
    name: str,
    create_method: Callable[[float], QuantityT],
    phase_streams: tuple[
        FormulaEngine[QuantityT],
        FormulaEngine[QuantityT],
        FormulaEngine[QuantityT],
    ],
) -> None

Create a FormulaEngine3Phase instance.

PARAMETER DESCRIPTION
name

A name for the formula.

TYPE: str

create_method

A method to generate the output Sample value with. If the formula is for generating power values, this would be Power.from_watts, for example.

TYPE: Callable[[float], QuantityT]

phase_streams

output streams of formula engines running per-phase formulas.

TYPE: tuple[FormulaEngine[QuantityT], FormulaEngine[QuantityT], FormulaEngine[QuantityT]]

Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
def __init__(
    self,
    name: str,
    create_method: Callable[[float], QuantityT],
    phase_streams: tuple[
        FormulaEngine[QuantityT],
        FormulaEngine[QuantityT],
        FormulaEngine[QuantityT],
    ],
) -> None:
    """Create a `FormulaEngine3Phase` instance.

    Args:
        name: A name for the formula.
        create_method: A method to generate the output `Sample` value with.  If the
            formula is for generating power values, this would be
            `Power.from_watts`, for example.
        phase_streams: output streams of formula engines running per-phase formulas.
    """
    self._higher_order_builder = HigherOrderFormulaBuilder3Phase
    self._name: str = name
    self._create_method: Callable[[float], QuantityT] = create_method
    self._channel: Broadcast[Sample3Phase[QuantityT]] = Broadcast(name=self._name)
    self._task: asyncio.Task[None] | None = None
    self._streams: tuple[
        FormulaEngine[QuantityT],
        FormulaEngine[QuantityT],
        FormulaEngine[QuantityT],
    ] = phase_streams
__mul__ ¤
__mul__(
    other: (
        FormulaEngine3Phase[QuantityT]
        | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula builder that multiplies (data in) self with other.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]

RETURNS DESCRIPTION
HigherOrderFormulaBuilder3Phase[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
def __mul__(
    self: FormulaEngine3Phase[QuantityT],
    other: (
        FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """Return a formula builder that multiplies (data in) `self` with `other`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method) * other
__sub__ ¤
__sub__(
    other: (
        FormulaEngine3Phase[QuantityT]
        | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula builder that subtracts (data in) other from self.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]

RETURNS DESCRIPTION
HigherOrderFormulaBuilder3Phase[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def __sub__(
    self: FormulaEngine3Phase[QuantityT],
    other: (
        FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """Return a formula builder that subtracts (data in) `other` from `self`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method) - other
__truediv__ ¤
__truediv__(
    other: (
        FormulaEngine3Phase[QuantityT]
        | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula builder that divides (data in) self by other.

PARAMETER DESCRIPTION
other

A formula receiver, or a formula builder instance corresponding to a sub-expression.

TYPE: FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]

RETURNS DESCRIPTION
HigherOrderFormulaBuilder3Phase[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
def __truediv__(
    self: FormulaEngine3Phase[QuantityT],
    other: (
        FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """Return a formula builder that divides (data in) `self` by `other`.

    Args:
        other: A formula receiver, or a formula builder instance corresponding to a
            sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method) / other
consumption ¤
consumption() -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula builder that applies the consumption operator on self.

The consumption operator returns either the identity if the power value is positive or 0.

Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
542
543
544
545
546
547
548
549
550
551
def consumption(
    self: FormulaEngine3Phase[QuantityT],
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """
    Return a formula builder that applies the consumption operator on `self`.

    The consumption operator returns either the identity if the power value is
    positive or 0.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method).consumption()
max ¤
max(
    other: (
        FormulaEngine3Phase[QuantityT]
        | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula engine that outputs the maximum of self and other.

PARAMETER DESCRIPTION
other

A formula receiver, a formula builder or a QuantityT instance corresponding to a sub-expression.

TYPE: FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]

RETURNS DESCRIPTION
HigherOrderFormulaBuilder3Phase[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def max(
    self: FormulaEngine3Phase[QuantityT],
    other: (
        FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """Return a formula engine that outputs the maximum of `self` and `other`.

    Args:
        other: A formula receiver, a formula builder or a QuantityT instance
            corresponding to a sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method).max(other)
min ¤
min(
    other: (
        FormulaEngine3Phase[QuantityT]
        | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula engine that outputs the minimum of self and other.

PARAMETER DESCRIPTION
other

A formula receiver, a formula builder or a QuantityT instance corresponding to a sub-expression.

TYPE: FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]

RETURNS DESCRIPTION
HigherOrderFormulaBuilder3Phase[QuantityT]

A formula builder that can take further expressions, or can be built into a formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
def min(
    self: FormulaEngine3Phase[QuantityT],
    other: (
        FormulaEngine3Phase[QuantityT] | HigherOrderFormulaBuilder3Phase[QuantityT]
    ),
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """Return a formula engine that outputs the minimum of `self` and `other`.

    Args:
        other: A formula receiver, a formula builder or a QuantityT instance
            corresponding to a sub-expression.

    Returns:
        A formula builder that can take further expressions, or can be built
            into a formula engine.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method).min(other)
new_receiver ¤
new_receiver(
    name: str | None = None, max_size: int = 50
) -> Receiver[Sample3Phase[QuantityT]]

Create a new receiver that streams the output of the formula engine.

PARAMETER DESCRIPTION
name

An optional name for the receiver.

TYPE: str | None DEFAULT: None

max_size

The size of the receiver's buffer.

TYPE: int DEFAULT: 50

RETURNS DESCRIPTION
Receiver[Sample3Phase[QuantityT]]

A receiver that streams output Samples from the formula engine.
Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
def new_receiver(
    self, name: str | None = None, max_size: int = 50
) -> Receiver[Sample3Phase[QuantityT]]:
    """Create a new receiver that streams the output of the formula engine.

    Args:
        name: An optional name for the receiver.
        max_size: The size of the receiver's buffer.

    Returns:
        A receiver that streams output `Sample`s from the formula engine.
    """
    if self._task is None:
        self._task = asyncio.create_task(self._run())

    recv = self._channel.new_receiver(name=name, limit=max_size)

    # This is a hack to ensure that the lifetime of the engine is tied to the
    # lifetime of the receiver.  This is necessary because the engine is a task that
    # runs forever, and in cases where higher order built for example with the below
    # idiom, the user would hold no references to the engine and it could get
    # garbage collected before the receiver.  This behaviour is explained in the
    # `asyncio.create_task` docs here:
    # https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
    #
    #     formula = (grid_power_engine + bat_power_engine).build().new_receiver()
    recv._engine_reference = self  # type: ignore # pylint: disable=protected-access
    return recv
production ¤
production() -> HigherOrderFormulaBuilder3Phase[QuantityT]

Return a formula builder that applies the production operator on self.

The production operator returns either the absolute value if the power value is negative or 0.

Source code in frequenz/sdk/timeseries/formula_engine/_formula_engine.py 
553
554
555
556
557
558
559
560
561
562
def production(
    self: FormulaEngine3Phase[QuantityT],
) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
    """
    Return a formula builder that applies the production operator on `self`.

    The production operator returns either the absolute value if the power value is
    negative or 0.
    """
    return HigherOrderFormulaBuilder3Phase(self, self._create_method).production()