From 3b20d69437e38ed168efce20e40fc9f4861d376b Mon Sep 17 00:00:00 2001 From: Sahas Subramanian Date: Thu, 29 Oct 2026 13:23:59 +0000 Subject: [PATCH 1/2] Share reference stores between pools for all and explicit components A pool created without component IDs, or with an empty set, was keyed by an empty set, so it didn't share its reference store with the power manager's pool for the same components, and each tracked the power bounds of the same components on its own. Signed-off-by: Sahas Subramanian --- RELEASE_NOTES.md | 1 + src/frequenz/sdk/microgrid/_data_pipeline.py | 30 +++------ .../_component_pool_reference_store.py | 45 +++++++++----- tests/microgrid/test_datapipeline.py | 62 +++++++++++++++++++ 4 files changed, 104 insertions(+), 34 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 45c015e56..7ad5c9d7d 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -19,3 +19,4 @@ * `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. diff --git a/src/frequenz/sdk/microgrid/_data_pipeline.py b/src/frequenz/sdk/microgrid/_data_pipeline.py index 7ed14cf65..87d27d959 100644 --- a/src/frequenz/sdk/microgrid/_data_pipeline.py +++ b/src/frequenz/sdk/microgrid/_data_pipeline.py @@ -298,10 +298,7 @@ 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}" if pool_key in self._known_pool_keys: @@ -335,7 +332,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,10 +369,7 @@ 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}" if pool_key in self._known_pool_keys: @@ -408,7 +402,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,10 +441,7 @@ 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}" if pool_key in self._known_pool_keys: @@ -485,7 +476,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,10 +517,9 @@ 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}" if pool_key in self._known_pool_keys: @@ -563,7 +553,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( diff --git a/src/frequenz/sdk/timeseries/component_pool/_component_pool_reference_store.py b/src/frequenz/sdk/timeseries/component_pool/_component_pool_reference_store.py index 098375f63..9306c3141 100644 --- a/src/frequenz/sdk/timeseries/component_pool/_component_pool_reference_store.py +++ b/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]: diff --git a/tests/microgrid/test_datapipeline.py b/tests/microgrid/test_datapipeline.py index 40c2c50aa..0f4229086 100644 --- a/tests/microgrid/test_datapipeline.py +++ b/tests/microgrid/test_datapipeline.py @@ -5,7 +5,9 @@ import asyncio import re +from collections.abc import Callable from datetime import timedelta +from typing import Any import async_solipsism import pytest @@ -23,10 +25,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 +243,60 @@ 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 From a72c2aacf08f504f2f259862a17adfc674273581 Mon Sep 17 00:00:00 2001 From: Sahas Subramanian Date: Thu, 29 Oct 2026 13:23:59 +0000 Subject: [PATCH 2/2] Compare duplicate pools by value, not by a string of their IDs The data pipeline warns when the same pool is created twice at the same priority, but it compared a string of the component IDs. Two equal sets of component IDs can print in different orders, so the warning was sometimes missed. The key now also holds the pool's reference store class, so pools of different types that have no components don't warn about each other. The warning now names the components the pool resolved to, not the `component_ids` argument, which is `None` or an empty set when no IDs are given. It also names the right `new_*_pool()` function. Signed-off-by: Sahas Subramanian --- RELEASE_NOTES.md | 3 ++ src/frequenz/sdk/microgrid/_data_pipeline.py | 35 +++++++------ tests/microgrid/test_datapipeline.py | 53 ++++++++++++++++++++ 3 files changed, 76 insertions(+), 15 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 7ad5c9d7d..fdcacc038 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -20,3 +20,6 @@ * `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. diff --git a/src/frequenz/sdk/microgrid/_data_pipeline.py b/src/frequenz/sdk/microgrid/_data_pipeline.py index 87d27d959..9ba36e1f5 100644 --- a/src/frequenz/sdk/microgrid/_data_pipeline.py +++ b/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. """ @@ -300,16 +305,16 @@ def new_ev_charger_pool( 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: @@ -371,16 +376,16 @@ def new_pv_pool( 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: @@ -443,16 +448,16 @@ def new_battery_pool( 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: @@ -521,16 +526,16 @@ def new_steam_boiler_pool( 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: diff --git a/tests/microgrid/test_datapipeline.py b/tests/microgrid/test_datapipeline.py index 0f4229086..fea4876c5 100644 --- a/tests/microgrid/test_datapipeline.py +++ b/tests/microgrid/test_datapipeline.py @@ -4,6 +4,7 @@ """Basic tests for the DataPipeline.""" import asyncio +import logging import re from collections.abc import Callable from datetime import timedelta @@ -300,3 +301,55 @@ async def test_pools_for_all_components_share_reference_stores( # 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