Skip to content

Formula Engine¤

FormulaEngine ¤

FormulaEngines are a part of the SDK's data pipeline, and provide a way for the SDK to apply formulas on resampled data streams.

They 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.battery_pool()

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 battery_pool().power and 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

logical_meter = microgrid.logical_meter()
battery_pool = microgrid.battery_pool()
ev_charger_pool = microgrid.ev_charger_pool()
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=}")

FormulaEngine3Phase ¤

A FormulaEngine3Phase is similar to a FormulaEngine, except that they stream 3-phase samples. All the current formulas (like Grid.current, EVChargerPool.current, etc.) are implemented as 3-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.ev_charger_pool()

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

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

from frequenz.sdk import microgrid

logical_meter = microgrid.logical_meter()
ev_charger_pool = microgrid.ev_charger_pool()
grid = microgrid.grid()

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

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