Add steam boiler pool

Signed-off-by: Simon Völcker <simon.voelcker@frequenz.com>

Author: simonvoelcker

pyproject.toml

@@ -29,7 +29,7 @@ dependencies = [
   # Make sure to update the mkdocs.yml file when
   # changing the version
   # (plugins.mkdocstrings.handlers.python.import)
-  "frequenz-client-microgrid >= 0.18.1, < 0.19.0",
+  "frequenz-client-microgrid >= 0.18.2, < 0.19.0",
   "frequenz-microgrid-component-graph >= 0.3.4, < 0.4",
   "frequenz-client-common >= 0.3.6, < 0.4.0",
   "frequenz-channels >= 1.6.1, < 2.0.0",

src/frequenz/sdk/timeseries/steam_boiler_pool/__init__.py

@@ -0,0 +1,13 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Interactions with steam boilers."""
+
+from ._result_types import SteamBoilerPoolReport
+from ._steam_boiler_pool import SteamBoilerPool, SteamBoilerPoolError
+
+__all__ = [
+    "SteamBoilerPool",
+    "SteamBoilerPoolError",
+    "SteamBoilerPoolReport",
+]

src/frequenz/sdk/timeseries/steam_boiler_pool/_result_types.py

@@ -0,0 +1,27 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Types for exposing steam boiler pool reports."""
+
+import typing
+
+from frequenz.quantities import Power
+
+from .. import Bounds
+from ..component_pool._component_pool_report import ComponentPoolReport
+
+
+class SteamBoilerPoolReport(ComponentPoolReport, typing.Protocol):
+    """A status report for a steam boiler pool."""
+
+    @property
+    def target_power(self) -> Power | None:
+        """The currently set power for the steam boilers."""
+
+    @property
+    def bounds(self) -> Bounds[Power] | None:
+        """The usable bounds for the steam boilers.
+
+        These bounds are adjusted to any restrictions placed by actors with higher
+        priorities.
+        """

src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py

@@ -0,0 +1,47 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Interactions with pools of steam boilers."""
+
+from frequenz.quantities import Power
+from typing_extensions import override
+
+from ...microgrid import connection_manager
+from ..component_pool import ComponentPool
+from ..formulas import Formula
+from ._result_types import SteamBoilerPoolReport
+from ._steam_boiler_pool_reference_store import SteamBoilerPoolReferenceStore
+
+
+class SteamBoilerPoolError(Exception):
+    """An error that occurred in any of the SteamBoilerPool methods."""
+
+
+class SteamBoilerPool(
+    ComponentPool[SteamBoilerPoolReferenceStore, SteamBoilerPoolReport]
+):
+    """An interface for interaction with pools of steam boilers."""
+
+    @property
+    @override
+    def power(self) -> Formula[Power]:
+        """Fetch the total power for the steam boilers in the pool.
+
+        This formula produces values that are in the Passive Sign Convention (PSC).
+
+        If a formula to calculate steam boiler power 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 the total power of all steam boilers.
+        """
+        return self._pool_ref_store.formula_pool.from_power_formula(
+            "steam_boiler_power",
+            # TODO component graph update needed, still under review
+            connection_manager.get().component_graph.steam_boiler_formula(
+                self._pool_ref_store.component_ids
+            ),
+        )

src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool_reference_store.py

@@ -0,0 +1,53 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Manages shared state/tasks for a set of steam boilers."""
+
+import uuid
+from typing import Type
+
+from frequenz.client.microgrid.component import (
+    Component,
+    SteamBoiler,  # TODO client 0.19.1 needed, under review
+)
+from typing_extensions import override
+
+from ..component_pool._component_pool_reference_store import ComponentPoolReferenceStore
+
+
+class SteamBoilerPoolReferenceStore(ComponentPoolReferenceStore):
+    """A class for maintaining the shared state/tasks for a set of pools of steam boilers.
+
+    This includes ownership of
+    - the formula pool and metric calculators.
+    - the tasks for calculating system bounds for the steam boilers.
+
+    These are independent of the priority of the actors and can be shared between
+    multiple users of the same set of steam boilers.
+
+    They are exposed through the SteamBoilerPool class.
+    """
+
+    @staticmethod
+    def get_component_class() -> Type[Component]:
+        """Class of the component type."""
+        return SteamBoiler
+
+    @staticmethod
+    def get_pool_type_name() -> str:
+        """Name of the pool type, for display purposes."""
+        return "SteamBoilerPool"
+
+    @staticmethod
+    def get_component_type_name_plural() -> str:
+        """Name of the component type, for display purposes."""
+        return "steam boilers"
+
+    @override
+    def get_namespace(self) -> str:
+        """Namespace to use with the data pipeline."""
+        return f"steam-boiler-pool-{uuid.uuid4()}"
+
+    @override
+    def create_bounds_tracker(self) -> None:
+        """Create the bounds tracker for the pool."""