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.actor import ResamplerConfig
from frequenz.client.microgrid import ComponentMetricId


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

logical_meter = (
    microgrid.logical_meter()
    .start_formula("#1001 + #1002", ComponentMetricId.ACTIVE_POWER)
    .new_receiver()
)

async for power in logical_meter:
    print(power.value)
Source code in frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
 19
 20
 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
140
141
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.actor import ResamplerConfig
        from frequenz.client.microgrid import ComponentMetricId


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

        logical_meter = (
            microgrid.logical_meter()
            .start_formula("#1001 + #1002", ComponentMetricId.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 = f"logical-meter-{uuid.uuid4()}"
        self._formula_pool = FormulaEnginePool(
            self._namespace,
            self._channel_registry,
            self._resampler_subscription_sender,
        )

    def start_formula(
        self,
        formula: str,
        component_metric_id: ComponentMetricId,
        *,
        nones_are_zeros: bool = False,
    ) -> FormulaEngine[Quantity]:
        """Start execution of the given formula.

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

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

        Args:
            formula: formula to execute.
            component_metric_id: The metric ID to use when fetching receivers from the
                resampling actor.
            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
                False, the returned value will be a None.

        Returns:
            A FormulaEngine that applies the formula and streams values.
        """
        return self._formula_pool.from_string(
            formula, component_metric_id, nones_are_zeros=nones_are_zeros
        )

    @property
    def chp_power(self) -> FormulaEngine[Power]:
        """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_power_formula_generator(
            "chp_power",
            CHPPowerFormula,
        )
        assert isinstance(engine, FormulaEngine)
        return engine

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

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 DESCRIPTION
FormulaEngine[Power]

A FormulaEngine 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 frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
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
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 = f"logical-meter-{uuid.uuid4()}"
    self._formula_pool = FormulaEnginePool(
        self._namespace,
        self._channel_registry,
        self._resampler_subscription_sender,
    )
start_formula ¤
start_formula(
    formula: str,
    component_metric_id: ComponentMetricId,
    *,
    nones_are_zeros: bool = False
) -> FormulaEngine[Quantity]

Start execution of the given formula.

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

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

PARAMETER DESCRIPTION
formula

formula to execute.

TYPE: str

component_metric_id

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

TYPE: ComponentMetricId

nones_are_zeros

Whether to treat None values from the stream as 0s. If False, the returned value will be a None.

TYPE: bool DEFAULT: False

RETURNS DESCRIPTION
FormulaEngine[Quantity]

A FormulaEngine that applies the formula and streams values.
Source code in frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
 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
def start_formula(
    self,
    formula: str,
    component_metric_id: ComponentMetricId,
    *,
    nones_are_zeros: bool = False,
) -> FormulaEngine[Quantity]:
    """Start execution of the given formula.

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

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

    Args:
        formula: formula to execute.
        component_metric_id: The metric ID to use when fetching receivers from the
            resampling actor.
        nones_are_zeros: Whether to treat None values from the stream as 0s.  If
            False, the returned value will be a None.

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

Stop all formula engines.

Source code in frequenz/sdk/timeseries/logical_meter/_logical_meter.py 
139
140
141
async def stop(self) -> None:
    """Stop all formula engines."""
    await self._formula_pool.stop()