Share reference stores between pools for all and explicit components

RELEASE_NOTES.md

@@ -19,3 +19,7 @@
 * `new_battery_pool(component_ids=set())` uses all batteries again, like before v1.0.0-rc2206.
 * `new_pv_pool()`, `new_ev_charger_pool()` and `new_steam_boiler_pool()` now also treat an empty `component_ids` set as "all components". Before, it gave a pool with no components, unless a pool for all components of that type had already been created (both pools then shared one cache entry).
 * Because of that shared cache entry, the result used to depend on which pool was created first. Now `None` and `set()` always give the same pool.
+* A pool created without component IDs now shares its state with a pool created for the explicit IDs of all the components of its type, like the one the power manager creates to track the pool's power bounds. Before, each of them tracked the power bounds of the same components on its own.
+* `new_battery_pool()`, `new_pv_pool()`, `new_ev_charger_pool()` and `new_steam_boiler_pool()` now always log the warning that a pool was already created when the same pool is created twice at the same priority. The warning now names the pool's components.
+  - Before, the warning was missed when a pool was created without IDs and again for all the IDs of its type, and sometimes for two equal sets of IDs that were built in different ways.
+  - Creating pools of different types without IDs at the same priority also no longer logs the warning by mistake.

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -47,6 +47,9 @@
     from ..timeseries.battery_pool._battery_pool_reference_store import (
         BatteryPoolReferenceStore,
     )
+    from ..timeseries.component_pool._component_pool_reference_store import (
+        ComponentPoolReferenceStore,
+    )
     from ..timeseries.consumer import Consumer
     from ..timeseries.ev_charger_pool import EVChargerPool
     from ..timeseries.ev_charger_pool._ev_charger_pool_reference_store import (
@@ -167,10 +170,12 @@ def __init__(
         self._frequency_instance: GridFrequency | None = None
         self._voltage_instance: VoltageStreamer | None = None
 
-        self._known_pool_keys: set[str] = set()
-        """A set of keys for corresponding to created EVChargerPool instances.
+        self._known_pool_keys: set[
+            tuple[type[ComponentPoolReferenceStore], frozenset[ComponentId], int]
+        ] = set()
+        """The reference store class, component IDs and priority of each pool created.
 
-        This is used to warn the user if they try to create a new EVChargerPool instance
+        This is used to warn the user if they try to create a new pool of the same type
         for the same set of component IDs, and with the same priority.
         """
 
@@ -298,21 +303,18 @@ def new_ev_charger_pool(
         if not self._ev_power_wrapper.started:
             self._ev_power_wrapper.start()
 
-        # We use frozenset to make a hashable key from the input set.
-        ref_store_key: frozenset[ComponentId] = frozenset()
-        if component_ids is not None:
-            ref_store_key = frozenset(component_ids)
+        ref_store_key = EVChargerPoolReferenceStore.resolve_component_ids(component_ids)
 
-        pool_key = f"{ref_store_key}-{priority}"
+        pool_key = (EVChargerPoolReferenceStore, ref_store_key, priority)
         if pool_key in self._known_pool_keys:
             _logger.warning(
                 "An EVChargerPool instance was already created for ev_charger_ids=%s "
-                "and priority=%s using `microgrid.ev_charger_pool(...)`."
+                "and priority=%s using `microgrid.new_ev_charger_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.",
-                component_ids,
+                sorted(ref_store_key),
                 priority,
             )
         else:
@@ -335,7 +337,7 @@ def new_ev_charger_pool(
                     power_distribution_results_fetcher=(
                         self._ev_power_wrapper.distribution_results_fetcher()
                     ),
-                    component_ids=component_ids,
+                    component_ids=ref_store_key,
                 )
             )
         return EVChargerPool(
@@ -372,21 +374,18 @@ def new_pv_pool(
         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[ComponentId] = frozenset()
-        if component_ids is not None:
-            ref_store_key = frozenset(component_ids)
+        ref_store_key = PVPoolReferenceStore.resolve_component_ids(component_ids)
 
-        pool_key = f"{ref_store_key}-{priority}"
+        pool_key = (PVPoolReferenceStore, 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(...)`."
+                "priority=%s using `microgrid.new_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.",
-                component_ids,
+                sorted(ref_store_key),
                 priority,
             )
         else:
@@ -408,7 +407,7 @@ def new_pv_pool(
                 power_distribution_results_fetcher=(
                     self._pv_power_wrapper.distribution_results_fetcher()
                 ),
-                component_ids=component_ids,
+                component_ids=ref_store_key,
             )
 
         return PVPool(
@@ -447,21 +446,18 @@ def new_battery_pool(
         if not self._battery_power_wrapper.started:
             self._battery_power_wrapper.start()
 
-        # We use frozenset to make a hashable key from the input set.
-        ref_store_key: frozenset[ComponentId] = frozenset()
-        if component_ids is not None:
-            ref_store_key = frozenset(component_ids)
+        ref_store_key = BatteryPoolReferenceStore.resolve_component_ids(component_ids)
 
-        pool_key = f"{ref_store_key}-{priority}"
+        pool_key = (BatteryPoolReferenceStore, ref_store_key, priority)
         if pool_key in self._known_pool_keys:
             _logger.warning(
                 "A BatteryPool instance was already created for battery_ids=%s and "
-                "priority=%s using `microgrid.battery_pool(...)`."
+                "priority=%s using `microgrid.new_battery_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.",
-                component_ids,
+                sorted(ref_store_key),
                 priority,
             )
         else:
@@ -485,7 +481,7 @@ def new_battery_pool(
                         self._battery_power_wrapper.distribution_results_fetcher()
                     ),
                     min_update_interval=self._resampler_config.resampling_period,
-                    component_ids=component_ids,
+                    component_ids=ref_store_key,
                 )
             )
 
@@ -526,21 +522,20 @@ def new_steam_boiler_pool(
         if not self._steam_boiler_power_wrapper.started:
             self._steam_boiler_power_wrapper.start()
 
-        # We use frozenset to make a hashable key from the input set.
-        ref_store_key: frozenset[ComponentId] = frozenset()
-        if component_ids is not None:
-            ref_store_key = frozenset(component_ids)
+        ref_store_key = SteamBoilerPoolReferenceStore.resolve_component_ids(
+            component_ids
+        )
 
-        pool_key = f"{ref_store_key}-{priority}"
+        pool_key = (SteamBoilerPoolReferenceStore, ref_store_key, priority)
         if pool_key in self._known_pool_keys:
             _logger.warning(
                 "A SteamBoilerPool instance was already created for steam_boiler_ids=%s "
-                "and priority=%s using `microgrid.steam_boiler_pool(...)`."
+                "and priority=%s using `microgrid.new_steam_boiler_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.",
-                component_ids,
+                sorted(ref_store_key),
                 priority,
             )
         else:
@@ -563,7 +558,7 @@ def new_steam_boiler_pool(
                     power_distribution_results_fetcher=(
                         self._steam_boiler_power_wrapper.distribution_results_fetcher()
                     ),
-                    component_ids=component_ids,
+                    component_ids=ref_store_key,
                 )
             )
         return SteamBoilerPool(

src/frequenz/sdk/timeseries/component_pool/_component_pool_reference_store.py

@@ -66,27 +66,17 @@ def __init__(  # pylint: disable=too-many-arguments
         self.power_manager_bounds_subs_sender = power_manager_bounds_subs_sender
         self.power_distribution_results_fetcher = power_distribution_results_fetcher
 
-        graph = connection_manager.get().component_graph
-        all_components = frozenset(
-            {
-                inv.id
-                for inv in graph.components(matching_types=self.get_component_class())
-            }
+        self.component_ids: frozenset[ComponentId] = self.resolve_component_ids(
+            component_ids
         )
-
-        # A None or empty component_ids means "use all components", matching the
-        # frozenset() key both values map to in `_DataPipeline.new_*_pool()`.
         if component_ids:
-            self.component_ids: frozenset[ComponentId] = frozenset(component_ids)
-            if not self.component_ids.issubset(all_components):
-                unknown_ids = self.component_ids - all_components
+            unknown_ids = self.component_ids - self.all_component_ids()
+            if unknown_ids:
                 raise ValueError(
                     f"Unable to create {self.get_pool_type_name()}. These component IDs "
                     + f"are either not {self.get_component_type_name_plural()} or are unknown: "
                     + f"{unknown_ids}"
                 )
-        else:
-            self.component_ids = all_components
 
         self.power_bounds_subs: dict[str, asyncio.Task[None]] = {}
 
@@ -104,6 +94,33 @@ def __init__(  # pylint: disable=too-many-arguments
         self.bounds_tracker: BackgroundService | None = None
         self.create_bounds_tracker()
 
+    @classmethod
+    def resolve_component_ids(
+        cls, component_ids: abc.Set[ComponentId] | None
+    ) -> frozenset[ComponentId]:
+        """Return the given IDs, or all IDs of this pool's type if none are given.
+
+        The data pipeline keys the reference stores by these IDs, so a pool for all
+        components shares its store with a pool created for their explicit IDs.
+
+        Args:
+            component_ids: The requested IDs, or `None` or an empty set for all
+                components of this pool's type.
+
+        Returns:
+            The requested IDs, or the IDs of all components of this pool's type.
+        """
+        return frozenset(component_ids) if component_ids else cls.all_component_ids()
+
+    @classmethod
+    def all_component_ids(cls) -> frozenset[ComponentId]:
+        """Return the IDs of all components of this pool's type in the microgrid."""
+        graph = connection_manager.get().component_graph
+        return frozenset(
+            component.id
+            for component in graph.components(matching_types=cls.get_component_class())
+        )
+
     @staticmethod
     @abstractmethod
     def get_component_class() -> Type[Component]:

tests/microgrid/test_datapipeline.py

@@ -4,8 +4,11 @@
 """Basic tests for the DataPipeline."""
 
 import asyncio
+import logging
 import re
+from collections.abc import Callable
 from datetime import timedelta
+from typing import Any
 
 import async_solipsism
 import pytest
@@ -23,10 +26,13 @@
 )
 from pytest_mock import MockerFixture
 
+from frequenz.sdk import microgrid
 from frequenz.sdk.microgrid import _data_pipeline
 from frequenz.sdk.microgrid._data_pipeline import _DataPipeline
 from frequenz.sdk.timeseries import ResamplerConfig2
+from frequenz.sdk.timeseries.component_pool import ComponentPool
 
+from ..timeseries.mock_microgrid import MockMicrogrid
 from ..utils.mock_microgrid_client import MockMicrogridClient
 
 
@@ -238,3 +244,112 @@ async def test_empty_component_ids_matches_none_in_either_call_order(
     assert steam_boiler_empty_second.component_ids == all_steam_boilers
 
     await datapipeline._stop()
+
+
+_POOL_TYPES = [
+    pytest.param(
+        microgrid.new_battery_pool,
+        MockMicrogrid.add_batteries,
+        "battery_ids",
+        id="battery",
+    ),
+    pytest.param(
+        microgrid.new_pv_pool,
+        MockMicrogrid.add_solar_inverters,
+        "pv_inverter_ids",
+        id="pv",
+    ),
+    pytest.param(
+        microgrid.new_ev_charger_pool,
+        MockMicrogrid.add_ev_chargers,
+        "evc_ids",
+        id="ev_charger",
+    ),
+    pytest.param(
+        microgrid.new_steam_boiler_pool,
+        MockMicrogrid.add_steam_boilers,
+        "steam_boiler_ids",
+        id="steam_boiler",
+    ),
+]
+"""The pool factory, the mock microgrid method that adds its components, and the
+mock microgrid attribute that holds their IDs, for each pool type."""
+
+
+@pytest.mark.parametrize("new_pool, add_components, ids_attr", _POOL_TYPES)
+async def test_pools_for_all_components_share_reference_stores(
+    mocker: MockerFixture,
+    new_pool: Callable[..., ComponentPool[Any, Any]],
+    add_components: Callable[[MockMicrogrid, int], None],
+    ids_attr: str,
+) -> None:
+    """Test that a pool without IDs and a pool given all IDs share a reference store.
+
+    The power manager creates pools for the explicit IDs it gets in proposals and
+    bounds subscriptions.  A pool created without IDs, or with an empty set, must
+    share its reference store with the pool the power manager creates for the same
+    components, instead of tracking the power bounds of those components on its own.
+    """
+    mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker)
+    add_components(mockgrid, 2)
+    async with mockgrid:
+        all_pool = new_pool(priority=5)
+        explicit_pool = new_pool(
+            priority=6, component_ids=set(getattr(mockgrid, ids_attr))
+        )
+        empty_set_pool = new_pool(priority=7, component_ids=set())
+        # pylint: disable=protected-access
+        assert explicit_pool._pool_ref_store is all_pool._pool_ref_store
+        assert empty_set_pool._pool_ref_store is all_pool._pool_ref_store
+
+
+@pytest.mark.parametrize("new_pool, add_components, ids_attr", _POOL_TYPES)
+async def test_pool_for_all_components_warns_like_explicit_ids(
+    mocker: MockerFixture,
+    caplog: pytest.LogCaptureFixture,
+    new_pool: Callable[..., ComponentPool[Any, Any]],
+    add_components: Callable[[MockMicrogrid, int], None],
+    ids_attr: str,
+) -> None:
+    """Test that pools made without IDs and with all IDs warn as duplicates.
+
+    The pool created without IDs resolves to all the IDs of its type, so at the same
+    priority both pools have the same key.  Each warning names the resolved IDs.
+    """
+    mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker)
+    # With 6 components, the copy made of the set passed in has a smaller table than
+    # the set read from the component graph.  So the two sets print in different
+    # orders, and keys made from their text would not match.
+    add_components(mockgrid, 6)
+    async with mockgrid:
+        component_ids = getattr(mockgrid, ids_attr)
+        new_pool(priority=5)
+        with caplog.at_level(logging.WARNING):
+            new_pool(priority=5, component_ids=set(component_ids))
+            new_pool(priority=5)
+        warnings = [
+            r.getMessage()
+            for r in caplog.records
+            if "already created" in r.getMessage()
+        ]
+        assert len(warnings) == 2
+        for message in warnings:
+            assert str(sorted(component_ids)) in message
+            assert f"microgrid.{new_pool.__name__}(...)" in message
+
+
+async def test_pools_of_different_types_without_components_dont_warn(
+    mocker: MockerFixture, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test that empty pools of different types, made without IDs, don't warn."""
+    mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker)
+    # There are no PV inverters, EV chargers or steam boilers, so all three pools
+    # resolve to the same empty set, and only the pool type tells them apart.
+    mockgrid.add_batteries(1)
+    async with mockgrid:
+        with caplog.at_level(logging.WARNING):
+            microgrid.new_pv_pool(priority=5)
+            microgrid.new_ev_charger_pool(priority=5)
+            microgrid.new_steam_boiler_pool(priority=5)
+        warnings = [r for r in caplog.records if "already created" in r.getMessage()]
+        assert not warnings