Move consumer and producer formulas out from logical meter

RELEASE_NOTES.md

@@ -44,6 +44,23 @@
     grid_current_recv = grid.current.new_receiver()
     ```
 
+- Consumer and producer power formulas were moved from `microgrid.logical_meter()` to `microgrid.consumer()` and `microgrid.producer()`, respectively.
+
+    Previously,
+
+    ```python
+    logical_meter = microgrid.logical_meter()
+    consumer_power_recv = logical_meter.consumer_power.new_receiver()
+    producer_power_recv = logical_meter.producer_power.new_receiver()
+    ```
+
+    Now,
+
+    ```python
+    consumer_power_recv = microgrid.consumer().power.new_receiver()
+    producer_power_recv = microgrid.producer().power.new_receiver()
+    ```
+
 - The `ComponentGraph.components()` parameters `component_id` and `component_category` were renamed to `component_ids` and `component_categories`, respectively.
 
 - The `GridFrequency.component` property was renamed to `GridFrequency.source`

docs/user-guide/glossary.md

@@ -167,7 +167,7 @@ the component graph are:
 
   - figure out how to calculate high level metrics like
 [`grid_power`][frequenz.sdk.timeseries.grid.Grid.power],
-[`consumer_power`][frequenz.sdk.timeseries.logical_meter.LogicalMeter.consumer_power],
+[`consumer_power`][frequenz.sdk.timeseries.consumer.Consumer.power],
 etc. for a microgrid, using the available components.
   - identify the available {{glossary("battery", "batteries")}} or
     {{glossary("EV charger", "EV chargers")}} at a site that can be controlled.

src/frequenz/sdk/microgrid/__init__.py

@@ -68,8 +68,7 @@
 
 This is the main power consumer at the site of a microgrid, and often the
 {{glossary("load")}} the microgrid is built to support.  The power drawn by the consumer
-is available through
-[`consumer_power`][frequenz.sdk.timeseries.logical_meter.LogicalMeter.consumer_power]
+is available through [`consumer_power`][frequenz.sdk.timeseries.consumer.Consumer.power]
 
 In locations without a consumer, this method streams zero values.
 
@@ -80,7 +79,7 @@
 similarly the total CHP production in a site can be streamed through
 [`chp_power`][frequenz.sdk.timeseries.logical_meter.LogicalMeter.chp_power].  And total
 producer power is available through
-[`producer_power`][frequenz.sdk.timeseries.logical_meter.LogicalMeter.producer_power].
+[`producer_power`][frequenz.sdk.timeseries.producer.Producer.power].
 
 As is the case with the other methods, if PV Arrays or CHPs are not available in a
 microgrid, the corresponding methods stream zero values.
@@ -126,10 +125,12 @@
 from . import _data_pipeline, client, component, connection_manager, metadata
 from ._data_pipeline import (
     battery_pool,
+    consumer,
     ev_charger_pool,
     frequency,
     grid,
     logical_meter,
+    producer,
     voltage,
 )
 
@@ -150,11 +151,13 @@ async def initialize(host: str, port: int, resampler_config: ResamplerConfig) ->
     "initialize",
     "client",
     "component",
+    "consumer",
     "battery_pool",
     "ev_charger_pool",
     "grid",
     "frequency",
     "logical_meter",
     "metadata",
+    "producer",
     "voltage",
 ]

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -46,8 +46,10 @@
     from ..timeseries.battery_pool._battery_pool_reference_store import (
         BatteryPoolReferenceStore,
     )
+    from ..timeseries.consumer import Consumer
     from ..timeseries.ev_charger_pool import EVChargerPool
     from ..timeseries.logical_meter import LogicalMeter
+    from ..timeseries.producer import Producer
 
 
 _REQUEST_RECV_BUFFER_SIZE = 500
@@ -118,18 +120,16 @@ def __init__(
         self._power_managing_actor: _power_managing.PowerManagingActor | None = None
 
         self._logical_meter: LogicalMeter | None = None
+        self._consumer: Consumer | None = None
+        self._producer: Producer | None = None
         self._grid: Grid | None = None
         self._ev_charger_pools: dict[frozenset[int], EVChargerPool] = {}
         self._battery_pools: dict[frozenset[int], BatteryPoolReferenceStore] = {}
         self._frequency_instance: GridFrequency | None = None
         self._voltage_instance: VoltageStreamer | None = None
 
     def frequency(self) -> GridFrequency:
-        """Fetch the grid frequency for the microgrid.
-
-        Returns:
-            The GridFrequency instance.
-        """
+        """Return the grid frequency measuring point."""
         if self._frequency_instance is None:
             self._frequency_instance = GridFrequency(
                 self._data_sourcing_request_sender(),
@@ -139,11 +139,7 @@ def frequency(self) -> GridFrequency:
         return self._frequency_instance
 
     def voltage(self) -> VoltageStreamer:
-        """Fetch the 3-phase voltage for the microgrid.
-
-        Returns:
-            The VoltageStreamer instance.
-        """
+        """Return the 3-phase voltage measuring point."""
         if not self._voltage_instance:
             self._voltage_instance = VoltageStreamer(
                 self._resampling_request_sender(),
@@ -153,13 +149,7 @@ def voltage(self) -> VoltageStreamer:
         return self._voltage_instance
 
     def logical_meter(self) -> LogicalMeter:
-        """Return the logical meter instance.
-
-        If a LogicalMeter instance doesn't exist, a new one is created and returned.
-
-        Returns:
-            A logical meter instance.
-        """
+        """Return the logical meter of the microgrid."""
         from ..timeseries.logical_meter import LogicalMeter
 
         if self._logical_meter is None:
@@ -169,6 +159,28 @@ def logical_meter(self) -> LogicalMeter:
             )
         return self._logical_meter
 
+    def consumer(self) -> Consumer:
+        """Return the consumption measuring point of the microgrid."""
+        from ..timeseries.consumer import Consumer
+
+        if self._consumer is None:
+            self._consumer = Consumer(
+                channel_registry=self._channel_registry,
+                resampler_subscription_sender=self._resampling_request_sender(),
+            )
+        return self._consumer
+
+    def producer(self) -> Producer:
+        """Return the production measuring point of the microgrid."""
+        from ..timeseries.producer import Producer
+
+        if self._producer is None:
+            self._producer = Producer(
+                channel_registry=self._channel_registry,
+                resampler_subscription_sender=self._resampling_request_sender(),
+            )
+        return self._producer
+
     def ev_charger_pool(
         self,
         ev_charger_ids: set[int] | None = None,
@@ -201,13 +213,7 @@ def ev_charger_pool(
         return self._ev_charger_pools[key]
 
     def grid(self) -> Grid:
-        """Return the grid instance.
-
-        If a Grid instance doesn't exist, a new one is created and returned.
-
-        Returns:
-            A Grid instance.
-        """
+        """Return the grid measuring point."""
         if self._grid is None:
             initialize_grid(
                 channel_registry=self._channel_registry,
@@ -419,32 +425,28 @@ async def initialize(resampler_config: ResamplerConfig) -> None:
 
 
 def frequency() -> GridFrequency:
-    """Return the grid frequency.
-
-    Returns:
-        The grid frequency.
-    """
+    """Return the grid frequency measuring point."""
     return _get().frequency()
 
 
 def voltage() -> VoltageStreamer:
-    """Return the 3-phase voltage for the microgrid.
-
-    Returns:
-        The 3-phase voltage.
-    """
+    """Return the 3-phase voltage measuring point."""
     return _get().voltage()
 
 
 def logical_meter() -> LogicalMeter:
-    """Return the logical meter instance.
+    """Return the logical meter of the microgrid."""
+    return _get().logical_meter()
 
-    If a LogicalMeter instance doesn't exist, a new one is created and returned.
 
-    Returns:
-        A logical meter instance.
-    """
-    return _get().logical_meter()
+def consumer() -> Consumer:
+    """Return the [`Consumption`][frequenz.sdk.timeseries.consumer.Consumer] measuring point."""
+    return _get().consumer()
+
+
+def producer() -> Producer:
+    """Return the [`Production`][frequenz.sdk.timeseries.producer.Producer] measuring point."""
+    return _get().producer()
 
 
 def ev_charger_pool(ev_charger_ids: set[int] | None = None) -> EVChargerPool:
@@ -492,11 +494,7 @@ def battery_pool(
 
 
 def grid() -> Grid:
-    """Return the grid instance.
-
-    Returns:
-        The Grid instance.
-    """
+    """Return the grid measuring point."""
     return _get().grid()
 
 

src/frequenz/sdk/timeseries/consumer.py

@@ -0,0 +1,100 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""The logical component for calculating high level consumer metrics for a microgrid."""
+
+import uuid
+
+from frequenz.channels import Sender
+
+from ..actor import ChannelRegistry, ComponentMetricRequest
+from ._quantities import Power
+from .formula_engine import FormulaEngine
+from .formula_engine._formula_engine_pool import FormulaEnginePool
+from .formula_engine._formula_generators import ConsumerPowerFormula
+
+
+class Consumer:
+    """Calculate high level consumer metrics in a microgrid.
+
+    Under normal circumstances this is expected to correspond to the gross
+    consumption of the site excluding active parts and battery.
+
+    Consumer 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
+        `Consumer` instances are not meant to be created directly by users.
+        Use the [`microgrid.consumer`][frequenz.sdk.microgrid.consumer] method
+        for creating `Consumer` instances.
+
+    Example:
+        ```python
+        from datetime import timedelta
+
+        from frequenz.sdk import microgrid
+        from frequenz.sdk.timeseries import ResamplerConfig
+
+        await microgrid.initialize(
+            "127.0.0.1",
+            50051,
+            ResamplerConfig(resampling_period=timedelta(seconds=1.0))
+        )
+
+        consumer = microgrid.consumer()
+
+        # Get a receiver for a builtin formula
+        consumer_power_recv = consumer.power.new_receiver()
+        async for consumer_power_sample in consumer_power_recv:
+            print(consumer_power_sample)
+        ```
+    """
+
+    _formula_pool: FormulaEnginePool
+    """The formula engine pool to generate consumer metrics."""
+
+    def __init__(
+        self,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
+    ) -> None:
+        """Initialize the consumer formula generator.
+
+        Args:
+            channel_registry: The channel registry to use for the consumer.
+            resampler_subscription_sender: The sender to use for resampler subscriptions.
+        """
+        namespace = f"consumer-{uuid.uuid4()}"
+        self._formula_pool = FormulaEnginePool(
+            namespace,
+            channel_registry,
+            resampler_subscription_sender,
+        )
+
+    @property
+    def power(self) -> FormulaEngine[Power]:
+        """Fetch the consumer power for the microgrid.
+
+        This formula produces values that are in the Passive Sign Convention (PSC).
+
+        It will start the formula engine to calculate consumer power if it is
+        not already running.
+
+        A receiver from the formula engine can be created using the
+        `new_receiver` method.
+
+        Returns:
+            A FormulaEngine that will calculate and stream consumer power.
+        """
+        engine = self._formula_pool.from_power_formula_generator(
+            "consumer_power",
+            ConsumerPowerFormula,
+        )
+        assert isinstance(engine, FormulaEngine)
+        return engine
+
+    async def stop(self) -> None:
+        """Stop all formula engines."""
+        await self._formula_pool.stop()

src/frequenz/sdk/timeseries/formula_engine/_formula_engine.py

@@ -232,7 +232,7 @@ class FormulaEngine(
 
     They are used in the SDK to calculate and stream metrics like
     [`grid_power`][frequenz.sdk.timeseries.grid.Grid.power],
-    [`consumer_power`][frequenz.sdk.timeseries.logical_meter.LogicalMeter.consumer_power],
+    [`consumer_power`][frequenz.sdk.timeseries.consumer.Consumer.power],
     etc., which are building blocks of the
     [Frequenz SDK Microgrid Model][frequenz.sdk.microgrid--frequenz-sdk-microgrid-model].
 
@@ -332,10 +332,8 @@ def from_receiver(
             from frequenz.sdk.timeseries import Power
 
             async def run() -> None:
-                producer_power_engine = microgrid.logical_meter().producer_power
-                consumer_power_recv = (
-                    microgrid.logical_meter().consumer_power.new_receiver()
-                )
+                producer_power_engine = microgrid.producer().power
+                consumer_power_recv = microgrid.consumer().power.new_receiver()
 
                 excess_power_recv = (
                     (