Skip to content

logical_meter

frequenz.sdk.timeseries.logical_meter ¤

A logical meter for calculating high level metrics for a microgrid.

Classes¤

frequenz.sdk.timeseries.logical_meter.LogicalMeter ¤

A logical meter for calculating high level metrics in a microgrid.

LogicalMeter provides methods for fetching power values from different points in the microgrid. These methods return FormulaReceiver objects, which can be used like normal Receivers, but can also be composed to form higher-order formula streams.

Note

LogicalMeter instances are not meant to be created directly by users. Use the microgrid.logical_meter method for creating LogicalMeter instances.

Example 
from datetime import timedelta

from frequenz.sdk import microgrid
from frequenz.sdk.timeseries import ResamplerConfig2
from frequenz.client.microgrid.metrics import Metric


await microgrid.initialize(
    "grpc://microgrid.sandbox.api.frequenz.io:62060",
    ResamplerConfig2(resampling_period=timedelta(seconds=1)),
)

logical_meter = (
    microgrid.logical_meter()
    .start_formula("#1001 + #1002", Metric.AC_ACTIVE_POWER)
    .new_receiver()
)

async for power in logical_meter:
    print(power.value)
Source code in src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 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
class LogicalMeter:
    """A logical meter for calculating high level metrics in a microgrid.

    LogicalMeter provides methods for fetching power values from different points in the
    microgrid.  These methods return `FormulaReceiver` objects, which can be used like
    normal `Receiver`s, but can also be composed to form higher-order formula streams.

    !!! note
        `LogicalMeter` instances are not meant to be created directly by users.  Use the
        [`microgrid.logical_meter`][frequenz.sdk.microgrid.logical_meter] method for
        creating `LogicalMeter` instances.

    Example:
        ```python
        from datetime import timedelta

        from frequenz.sdk import microgrid
        from frequenz.sdk.timeseries import ResamplerConfig2
        from frequenz.client.microgrid.metrics import Metric


        await microgrid.initialize(
            "grpc://microgrid.sandbox.api.frequenz.io:62060",
            ResamplerConfig2(resampling_period=timedelta(seconds=1)),
        )

        logical_meter = (
            microgrid.logical_meter()
            .start_formula("#1001 + #1002", Metric.AC_ACTIVE_POWER)
            .new_receiver()
        )

        async for power in logical_meter:
            print(power.value)
        ```
    """

    def __init__(
        self,
        channel_registry: ChannelRegistry,
        resampler_subscription_sender: Sender[ComponentMetricRequest],
    ) -> None:
        """Create a `LogicalMeter` instance.

        !!! note
            `LogicalMeter` instances are not meant to be created directly by users.  Use
            the [`microgrid.logical_meter`][frequenz.sdk.microgrid.logical_meter] method
            for creating `LogicalMeter` instances.

        Args:
            channel_registry: A channel registry instance shared with the resampling
                actor.
            resampler_subscription_sender: A sender for sending metric requests to the
                resampling actor.
        """
        self._channel_registry: ChannelRegistry = channel_registry
        self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
            resampler_subscription_sender
        )

        # Use a randomly generated uuid to create a unique namespace name for the local
        # meter to use when communicating with the resampling actor.
        self._namespace: str = f"logical-meter-{uuid.uuid4()}"
        self._formula_pool: FormulaPool = FormulaPool(
            self._namespace,
            self._channel_registry,
            self._resampler_subscription_sender,
        )

    def start_formula(
        self,
        formula: str,
        metric: Metric,
    ) -> Formula[Quantity]:
        """Start execution of the given formula.

        Formulas can have Component IDs that are preceded by a pound symbol(`#`),
        constant values and these operators: `+`, `-`, `*`, `/`, `(`, `)`.

        These functions are also supported: `COALESCE`, `MAX`, `MIN`.

        For example, the input string: `#20 + #5` is a formula for adding metrics from
        two components with ids 20 and 5.

        A more detailed description of the formula syntax with examples can be found
        [here](https://github.com/frequenz-floss/frequenz-microgrid-formula-engine-rs?tab=readme-ov-file#formula-syntax-overview).

        Args:
            formula: formula to execute.
            metric: The metric to use when fetching receivers from the resampling actor.

        Returns:
            A Formula that applies the formula and streams values.
        """
        return self._formula_pool.from_string(formula, metric)

    @property
    def chp_power(self) -> Formula[Power]:
        """Fetch the CHP power production in the microgrid.

        This formula produces values that are in the Passive Sign Convention (PSC).

        If a formula to calculate CHP power production is not already running, it
        will be started.

        A receiver from the formula can be created using the `new_receiver`
        method.

        Returns:
            A Formula that will calculate and stream CHP power production.
        """
        return self._formula_pool.from_power_formula(
            channel_key="chp_power",
            formula_str=connection_manager.get().component_graph.chp_formula(None),
        )

    async def stop(self) -> None:
        """Stop all formulas."""
        await self._formula_pool.stop()
