Move reusable parts of EVChargerPool to EVChargerPoolReferenceStore

This allows the efficient creation of multiple `EVChargerPool`
instances for the same set of EV chargers, but with different
priorities, while reusing the formulas and bounds by having only a
single instance of the `EVChargerPoolReferenceStore` class for any
given set of EV chargers.

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

Author: shsms

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -38,6 +38,9 @@
     )
     from ..timeseries.consumer import Consumer
     from ..timeseries.ev_charger_pool import EVChargerPool
+    from ..timeseries.ev_charger_pool._ev_charger_pool_reference_store import (
+        EVChargerPoolReferenceStore,
+    )
     from ..timeseries.logical_meter import LogicalMeter
     from ..timeseries.producer import Producer
 
@@ -100,7 +103,7 @@ def __init__(
         self._consumer: Consumer | None = None
         self._producer: Producer | None = None
         self._grid: Grid | None = None
-        self._ev_charger_pools: dict[frozenset[int], EVChargerPool] = {}
+        self._ev_charger_pools: dict[frozenset[int], EVChargerPoolReferenceStore] = {}
         self._battery_pools: dict[frozenset[int], BatteryPoolReferenceStore] = {}
         self._frequency_instance: GridFrequency | None = None
         self._voltage_instance: VoltageStreamer | None = None
@@ -161,6 +164,8 @@ def producer(self) -> Producer:
     def ev_charger_pool(
         self,
         ev_charger_ids: abc.Set[int] | None = None,
+        name: str | None = None,
+        priority: int = -sys.maxsize - 1,
     ) -> EVChargerPool:
         """Return the corresponding EVChargerPool instance for the given ids.
 
@@ -170,11 +175,17 @@ def ev_charger_pool(
         Args:
             ev_charger_ids: Optional set of IDs of EV Chargers to be managed by the
                 EVChargerPool.
+            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:
             An EVChargerPool instance.
         """
         from ..timeseries.ev_charger_pool import EVChargerPool
+        from ..timeseries.ev_charger_pool._ev_charger_pool_reference_store import (
+            EVChargerPoolReferenceStore,
+        )
 
         if not self._ev_power_wrapper.started:
             self._ev_power_wrapper.start()
@@ -185,15 +196,15 @@ def ev_charger_pool(
             key = frozenset(ev_charger_ids)
 
         if key not in self._ev_charger_pools:
-            self._ev_charger_pools[key] = EVChargerPool(
+            self._ev_charger_pools[key] = EVChargerPoolReferenceStore(
                 channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
                 status_receiver=self._ev_power_wrapper.status_channel.new_receiver(
                     limit=1
                 ),
                 component_ids=ev_charger_ids,
             )
-        return self._ev_charger_pools[key]
+        return EVChargerPool(self._ev_charger_pools[key], name, priority)
 
     def grid(self) -> Grid:
         """Return the grid measuring point."""
@@ -362,21 +373,32 @@ def producer() -> Producer:
     return _get().producer()
 
 
-def ev_charger_pool(ev_charger_ids: abc.Set[int] | None = None) -> EVChargerPool:
-    """Return the corresponding EVChargerPool instance for the given ids.
+def ev_charger_pool(
+    ev_charger_ids: abc.Set[int] | None = None,
+    name: str | None = None,
+    priority: int = -sys.maxsize - 1,
+) -> EVChargerPool:
+    """Return a new `EVChargerPool` instance for the given parameters.
 
-    If an EVChargerPool instance for the given ids doesn't exist, a new one is
-    created and returned.
+    The priority value is used to resolve conflicts when multiple actors are trying to
+    propose different power values for the same set of EV chargers.
+
+    !!! note
+        When specifying priority, bigger values indicate higher priority. The default
+        priority is the lowest possible value.
 
     Args:
         ev_charger_ids: Optional set of IDs of EV Chargers to be managed by the
             EVChargerPool.  If not specified, all EV Chargers 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:
-        An EVChargerPool instance.
+        An `EVChargerPool` instance.
     """
-    return _get().ev_charger_pool(ev_charger_ids)
+    return _get().ev_charger_pool(ev_charger_ids, name, priority)
 
 
 def battery_pool(

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

@@ -4,10 +4,8 @@
 """Interactions with EV Chargers."""
 
 from ._ev_charger_pool import EVChargerPool, EVChargerPoolError
-from ._set_current_bounds import ComponentCurrentLimit
 
 __all__ = [
-    "ComponentCurrentLimit",
     "EVChargerPool",
     "EVChargerPoolError",
 ]

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

@@ -7,23 +7,16 @@
 import uuid
 from collections import abc
 
-from frequenz.channels import Broadcast, Receiver, Sender
-from frequenz.client.microgrid import ComponentCategory
-
 from ..._internal._channels import ReceiverFetcher
-from ...actor import ChannelRegistry, ComponentMetricRequest
-from ...actor.power_distributing import ComponentPoolStatus
-from ...microgrid import connection_manager
 from .._base_types import SystemBounds
 from .._quantities import Current, Power
 from ..formula_engine import FormulaEngine, FormulaEngine3Phase
-from ..formula_engine._formula_engine_pool import FormulaEnginePool
 from ..formula_engine._formula_generators import (
     EVChargerCurrentFormula,
     EVChargerPowerFormula,
     FormulaGeneratorConfig,
 )
-from ._system_bounds_tracker import EVCSystemBoundsTracker
+from ._ev_charger_pool_reference_store import EVChargerPoolReferenceStore
 
 
 class EVChargerPoolError(Exception):
@@ -47,10 +40,9 @@ class EVChargerPool:
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
-        channel_registry: ChannelRegistry,
-        resampler_subscription_sender: Sender[ComponentMetricRequest],
-        status_receiver: Receiver[ComponentPoolStatus],
-        component_ids: abc.Set[int] | None = None,
+        ev_charger_pool_ref: EVChargerPoolReferenceStore,
+        name: str | None,
+        priority: int,
     ) -> None:
         """Create an `EVChargerPool` instance.
 
@@ -60,46 +52,15 @@ def __init__(  # pylint: disable=too-many-arguments
             method for creating `EVChargerPool` instances.
 
         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 EV Chargers in
-                the pool.
-            component_ids: An optional list of component_ids belonging to this pool.  If
-                not specified, IDs of all EV Chargers in the microgrid will be fetched
-                from the component graph.
+            ev_charger_pool_ref: The EV charger 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._channel_registry: ChannelRegistry = channel_registry
-        self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
-            resampler_subscription_sender
-        )
-        self._status_receiver: Receiver[ComponentPoolStatus] = status_receiver
-        self._component_ids: abc.Set[int] = set()
-        if component_ids is not None:
-            self._component_ids = component_ids
-        else:
-            graph = connection_manager.get().component_graph
-            self._component_ids = {
-                evc.component_id
-                for evc in graph.components(
-                    component_categories={ComponentCategory.EV_CHARGER}
-                )
-            }
-        self._namespace: str = f"ev-charger-pool-{uuid.uuid4()}"
-        self._formula_pool: FormulaEnginePool = FormulaEnginePool(
-            self._namespace,
-            self._channel_registry,
-            self._resampler_subscription_sender,
-        )
-
-        self._bounds_channel: Broadcast[SystemBounds] = Broadcast(
-            name=f"System Bounds for EV Chargers: {component_ids}"
-        )
-        self._bounds_tracker: EVCSystemBoundsTracker = EVCSystemBoundsTracker(
-            self.component_ids, self._status_receiver, self._bounds_channel.new_sender()
-        )
-        self._bounds_tracker.start()
+        self._ev_charger_pool = ev_charger_pool_ref
+        unique_id = 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[int]:
@@ -108,7 +69,7 @@ def component_ids(self) -> abc.Set[int]:
         Returns:
             Set of managed component IDs.
         """
-        return self._component_ids
+        return self._ev_charger_pool.component_ids
 
     @property
     def current(self) -> FormulaEngine3Phase[Current]:
@@ -126,10 +87,14 @@ def current(self) -> FormulaEngine3Phase[Current]:
             A FormulaEngine that will calculate and stream the total current of all EV
                 Chargers.
         """
-        engine = self._formula_pool.from_3_phase_current_formula_generator(
-            "ev_charger_total_current",
-            EVChargerCurrentFormula,
-            FormulaGeneratorConfig(component_ids=self._component_ids),
+        engine = (
+            self._ev_charger_pool.formula_pool.from_3_phase_current_formula_generator(
+                "ev_charger_total_current",
+                EVChargerCurrentFormula,
+                FormulaGeneratorConfig(
+                    component_ids=self._ev_charger_pool.component_ids
+                ),
+            )
         )
         assert isinstance(engine, FormulaEngine3Phase)
         return engine
@@ -150,21 +115,21 @@ def power(self) -> FormulaEngine[Power]:
             A FormulaEngine that will calculate and stream the total power of all EV
                 Chargers.
         """
-        engine = self._formula_pool.from_power_formula_generator(
+        engine = self._ev_charger_pool.formula_pool.from_power_formula_generator(
             "ev_charger_power",
             EVChargerPowerFormula,
             FormulaGeneratorConfig(
-                component_ids=self._component_ids,
+                component_ids=self._ev_charger_pool.component_ids,
             ),
         )
         assert isinstance(engine, FormulaEngine)
         return engine
 
     async def stop(self) -> None:
         """Stop all tasks and channels owned by the EVChargerPool."""
-        await self._formula_pool.stop()
+        await self._ev_charger_pool.stop()
 
     @property
     def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
         """Return a receiver for the system power bounds."""
-        return self._bounds_channel
+        return self._ev_charger_pool.bounds_channel

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

@@ -0,0 +1,91 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Manages shared state/tasks for a set of EV chargers."""
+
+
+import uuid
+from collections import abc
+
+from frequenz.channels import Broadcast, Receiver, Sender
+from frequenz.client.microgrid import ComponentCategory
+
+from ...actor import ChannelRegistry, ComponentMetricRequest
+from ...actor.power_distributing import ComponentPoolStatus
+from ...microgrid import connection_manager
+from .._base_types import SystemBounds
+from ..formula_engine._formula_engine_pool import FormulaEnginePool
+from ._system_bounds_tracker import EVCSystemBoundsTracker
+
+
+class EVChargerPoolReferenceStore:
+    """A class for maintaining the shared state/tasks for a set of pool of EV chargers.
+
+    This includes ownership of
+    - the formula engine pool and metric calculators.
+    - the tasks for calculating system bounds for the EV chargers.
+
+    These are independent of the priority of the actors and can be shared between
+    multiple users of the same set of EV chargers.
+
+    They are exposed through the EVChargerPool class.
+    """
+
+    def __init__(
+        self,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
+        status_receiver: Receiver[ComponentPoolStatus],
+        component_ids: abc.Set[int] | None = None,
+    ):
+        """Create an instance of the class.
+
+        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 EV Chargers in
+                the pool.
+            component_ids: An optional list of component_ids belonging to this pool.  If
+                not specified, IDs of all EV Chargers in the microgrid will be fetched
+                from the component graph.
+        """
+        self.channel_registry = channel_registry
+        self.resampler_subscription_sender = resampler_subscription_sender
+        self.status_receiver = status_receiver
+
+        if component_ids is not None:
+            self.component_ids: frozenset[int] = frozenset(component_ids)
+        else:
+            graph = connection_manager.get().component_graph
+            self.component_ids = frozenset(
+                {
+                    evc.component_id
+                    for evc in graph.components(
+                        component_categories={ComponentCategory.EV_CHARGER}
+                    )
+                }
+            )
+
+        self.namespace: str = f"ev-charger-pool-{uuid.uuid4()}"
+        self.formula_pool = FormulaEnginePool(
+            self.namespace,
+            self.channel_registry,
+            self.resampler_subscription_sender,
+        )
+
+        self.bounds_channel: Broadcast[SystemBounds] = Broadcast(
+            name=f"System Bounds for EV Chargers: {component_ids}"
+        )
+        self.bounds_tracker: EVCSystemBoundsTracker = EVCSystemBoundsTracker(
+            self.component_ids,
+            self.status_receiver,
+            self.bounds_channel.new_sender(),
+        )
+        self.bounds_tracker.start()
+
+    async def stop(self) -> None:
+        """Stop all tasks and channels owned by the EVChargerPool."""
+        await self.formula_pool.stop()
+        await self.bounds_tracker.stop()