diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 45c015e56..fdcacc038 100644 --- a/RELEASE_NOTES.md +++ b/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. diff --git a/src/frequenz/sdk/microgrid/_data_pipeline.py b/src/frequenz/sdk/microgrid/_data_pipeline.py index 7ed14cf65..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. """ @@ -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( 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..fea4876c5 100644 --- a/tests/microgrid/test_datapipeline.py +++ b/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