Make AbstractPool generic over ref store and report types.

Extract abstract base class for reports.

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

Author: simonvoelcker

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

@@ -5,8 +5,10 @@
 
 from ._abstract_pool import AbstractPool
 from ._abstract_pool_reference_store import AbstractPoolReferenceStore
+from ._abstract_pool_report import AbstractPoolReport
 
 __all__ = [
     "AbstractPool",
     "AbstractPoolReferenceStore",
+    "AbstractPoolReport",
 ]

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

@@ -7,6 +7,7 @@
 import uuid
 from abc import ABC, abstractmethod
 from collections import abc
+from typing import Generic, TypeVar, cast
 
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.quantities import Power
@@ -15,19 +16,22 @@
 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
 
+from ._abstract_pool_reference_store import AbstractPoolReferenceStore
+from ._abstract_pool_report import AbstractPoolReport
 
-class AbstractPool(ABC):
+RefStoreT = TypeVar("RefStoreT", bound=AbstractPoolReferenceStore)
+ReportT = TypeVar("ReportT", bound=AbstractPoolReport)
+
+
+class AbstractPool(ABC, Generic[RefStoreT, ReportT]):
     """Abstract base class for component pools."""
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
         *,
-        pool_ref_store: AbstractPoolReferenceStore,
+        pool_ref_store: RefStoreT,
         name: str | None,
         priority: int,
     ) -> None:
@@ -97,7 +101,7 @@ def power(self) -> Formula[Power]:
         """
 
     @property
-    def power_status(self) -> ReceiverFetcher[_power_managing._Report]:
+    def power_status(self) -> ReceiverFetcher[ReportT]:
         """Get a receiver to receive new power status reports when they change.
 
         These include
@@ -124,7 +128,7 @@ def power_status(self) -> ReceiverFetcher[_power_managing._Report]:
         )
         channel.resend_latest = True
 
-        return channel
+        return cast(ReceiverFetcher[ReportT], channel)
 
     @property
     def power_distribution_results(self) -> ReceiverFetcher[_power_distributing.Result]:

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

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

src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

@@ -28,11 +28,12 @@
     SoCCalculator,
     TemperatureCalculator,
 )
+from .messages import BatteryPoolReport
 
 # pylint: disable=protected-access
 
 
-class BatteryPool(AbstractPool):
+class BatteryPool(AbstractPool[BatteryPoolReferenceStore, BatteryPoolReport]):
     """An interface for interaction with pools of batteries.
 
     Provides:

src/frequenz/sdk/timeseries/battery_pool/messages.py

@@ -16,11 +16,12 @@
     Success,
 )
 from .._base_types import Bounds
+from ..abstract_pool import AbstractPoolReport
 
 
 # This class is used to expose the generic reports from the PowerManager with specific
 # documentation for the battery pool.
-class BatteryPoolReport(typing.Protocol):
+class BatteryPoolReport(AbstractPoolReport, typing.Protocol):
     """A status report for a battery pool."""
 
     @property

src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py

@@ -10,13 +10,15 @@
 from ...timeseries import Bounds
 from ..abstract_pool import AbstractPool
 from ..formulas import Formula, Formula3Phase
+from ._ev_charger_pool_reference_store import EVChargerPoolReferenceStore
+from ._result_types import EVChargerPoolReport
 
 
 class EVChargerPoolError(Exception):
     """An error that occurred in any of the EVChargerPool methods."""
 
 
-class EVChargerPool(AbstractPool):
+class EVChargerPool(AbstractPool[EVChargerPoolReferenceStore, EVChargerPoolReport]):
     """An interface for interaction with pools of EV Chargers.
 
     Provides:

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

@@ -7,10 +7,11 @@
 
 from frequenz.quantities import Power
 
-from .._base_types import Bounds
+from .. import Bounds
+from ..abstract_pool import AbstractPoolReport
 
 
-class EVChargerPoolReport(typing.Protocol):
+class EVChargerPoolReport(AbstractPoolReport, typing.Protocol):
     """A status report for an EV chargers pool."""
 
     @property

src/frequenz/sdk/timeseries/pv_pool/_pv_pool.py

@@ -11,13 +11,15 @@
 from ...timeseries import Bounds
 from ..abstract_pool import AbstractPool
 from ..formulas import Formula
+from ._pv_pool_reference_store import PVPoolReferenceStore
+from ._result_types import PVPoolReport
 
 
 class PVPoolError(Exception):
     """An error that occurred in any of the PVPool methods."""
 
 
-class PVPool(AbstractPool):
+class PVPool(AbstractPool[PVPoolReferenceStore, PVPoolReport]):
     """An interface for interaction with pools of PV inverters.
 
     Provides:

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

@@ -8,9 +8,10 @@
 from frequenz.quantities import Power
 
 from .._base_types import Bounds
+from ..abstract_pool import AbstractPoolReport
 
 
-class PVPoolReport(typing.Protocol):
+class PVPoolReport(AbstractPoolReport, typing.Protocol):
     """A status report for a PV pool."""
 
     @property