Extract abstract base classes from pools and pool ref stores

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

Author: simonvoelcker

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

@@ -0,0 +1,12 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Manage a pool of components."""
+
+from ._abstract_pool import AbstractPool
+from ._abstract_pool_reference_store import AbstractPoolReferenceStore
+
+__all__ = [
+    "AbstractPool",
+    "AbstractPoolReferenceStore",
+]

src/frequenz/sdk/timeseries/abstract_pool/_abstract_pool.py

@@ -0,0 +1,156 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Abstract base class for component pools."""
+
+import asyncio
+import uuid
+from abc import ABC, abstractmethod
+from collections import abc
+
+from frequenz.client.common.microgrid.components import ComponentId
+from frequenz.quantities import Power
+
+from frequenz.sdk._internal._channels import MappingReceiverFetcher, ReceiverFetcher
+from frequenz.sdk.microgrid import _power_distributing, _power_managing
+from frequenz.sdk.timeseries import Bounds
+from frequenz.sdk.timeseries._base_types import SystemBounds
+from frequenz.sdk.timeseries.abstract_pool._abstract_pool_reference_store import (
+    AbstractPoolReferenceStore,
+)
+from frequenz.sdk.timeseries.formulas import Formula
+
+
+class AbstractPool(ABC):
+    """Abstract base class for component pools."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        *,
+        pool_ref_store: AbstractPoolReferenceStore,
+        name: str | None,
+        priority: int,
+    ) -> None:
+        """Create an `AbstractPool` instance.
+
+        Args:
+            pool_ref_store: The pool reference store instance.
+            name: An optional name used to identify this instance of the pool or a
+                corresponding actor in the logs.
+            priority: The priority of the actor using this wrapper.
+        """
+        self._pool_ref_store = pool_ref_store
+        unique_id = str(uuid.uuid4())
+        self._source_id = unique_id if name is None else f"{name}-{unique_id}"
+        self._priority = priority
+
+    @property
+    def component_ids(self) -> abc.Set[ComponentId]:
+        """Return component IDs of all component IDs managed by this pool.
+
+        Returns:
+            Set of managed component IDs.
+        """
+        return self._pool_ref_store.component_ids
+
+    async def propose_power(
+        self,
+        power: Power | None,
+        bounds: Bounds[Power | None] = Bounds(None, None),
+    ) -> None:
+        """Send a proposal to the power manager for the pool's underlying components.
+
+        This proposal is for the maximum power that can be set for the components in
+        the pool. The actual production or consumption might be lower.
+
+        Details on how the power manager handles proposals can be found in the
+        [Microgrid][frequenz.sdk.microgrid--setting-power] documentation.
+
+        Args:
+            power: The power to propose.  If `None`,
+                this proposal will not have any effect on the target power, unless
+                bounds are specified.  When specified without bounds, bounds for lower
+                priority actors will be shifted by this power.  If both are `None`, it
+                is equivalent to not having a proposal or withdrawing a previous one.
+            bounds: The power bounds for the proposal. When specified, these bounds will
+                limit the bounds for lower priority actors.
+        """
+        await self._pool_ref_store.power_manager_requests_sender.send(
+            _power_managing.Proposal(
+                source_id=self._source_id,
+                preferred_power=power,
+                bounds=bounds,
+                component_ids=self._pool_ref_store.component_ids,
+                priority=self._priority,
+                creation_time=asyncio.get_running_loop().time(),
+            )
+        )
+
+    @property
+    @abstractmethod
+    def power(self) -> Formula[Power]:
+        """Fetch the total power for the components in the pool.
+
+        Returns:
+            A Formula that will calculate and stream the total power of all
+            components in the pool.
+        """
+
+    @property
+    def power_status(self) -> ReceiverFetcher[_power_managing._Report]:
+        """Get a receiver to receive new power status reports when they change.
+
+        These include
+          - the current inclusion/exclusion bounds available for the pool's priority,
+          - the current target power for the pool's set of components,
+          - the result of the last distribution request for the pool's set of components,.
+
+        Returns:
+            A receiver that will stream power status reports for the pool's priority.
+        """
+        sub = _power_managing.ReportRequest(
+            source_id=self._source_id,
+            priority=self._priority,
+            component_ids=self._pool_ref_store.component_ids,
+        )
+        self._pool_ref_store.power_bounds_subs[sub.get_channel_name()] = (
+            asyncio.create_task(
+                self._pool_ref_store.power_manager_bounds_subs_sender.send(sub)
+            )
+        )
+        channel = self._pool_ref_store.channel_registry.get_or_create(
+            _power_managing._Report,  # pylint: disable=protected-access
+            sub.get_channel_name(),
+        )
+        channel.resend_latest = True
+
+        return channel
+
+    @property
+    def power_distribution_results(self) -> ReceiverFetcher[_power_distributing.Result]:
+        """Get a receiver to receive power distribution results.
+
+        Returns:
+            A receiver that will stream power distribution results for the pool's set of
+            components.
+        """
+        return MappingReceiverFetcher(
+            self._pool_ref_store.power_distribution_results_fetcher,
+            lambda recv: recv.filter(
+                lambda x: x.request.component_ids == self._pool_ref_store.component_ids
+            ),
+        )
+
+    @property
+    def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
+        """Return a receiver fetcher for the system power bounds."""
+        return self._pool_ref_store.bounds_channel
+
+    async def stop(self) -> None:
+        """Stop all tasks and channels owned by the pool."""
+        # This was closing the pool_ref_store, which is not correct, because those are
+        # shared.
+        #
+        # This method will do until we have a mechanism to track the resources created
+        # through it. It can also eventually cleanup the pool_ref_store, when it is
+        # holding the last reference to it.

src/frequenz/sdk/timeseries/abstract_pool/_abstract_pool_reference_store.py

@@ -0,0 +1,126 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Abstract base class for pool reference stores."""
+
+import asyncio
+from abc import ABC, abstractmethod
+from collections import abc
+from typing import Type
+
+from frequenz.channels import Broadcast, Receiver, Sender
+from frequenz.client.common.microgrid.components import ComponentId
+from frequenz.client.microgrid.component import Component
+
+from frequenz.sdk._internal._channels import ChannelRegistry, ReceiverFetcher
+from frequenz.sdk.actor import BackgroundService
+from frequenz.sdk.microgrid import connection_manager
+from frequenz.sdk.microgrid._data_sourcing import ComponentMetricRequest
+from frequenz.sdk.microgrid._power_distributing import ComponentPoolStatus, Result
+from frequenz.sdk.microgrid._power_managing import Proposal, ReportRequest
+from frequenz.sdk.timeseries._base_types import SystemBounds
+from frequenz.sdk.timeseries.formulas._formula_pool import FormulaPool
+
+
+class AbstractPoolReferenceStore(ABC):
+    """Abstract base class for pool reference stores."""
+
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        *,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
+        status_receiver: Receiver[ComponentPoolStatus],
+        power_manager_requests_sender: Sender[Proposal],
+        power_manager_bounds_subs_sender: Sender[ReportRequest],
+        power_distribution_results_fetcher: ReceiverFetcher[Result],
+        component_ids: abc.Set[ComponentId] | None = None,
+    ):
+        """Initialize this instance.
+
+        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.
+            status_receiver: A receiver that streams the status of the components in
+                the pool.
+            power_manager_requests_sender: A Channel sender for sending power
+                requests to the power managing actor.
+            power_manager_bounds_subs_sender: A Channel sender for sending power bounds
+                subscription requests to the power managing actor.
+            power_distribution_results_fetcher: A ReceiverFetcher for the results from
+                the power distributing actor.
+            component_ids: An optional list of component_ids belonging to this pool.  If
+                not specified, IDs of all components of the components type of this pool
+                in the microgrid will be fetched from the component graph.
+
+        Raises:
+            ValueError: If any of the provided component_ids are not of correct type or
+                are unknown to the component graph.
+        """
+        self.channel_registry = channel_registry
+        self.resampler_subscription_sender = resampler_subscription_sender
+        self.status_receiver = status_receiver
+        self.power_manager_requests_sender = power_manager_requests_sender
+        self.power_manager_bounds_subs_sender = power_manager_bounds_subs_sender
+        self.power_distribution_results_fetcher = power_distribution_results_fetcher
+
+        graph = connection_manager.get().component_graph
+        all_components = frozenset(
+            {
+                inv.id
+                for inv in graph.components(matching_types=self.get_component_class())
+            }
+        )
+
+        if component_ids is not None:
+            self.component_ids: frozenset[ComponentId] = frozenset(component_ids)
+            if not self.component_ids.issubset(all_components):
+                unknown_ids = self.component_ids - all_components
+                raise ValueError(
+                    f"Unable to create {self.get_pool_type_name()}. These component IDs "
+                    + f"are either not {self.get_component_type_name_plural()} or are unknown: "
+                    + f"{unknown_ids}"
+                )
+        else:
+            self.component_ids = all_components
+
+        self.power_bounds_subs: dict[str, asyncio.Task[None]] = {}
+
+        self.namespace = self.get_namespace()
+        self.formula_pool = FormulaPool(
+            self.namespace,
+            self.channel_registry,
+            self.resampler_subscription_sender,
+        )
+        self.bounds_channel: Broadcast[SystemBounds] = Broadcast(
+            name=f"System Bounds for {self.get_component_type_name_plural()}: {self.component_ids}",
+            resend_latest=True,
+        )
+
+        self.bounds_tracker: BackgroundService | None = None
+        self.create_bounds_tracker()
+
+    @staticmethod
+    @abstractmethod
+    def get_component_class() -> Type[Component]:
+        """Class of the component type."""
+
+    @staticmethod
+    @abstractmethod
+    def get_pool_type_name() -> str:
+        """Name of the pool type, for display purposes."""
+
+    @staticmethod
+    @abstractmethod
+    def get_component_type_name_plural() -> str:
+        """Name of the component type, for display purposes."""
+
+    @abstractmethod
+    def get_namespace(self) -> str:
+        """Namespace to use with the data pipeline."""
+
+    @abstractmethod
+    def create_bounds_tracker(self) -> None:
+        """Create the bounds tracker for the pool."""