Attributes¤
chp_power property ¤
chp_power: Formula[Power]

Fetch the CHP power production in the microgrid.

This formula produces values that are in the Passive Sign Convention (PSC).

If a formula to calculate CHP power production is not already running, it will be started.

A receiver from the formula can be created using the new_receiver method.

RETURNS DESCRIPTION
Formula[Power]

A Formula that will calculate and stream CHP power production.
Functions¤
__init__ ¤
__init__(
    channel_registry: ChannelRegistry,
    resampler_subscription_sender: Sender[
        ComponentMetricRequest
    ],
) -> None

Create a LogicalMeter instance.

Note

LogicalMeter instances are not meant to be created directly by users. Use the microgrid.logical_meter method for creating LogicalMeter instances.

PARAMETER DESCRIPTION
channel_registry

A channel registry instance shared with the resampling actor.

TYPE: ChannelRegistry

resampler_subscription_sender

A sender for sending metric requests to the resampling actor.

TYPE: Sender[ComponentMetricRequest]

Source code in src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
58
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
def __init__(
    self,
    channel_registry: ChannelRegistry,
    resampler_subscription_sender: Sender[ComponentMetricRequest],
) -> None:
    """Create a `LogicalMeter` instance.

    !!! note
        `LogicalMeter` instances are not meant to be created directly by users.  Use
        the [`microgrid.logical_meter`][frequenz.sdk.microgrid.logical_meter] method
        for creating `LogicalMeter` instances.

    Args:
        channel_registry: A channel registry instance shared with the resampling
            actor.
        resampler_subscription_sender: A sender for sending metric requests to the
            resampling actor.
    """
    self._channel_registry: ChannelRegistry = channel_registry
    self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
        resampler_subscription_sender
    )

    # Use a randomly generated uuid to create a unique namespace name for the local
    # meter to use when communicating with the resampling actor.
    self._namespace: str = f"logical-meter-{uuid.uuid4()}"
    self._formula_pool: FormulaPool = FormulaPool(
        self._namespace,
        self._channel_registry,
        self._resampler_subscription_sender,
    )
start_formula ¤
start_formula(
    formula: str, metric: Metric
) -> Formula[Quantity]

Start execution of the given formula.

Formulas can have Component IDs that are preceded by a pound symbol(#), constant values and these operators: +, -, *, /, (, ).

These functions are also supported: COALESCE, MAX, MIN.

For example, the input string: #20 + #5 is a formula for adding metrics from two components with ids 20 and 5.

A more detailed description of the formula syntax with examples can be found here.

PARAMETER DESCRIPTION
formula

formula to execute.

TYPE: str

metric

The metric to use when fetching receivers from the resampling actor.

TYPE: Metric

RETURNS DESCRIPTION
Formula[Quantity]

A Formula that applies the formula and streams values.
Source code in src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
 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
def start_formula(
    self,
    formula: str,
    metric: Metric,
) -> Formula[Quantity]:
    """Start execution of the given formula.

    Formulas can have Component IDs that are preceded by a pound symbol(`#`),
    constant values and these operators: `+`, `-`, `*`, `/`, `(`, `)`.

    These functions are also supported: `COALESCE`, `MAX`, `MIN`.

    For example, the input string: `#20 + #5` is a formula for adding metrics from
    two components with ids 20 and 5.

    A more detailed description of the formula syntax with examples can be found
    [here](https://github.com/frequenz-floss/frequenz-microgrid-formula-engine-rs?tab=readme-ov-file#formula-syntax-overview).

    Args:
        formula: formula to execute.
        metric: The metric to use when fetching receivers from the resampling actor.

    Returns:
        A Formula that applies the formula and streams values.
    """
    return self._formula_pool.from_string(formula, metric)
stop async ¤
stop() -> None

Stop all formulas.

Source code in src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
137
138
139
async def stop(self) -> None:
    """Stop all formulas."""
    await self._formula_pool.stop()