Support sending power proposals to PowerManager from the PVPool

This is done through the new `PVPool.propose_power` method.

This commit also adds a `component_ids` property, that exposes the IDs
of the PV inverters managed by this PVPool.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

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

@@ -6,10 +6,14 @@
 import asyncio
 import typing
 import uuid
+from collections import abc
+from datetime import timedelta
 
 from ..._internal._channels import ReceiverFetcher
 from ...actor import _power_managing
+from ...timeseries import Bounds
 from .._base_types import SystemBounds
+from .._quantities import Power
 from ._pv_pool_reference_store import PVPoolReferenceStore
 from ._result_types import PVPoolReport
 
@@ -54,6 +58,79 @@ def __init__(  # pylint: disable=too-many-arguments
         self._source_id = str(unique_id) if name is None else f"{name}-{unique_id}"
         self._priority = priority
 
+    async def propose_power(
+        self,
+        power: Power | None,
+        *,
+        request_timeout: timedelta = timedelta(seconds=5.0),
+        bounds: Bounds[Power | None] = Bounds(None, None),
+    ) -> None:
+        """Send a proposal to the power manager for the pool's set of PV inverters.
+
+        This proposal is for the maximum power that can be set for the PV inverters in
+        the pool.  The actual production might be lower.
+
+        Power values need to follow the Passive Sign Convention (PSC). That is, positive
+        values indicate charge power and negative values indicate discharge power.
+        Only discharge powers are allowed for PV inverters.
+
+        If the same PV inverters are shared by multiple actors, the power manager will
+        consider the priority of the actors, the bounds they set, and their preferred
+        power, when calculating the target power for the PV inverters.
+
+        The preferred power of lower priority actors will take precedence as long as
+        they respect the bounds set by higher priority actors.  If lower priority actors
+        request power values outside of the bounds set by higher priority actors, the
+        target power will be the closest value to the preferred power that is within the
+        bounds.
+
+        When there are no other actors trying to use the same PV inverters, the actor's
+        preferred power would be set as the target power, as long as it falls within the
+        system power bounds for the PV inverters.
+
+        The result of the request can be accessed using the receiver returned from the
+        [`power_status`][frequenz.sdk.timeseries.pv_pool.PVPool.power_status]
+        method, which also streams the bounds that an actor should comply with, based on
+        its priority.
+
+        Args:
+            power: The power to propose for the PV inverters in the pool.  If `None`,
+                this proposal will not have any effect on the target power, unless
+                bounds are specified.  If both are `None`, it is equivalent to not
+                having a proposal or withdrawing a previous one.
+            request_timeout: The timeout for the request.
+            bounds: The power bounds for the proposal.  These bounds will apply to
+                actors with a lower priority, and can be overridden by bounds from
+                actors with a higher priority.  If None, the power bounds will be set to
+                the maximum power of the batteries in the pool.  This is currently and
+                experimental feature.
+
+        Raises:
+            PVPoolError: If a charge power for PV inverters is requested.
+        """
+        if power is not None and power > Power.zero():
+            raise PVPoolError("Charge powers for PV inverters is not supported.")
+        await self._pv_pool_ref.power_manager_requests_sender.send(
+            _power_managing.Proposal(
+                source_id=self._source_id,
+                preferred_power=power,
+                bounds=bounds,
+                component_ids=self._pv_pool_ref.component_ids,
+                priority=self._priority,
+                creation_time=asyncio.get_running_loop().time(),
+                request_timeout=request_timeout,
+            )
+        )
+
+    @property
+    def component_ids(self) -> abc.Set[int]:
+        """Return component IDs of all PV inverters managed by this PVPool.
+
+        Returns:
+            Set of managed component IDs.
+        """
+        return self._pv_pool_ref.component_ids
+
     @property
     def power_status(self) -> ReceiverFetcher[PVPoolReport]:
         """Get a receiver to receive new power status reports when they change.