Cleanup *Report classes as simple protocols with only methods (#981)

This was required to remove the explicit casting of the `_Report` type
to the `*PoolReport` protocol types.

Closes #823

Author: shsms

src/frequenz/sdk/actor/_power_managing/_base_classes.py

@@ -46,60 +46,8 @@ def get_channel_name(self) -> str:
         return f"power_manager.report.{self.component_ids=}.{self.priority=}"
 
 
-class Report(typing.Protocol):
-    """Current PowerManager report for a set of components.
-
-    This protocol can be specialized by different component pools to provide more
-    specific details and documentation for the reports.
-    """
-
-    @property
-    def bounds(self) -> timeseries.Bounds[Power] | None:
-        """The bounds for the components.
-
-        These bounds are adjusted to any restrictions placed by actors with higher
-        priorities.
-
-        There might be exclusion zones within these bounds. If necessary, the
-        `adjust_to_bounds` method may be used to check if a desired power value fits the
-        bounds, or to get the closest possible power values that do fit the bounds.
-        """
-
-    @abc.abstractmethod
-    def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
-        """Adjust a power value to the bounds.
-
-        This method can be used to adjust a desired power value to the power bounds
-        available to the actor.
-
-        If the given power value falls within the usable bounds, it will be returned
-        unchanged.
-
-        If it falls outside the usable bounds, the closest possible value on the
-        corresponding side will be returned.  For example, if the given power is lower
-        than the lowest usable power, only the lowest usable power will be returned, and
-        similarly for the highest usable power.
-
-        If the given power falls within an exclusion zone that's contained within the
-        usable bounds, the closest possible power values on both sides will be returned.
-
-        !!! note
-            It is completely optional to use this method to adjust power values before
-            proposing them, because the PowerManager will do this automatically.  This
-            method is provided for convenience, and for granular control when there are
-            two possible power values, both of which fall within the available bounds.
-
-        Args:
-            power: The power value to adjust.
-
-        Returns:
-            A tuple of the closest power values to the desired power that fall within
-                the available bounds for the actor.
-        """
-
-
 @dataclasses.dataclass(frozen=True, kw_only=True)
-class _Report(Report):
+class _Report:
     """Current PowerManager report for a set of components."""
 
     target_power: Power | None

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

@@ -12,7 +12,6 @@
 import uuid
 from collections import abc
 from datetime import timedelta
-from typing import cast
 
 from ... import timeseries
 from ..._internal._channels import ReceiverFetcher
@@ -401,9 +400,7 @@ def power_status(self) -> ReceiverFetcher[BatteryPoolReport]:
         )
         channel.resend_latest = True
 
-        # More details on why the cast is needed here:
-        # https://github.com/frequenz-floss/frequenz-sdk-python/issues/823
-        return cast(ReceiverFetcher[BatteryPoolReport], channel)
+        return channel
 
     @property
     def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:

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

@@ -4,26 +4,28 @@
 """Types for exposing battery pool reports."""
 
 import abc
+import typing
 
 from ...actor import power_distributing
-from ...actor._power_managing._base_classes import Report
 from .._base_types import Bounds
 from .._quantities import Power
 
 
 # This class is used to expose the generic reports from the PowerManager with specific
 # documentation for the battery pool.
-class BatteryPoolReport(Report):
+class BatteryPoolReport(typing.Protocol):
     """A status report for a battery pool."""
 
-    target_power: Power | None
-    """The currently set power for the batteries."""
+    @property
+    def target_power(self) -> Power | None:
+        """The currently set power for the batteries."""
 
-    distribution_result: power_distributing.Result | None
-    """The result of the last power distribution.
+    @property
+    def distribution_result(self) -> power_distributing.Result | None:
+        """The result of the last power distribution.
 
-    This is `None` if no power distribution has been performed yet.
-    """
+        This is `None` if no power distribution has been performed yet.
+        """
 
     @property
     def bounds(self) -> Bounds[Power] | None:

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

@@ -5,7 +5,6 @@
 
 
 import asyncio
-import typing
 import uuid
 from collections import abc
 from datetime import timedelta
@@ -230,9 +229,7 @@ def power_status(self) -> ReceiverFetcher[EVChargerPoolReport]:
         )
         channel.resend_latest = True
 
-        # More details on why the cast is needed here:
-        # https://github.com/frequenz-floss/frequenz-sdk-python/issues/823
-        return typing.cast(ReceiverFetcher[EVChargerPoolReport], channel)
+        return channel
 
     async def stop(self) -> None:
         """Stop all tasks and channels owned by the EVChargerPool."""

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

@@ -13,14 +13,16 @@
 class EVChargerPoolReport(typing.Protocol):
     """A status report for an EV chargers pool."""
 
-    target_power: Power | None
-    """The currently set power for the EV chargers."""
+    @property
+    def target_power(self) -> Power | None:
+        """The currently set power for the EV chargers."""
 
-    distribution_result: power_distributing.Result | None
-    """The result of the last power distribution.
+    @property
+    def distribution_result(self) -> power_distributing.Result | None:
+        """The result of the last power distribution.
 
-    This is `None` if no power distribution has been performed yet.
-    """
+        This is `None` if no power distribution has been performed yet.
+        """
 
     @property
     def bounds(self) -> Bounds[Power] | None:

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

@@ -4,7 +4,6 @@
 """Interactions with pools of PV inverters."""
 
 import asyncio
-import typing
 import uuid
 from collections import abc
 from datetime import timedelta
@@ -189,9 +188,7 @@ def power_status(self) -> ReceiverFetcher[PVPoolReport]:
         )
         channel.resend_latest = True
 
-        # More details on why the cast is needed here:
-        # https://github.com/frequenz-floss/frequenz-sdk-python/issues/823
-        return typing.cast(ReceiverFetcher[PVPoolReport], channel)
+        return channel
 
     async def stop(self) -> None:
         """Stop all tasks and channels owned by the PVPool."""

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

@@ -13,14 +13,16 @@
 class PVPoolReport(typing.Protocol):
     """A status report for a PV pool."""
 
-    target_power: Power | None
-    """The currently set power for the PV inverters."""
+    @property
+    def target_power(self) -> Power | None:
+        """The currently set power for the PV inverters."""
 
-    distribution_result: power_distributing.Result | None
-    """The result of the last power distribution.
+    @property
+    def distribution_result(self) -> power_distributing.Result | None:
+        """The result of the last power distribution.
 
-    This is `None` if no power distribution has been performed yet.
-    """
+        This is `None` if no power distribution has been performed yet.
+        """
 
     @property
     def bounds(self) -> Bounds[Power] | None: