Expose the PVPool through microgrid.pv_pool()

There is a plan to rename these constructors to
`microgrid.new_*_pool()`.  Not doing that now, so that it can be done
together for all the pools in a separate PR.

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

Author: shsms

src/frequenz/sdk/microgrid/__init__.py

@@ -124,6 +124,7 @@
     grid,
     logical_meter,
     producer,
+    pv_pool,
     voltage,
 )
 
@@ -149,5 +150,6 @@ async def initialize(host: str, port: int, resampler_config: ResamplerConfig) ->
     "frequency",
     "logical_meter",
     "producer",
+    "pv_pool",
     "voltage",
 ]

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -17,7 +17,7 @@
 from dataclasses import dataclass
 
 from frequenz.channels import Broadcast, Sender
-from frequenz.client.microgrid import ComponentCategory
+from frequenz.client.microgrid import ComponentCategory, InverterType
 
 from ..actor._actor import Actor
 from ..timeseries._grid_frequency import GridFrequency
@@ -44,6 +44,8 @@
     )
     from ..timeseries.logical_meter import LogicalMeter
     from ..timeseries.producer import Producer
+    from ..timeseries.pv_pool import PVPool
+    from ..timeseries.pv_pool._pv_pool_reference_store import PVPoolReferenceStore
 
 _logger = logging.getLogger(__name__)
 
@@ -101,6 +103,9 @@ def __init__(
         self._ev_power_wrapper = PowerWrapper(
             ComponentCategory.EV_CHARGER, None, self._channel_registry
         )
+        self._pv_power_wrapper = PowerWrapper(
+            ComponentCategory.INVERTER, InverterType.SOLAR, self._channel_registry
+        )
 
         self._logical_meter: LogicalMeter | None = None
         self._consumer: Consumer | None = None
@@ -112,6 +117,7 @@ def __init__(
         self._battery_pool_reference_stores: dict[
             frozenset[int], BatteryPoolReferenceStore
         ] = {}
+        self._pv_pool_reference_stores: dict[frozenset[int], PVPoolReferenceStore] = {}
         self._frequency_instance: GridFrequency | None = None
         self._voltage_instance: VoltageStreamer | None = None
 
@@ -245,6 +251,71 @@ def ev_charger_pool(
             self._ev_charger_pool_reference_stores[ref_store_key], name, priority
         )
 
+    def pv_pool(
+        self,
+        pv_inverter_ids: abc.Set[int] | None = None,
+        name: str | None = None,
+        priority: int = -sys.maxsize - 1,
+    ) -> PVPool:
+        """Return a new `PVPool` instance for the given ids.
+
+        If a `PVPoolReferenceStore` instance for the given PV inverter ids doesn't
+        exist, a new one is created and used for creating the `PVPool`.
+
+        Args:
+            pv_inverter_ids: Optional set of IDs of PV inverters to be managed by the
+                `PVPool`.
+            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 making the call.
+
+        Returns:
+            A `PVPool` instance.
+        """
+        from ..timeseries.pv_pool import PVPool
+        from ..timeseries.pv_pool._pv_pool_reference_store import PVPoolReferenceStore
+
+        if not self._pv_power_wrapper.started:
+            self._pv_power_wrapper.start()
+
+        # We use frozenset to make a hashable key from the input set.
+        ref_store_key: frozenset[int] = frozenset()
+        if pv_inverter_ids is not None:
+            ref_store_key = frozenset(pv_inverter_ids)
+
+        pool_key = f"{ref_store_key}-{priority}"
+        if pool_key in self._known_pool_keys:
+            _logger.warning(
+                "A PVPool instance was already created for pv_inverter_ids=%s and "
+                "priority=%s using `microgrid.pv_pool(...)`."
+                "\n  Hint: If the multiple instances are created from the same actor, "
+                "consider reusing the same instance."
+                "\n  Hint: If the instances are created from different actors, "
+                "consider using different priorities to distinguish them.",
+                pv_inverter_ids,
+                priority,
+            )
+        else:
+            self._known_pool_keys.add(pool_key)
+
+        if ref_store_key not in self._pv_pool_reference_stores:
+            self._pv_pool_reference_stores[ref_store_key] = PVPoolReferenceStore(
+                channel_registry=self._channel_registry,
+                resampler_subscription_sender=self._resampling_request_sender(),
+                status_receiver=(
+                    self._pv_power_wrapper.status_channel.new_receiver(limit=1)
+                ),
+                power_manager_requests_sender=(
+                    self._pv_power_wrapper.proposal_channel.new_sender()
+                ),
+                power_manager_bounds_subs_sender=(
+                    self._pv_power_wrapper.bounds_subscription_channel.new_sender()
+                ),
+                component_ids=pv_inverter_ids,
+            )
+
+        return PVPool(self._pv_pool_reference_stores[ref_store_key], name, priority)
+
     def grid(self) -> Grid:
         """Return the grid measuring point."""
         if self._grid is None:
@@ -504,6 +575,43 @@ def battery_pool(
     return _get().battery_pool(battery_ids, name, priority)
 
 
+def pv_pool(
+    pv_inverter_ids: abc.Set[int] | None = None,
+    name: str | None = None,
+    priority: int = -sys.maxsize - 1,
+) -> PVPool:
+    """Return a new `PVPool` instance for the given parameters.
+
+    The priority value is used to resolve conflicts when multiple actors are trying to
+    propose different power values for the same set of PV inverters.
+
+    !!! note
+        When specifying priority, bigger values indicate higher priority. The default
+        priority is the lowest possible value.
+
+        It is recommended to reuse the same instance of the `PVPool` within the same
+        actor, unless they are managing different sets of PV inverters.
+
+        In deployments with multiple actors managing the same set of PV inverters, it is
+        recommended to use different priorities to distinguish between them.  If not,
+        a random prioritization will be imposed on them to resolve conflicts, which may
+        lead to unexpected behavior like longer duration to converge on the desired
+        power.
+
+    Args:
+        pv_inverter_ids: Optional set of IDs of PV inverters to be managed by the
+            `PVPool`. If not specified, all PV inverters available in the component
+            graph are used.
+        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 making the call.
+
+    Returns:
+        A `PVPool` instance.
+    """
+    return _get().pv_pool(pv_inverter_ids, name, priority)
+
+
 def grid() -> Grid:
     """Return the grid measuring point."""
     return _get().grid()

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

@@ -2,3 +2,10 @@
 # Copyright © 2024 Frequenz Energy-as-a-Service GmbH
 
 """Interactions with PV inverters."""
+
+from ._pv_pool import PVPool, PVPoolError
+
+__all__ = [
+    "PVPool",
+    "PVPoolError",
+]