Add a steam boiler pool (#1429)

Steam boilers are controllable electrical loads the SDK could not
monitor or
manage. This adds a steam boiler pool with power readings and power
control,
following the structure of the EV charger and PV pools.

## Changes

- New public API: `microgrid.new_steam_boiler_pool()` returning a
  `SteamBoilerPool` with `power`, `power_status`, and a consumption-only
  `propose_power` (negative power is rejected).
- `SteamBoilerData` component data type and data sourcing support.
- Power manager wiring using the plain matryoshka algorithm, with
  `DefaultPower.ZERO` when no actor is proposing.
- `SteamBoilerManager` splits the target power equally across the
working
boilers, keeping every allocation inside the boiler's inclusion bounds:
a
share below a boiler's minimum operating power is raised to the minimum,
or
the boiler is kept off. Like the PV manager, it subtracts the measured
draw
  of unreachable boilers from the target.
- `SteamBoilerStatusTracker` for per-boiler health, mirroring the PV
tracker.

## Worth a look

- The single-pass allocation reports power as excess when a high-minimum
  boiler can't be afforded instead of topping up earlier boilers
  (`test_unaffordable_minimum_strands_excess` pins this).
- With no usable boilers a request is dropped with only an error log —
the
same behavior as the PV and EV managers; a follow-up could make all
three
  send an `Error` result the way the battery manager does.

Author: shsms

RELEASE_NOTES.md

@@ -12,7 +12,7 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* Added `SteamBoilerPool` for monitoring and controlling pools of steam boilers, available via `microgrid.new_steam_boiler_pool()`.
 
 ## Bug Fixes
 

src/frequenz/sdk/microgrid/__init__.py

@@ -138,14 +138,30 @@
 which accepts values in the {{glossary("psc", "Passive Sign Convention")}} and supports
 only charging.
 
+## Steam Boilers
+
+The [`steam_boiler_pool`][frequenz.sdk.microgrid.new_steam_boiler_pool] offers a
+[`power`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.power] method that
+streams the total power measured for all the steam boilers at a site.
+
+The `steam_boiler_pool` also provides available power bounds through the
+[`power_status`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.power_status]
+method.
+
+The `steam_boiler_pool` also provides a control method
+[`propose_power`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.propose_power],
+which accepts values in the {{glossary("psc", "Passive Sign Convention")}} and supports
+only consumption.
+
 # Component pools
 
 The SDK provides a unified interface for interacting with sets of Batteries, EV
-chargers and PV arrays, through their corresponding `Pool`s.
+chargers, PV arrays and steam boilers, through their corresponding `Pool`s.
 
 * [Battery pool][frequenz.sdk.microgrid.new_battery_pool]
 * [EV charger pool][frequenz.sdk.microgrid.new_ev_charger_pool]
 * [PV pool][frequenz.sdk.microgrid.new_pv_pool]
+* [Steam boiler pool][frequenz.sdk.microgrid.new_steam_boiler_pool]
 
 All of them provide support for streaming aggregated data and for setting the
 power values of the components.
@@ -366,6 +382,7 @@
 | Batteries          | Zero                                                 |
 | PV                 | Max production (Min power according to PSC)          |
 | EV Chargers        | Max consumption (Max power according to PSC)         |
+| Steam Boilers      | Zero                                                 |
 """  # noqa: D205, D400, E501
 
 from datetime import timedelta
@@ -382,6 +399,7 @@
     new_battery_pool,
     new_ev_charger_pool,
     new_pv_pool,
+    new_steam_boiler_pool,
     producer,
     voltage_per_phase,
 )
@@ -442,6 +460,7 @@ async def initialize(
     "new_battery_pool",
     "new_ev_charger_pool",
     "new_pv_pool",
+    "new_steam_boiler_pool",
     "producer",
     "voltage_per_phase",
 ]

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -19,7 +19,12 @@
 
 from frequenz.channels import Broadcast, Sender
 from frequenz.client.common.microgrid.components import ComponentId
-from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter
+from frequenz.client.microgrid.component import (
+    Battery,
+    EvCharger,
+    SolarInverter,
+    SteamBoiler,
+)
 
 from .._internal._channels import ChannelRegistry
 from ..actor._actor import Actor
@@ -52,6 +57,10 @@
     from ..timeseries.producer import Producer
     from ..timeseries.pv_pool import PVPool
     from ..timeseries.pv_pool._pv_pool_reference_store import PVPoolReferenceStore
+    from ..timeseries.steam_boiler_pool import SteamBoilerPool
+    from ..timeseries.steam_boiler_pool._steam_boiler_pool_reference_store import (
+        SteamBoilerPoolReferenceStore,
+    )
 
 _logger = logging.getLogger(__name__)
 
@@ -131,6 +140,13 @@ def __init__(
             # https://github.com/frequenz-floss/frequenz-sdk-python/issues/1285
             component_class=SolarInverter,
         )
+        self._steam_boiler_power_wrapper = PowerWrapper(
+            self._channel_registry,
+            api_power_request_timeout=api_power_request_timeout,
+            power_manager_algorithm=PowerManagerAlgorithm.MATRYOSHKA,
+            default_power=DefaultPower.ZERO,
+            component_class=SteamBoiler,
+        )
 
         self._logical_meter: LogicalMeter | None = None
         self._consumer: Consumer | None = None
@@ -145,6 +161,9 @@ def __init__(
         self._pv_pool_reference_stores: dict[
             frozenset[ComponentId], PVPoolReferenceStore
         ] = {}
+        self._steam_boiler_pool_reference_stores: dict[
+            frozenset[ComponentId], SteamBoilerPoolReferenceStore
+        ] = {}
         self._frequency_instance: GridFrequency | None = None
         self._voltage_instance: VoltageStreamer | None = None
 
@@ -476,6 +495,83 @@ def new_battery_pool(
             priority=priority,
         )
 
+    def new_steam_boiler_pool(
+        self,
+        *,
+        priority: int,
+        component_ids: abc.Set[ComponentId] | None = None,
+        name: str | None = None,
+    ) -> SteamBoilerPool:
+        """Return a new `SteamBoilerPool` instance for the given ids.
+
+        If a `SteamBoilerPoolReferenceStore` instance for the given steam boiler ids
+        doesn't exist, a new one is created and used for creating the
+        `SteamBoilerPool`.
+
+        Args:
+            priority: The priority of the actor making the call.
+            component_ids: Optional set of IDs of steam boilers to be managed by the
+                SteamBoilerPool.
+            name: An optional name used to identify this instance of the pool or a
+                corresponding actor in the logs.
+
+        Returns:
+            A SteamBoilerPool instance.
+        """
+        from ..timeseries.steam_boiler_pool import SteamBoilerPool
+        from ..timeseries.steam_boiler_pool._steam_boiler_pool_reference_store import (
+            SteamBoilerPoolReferenceStore,
+        )
+
+        if not self._steam_boiler_power_wrapper.started:
+            self._steam_boiler_power_wrapper.start()
+
+        # We use frozenset to make a hashable key from the input set.
+        ref_store_key: frozenset[ComponentId] = frozenset()
+        if component_ids is not None:
+            ref_store_key = frozenset(component_ids)
+
+        pool_key = f"{ref_store_key}-{priority}"
+        if pool_key in self._known_pool_keys:
+            _logger.warning(
+                "A SteamBoilerPool instance was already created for steam_boiler_ids=%s "
+                "and priority=%s using `microgrid.steam_boiler_pool(...)`."
+                "\n  Hint: If the multiple instances are created from the same actor, "
+                "consider reusing the same instance."
+                "\n  Hint: If the instances are created from different actors, "
+                "consider using different priorities to distinguish them.",
+                component_ids,
+                priority,
+            )
+        else:
+            self._known_pool_keys.add(pool_key)
+
+        if ref_store_key not in self._steam_boiler_pool_reference_stores:
+            self._steam_boiler_pool_reference_stores[ref_store_key] = (
+                SteamBoilerPoolReferenceStore(
+                    channel_registry=self._channel_registry,
+                    resampler_subscription_sender=self._resampling_request_sender(),
+                    status_receiver=self._steam_boiler_power_wrapper.status_channel.new_receiver(
+                        limit=1
+                    ),
+                    power_manager_requests_sender=(
+                        self._steam_boiler_power_wrapper.proposal_channel.new_sender()
+                    ),
+                    power_manager_bounds_subs_sender=(
+                        self._steam_boiler_power_wrapper.bounds_subscription_channel.new_sender()
+                    ),
+                    power_distribution_results_fetcher=(
+                        self._steam_boiler_power_wrapper.distribution_results_fetcher()
+                    ),
+                    component_ids=component_ids,
+                )
+            )
+        return SteamBoilerPool(
+            pool_ref_store=self._steam_boiler_pool_reference_stores[ref_store_key],
+            name=name,
+            priority=priority,
+        )
+
     def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]:
         """Return a Sender for sending requests to the data sourcing actor.
 
@@ -532,12 +628,15 @@ async def _stop(self) -> None:
         await self._battery_power_wrapper.stop()
         await self._ev_power_wrapper.stop()
         await self._pv_power_wrapper.stop()
+        await self._steam_boiler_power_wrapper.stop()
         for pool in self._battery_pool_reference_stores.values():
             await pool.stop()
         for evpool in self._ev_charger_pool_reference_stores.values():
             await evpool.stop()
         for pvpool in self._pv_pool_reference_stores.values():
             await pvpool.stop()
+        for steam_boiler_pool in self._steam_boiler_pool_reference_stores.values():
+            await steam_boiler_pool.stop()
 
 
 _DATA_PIPELINE: _DataPipeline | None = None
@@ -724,6 +823,45 @@ def new_pv_pool(
     return _get().new_pv_pool(priority=priority, component_ids=component_ids, name=name)
 
 
+def new_steam_boiler_pool(
+    *,
+    priority: int,
+    component_ids: abc.Set[ComponentId] | None = None,
+    name: str | None = None,
+) -> SteamBoilerPool:
+    """Return a new `SteamBoilerPool` instance for the given parameters.
+
+    The priority value is used to resolve conflicts when multiple actors are trying to
+    propose different power values for the same set of steam boilers.
+
+    !!! note
+        When specifying priority, bigger values indicate higher priority.
+
+        It is recommended to reuse the same instance of the `SteamBoilerPool` within the same
+        actor, unless they are managing different sets of steam boilers.
+
+        In deployments with multiple actors managing the same set of steam boilers, it is
+        recommended to use different priorities to distinguish between them.  If not,
+        a random prioritization will be imposed on them to resolve conflicts, which may
+        lead to unexpected behavior like longer duration to converge on the desired
+        power.
+
+    Args:
+        priority: The priority of the actor making the call.
+        component_ids: Optional set of IDs of steam boilers to be managed by the
+            `SteamBoilerPool`. If not specified, all steam boilers available in the component
+            graph are used.
+        name: An optional name used to identify this instance of the pool or a
+            corresponding actor in the logs.
+
+    Returns:
+        A `SteamBoilerPool` instance.
+    """
+    return _get().new_steam_boiler_pool(
+        priority=priority, component_ids=component_ids, name=name
+    )
+
+
 def grid() -> Grid:
     """Return the grid measuring point."""
     return _get().grid()

src/frequenz/sdk/microgrid/_data_sourcing/microgrid_api_source.py

@@ -24,6 +24,7 @@
     EVChargerData,
     InverterData,
     MeterData,
+    SteamBoilerData,
     TransitionalMetric,
 )
 from ._component_metric_request import ComponentMetricRequest
@@ -152,6 +153,38 @@
     Metric.AC_REACTIVE_POWER_PHASE_3: lambda msg: msg.reactive_power_per_phase[2],
 }
 
+_STEAM_BOILER_DATA_METHODS: dict[
+    Metric | TransitionalMetric, Callable[[SteamBoilerData], float]
+] = {
+    Metric.AC_ACTIVE_POWER: lambda msg: msg.active_power,
+    Metric.AC_ACTIVE_POWER_PHASE_1: lambda msg: msg.active_power_per_phase[0],
+    Metric.AC_ACTIVE_POWER_PHASE_2: lambda msg: msg.active_power_per_phase[1],
+    Metric.AC_ACTIVE_POWER_PHASE_3: lambda msg: msg.active_power_per_phase[2],
+    TransitionalMetric.ACTIVE_POWER_INCLUSION_LOWER_BOUND: lambda msg: (
+        msg.active_power_inclusion_lower_bound
+    ),
+    TransitionalMetric.ACTIVE_POWER_EXCLUSION_LOWER_BOUND: lambda msg: (
+        msg.active_power_exclusion_lower_bound
+    ),
+    TransitionalMetric.ACTIVE_POWER_EXCLUSION_UPPER_BOUND: lambda msg: (
+        msg.active_power_exclusion_upper_bound
+    ),
+    TransitionalMetric.ACTIVE_POWER_INCLUSION_UPPER_BOUND: lambda msg: (
+        msg.active_power_inclusion_upper_bound
+    ),
+    Metric.AC_CURRENT_PHASE_1: lambda msg: msg.current_per_phase[0],
+    Metric.AC_CURRENT_PHASE_2: lambda msg: msg.current_per_phase[1],
+    Metric.AC_CURRENT_PHASE_3: lambda msg: msg.current_per_phase[2],
+    Metric.AC_VOLTAGE_PHASE_1_N: lambda msg: msg.voltage_per_phase[0],
+    Metric.AC_VOLTAGE_PHASE_2_N: lambda msg: msg.voltage_per_phase[1],
+    Metric.AC_VOLTAGE_PHASE_3_N: lambda msg: msg.voltage_per_phase[2],
+    Metric.AC_FREQUENCY: lambda msg: msg.frequency,
+    Metric.AC_REACTIVE_POWER: lambda msg: msg.reactive_power,
+    Metric.AC_REACTIVE_POWER_PHASE_1: lambda msg: msg.reactive_power_per_phase[0],
+    Metric.AC_REACTIVE_POWER_PHASE_2: lambda msg: msg.reactive_power_per_phase[1],
+    Metric.AC_REACTIVE_POWER_PHASE_3: lambda msg: msg.reactive_power_per_phase[2],
+}
+
 
 class MicrogridApiSource:
     """Fetches requested metrics from the Microgrid API.
@@ -306,6 +339,31 @@ async def _check_chp_request(
                 connection_manager.get().api_client, comp_id
             )
 
+    async def _check_steam_boiler_request(
+        self,
+        comp_id: ComponentId,
+        requests: dict[Metric | TransitionalMetric, list[ComponentMetricRequest]],
+    ) -> None:
+        """Check if the requests are valid steam boiler metrics.
+
+        Raises:
+            ValueError: if the requested metric is not available for steam boilers.
+
+        Args:
+            comp_id: The id of the requested component.
+            requests: A list of metric requests received from external actors
+                for the given steam boiler.
+        """
+        for metric in requests:
+            if metric not in _STEAM_BOILER_DATA_METHODS:
+                err = f"Unknown metric {metric} for steam boiler id {comp_id}"
+                _logger.error(err)
+                raise ValueError(err)
+        if comp_id not in self.comp_data_receivers:
+            self.comp_data_receivers[comp_id] = SteamBoilerData.subscribe(
+                connection_manager.get().api_client, comp_id
+            )
+
     async def _check_meter_request(
         self,
         comp_id: ComponentId,
@@ -362,6 +420,8 @@ async def _check_requested_component_and_metrics(
             await self._check_meter_request(comp_id, requests)
         elif category == ComponentCategory.CHP:
             await self._check_chp_request(comp_id, requests)
+        elif category == ComponentCategory.STEAM_BOILER:
+            await self._check_steam_boiler_request(comp_id, requests)
         else:
             err = f"Unknown component category {category}"
             _logger.error(err)
@@ -393,6 +453,8 @@ def _get_data_extraction_method(
             return _EV_CHARGER_DATA_METHODS[metric]
         if category == ComponentCategory.CHP:
             return _CHP_DATA_METHODS[metric]
+        if category == ComponentCategory.STEAM_BOILER:
+            return _STEAM_BOILER_DATA_METHODS[metric]
         err = f"Unknown component category {category}"
         _logger.error(err)
         raise ValueError(err)