diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 61ee6f2ad..f87cb2812 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -10,7 +10,7 @@ ## New Features - +* Added `SteamBoilerPool` for monitoring and controlling pools of steam boilers, available via `microgrid.new_steam_boiler_pool()`. ## Bug Fixes diff --git a/src/frequenz/sdk/microgrid/__init__.py b/src/frequenz/sdk/microgrid/__init__.py index 3bddf8ee5..244212e29 100644 --- a/src/frequenz/sdk/microgrid/__init__.py +++ b/src/frequenz/sdk/microgrid/__init__.py @@ -138,14 +138,30 @@ which accepts values in the {{glossary("psc", "Passive Sign Convention")}} and supports only charging. +## Steam Boilers + +The [`steam_boiler_pool`][frequenz.sdk.microgrid.new_steam_boiler_pool] offers a +[`power`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.power] method that +streams the total power measured for all the steam boilers at a site. + +The `steam_boiler_pool` also provides available power bounds through the +[`power_status`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.power_status] +method. + +The `steam_boiler_pool` also provides a control method +[`propose_power`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.propose_power], +which accepts values in the {{glossary("psc", "Passive Sign Convention")}} and supports +only consumption. + # Component pools The SDK provides a unified interface for interacting with sets of Batteries, EV -chargers and PV arrays, through their corresponding `Pool`s. +chargers, PV arrays and steam boilers, through their corresponding `Pool`s. * [Battery pool][frequenz.sdk.microgrid.new_battery_pool] * [EV charger pool][frequenz.sdk.microgrid.new_ev_charger_pool] * [PV pool][frequenz.sdk.microgrid.new_pv_pool] +* [Steam boiler pool][frequenz.sdk.microgrid.new_steam_boiler_pool] All of them provide support for streaming aggregated data and for setting the power values of the components. @@ -366,6 +382,7 @@ | Batteries | Zero | | PV | Max production (Min power according to PSC) | | EV Chargers | Max consumption (Max power according to PSC) | +| Steam Boilers | Zero | """ # noqa: D205, D400, E501 from datetime import timedelta @@ -382,6 +399,7 @@ new_battery_pool, new_ev_charger_pool, new_pv_pool, + new_steam_boiler_pool, producer, voltage_per_phase, ) @@ -442,6 +460,7 @@ async def initialize( "new_battery_pool", "new_ev_charger_pool", "new_pv_pool", + "new_steam_boiler_pool", "producer", "voltage_per_phase", ] diff --git a/src/frequenz/sdk/microgrid/_data_pipeline.py b/src/frequenz/sdk/microgrid/_data_pipeline.py index 4c233e7ae..db8a8f736 100644 --- a/src/frequenz/sdk/microgrid/_data_pipeline.py +++ b/src/frequenz/sdk/microgrid/_data_pipeline.py @@ -19,7 +19,12 @@ from frequenz.channels import Broadcast, Sender from frequenz.client.common.microgrid.components import ComponentId -from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter +from frequenz.client.microgrid.component import ( + Battery, + EvCharger, + SolarInverter, + SteamBoiler, +) from .._internal._channels import ChannelRegistry from ..actor._actor import Actor @@ -52,6 +57,10 @@ from ..timeseries.producer import Producer from ..timeseries.pv_pool import PVPool from ..timeseries.pv_pool._pv_pool_reference_store import PVPoolReferenceStore + from ..timeseries.steam_boiler_pool import SteamBoilerPool + from ..timeseries.steam_boiler_pool._steam_boiler_pool_reference_store import ( + SteamBoilerPoolReferenceStore, + ) _logger = logging.getLogger(__name__) @@ -131,6 +140,13 @@ def __init__( # https://github.com/frequenz-floss/frequenz-sdk-python/issues/1285 component_class=SolarInverter, ) + self._steam_boiler_power_wrapper = PowerWrapper( + self._channel_registry, + api_power_request_timeout=api_power_request_timeout, + power_manager_algorithm=PowerManagerAlgorithm.MATRYOSHKA, + default_power=DefaultPower.ZERO, + component_class=SteamBoiler, + ) self._logical_meter: LogicalMeter | None = None self._consumer: Consumer | None = None @@ -145,6 +161,9 @@ def __init__( self._pv_pool_reference_stores: dict[ frozenset[ComponentId], PVPoolReferenceStore ] = {} + self._steam_boiler_pool_reference_stores: dict[ + frozenset[ComponentId], SteamBoilerPoolReferenceStore + ] = {} self._frequency_instance: GridFrequency | None = None self._voltage_instance: VoltageStreamer | None = None @@ -476,6 +495,83 @@ def new_battery_pool( priority=priority, ) + def new_steam_boiler_pool( + self, + *, + priority: int, + component_ids: abc.Set[ComponentId] | None = None, + name: str | None = None, + ) -> SteamBoilerPool: + """Return a new `SteamBoilerPool` instance for the given ids. + + If a `SteamBoilerPoolReferenceStore` instance for the given steam boiler ids + doesn't exist, a new one is created and used for creating the + `SteamBoilerPool`. + + Args: + priority: The priority of the actor making the call. + component_ids: Optional set of IDs of steam boilers to be managed by the + SteamBoilerPool. + name: An optional name used to identify this instance of the pool or a + corresponding actor in the logs. + + Returns: + A SteamBoilerPool instance. + """ + from ..timeseries.steam_boiler_pool import SteamBoilerPool + from ..timeseries.steam_boiler_pool._steam_boiler_pool_reference_store import ( + SteamBoilerPoolReferenceStore, + ) + + 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) + + pool_key = f"{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(...)`." + "\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, + priority, + ) + else: + self._known_pool_keys.add(pool_key) + + if ref_store_key not in self._steam_boiler_pool_reference_stores: + self._steam_boiler_pool_reference_stores[ref_store_key] = ( + SteamBoilerPoolReferenceStore( + channel_registry=self._channel_registry, + resampler_subscription_sender=self._resampling_request_sender(), + status_receiver=self._steam_boiler_power_wrapper.status_channel.new_receiver( + limit=1 + ), + power_manager_requests_sender=( + self._steam_boiler_power_wrapper.proposal_channel.new_sender() + ), + power_manager_bounds_subs_sender=( + self._steam_boiler_power_wrapper.bounds_subscription_channel.new_sender() + ), + power_distribution_results_fetcher=( + self._steam_boiler_power_wrapper.distribution_results_fetcher() + ), + component_ids=component_ids, + ) + ) + return SteamBoilerPool( + pool_ref_store=self._steam_boiler_pool_reference_stores[ref_store_key], + name=name, + priority=priority, + ) + def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]: """Return a Sender for sending requests to the data sourcing actor. @@ -532,12 +628,15 @@ async def _stop(self) -> None: await self._battery_power_wrapper.stop() await self._ev_power_wrapper.stop() await self._pv_power_wrapper.stop() + await self._steam_boiler_power_wrapper.stop() for pool in self._battery_pool_reference_stores.values(): await pool.stop() for evpool in self._ev_charger_pool_reference_stores.values(): await evpool.stop() for pvpool in self._pv_pool_reference_stores.values(): await pvpool.stop() + for steam_boiler_pool in self._steam_boiler_pool_reference_stores.values(): + await steam_boiler_pool.stop() _DATA_PIPELINE: _DataPipeline | None = None @@ -724,6 +823,45 @@ def new_pv_pool( return _get().new_pv_pool(priority=priority, component_ids=component_ids, name=name) +def new_steam_boiler_pool( + *, + priority: int, + component_ids: abc.Set[ComponentId] | None = None, + name: str | None = None, +) -> SteamBoilerPool: + """Return a new `SteamBoilerPool` 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 steam boilers. + + !!! note + When specifying priority, bigger values indicate higher priority. + + It is recommended to reuse the same instance of the `SteamBoilerPool` within the same + actor, unless they are managing different sets of steam boilers. + + In deployments with multiple actors managing the same set of steam boilers, 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: + priority: The priority of the actor making the call. + component_ids: Optional set of IDs of steam boilers to be managed by the + `SteamBoilerPool`. If not specified, all steam boilers 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. + + Returns: + A `SteamBoilerPool` instance. + """ + return _get().new_steam_boiler_pool( + priority=priority, component_ids=component_ids, name=name + ) + + def grid() -> Grid: """Return the grid measuring point.""" return _get().grid() diff --git a/src/frequenz/sdk/microgrid/_data_sourcing/microgrid_api_source.py b/src/frequenz/sdk/microgrid/_data_sourcing/microgrid_api_source.py index d43da8e58..d98cfea84 100644 --- a/src/frequenz/sdk/microgrid/_data_sourcing/microgrid_api_source.py +++ b/src/frequenz/sdk/microgrid/_data_sourcing/microgrid_api_source.py @@ -24,6 +24,7 @@ EVChargerData, InverterData, MeterData, + SteamBoilerData, TransitionalMetric, ) from ._component_metric_request import ComponentMetricRequest @@ -152,6 +153,38 @@ Metric.AC_REACTIVE_POWER_PHASE_3: lambda msg: msg.reactive_power_per_phase[2], } +_STEAM_BOILER_DATA_METHODS: dict[ + Metric | TransitionalMetric, Callable[[SteamBoilerData], float] +] = { + Metric.AC_ACTIVE_POWER: lambda msg: msg.active_power, + Metric.AC_ACTIVE_POWER_PHASE_1: lambda msg: msg.active_power_per_phase[0], + Metric.AC_ACTIVE_POWER_PHASE_2: lambda msg: msg.active_power_per_phase[1], + Metric.AC_ACTIVE_POWER_PHASE_3: lambda msg: msg.active_power_per_phase[2], + TransitionalMetric.ACTIVE_POWER_INCLUSION_LOWER_BOUND: lambda msg: ( + msg.active_power_inclusion_lower_bound + ), + TransitionalMetric.ACTIVE_POWER_EXCLUSION_LOWER_BOUND: lambda msg: ( + msg.active_power_exclusion_lower_bound + ), + TransitionalMetric.ACTIVE_POWER_EXCLUSION_UPPER_BOUND: lambda msg: ( + msg.active_power_exclusion_upper_bound + ), + TransitionalMetric.ACTIVE_POWER_INCLUSION_UPPER_BOUND: lambda msg: ( + msg.active_power_inclusion_upper_bound + ), + Metric.AC_CURRENT_PHASE_1: lambda msg: msg.current_per_phase[0], + Metric.AC_CURRENT_PHASE_2: lambda msg: msg.current_per_phase[1], + Metric.AC_CURRENT_PHASE_3: lambda msg: msg.current_per_phase[2], + Metric.AC_VOLTAGE_PHASE_1_N: lambda msg: msg.voltage_per_phase[0], + Metric.AC_VOLTAGE_PHASE_2_N: lambda msg: msg.voltage_per_phase[1], + Metric.AC_VOLTAGE_PHASE_3_N: lambda msg: msg.voltage_per_phase[2], + Metric.AC_FREQUENCY: lambda msg: msg.frequency, + Metric.AC_REACTIVE_POWER: lambda msg: msg.reactive_power, + Metric.AC_REACTIVE_POWER_PHASE_1: lambda msg: msg.reactive_power_per_phase[0], + Metric.AC_REACTIVE_POWER_PHASE_2: lambda msg: msg.reactive_power_per_phase[1], + Metric.AC_REACTIVE_POWER_PHASE_3: lambda msg: msg.reactive_power_per_phase[2], +} + class MicrogridApiSource: """Fetches requested metrics from the Microgrid API. @@ -306,6 +339,31 @@ async def _check_chp_request( connection_manager.get().api_client, comp_id ) + async def _check_steam_boiler_request( + self, + comp_id: ComponentId, + requests: dict[Metric | TransitionalMetric, list[ComponentMetricRequest]], + ) -> None: + """Check if the requests are valid steam boiler metrics. + + Raises: + ValueError: if the requested metric is not available for steam boilers. + + Args: + comp_id: The id of the requested component. + requests: A list of metric requests received from external actors + for the given steam boiler. + """ + for metric in requests: + if metric not in _STEAM_BOILER_DATA_METHODS: + err = f"Unknown metric {metric} for steam boiler id {comp_id}" + _logger.error(err) + raise ValueError(err) + if comp_id not in self.comp_data_receivers: + self.comp_data_receivers[comp_id] = SteamBoilerData.subscribe( + connection_manager.get().api_client, comp_id + ) + async def _check_meter_request( self, comp_id: ComponentId, @@ -362,6 +420,8 @@ async def _check_requested_component_and_metrics( await self._check_meter_request(comp_id, requests) elif category == ComponentCategory.CHP: await self._check_chp_request(comp_id, requests) + elif category == ComponentCategory.STEAM_BOILER: + await self._check_steam_boiler_request(comp_id, requests) else: err = f"Unknown component category {category}" _logger.error(err) @@ -393,6 +453,8 @@ def _get_data_extraction_method( return _EV_CHARGER_DATA_METHODS[metric] if category == ComponentCategory.CHP: return _CHP_DATA_METHODS[metric] + if category == ComponentCategory.STEAM_BOILER: + return _STEAM_BOILER_DATA_METHODS[metric] err = f"Unknown component category {category}" _logger.error(err) raise ValueError(err) diff --git a/src/frequenz/sdk/microgrid/_old_component_data.py b/src/frequenz/sdk/microgrid/_old_component_data.py index 84ea59b5e..c30f626db 100644 --- a/src/frequenz/sdk/microgrid/_old_component_data.py +++ b/src/frequenz/sdk/microgrid/_old_component_data.py @@ -1130,6 +1130,270 @@ def to_samples(self) -> ComponentDataSamples: ) +@dataclass(kw_only=True) +class SteamBoilerData(ComponentData): # pylint: disable=too-many-instance-attributes + """A wrapper class for holding steam boiler data. + + Steam boilers are controllable electrical loads. Their active power and its + inclusion/exclusion bounds, along with the per-phase power, current, voltage and + frequency, are reported through the corresponding `AC_*` metrics. + """ + + active_power: float = 0.0 + """The total active 3-phase AC power, in Watts (W). + + Represented in the passive sign convention. + + * Positive means consumption from the grid. + * Negative means supply into the grid. + """ + + active_power_per_phase: PhaseTuple = (0.0, 0.0, 0.0) + """The per-phase AC active power for phase 1, 2, and 3 respectively, in Watt (W). + + Represented in the passive sign convention. + + * Positive means consumption from the grid. + * Negative means supply into the grid. + """ + + reactive_power: float = 0.0 + """The total reactive 3-phase AC power, in Volt-Ampere Reactive (VAr). + + * Positive power means capacitive (current leading w.r.t. voltage). + * Negative power means inductive (current lagging w.r.t. voltage). + """ + + reactive_power_per_phase: PhaseTuple = (0.0, 0.0, 0.0) + """The per-phase AC reactive power, in Volt-Ampere Reactive (VAr). + + The provided values are for phase 1, 2, and 3 respectively. + + * Positive power means capacitive (current leading w.r.t. voltage). + * Negative power means inductive (current lagging w.r.t. voltage). + """ + + current_per_phase: PhaseTuple = (0.0, 0.0, 0.0) + """AC current in Amperes (A) for phase/line 1, 2 and 3 respectively. + + Represented in the passive sign convention. + + * Positive means consumption from the grid. + * Negative means supply into the grid. + """ + + voltage_per_phase: PhaseTuple = (0.0, 0.0, 0.0) + """The AC voltage in Volts (V) between the line and the neutral wire for + phase/line 1, 2 and 3 respectively. + """ + + active_power_inclusion_lower_bound: float = 0.0 + """Lower inclusion bound for steam boiler power in watts. + + This is the lower limit of the range within which power requests are allowed for the + steam boiler. + + See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and + [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more + details. + """ + + active_power_exclusion_lower_bound: float = 0.0 + """Lower exclusion bound for steam boiler power in watts. + + This is the lower limit of the range within which power requests are not allowed for + the steam boiler. + + See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and + [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more + details. + """ + + active_power_inclusion_upper_bound: float = 0.0 + """Upper inclusion bound for steam boiler power in watts. + + This is the upper limit of the range within which power requests are allowed for the + steam boiler. + + See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and + [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more + details. + """ + + active_power_exclusion_upper_bound: float = 0.0 + """Upper exclusion bound for steam boiler power in watts. + + This is the upper limit of the range within which power requests are not allowed for + the steam boiler. + + See [`frequenz.api.common.metrics_pb2.Metric.system_inclusion_bounds`][] and + [`frequenz.api.common.metrics_pb2.Metric.system_exclusion_bounds`][] for more + details. + """ + + frequency: float = 0.0 + """AC frequency, in Hertz (Hz).""" + + CATEGORY: ClassVar[ComponentCategory] = ComponentCategory.STEAM_BOILER + + METRICS: ClassVar[frozenset[Metric]] = frozenset( + [ + Metric.AC_ACTIVE_POWER, + Metric.AC_ACTIVE_POWER_PHASE_1, + Metric.AC_ACTIVE_POWER_PHASE_2, + Metric.AC_ACTIVE_POWER_PHASE_3, + Metric.AC_REACTIVE_POWER, + Metric.AC_REACTIVE_POWER_PHASE_1, + Metric.AC_REACTIVE_POWER_PHASE_2, + Metric.AC_REACTIVE_POWER_PHASE_3, + Metric.AC_CURRENT_PHASE_1, + Metric.AC_CURRENT_PHASE_2, + Metric.AC_CURRENT_PHASE_3, + Metric.AC_VOLTAGE_PHASE_1_N, + Metric.AC_VOLTAGE_PHASE_2_N, + Metric.AC_VOLTAGE_PHASE_3_N, + Metric.AC_FREQUENCY, + ] + ) + """The metrics of this component.""" + + @override + @classmethod + # pylint: disable-next=too-many-branches + def from_samples(cls, samples: ComponentDataSamples) -> Self: + """Create a new instance from a component data object.""" + self = cls._from_samples(cls, samples) + + active_power_per_phase: list[float] = [0.0, 0.0, 0.0] + reactive_power_per_phase: list[float] = [0.0, 0.0, 0.0] + current_per_phase: list[float] = [0.0, 0.0, 0.0] + voltage_per_phase: list[float] = [0.0, 0.0, 0.0] + + for sample in samples.metric_samples: + value = sample.as_single_value() or 0.0 + match sample.metric: + case _M.AC_ACTIVE_POWER: + self.active_power = value + ( + self.active_power_inclusion_lower_bound, + self.active_power_inclusion_upper_bound, + self.active_power_exclusion_lower_bound, + self.active_power_exclusion_upper_bound, + ) = _bound_ranges_to_inclusion_exclusion( + sample.bounds, "AC_ACTIVE_POWER", sample + ) + case _M.AC_ACTIVE_POWER_PHASE_1: + active_power_per_phase[0] = value + case _M.AC_ACTIVE_POWER_PHASE_2: + active_power_per_phase[1] = value + case _M.AC_ACTIVE_POWER_PHASE_3: + active_power_per_phase[2] = value + case _M.AC_REACTIVE_POWER: + self.reactive_power = value + case _M.AC_REACTIVE_POWER_PHASE_1: + reactive_power_per_phase[0] = value + case _M.AC_REACTIVE_POWER_PHASE_2: + reactive_power_per_phase[1] = value + case _M.AC_REACTIVE_POWER_PHASE_3: + reactive_power_per_phase[2] = value + case _M.AC_CURRENT_PHASE_1: + current_per_phase[0] = value + case _M.AC_CURRENT_PHASE_2: + current_per_phase[1] = value + case _M.AC_CURRENT_PHASE_3: + current_per_phase[2] = value + case _M.AC_VOLTAGE_PHASE_1_N: + voltage_per_phase[0] = value + case _M.AC_VOLTAGE_PHASE_2_N: + voltage_per_phase[1] = value + case _M.AC_VOLTAGE_PHASE_3_N: + voltage_per_phase[2] = value + case _M.AC_FREQUENCY: + self.frequency = value + case unexpected: + _logger.warning( + "Unexpected metric %s in steam boiler data sample: %r", + unexpected, + sample, + ) + + self.active_power_per_phase = cast(PhaseTuple, tuple(active_power_per_phase)) + self.reactive_power_per_phase = cast( + PhaseTuple, tuple(reactive_power_per_phase) + ) + self.current_per_phase = cast(PhaseTuple, tuple(current_per_phase)) + self.voltage_per_phase = cast(PhaseTuple, tuple(voltage_per_phase)) + + return self + + @override + def to_samples(self) -> ComponentDataSamples: + """Convert the component data to a component data object.""" + return ComponentDataSamples( + component_id=self.component_id, + metric_samples=[ + MetricSample( + sampled_at=self.timestamp, + metric=Metric.AC_ACTIVE_POWER, + value=self.active_power, + bounds=_inclusion_exclusion_bounds_to_ranges( + self.active_power_inclusion_lower_bound, + self.active_power_inclusion_upper_bound, + self.active_power_exclusion_lower_bound, + self.active_power_exclusion_upper_bound, + ), + ), + *( + MetricSample( + sampled_at=self.timestamp, metric=metric, value=value, bounds=[] + ) + for metric, value in [ + ( + Metric.AC_ACTIVE_POWER_PHASE_1, + self.active_power_per_phase[0], + ), + ( + Metric.AC_ACTIVE_POWER_PHASE_2, + self.active_power_per_phase[1], + ), + ( + Metric.AC_ACTIVE_POWER_PHASE_3, + self.active_power_per_phase[2], + ), + (Metric.AC_REACTIVE_POWER, self.reactive_power), + ( + Metric.AC_REACTIVE_POWER_PHASE_1, + self.reactive_power_per_phase[0], + ), + ( + Metric.AC_REACTIVE_POWER_PHASE_2, + self.reactive_power_per_phase[1], + ), + ( + Metric.AC_REACTIVE_POWER_PHASE_3, + self.reactive_power_per_phase[2], + ), + (Metric.AC_CURRENT_PHASE_1, self.current_per_phase[0]), + (Metric.AC_CURRENT_PHASE_2, self.current_per_phase[1]), + (Metric.AC_CURRENT_PHASE_3, self.current_per_phase[2]), + (Metric.AC_VOLTAGE_PHASE_1_N, self.voltage_per_phase[0]), + (Metric.AC_VOLTAGE_PHASE_2_N, self.voltage_per_phase[1]), + (Metric.AC_VOLTAGE_PHASE_3_N, self.voltage_per_phase[2]), + (Metric.AC_FREQUENCY, self.frequency), + ] + ), + ], + states=[ + ComponentStateSample( + sampled_at=self.timestamp, + states=frozenset(self.states), + warnings=frozenset(self.warnings), + errors=frozenset(self.errors), + ) + ], + ) + + @dataclass(kw_only=True) class EVChargerData(ComponentData): # pylint: disable=too-many-instance-attributes """A wrapper class for holding ev_charger data.""" diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/__init__.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/__init__.py index 606dfea99..4bdb43dd9 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/__init__.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/__init__.py @@ -7,10 +7,12 @@ from ._component_manager import ComponentManager from ._ev_charger_manager import EVChargerManager from ._pv_inverter_manager import PVManager +from ._steam_boiler_manager import SteamBoilerManager __all__ = [ "BatteryManager", "ComponentManager", "EVChargerManager", "PVManager", + "SteamBoilerManager", ] diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_steam_boiler_manager/__init__.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_steam_boiler_manager/__init__.py new file mode 100644 index 000000000..4512d5962 --- /dev/null +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_steam_boiler_manager/__init__.py @@ -0,0 +1,8 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Manage steam boilers for the power distributor.""" + +from ._steam_boiler_manager import SteamBoilerManager + +__all__ = ["SteamBoilerManager"] diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_steam_boiler_manager/_steam_boiler_manager.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_steam_boiler_manager/_steam_boiler_manager.py new file mode 100644 index 000000000..5292065e5 --- /dev/null +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_steam_boiler_manager/_steam_boiler_manager.py @@ -0,0 +1,239 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Manage steam boilers for the power distributor.""" + +import asyncio +import collections.abc +import logging +from datetime import timedelta + +from frequenz.channels import LatestValueCache, Sender +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.client.microgrid.component import SteamBoiler +from frequenz.quantities import Power +from typing_extensions import override + +from ....._internal._math import is_close_to_zero +from .... import connection_manager +from ...._old_component_data import SteamBoilerData +from ..._component_pool_status_tracker import ComponentPoolStatusTracker +from ..._component_status import ComponentPoolStatus, SteamBoilerStatusTracker +from ...request import Request +from ...result import Result, Success +from .._component_manager import ComponentManager +from .._utils import _set_component_power + +_logger = logging.getLogger(__name__) + + +class SteamBoilerManager(ComponentManager): + """Manage steam boilers for the power distributor.""" + + @override + def __init__( + self, + component_pool_status_sender: Sender[ComponentPoolStatus], + results_sender: Sender[Result], + api_power_request_timeout: timedelta, + ) -> None: + """Initialize this instance. + + Args: + component_pool_status_sender: Channel for sending information about which + components are expected to be working. + results_sender: Channel for sending results of power distribution. + api_power_request_timeout: Timeout to use when making power requests to + the microgrid API. + """ + super().__init__() + self._results_sender = results_sender + self._api_power_request_timeout = api_power_request_timeout + self._steam_boiler_ids = self._get_steam_boiler_ids() + + self._component_pool_status_tracker = ( + ComponentPoolStatusTracker( + component_ids=self._steam_boiler_ids, + component_status_sender=component_pool_status_sender, + max_data_age=timedelta(seconds=10.0), + max_blocking_duration=timedelta(seconds=30.0), + component_status_tracker_type=SteamBoilerStatusTracker, + ) + if self._steam_boiler_ids + else None + ) + self._component_data_caches: dict[ + ComponentId, LatestValueCache[SteamBoilerData] + ] = {} + + self._task: asyncio.Task[None] | None = None + + @override + def component_ids(self) -> collections.abc.Set[ComponentId]: + """Return the set of steam boiler ids.""" + return self._steam_boiler_ids + + @override + async def start(self) -> None: + """Start the steam boiler manager.""" + self._component_data_caches = { + steam_boiler_id: LatestValueCache( + SteamBoilerData.subscribe( + connection_manager.get().api_client, steam_boiler_id + ), + unique_id=( + f"{type(self).__name__}«{hex(id(self))}»:" + f"steam_boiler«{steam_boiler_id}»" + ), + ) + for steam_boiler_id in self._steam_boiler_ids + } + + @override + async def stop(self) -> None: + """Stop the steam boiler manager.""" + await asyncio.gather( + *[cache.stop() for cache in self._component_data_caches.values()] + ) + if self._component_pool_status_tracker: + await self._component_pool_status_tracker.stop() + await self._stop_all_unreachable_power_subscriptions() + + @override + async def distribute_power(self, request: Request) -> None: + """Distribute the requested power to the steam boilers. + + Args: + request: Request to get the distribution for. + + Raises: + ValueError: If no steam boilers are present in the component graph, but + component_ids are provided in the request. + """ + remaining_power = request.power + allocations: dict[ComponentId, Power] = {} + if not self._component_pool_status_tracker: + if not request.component_ids: + await self._results_sender.send( + Success( + succeeded_components=set(), + succeeded_power=Power.zero(), + excess_power=remaining_power, + request=request, + ) + ) + return + raise ValueError( + "Cannot distribute power to steam boilers - None found in the component graph." + ) + + working = self._component_pool_status_tracker.get_working_components( + request.component_ids + ) + await self._subscribe_to_unreachable_power(request.component_ids, working) + unreachable_power = self._unreachable_power(request.component_ids) + if unreachable_power is not None: + remaining_power -= unreachable_power + _logger.debug( + "Excluding %s measured on unreachable steam boilers from the power " + "to distribute on working steam boilers.", + unreachable_power, + ) + + working_components: list[ComponentId] = [] + for comp_id in working: + if self._component_data_caches[comp_id].has_value(): + working_components.append(comp_id) + else: + _logger.warning( + "Excluding steam boiler %s from power distribution due to " + "lack of data since startup.", + comp_id, + ) + + num_components = len(working_components) + if num_components == 0: + _logger.error( + "No steam boilers available for power distribution. Aborting." + ) + return + + working_components.sort( + key=lambda comp_id: self._component_data_caches[comp_id] + .get() + .active_power_inclusion_upper_bound, + ) + + for idx, comp_id in enumerate(working_components): + # When no power is left to distribute, + # set power to zero for all remaining steam boilers. + if remaining_power < Power.zero() or is_close_to_zero( + remaining_power.as_watts() + ): + allocations[comp_id] = Power.zero() + continue + + component_data = self._component_data_caches[comp_id] + if not component_data.has_value(): + allocations[comp_id] = Power.zero() + continue + + # Never allocate more than a steam boiler's upper power bound. + upper_bound = Power.from_watts( + component_data.get().active_power_inclusion_upper_bound + ) + allocated_power = min( + remaining_power, + remaining_power / float(num_components - idx), + upper_bound, + ) + # A boiler with a minimum operating power can't run below it: raise + # the allocation to the minimum when the remaining power covers it, + # otherwise keep the boiler off. Power that a kept-off boiler's + # share would have used is not redistributed to earlier boilers; + # it is reported as excess power in the result. + lower_bound = Power.from_watts( + component_data.get().active_power_inclusion_lower_bound + ) + if Power.zero() < allocated_power < lower_bound: + allocated_power = ( + lower_bound + if lower_bound <= upper_bound and remaining_power >= lower_bound + else Power.zero() + ) + allocations[comp_id] = allocated_power + remaining_power -= allocated_power + + _logger.debug( + "Distributing %s to steam boilers %s", + request.power, + allocations, + ) + + result = await _set_component_power( + request=request, + target_power=request.power, + allocations=allocations, + api_request_timeout=self._api_power_request_timeout, + remaining_power=remaining_power, + component_category="steam boiler", + ) + await self._results_sender.send(result) + + @override + def _unreachable_power_formula( + self, component_ids: collections.abc.Set[ComponentId] + ) -> str: + """Return the formula for the active power of the given steam boilers.""" + return connection_manager.get().component_graph.steam_boiler_formula( + component_ids + ) + + def _get_steam_boiler_ids(self) -> collections.abc.Set[ComponentId]: + """Return the IDs of all steam boilers present in the component graph.""" + return { + boiler.id + for boiler in connection_manager.get().component_graph.components( + matching_types=SteamBoiler + ) + } diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/__init__.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/__init__.py index 4794cbf34..55a1cb966 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/__init__.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/__init__.py @@ -13,6 +13,7 @@ ) from ._ev_charger_status_tracker import EVChargerStatusTracker from ._pv_inverter_status_tracker import PVInverterStatusTracker +from ._steam_boiler_status_tracker import SteamBoilerStatusTracker __all__ = [ "BatteryStatusTracker", @@ -23,4 +24,5 @@ "EVChargerStatusTracker", "PVInverterStatusTracker", "SetPowerResult", + "SteamBoilerStatusTracker", ] diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py new file mode 100644 index 000000000..f12049480 --- /dev/null +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py @@ -0,0 +1,191 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Background service that tracks the status of a steam boiler.""" + +import asyncio +import logging +from datetime import datetime, timedelta, timezone + +from frequenz.channels import Receiver, Sender, select, selected_from +from frequenz.channels.timer import SkipMissedAndDrift, Timer +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.client.microgrid.component import ComponentStateCode +from typing_extensions import override + +from ...._internal._asyncio import run_forever +from ....actor._background_service import BackgroundService +from ... import connection_manager +from ..._old_component_data import SteamBoilerData +from ._blocking_status import BlockingStatus +from ._component_status import ( + ComponentStatus, + ComponentStatusEnum, + ComponentStatusTracker, + SetPowerResult, +) + +_logger = logging.getLogger(__name__) + + +class SteamBoilerStatusTracker(ComponentStatusTracker, BackgroundService): + """Status tracker for steam boilers. + + It reports a steam boiler as `WORKING` or `NOT_WORKING` based on the status in the + received component data from the microgrid API. When no data is received for a + specific duration, the component is marked as `NOT_WORKING`. + + If it receives a power assignment failure from the PowerDistributor, when the + component is expected to be `WORKING`, it is marked as `UNCERTAIN` for a specific + interval, before being marked `WORKING` again. + """ + + @override + def __init__( # pylint: disable=too-many-arguments + self, + *, + component_id: ComponentId, + max_data_age: timedelta, + max_blocking_duration: timedelta, + status_sender: Sender[ComponentStatus], + set_power_result_receiver: Receiver[SetPowerResult], + ) -> None: + """Initialize this instance. + + Args: + component_id: ID of the steam boiler to monitor the status of. + max_data_age: max duration to wait for, before marking a component as + NOT_WORKING, unless new data arrives. + max_blocking_duration: duration for which the component status should be + UNCERTAIN if a request to the component failed unexpectedly. + status_sender: Sender to send the status of the steam boiler. + set_power_result_receiver: Receiver for the power assignment result. + """ + BackgroundService.__init__( + self, name=f"SteamBoilerStatusTracker({component_id})" + ) + self._component_id = component_id + self._max_data_age = max_data_age + self._status_sender = status_sender + self._set_power_result_receiver = set_power_result_receiver + + self._last_status = ComponentStatusEnum.NOT_WORKING + self._blocking_status = BlockingStatus( + min_duration=timedelta(seconds=1.0), + max_duration=max_blocking_duration, + ) + + @override + def start(self) -> None: + """Start the status tracker.""" + self._tasks.add(asyncio.create_task(run_forever(self._run))) + + def _is_working(self, steam_boiler_data: SteamBoilerData) -> bool: + """Return whether the given data indicates the steam boiler is working.""" + return bool( + { + ComponentStateCode.CHARGING, + ComponentStateCode.READY, + } + & steam_boiler_data.states + ) + + def _is_stale(self, steam_boiler_data: SteamBoilerData) -> bool: + """Return whether the given data is stale.""" + now = datetime.now(tz=timezone.utc) + stale = now - steam_boiler_data.timestamp > self._max_data_age + return stale + + def _handle_set_power_result( + self, set_power_result: SetPowerResult + ) -> ComponentStatusEnum: + """Handle a new set power result.""" + if self._component_id in set_power_result.succeeded: + return ComponentStatusEnum.WORKING + + self._blocking_status.block() + if self._last_status == ComponentStatusEnum.WORKING: + _logger.warning( + "Steam boiler %s is in UNCERTAIN state. Set power result: %s", + self._component_id, + set_power_result, + ) + return ComponentStatusEnum.UNCERTAIN + + def _handle_steam_boiler_data( + self, steam_boiler_data: SteamBoilerData + ) -> ComponentStatusEnum: + """Handle new steam boiler data.""" + if self._is_stale(steam_boiler_data): + if self._last_status == ComponentStatusEnum.WORKING: + _logger.warning( + "Steam boiler %s data is stale. Last timestamp: %s", + self._component_id, + steam_boiler_data.timestamp, + ) + return ComponentStatusEnum.NOT_WORKING + + if self._is_working(steam_boiler_data): + if self._last_status == ComponentStatusEnum.NOT_WORKING: + _logger.info( + "Steam boiler %s: state changed to WORKING.", + self._component_id, + ) + return ComponentStatusEnum.WORKING + + if self._last_status == ComponentStatusEnum.WORKING: + _logger.warning( + "Steam boiler %s is in NOT_WORKING state. Component states: %s", + self._component_id, + steam_boiler_data.states, + ) + return ComponentStatusEnum.NOT_WORKING + + async def _run(self) -> None: + """Run the status tracker.""" + steam_boiler_data_rx = SteamBoilerData.subscribe( + connection_manager.get().api_client, self._component_id + ) + set_power_result_rx = self._set_power_result_receiver + missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift()) + + # Send initial status + await self._status_sender.send( + ComponentStatus(self._component_id, self._last_status) + ) + + async for selected in select( + steam_boiler_data_rx, set_power_result_rx, missing_data_timer + ): + new_status = ComponentStatusEnum.NOT_WORKING + if selected_from(selected, steam_boiler_data_rx): + missing_data_timer.reset() + new_status = self._handle_steam_boiler_data(selected.message) + elif selected_from(selected, set_power_result_rx): + new_status = self._handle_set_power_result(selected.message) + elif selected_from(selected, missing_data_timer): + _logger.warning( + "No steam boiler %s data received for %s. " + "Setting status to NOT_WORKING.", + self._component_id, + self._max_data_age, + ) + + # Send status update if status changed + if ( + self._blocking_status.is_blocked() + and new_status != ComponentStatusEnum.NOT_WORKING + ): + new_status = ComponentStatusEnum.UNCERTAIN + + if new_status != self._last_status: + _logger.info( + "Steam boiler %s status changed from %s to %s", + self._component_id, + self._last_status, + new_status, + ) + self._last_status = new_status + await self._status_sender.send( + ComponentStatus(self._component_id, new_status) + ) diff --git a/src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py b/src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py index 90a14f078..45061b646 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/power_distributing.py @@ -16,7 +16,12 @@ from frequenz.channels import Receiver, Sender from frequenz.client.common.microgrid.components import ComponentId -from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter +from frequenz.client.microgrid.component import ( + Battery, + EvCharger, + SolarInverter, + SteamBoiler, +) from typing_extensions import override from ...actor._actor import Actor @@ -25,6 +30,7 @@ ComponentManager, EVChargerManager, PVManager, + SteamBoilerManager, ) from ._component_status import ComponentPoolStatus from .request import Request @@ -60,7 +66,7 @@ class PowerDistributingActor(Actor): # pylint: disable=too-many-instance-attrib def __init__( # pylint: disable=too-many-arguments self, - component_type: type[Battery | EvCharger | SolarInverter], + component_type: type[Battery | EvCharger | SolarInverter | SteamBoiler], requests_receiver: Receiver[Request], results_sender: Sender[Result], component_pool_status_sender: Sender[ComponentPoolStatus], @@ -114,6 +120,10 @@ def __init__( # pylint: disable=too-many-arguments self._component_manager = PVManager( component_pool_status_sender, results_sender, api_power_request_timeout ) + elif issubclass(component_type, SteamBoiler): + self._component_manager = SteamBoilerManager( + component_pool_status_sender, results_sender, api_power_request_timeout + ) else: assert_never(component_type) diff --git a/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py b/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py index d200bd0ff..1f712f254 100644 --- a/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py +++ b/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py @@ -14,7 +14,12 @@ from frequenz.channels import Receiver, Sender, select, selected_from from frequenz.channels.timer import SkipMissedAndDrift, Timer from frequenz.client.common.microgrid.components import ComponentId -from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter +from frequenz.client.microgrid.component import ( + Battery, + EvCharger, + SolarInverter, + SteamBoiler, +) from typing_extensions import override from ..._internal._asyncio import run_forever @@ -49,7 +54,7 @@ def __init__( # pylint: disable=too-many-arguments channel_registry: ChannelRegistry, algorithm: PowerManagerAlgorithm, default_power: DefaultPower, - component_class: type[Battery | EvCharger | SolarInverter], + component_class: type[Battery | EvCharger | SolarInverter | SteamBoiler], ): """Create a new instance of the power manager. @@ -160,6 +165,11 @@ def _add_system_bounds_tracker(self, component_ids: frozenset[ComponentId]) -> N priority=-sys.maxsize - 1, component_ids=component_ids ) bounds_receiver = pv_pool.system_power_bounds.new_receiver() + elif issubclass(self._component_class, SteamBoiler): + steam_boiler_pool = _data_pipeline.new_steam_boiler_pool( + priority=-sys.maxsize - 1, component_ids=component_ids + ) + bounds_receiver = steam_boiler_pool.system_power_bounds.new_receiver() else: _logger.error( "PowerManagingActor: Unsupported component class: %s", diff --git a/src/frequenz/sdk/microgrid/_power_wrapper.py b/src/frequenz/sdk/microgrid/_power_wrapper.py index 70e2e4e55..8ed1bd14f 100644 --- a/src/frequenz/sdk/microgrid/_power_wrapper.py +++ b/src/frequenz/sdk/microgrid/_power_wrapper.py @@ -9,7 +9,12 @@ from datetime import timedelta from frequenz.channels import Broadcast -from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter +from frequenz.client.microgrid.component import ( + Battery, + EvCharger, + SolarInverter, + SteamBoiler, +) from .._internal._channels import ChannelRegistry, ReceiverFetcher @@ -40,7 +45,7 @@ def __init__( # pylint: disable=too-many-arguments api_power_request_timeout: timedelta, power_manager_algorithm: PowerManagerAlgorithm, default_power: DefaultPower, - component_class: type[Battery | EvCharger | SolarInverter], + component_class: type[Battery | EvCharger | SolarInverter | SteamBoiler], ): """Initialize the power control. diff --git a/src/frequenz/sdk/timeseries/steam_boiler_pool/__init__.py b/src/frequenz/sdk/timeseries/steam_boiler_pool/__init__.py new file mode 100644 index 000000000..1be6ef82a --- /dev/null +++ b/src/frequenz/sdk/timeseries/steam_boiler_pool/__init__.py @@ -0,0 +1,13 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Interactions with pools of steam boilers.""" + +from ._result_types import SteamBoilerPoolReport +from ._steam_boiler_pool import SteamBoilerPool, SteamBoilerPoolError + +__all__ = [ + "SteamBoilerPool", + "SteamBoilerPoolError", + "SteamBoilerPoolReport", +] diff --git a/src/frequenz/sdk/timeseries/steam_boiler_pool/_result_types.py b/src/frequenz/sdk/timeseries/steam_boiler_pool/_result_types.py new file mode 100644 index 000000000..0c1e9ade2 --- /dev/null +++ b/src/frequenz/sdk/timeseries/steam_boiler_pool/_result_types.py @@ -0,0 +1,27 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Types for exposing steam boiler pool reports.""" + +import typing + +from frequenz.quantities import Power + +from .. import Bounds +from ..component_pool._component_pool_report import ComponentPoolReport + + +class SteamBoilerPoolReport(ComponentPoolReport, typing.Protocol): + """A status report for a steam boiler pool.""" + + @property + def target_power(self) -> Power | None: + """The currently set power for the steam boilers.""" + + @property + def bounds(self) -> Bounds[Power] | None: + """The usable bounds for the steam boilers. + + These bounds are adjusted to any restrictions placed by actors with higher + priorities. + """ diff --git a/src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py b/src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py new file mode 100644 index 000000000..63c1e6190 --- /dev/null +++ b/src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool.py @@ -0,0 +1,85 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Interactions with pools of steam boilers.""" + +from frequenz.quantities import Power +from typing_extensions import override + +from ...microgrid import connection_manager +from ...timeseries import Bounds +from ..component_pool import ComponentPool +from ..formulas import Formula +from ._result_types import SteamBoilerPoolReport +from ._steam_boiler_pool_reference_store import SteamBoilerPoolReferenceStore + + +class SteamBoilerPoolError(Exception): + """An error that occurred in any of the SteamBoilerPool methods.""" + + +class SteamBoilerPool( + ComponentPool[SteamBoilerPoolReferenceStore, SteamBoilerPoolReport] +): + """An interface for interaction with pools of steam boilers. + + Provides: + - Aggregate + [`power`][frequenz.sdk.timeseries.steam_boiler_pool.SteamBoilerPool.power] + measurements of the steam boilers in the pool. + """ + + @override + async def propose_power( + self, + power: Power | None, + bounds: Bounds[Power | None] = Bounds(None, None), + ) -> None: + """Send a proposal to the power manager for the pool's set of steam boilers. + + Steam boilers are controllable loads, so this proposal is for the power the + boilers in the pool should consume. + + Details on how the power manager handles proposals can be found in the + [Microgrid][frequenz.sdk.microgrid--setting-power] documentation. + + Args: + power: The power to propose for the steam boilers in the pool. If `None`, + this proposal will not have any effect on the target power, unless + bounds are specified. When specified without bounds, bounds for lower + priority actors will be shifted by this power. If both are `None`, it + is equivalent to not having a proposal or withdrawing a previous one. + bounds: The power bounds for the proposal. When specified, these bounds will + limit the bounds for lower priority actors. + + Raises: + SteamBoilerPoolError: If a discharge power for steam boilers is requested. + """ + if power is not None and power < Power.zero(): + raise SteamBoilerPoolError( + "Discharging from steam boilers is not supported." + ) + await super().propose_power(power, bounds=bounds) + + @property + @override + def power(self) -> Formula[Power]: + """Fetch the total power for the steam boilers in the pool. + + This formula produces values that are in the Passive Sign Convention (PSC). + + If a formula to calculate steam boiler power is not already running, it + will be started. + + A receiver from the formula can be created using the `new_receiver` + method. + + Returns: + A Formula that will calculate and stream the total power of all steam boilers. + """ + return self._pool_ref_store.formula_pool.from_power_formula( + "steam_boiler_power", + connection_manager.get().component_graph.steam_boiler_formula( + self._pool_ref_store.component_ids + ), + ) diff --git a/src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool_reference_store.py b/src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool_reference_store.py new file mode 100644 index 000000000..559463ba9 --- /dev/null +++ b/src/frequenz/sdk/timeseries/steam_boiler_pool/_steam_boiler_pool_reference_store.py @@ -0,0 +1,59 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Manages shared state/tasks for a set of steam boilers.""" + +import uuid +from typing import Type + +from frequenz.client.microgrid.component import Component, SteamBoiler +from typing_extensions import override + +from ..component_pool._component_pool_reference_store import ComponentPoolReferenceStore +from ._system_bounds_tracker import SteamBoilerSystemBoundsTracker + + +class SteamBoilerPoolReferenceStore(ComponentPoolReferenceStore): + """A class for maintaining the shared state/tasks for a set of pools of steam boilers. + + This includes ownership of + - the formula pool and metric calculators. + - the tasks for calculating system bounds for the steam boilers. + + These are independent of the priority of the actors and can be shared between + multiple users of the same set of steam boilers. + + They are exposed through the SteamBoilerPool class. + """ + + @staticmethod + def get_component_class() -> Type[Component]: + """Class of the component type.""" + return SteamBoiler + + @staticmethod + def get_pool_type_name() -> str: + """Name of the pool type, for display purposes.""" + return "SteamBoilerPool" + + @staticmethod + def get_component_type_name_plural() -> str: + """Name of the component type, for display purposes.""" + return "steam boilers" + + @override + def get_namespace(self) -> str: + """Namespace to use with the data pipeline.""" + return f"steam-boiler-pool-{uuid.uuid4()}" + + @override + def create_bounds_tracker(self) -> None: + """Create the bounds tracker for the pool.""" + # In locations without steam boilers, the bounds tracker will not be started. + if self.component_ids: + self.bounds_tracker = SteamBoilerSystemBoundsTracker( + self.component_ids, + self.status_receiver, + self.bounds_channel.new_sender(), + ) + self.bounds_tracker.start() diff --git a/src/frequenz/sdk/timeseries/steam_boiler_pool/_system_bounds_tracker.py b/src/frequenz/sdk/timeseries/steam_boiler_pool/_system_bounds_tracker.py new file mode 100644 index 000000000..4a3748641 --- /dev/null +++ b/src/frequenz/sdk/timeseries/steam_boiler_pool/_system_bounds_tracker.py @@ -0,0 +1,138 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""System bounds tracker for steam boilers.""" + +import asyncio +from collections import abc + +from frequenz.channels import Receiver, Sender, merge, select, selected_from +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.quantities import Power + +from ..._internal._asyncio import run_forever +from ...actor import BackgroundService +from ...microgrid import connection_manager +from ...microgrid._old_component_data import SteamBoilerData +from ...microgrid._power_distributing._component_status import ComponentPoolStatus +from .._base_types import Bounds, SystemBounds + + +class SteamBoilerSystemBoundsTracker(BackgroundService): + """Track the system bounds for steam boilers. + + System bounds are the aggregate bounds for the steam boilers in the pool that are + in a working state. They are calculated from the individual bounds received from + the microgrid API. + + The system bounds are sent to the `bounds_sender` whenever they change. + """ + + def __init__( + self, + component_ids: abc.Set[ComponentId], + status_receiver: Receiver[ComponentPoolStatus], + bounds_sender: Sender[SystemBounds], + ): + """Initialize the system bounds tracker. + + Args: + component_ids: The ids of the components to track. + status_receiver: A receiver that streams the status of the steam boilers in + the pool. + bounds_sender: A sender to send the system bounds to. + """ + super().__init__() + + self._component_ids = component_ids + self._status_receiver = status_receiver + self._bounds_sender = bounds_sender + self._latest_component_data: dict[ComponentId, SteamBoilerData] = {} + self._last_sent_bounds: SystemBounds | None = None + self._component_pool_status = ComponentPoolStatus(set(), set()) + + def start(self) -> None: + """Start the steam boiler system bounds tracker.""" + self._tasks.add(asyncio.create_task(run_forever(self._run))) + + async def _send_bounds(self) -> None: + """Calculate and send the aggregate system bounds if they have changed.""" + if not self._latest_component_data: + return + inclusion_bounds = Bounds( + lower=Power.from_watts( + sum( + data.active_power_inclusion_lower_bound + for data in self._latest_component_data.values() + ) + ), + upper=Power.from_watts( + sum( + data.active_power_inclusion_upper_bound + for data in self._latest_component_data.values() + ) + ), + ) + exclusion_bounds = Bounds( + lower=Power.from_watts( + sum( + data.active_power_exclusion_lower_bound + for data in self._latest_component_data.values() + ) + ), + upper=Power.from_watts( + sum( + data.active_power_exclusion_upper_bound + for data in self._latest_component_data.values() + ) + ), + ) + + if ( + self._last_sent_bounds is None + or self._last_sent_bounds.inclusion_bounds != inclusion_bounds + or self._last_sent_bounds.exclusion_bounds != exclusion_bounds + ): + self._last_sent_bounds = SystemBounds( + timestamp=max( + data.timestamp for data in self._latest_component_data.values() + ), + inclusion_bounds=inclusion_bounds, + exclusion_bounds=exclusion_bounds, + ) + await self._bounds_sender.send(self._last_sent_bounds) + + async def _run(self) -> None: + """Run the system bounds tracker.""" + api_client = connection_manager.get().api_client + status_rx = self._status_receiver + steam_boiler_data_rx = merge( + *( + SteamBoilerData.subscribe(api_client, component_id) + for component_id in self._component_ids + ) + ) + + async for selected in select(status_rx, steam_boiler_data_rx): + if selected_from(selected, status_rx): + self._component_pool_status = selected.message + to_remove: list[ComponentId] = [] + for comp_id in self._latest_component_data: + if ( + comp_id not in self._component_pool_status.working + and comp_id not in self._component_pool_status.uncertain + ): + to_remove.append(comp_id) + for comp_id in to_remove: + del self._latest_component_data[comp_id] + elif selected_from(selected, steam_boiler_data_rx): + data = selected.message + comp_id = data.component_id + if ( + comp_id not in self._component_pool_status.working + and comp_id not in self._component_pool_status.uncertain + ): + continue + self._latest_component_data[data.component_id] = data + + await self._send_bounds() diff --git a/tests/microgrid/power_distributing/_component_status/test_steam_boiler_status.py b/tests/microgrid/power_distributing/_component_status/test_steam_boiler_status.py new file mode 100644 index 000000000..51d7d7f30 --- /dev/null +++ b/tests/microgrid/power_distributing/_component_status/test_steam_boiler_status.py @@ -0,0 +1,224 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for SteamBoilerStatusTracker.""" + +import asyncio +from datetime import datetime, timedelta, timezone + +from frequenz.channels import Broadcast +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.client.microgrid.component import ComponentStateCode +from pytest_mock import MockerFixture + +from frequenz.sdk._internal._asyncio import cancel_and_await +from frequenz.sdk.microgrid._power_distributing._component_status import ( + ComponentStatus, + ComponentStatusEnum, + SetPowerResult, + SteamBoilerStatusTracker, +) + +from ....timeseries.mock_microgrid import MockMicrogrid +from ....utils.component_data_wrapper import SteamBoilerDataWrapper +from ....utils.receive_timeout import Timeout, receive_timeout + +_STEAM_BOILER_ID = ComponentId(3) + + +class TestSteamBoilerStatusTracker: + """Tests for SteamBoilerStatusTracker.""" + + async def test_status_changes(self, mocker: MockerFixture) -> None: + """Test that the status changes as expected.""" + mock_microgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mock_microgrid.add_steam_boilers(1) + + status_channel = Broadcast[ComponentStatus](name="steam_boiler_status") + set_power_result_channel = Broadcast[SetPowerResult](name="set_power_result") + set_power_result_sender = set_power_result_channel.new_sender() + + async with ( + mock_microgrid, + SteamBoilerStatusTracker( + component_id=_STEAM_BOILER_ID, + max_data_age=timedelta(seconds=0.2), + max_blocking_duration=timedelta(seconds=1), + status_sender=status_channel.new_sender(), + set_power_result_receiver=set_power_result_channel.new_receiver(), + ), + ): + status_receiver = status_channel.new_receiver() + # The status is initially not working. + assert ( + await status_receiver.receive() + ).value == ComponentStatusEnum.NOT_WORKING + + # When there's healthy steam boiler data, status should be working. + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.WORKING + ) + + # When it is charging, there should be no change in status + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.CHARGING}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) is Timeout + + # Steam boilers only consume: DISCHARGING is not a healthy state, so + # the status should be not working. + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.DISCHARGING}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.NOT_WORKING + ) + + # STANDBY is not a healthy state either: the status should stay + # not working. + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.STANDBY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) is Timeout + + # Get it back to working again + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.WORKING + ) + + # When there an error message, status should be not working + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.ERROR}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.NOT_WORKING + ) + + # Get it back to working again + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.WORKING + ) + + # When data with an old timestamp arrives, status should be not working + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc) - timedelta(seconds=1), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.NOT_WORKING + ) + + # Fresh data should bring it back to working + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.WORKING + ) + + # When there's no new data, status should be not working + assert await receive_timeout(status_receiver, 0.1) is Timeout + assert await receive_timeout(status_receiver, 0.2) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.NOT_WORKING + ) + + # Get it back to working again + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.WORKING + ) + + async def keep_sending_healthy_message() -> None: + """Keep sending healthy messages.""" + while True: + await mock_microgrid.mock_client.send( + SteamBoilerDataWrapper( + _STEAM_BOILER_ID, + datetime.now(tz=timezone.utc), + active_power=0.0, + states={ComponentStateCode.READY}, + ).to_samples() + ) + await asyncio.sleep(0.1) + + _keep_sending_healthy_message_task = asyncio.create_task( + keep_sending_healthy_message() + ) + # when there's a PowerDistributor failure for the component, status should + # become uncertain. + await set_power_result_sender.send( + SetPowerResult( + succeeded=set(), + failed={_STEAM_BOILER_ID}, + ) + ) + assert await receive_timeout(status_receiver) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.UNCERTAIN + ) + + # After the blocking duration, it should become working again. + assert await receive_timeout(status_receiver) is Timeout + assert await receive_timeout(status_receiver, 1.0) == ComponentStatus( + _STEAM_BOILER_ID, ComponentStatusEnum.WORKING + ) + await cancel_and_await(_keep_sending_healthy_message_task) diff --git a/tests/microgrid/power_distributing/test_steam_boiler_manager.py b/tests/microgrid/power_distributing/test_steam_boiler_manager.py new file mode 100644 index 000000000..c29078e3c --- /dev/null +++ b/tests/microgrid/power_distributing/test_steam_boiler_manager.py @@ -0,0 +1,339 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for the steam boiler manager's power distribution.""" + +from __future__ import annotations + +from collections import abc +from datetime import timedelta +from typing import cast +from unittest.mock import AsyncMock + +from frequenz.channels import Broadcast, LatestValueCache +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.quantities import Power +from pytest_mock import MockerFixture + +from frequenz.sdk.microgrid import connection_manager +from frequenz.sdk.microgrid._old_component_data import SteamBoilerData +from frequenz.sdk.microgrid._power_distributing import ComponentPoolStatus +from frequenz.sdk.microgrid._power_distributing._component_managers._steam_boiler_manager._steam_boiler_manager import ( # noqa: E501 + SteamBoilerManager, +) +from frequenz.sdk.microgrid._power_distributing._component_pool_status_tracker import ( + ComponentPoolStatusTracker, +) +from frequenz.sdk.microgrid._power_distributing.request import Request +from frequenz.sdk.microgrid._power_distributing.result import Result, Success + +from ...timeseries.mock_microgrid import MockMicrogrid + +# The tests drive the manager through its internal hooks and caches. +# pylint: disable=protected-access + +_STEAM_BOILER_MODULE = ( + "frequenz.sdk.microgrid._power_distributing._component_managers" + "._steam_boiler_manager._steam_boiler_manager" +) + + +def _steam_boiler_cache( + mocker: MockerFixture, *, lower_bound_w: float, upper_bound_w: float +) -> LatestValueCache[SteamBoilerData]: + """Build a steam boiler data cache with fixed inclusion bounds.""" + cache = mocker.MagicMock(spec=LatestValueCache) + cache.has_value.return_value = True + data = mocker.MagicMock(spec=SteamBoilerData) + data.active_power_inclusion_lower_bound = lower_bound_w + data.active_power_inclusion_upper_bound = upper_bound_w + cache.get.return_value = data + return cast(LatestValueCache[SteamBoilerData], cache) + + +async def _make_manager( + mocker: MockerFixture, + *, + working: abc.Set[ComponentId], + bounds: dict[ComponentId, tuple[float, float]], +) -> tuple[SteamBoilerManager, AsyncMock]: + """Build a steam boiler manager with mocked status, caches and API calls.""" + tracker = mocker.MagicMock(spec=ComponentPoolStatusTracker) + tracker.get_working_components.return_value = working + tracker.stop = mocker.AsyncMock() + mocker.patch( + f"{_STEAM_BOILER_MODULE}.ComponentPoolStatusTracker", return_value=tracker + ) + + status_channel = Broadcast[ComponentPoolStatus](name="steam_boiler_status") + results_channel = Broadcast[Result](name="steam_boiler_results") + manager = SteamBoilerManager( + component_pool_status_sender=status_channel.new_sender(), + results_sender=results_channel.new_sender(), + api_power_request_timeout=timedelta(seconds=1), + ) + manager._component_data_caches = { + boiler: _steam_boiler_cache( + mocker, + lower_bound_w=bounds[boiler][0], + upper_bound_w=bounds[boiler][1], + ) + for boiler in working + } + set_power = mocker.patch( + f"{_STEAM_BOILER_MODULE}._set_component_power", + new=mocker.AsyncMock(return_value=mocker.MagicMock(spec=Success)), + ) + return manager, set_power + + +class TestSteamBoilerUnreachablePowerDistribution: + """Tests for how the manager applies measured unreachable power.""" + + async def test_subtracts_unreachable_power_from_target( + self, mocker: MockerFixture + ) -> None: + """Measured unreachable boiler power is subtracted from the target.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(3) + async with mockgrid: + boiler_ids = set(mockgrid.steam_boiler_ids) + working = set(sorted(boiler_ids)[:2]) + manager, set_power = await _make_manager( + mocker, + working=working, + bounds={boiler: (0.0, 10000.0) for boiler in working}, + ) + mocker.patch.object( + manager, "_subscribe_to_unreachable_power", mocker.AsyncMock() + ) + mocker.patch.object( + manager, "_unreachable_power", return_value=Power.from_watts(300.0) + ) + + await manager.distribute_power( + Request(power=Power.from_watts(1000.0), component_ids=boiler_ids) + ) + + kwargs = set_power.call_args.kwargs + assert kwargs["target_power"].isclose(Power.from_watts(1000.0)) + allocated = Power.zero() + for power in kwargs["allocations"].values(): + allocated += power + # 1000 W requested minus the 300 W already drawn by the unreachable + # boiler leaves 700 W to split across the two reachable boilers. + assert allocated.isclose(Power.from_watts(700.0)) + + async def test_unreachable_power_exceeding_target( + self, mocker: MockerFixture + ) -> None: + """When unreachable draw exceeds the target, nothing is requested.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(3) + async with mockgrid: + boiler_ids = set(mockgrid.steam_boiler_ids) + working = set(sorted(boiler_ids)[:2]) + manager, set_power = await _make_manager( + mocker, + working=working, + bounds={boiler: (0.0, 10000.0) for boiler in working}, + ) + mocker.patch.object( + manager, "_subscribe_to_unreachable_power", mocker.AsyncMock() + ) + mocker.patch.object( + manager, "_unreachable_power", return_value=Power.from_watts(1200.0) + ) + + await manager.distribute_power( + Request(power=Power.from_watts(1000.0), component_ids=boiler_ids) + ) + + kwargs = set_power.call_args.kwargs + # The unreachable boilers already draw more than the target, so the + # reachable boilers are asked for nothing and the overdraw is reported + # as remaining power. + assert all(power == Power.zero() for power in kwargs["allocations"].values()) + assert kwargs["remaining_power"].isclose(Power.from_watts(-200.0)) + + +class TestSteamBoilerMinimumPowerDistribution: + """Tests for how the manager respects per-boiler minimum power bounds.""" + + async def test_shares_below_minimum_are_raised(self, mocker: MockerFixture) -> None: + """A share below a boiler's minimum is raised to the minimum.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(3) + async with mockgrid: + boiler_ids = sorted(mockgrid.steam_boiler_ids) + fixed_1, fixed_2, flexible = boiler_ids + manager, set_power = await _make_manager( + mocker, + working=set(boiler_ids), + bounds={ + fixed_1: (4000.0, 4000.0), + fixed_2: (4000.0, 4000.0), + flexible: (0.0, 20000.0), + }, + ) + mocker.patch.object( + manager, "_subscribe_to_unreachable_power", mocker.AsyncMock() + ) + mocker.patch.object(manager, "_unreachable_power", return_value=None) + + await manager.distribute_power( + Request(power=Power.from_watts(9000.0), component_ids=set(boiler_ids)) + ) + + allocations = set_power.call_args.kwargs["allocations"] + # The fixed-power boilers get their 4 kW minimum instead of an illegal + # 3 kW equal share; the flexible boiler takes the remaining 1 kW. + assert allocations[fixed_1].isclose(Power.from_watts(4000.0)) + assert allocations[fixed_2].isclose(Power.from_watts(4000.0)) + assert allocations[flexible].isclose(Power.from_watts(1000.0)) + + async def test_boiler_kept_off_when_minimum_unaffordable( + self, mocker: MockerFixture + ) -> None: + """A boiler whose minimum exceeds the remaining budget stays off.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(3) + async with mockgrid: + boiler_ids = sorted(mockgrid.steam_boiler_ids) + fixed_1, fixed_2, flexible = boiler_ids + manager, set_power = await _make_manager( + mocker, + working=set(boiler_ids), + bounds={ + fixed_1: (4000.0, 4000.0), + fixed_2: (4000.0, 4000.0), + flexible: (0.0, 20000.0), + }, + ) + mocker.patch.object( + manager, "_subscribe_to_unreachable_power", mocker.AsyncMock() + ) + mocker.patch.object(manager, "_unreachable_power", return_value=None) + + await manager.distribute_power( + Request(power=Power.from_watts(3000.0), component_ids=set(boiler_ids)) + ) + + allocations = set_power.call_args.kwargs["allocations"] + # 3 kW can't cover either fixed boiler's 4 kW minimum, so both stay + # off and the flexible boiler takes the full target. + assert allocations[fixed_1] == Power.zero() + assert allocations[fixed_2] == Power.zero() + assert allocations[flexible].isclose(Power.from_watts(3000.0)) + + async def test_unaffordable_minimum_strands_excess( + self, mocker: MockerFixture + ) -> None: + """Power freed by a kept-off boiler is not redistributed. + + The single-pass allocation visits boilers in ascending upper-bound + order, so power that a kept-off high-minimum boiler can't take is + reported as excess instead of topping up earlier boilers. This test + pins that known limitation. + """ + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(2) + async with mockgrid: + boiler_ids = sorted(mockgrid.steam_boiler_ids) + small, high_minimum = boiler_ids + manager, set_power = await _make_manager( + mocker, + working=set(boiler_ids), + bounds={ + small: (0.0, 1000.0), + high_minimum: (4000.0, 10000.0), + }, + ) + mocker.patch.object( + manager, "_subscribe_to_unreachable_power", mocker.AsyncMock() + ) + mocker.patch.object(manager, "_unreachable_power", return_value=None) + + await manager.distribute_power( + Request(power=Power.from_watts(4500.0), component_ids=set(boiler_ids)) + ) + + kwargs = set_power.call_args.kwargs + assert kwargs["allocations"][small].isclose(Power.from_watts(1000.0)) + assert kwargs["allocations"][high_minimum] == Power.zero() + assert kwargs["remaining_power"].isclose(Power.from_watts(3500.0)) + + async def test_inconsistent_bounds_keep_boiler_off( + self, mocker: MockerFixture + ) -> None: + """A boiler reporting a minimum above its maximum is kept off.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(1) + async with mockgrid: + boiler = mockgrid.steam_boiler_ids[0] + manager, set_power = await _make_manager( + mocker, + working={boiler}, + bounds={boiler: (2000.0, 1000.0)}, + ) + mocker.patch.object( + manager, "_subscribe_to_unreachable_power", mocker.AsyncMock() + ) + mocker.patch.object(manager, "_unreachable_power", return_value=None) + + await manager.distribute_power( + Request(power=Power.from_watts(5000.0), component_ids={boiler}) + ) + + # The bump to the minimum must never push the allocation past the + # reported maximum, so the boiler is kept off instead. + assert set_power.call_args.kwargs["allocations"][boiler] == Power.zero() + + +class TestSteamBoilerManagerWiring: + """Tests for the manager's unreachable-power wiring.""" + + async def test_unreachable_power_formula_delegates_to_graph( + self, mocker: MockerFixture + ) -> None: + """The unreachable-power formula comes from the steam boiler formula.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(2) + async with mockgrid: + boiler_ids = set(mockgrid.steam_boiler_ids) + manager, _ = await _make_manager( + mocker, + working=boiler_ids, + bounds={boiler: (0.0, 10000.0) for boiler in boiler_ids}, + ) + graph = connection_manager.get().component_graph + formula = manager._unreachable_power_formula(boiler_ids) + + # The compiled graph object can't be patched, so pin the + # delegation by comparing against the graph's own output. + assert formula == graph.steam_boiler_formula(boiler_ids) + for boiler in boiler_ids: + assert f"#{int(boiler)}" in formula + + async def test_stop_tears_down_unreachable_power_subscriptions( + self, mocker: MockerFixture + ) -> None: + """Stopping the manager stops the unreachable-power subscriptions.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(1) + async with mockgrid: + boiler = mockgrid.steam_boiler_ids[0] + manager, _ = await _make_manager( + mocker, + working={boiler}, + bounds={boiler: (0.0, 10000.0)}, + ) + stop_subscriptions = mocker.patch.object( + manager, + "_stop_all_unreachable_power_subscriptions", + mocker.AsyncMock(), + ) + + await manager.stop() + + stop_subscriptions.assert_awaited_once() diff --git a/tests/microgrid/test_data_sourcing.py b/tests/microgrid/test_data_sourcing.py index 310603ba7..d29fc110e 100644 --- a/tests/microgrid/test_data_sourcing.py +++ b/tests/microgrid/test_data_sourcing.py @@ -20,6 +20,7 @@ DcEvCharger, LiIonBattery, Meter, + SteamBoiler, ) from frequenz.client.microgrid.metrics import Metric from frequenz.quantities import Quantity @@ -35,6 +36,7 @@ EVChargerData, InverterData, MeterData, + SteamBoilerData, ) from frequenz.sdk.timeseries import Sample @@ -56,6 +58,7 @@ def mock_connection_manager(mocker: pytest_mock.MockFixture) -> mock.Mock: BatteryInverter(id=ComponentId(6), microgrid_id=_MICROGRID_ID), LiIonBattery(id=ComponentId(9), microgrid_id=_MICROGRID_ID), DcEvCharger(id=ComponentId(12), microgrid_id=_MICROGRID_ID), + SteamBoiler(id=ComponentId(15), microgrid_id=_MICROGRID_ID), ], ) @@ -75,6 +78,11 @@ def mock_connection_manager(mocker: pytest_mock.MockFixture) -> mock.Mock: "frequenz.sdk.microgrid._data_sourcing.microgrid_api_source.EVChargerData.subscribe", side_effect=_new_ev_charger_data_mock(ComponentId(12), starting_value=-13.0), ) + mocker.patch( + "frequenz.sdk.microgrid._data_sourcing" + ".microgrid_api_source.SteamBoilerData.subscribe", + side_effect=_new_steam_boiler_data_mock(ComponentId(15), starting_value=50.0), + ) mock_conn_manager = mock.MagicMock(name="connection_manager") mocker.patch( @@ -144,6 +152,14 @@ async def test_data_sourcing_actor( # pylint: disable=too-many-locals ).new_receiver() await req_sender.send(active_power_request_12) + active_power_request_15 = ComponentMetricRequest( + "test-namespace", ComponentId(15), Metric.AC_ACTIVE_POWER, None + ) + active_power_recv_15 = registry.get_or_create( + Sample[Quantity], active_power_request_15.get_channel_name() + ).new_receiver() + await req_sender.send(active_power_request_15) + for i in range(3): sample = await active_power_recv_4.receive() assert sample.value is not None @@ -169,6 +185,10 @@ async def test_data_sourcing_actor( # pylint: disable=too-many-locals assert sample.value is not None assert -13.0 + i == sample.value.base_value + sample = await active_power_recv_15.receive() + assert sample.value is not None + assert 50.0 + i == sample.value.base_value + async def test_duplicate_requests_do_not_block_new_receivers( mock_connection_manager: mock.Mock, # pylint: disable=redefined-outer-name,unused-argument @@ -359,6 +379,29 @@ def _new_ev_charger_data( ) +def _new_steam_boiler_data( + component_id: ComponentId, timestamp: datetime, value: float +) -> SteamBoilerData: + return SteamBoilerData( + component_id=component_id, + timestamp=timestamp, + active_power=value, + active_power_per_phase=(value, value, value), + current_per_phase=(value, value, value), + frequency=value, + reactive_power=value, + reactive_power_per_phase=(value, value, value), + voltage_per_phase=(value, value, value), + active_power_exclusion_lower_bound=value, + active_power_exclusion_upper_bound=value, + active_power_inclusion_lower_bound=value, + active_power_inclusion_upper_bound=value, + states={ComponentStateCode.UNSPECIFIED}, + errors=frozenset(), + warnings=frozenset(), + ) + + def _new_streamer_mock( name: str, constructor: Callable[[ComponentId, datetime, float], T], @@ -421,3 +464,15 @@ def _new_ev_charger_data_mock( component_id, starting_value, ) + + +def _new_steam_boiler_data_mock( + component_id: ComponentId, starting_value: float +) -> mock.Mock: + """Get a mock streamer for steam boiler data.""" + return _new_streamer_mock( + f"steam_boiler_data_mock(id={component_id}, starting_value={starting_value})", + _new_steam_boiler_data, + component_id, + starting_value, + ) diff --git a/tests/timeseries/_steam_boiler_pool/__init__.py b/tests/timeseries/_steam_boiler_pool/__init__.py new file mode 100644 index 000000000..3fbbdbed7 --- /dev/null +++ b/tests/timeseries/_steam_boiler_pool/__init__.py @@ -0,0 +1,4 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Test the steam boiler pool control methods.""" diff --git a/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py b/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py new file mode 100644 index 000000000..4c850b1cc --- /dev/null +++ b/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py @@ -0,0 +1,153 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for the `SteamBoilerPool`.""" + +import asyncio +from unittest.mock import MagicMock + +import pytest +from frequenz.channels import Broadcast +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.quantities import Power +from pytest_mock import MockerFixture + +from frequenz.sdk import microgrid, timeseries +from frequenz.sdk._internal._channels import ChannelRegistry +from frequenz.sdk.microgrid._power_managing import ReportRequest, _Report +from frequenz.sdk.timeseries.steam_boiler_pool import ( + SteamBoilerPool, + SteamBoilerPoolError, +) +from frequenz.sdk.timeseries.steam_boiler_pool._steam_boiler_pool_reference_store import ( + SteamBoilerPoolReferenceStore, +) +from tests.timeseries.mock_microgrid import MockMicrogrid + + +def _new_power_status_report(target_power_watts: float) -> _Report: + """Create a distinct report for power status assertions.""" + target_power = Power.from_watts(target_power_watts) + return _Report( + target_power=target_power, + _inclusion_bounds=timeseries.Bounds(target_power, target_power), + _exclusion_bounds=None, + ) + + +class TestSteamBoilerPool: + """Tests for the `SteamBoilerPool`.""" + + async def test_power( # pylint: disable=too-many-locals + self, + mocker: MockerFixture, + ) -> None: + """Test the power formula.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(3) + + async with mockgrid: + steam_boiler_pool = microgrid.new_steam_boiler_pool(priority=5) + power_receiver = steam_boiler_pool.power.new_receiver() + + # The boiler values sum to 15 W, not the 16 W on the meter, to + # check that the boilers are the primary source. + await mockgrid.mock_resampler.send_meter_power([16.0]) + await mockgrid.mock_resampler.send_steam_boiler_power([2.0, 6.0, 7.0]) + assert (await power_receiver.receive()).value == Power.from_watts(15.0) + + # When the boilers have no value, the formula falls back to the + # meter. The fallback subscription starts after the first `None`, + # so send twice and skip one output. + await mockgrid.mock_resampler.send_meter_power([16.0]) + await mockgrid.mock_resampler.send_steam_boiler_power([None, None, None]) + await mockgrid.mock_resampler.send_meter_power([16.0]) + await mockgrid.mock_resampler.send_steam_boiler_power([None, None, None]) + await power_receiver.receive() + assert (await power_receiver.receive()).value == Power.from_watts(16.0) + + async def test_power_without_meters( + self, + mocker: MockerFixture, + ) -> None: + """Test the power formula when the boilers have no meter of their own. + + Without an upstream meter the formula reads each steam boiler component + directly, which exercises the steam boiler path in the data sourcing actor. + """ + mockgrid = MockMicrogrid(grid_meter=False, mocker=mocker) + mockgrid.add_steam_boilers(3) + + async with mockgrid: + steam_boiler_pool = microgrid.new_steam_boiler_pool(priority=5) + power_receiver = steam_boiler_pool.power.new_receiver() + + await mockgrid.mock_resampler.send_steam_boiler_power([2.0, 3.0, 4.0]) + assert (await power_receiver.receive()).value == Power.from_watts(9.0) + + async def test_propose_discharge_power_is_rejected( + self, + mocker: MockerFixture, + ) -> None: + """Proposing a discharge (negative) power for steam boilers is rejected.""" + mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker) + mockgrid.add_steam_boilers(1) + + async with mockgrid: + steam_boiler_pool = microgrid.new_steam_boiler_pool(priority=5) + with pytest.raises(SteamBoilerPoolError): + await steam_boiler_pool.propose_power(Power.from_watts(-5.0)) + + +async def test_power_status_same_instance_subscriptions_work( + mocker: MockerFixture, +) -> None: + """Ensure same-instance power_status subscriptions share the same channel.""" + mock_cm = MagicMock() + mock_graph = MagicMock() + mock_graph.components.return_value = [ + MagicMock(id=ComponentId(12)), + MagicMock(id=ComponentId(22)), + ] + mock_cm.component_graph = mock_graph + mocker.patch( + "frequenz.sdk.microgrid.connection_manager._CONNECTION_MANAGER", + mock_cm, + ) + mocker.patch("frequenz.sdk.microgrid.connection_manager.get", return_value=mock_cm) + + registry = ChannelRegistry(name="steam_boiler-pool-test") + requests_channel = Broadcast[ReportRequest](name="steam_boiler-pool-requests") + requests_rx = requests_channel.new_receiver() + component_ids = frozenset({ComponentId(12), ComponentId(22)}) + pool = SteamBoilerPool( + pool_ref_store=SteamBoilerPoolReferenceStore( + channel_registry=registry, + resampler_subscription_sender=MagicMock(), + status_receiver=MagicMock(), + power_manager_requests_sender=MagicMock(), + power_manager_bounds_subs_sender=requests_channel.new_sender(), + power_distribution_results_fetcher=MagicMock(), + component_ids=component_ids, + ), + name="steam_boiler-pool", + priority=5, + ) + + first_status_rx = pool.power_status.new_receiver() + second_status_rx = pool.power_status.new_receiver() + + await asyncio.sleep(0) + + first_request = await asyncio.wait_for(requests_rx.receive(), timeout=1.0) + second_request = await asyncio.wait_for(requests_rx.receive(), timeout=1.0) + assert second_request.get_channel_name() == first_request.get_channel_name() + + await registry.get_or_create( + _Report, first_request.get_channel_name() + ).new_sender().send(_new_power_status_report(123.0)) + + first_report = await asyncio.wait_for(first_status_rx.receive(), timeout=1.0) + second_report = await asyncio.wait_for(second_status_rx.receive(), timeout=1.0) + assert first_report.target_power == Power.from_watts(123.0) + assert second_report.target_power == Power.from_watts(123.0) diff --git a/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool_control_methods.py b/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool_control_methods.py new file mode 100644 index 000000000..826392a01 --- /dev/null +++ b/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool_control_methods.py @@ -0,0 +1,369 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Test the steam boiler pool control methods.""" + +import asyncio +import typing +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock + +import async_solipsism +import pytest +from frequenz.channels import Receiver +from frequenz.client.common.microgrid.components import ComponentId +from frequenz.client.microgrid.component import ComponentStateCode +from frequenz.quantities import Power +from pytest_mock import MockerFixture + +from frequenz.sdk import microgrid +from frequenz.sdk.microgrid import _power_distributing +from frequenz.sdk.microgrid._data_pipeline import _DataPipeline +from frequenz.sdk.timeseries import ResamplerConfig2 +from frequenz.sdk.timeseries.steam_boiler_pool import SteamBoilerPoolReport + +from ...microgrid.fixtures import _Mocks +from ...utils.component_data_streamer import MockComponentDataStreamer +from ...utils.component_data_wrapper import SteamBoilerDataWrapper +from ..mock_microgrid import MockMicrogrid + + +@pytest.fixture +def event_loop_policy() -> async_solipsism.EventLoopPolicy: + """Event loop policy.""" + return async_solipsism.EventLoopPolicy() + + +@pytest.fixture +async def mocks(mocker: MockerFixture) -> typing.AsyncIterator[_Mocks]: + """Create the mocks.""" + mockgrid = MockMicrogrid(grid_meter=True) + mockgrid.add_steam_boilers(4) + await mockgrid.start(mocker) + + # pylint: disable=protected-access + if microgrid._data_pipeline._DATA_PIPELINE is not None: + microgrid._data_pipeline._DATA_PIPELINE = None + await microgrid._data_pipeline.initialize( + ResamplerConfig2(resampling_period=timedelta(seconds=0.1)) + ) + streamer = MockComponentDataStreamer(mockgrid.mock_client) + + dp = typing.cast(_DataPipeline, microgrid._data_pipeline._DATA_PIPELINE) + + _mocks = _Mocks( + mockgrid, + streamer, + dp._steam_boiler_power_wrapper.status_channel.new_sender(), + ) + try: + yield _mocks + finally: + await _mocks.stop() + + +class TestSteamBoilerPoolControl: + """Test control methods for the SteamBoilerPool.""" + + async def _init_steam_boilers(self, mocks: _Mocks) -> None: + now = datetime.now(tz=timezone.utc) + for idx, comp_id in enumerate(mocks.microgrid.steam_boiler_ids): + mocks.streamer.start_streaming( + SteamBoilerDataWrapper( + comp_id, + now, + states={ComponentStateCode.READY}, + active_power=0.0, + active_power_inclusion_lower_bound=0.0, + active_power_inclusion_upper_bound=10000.0 * (idx + 1), + ), + 0.05, + ) + + async def _fail_steam_boilers( + self, fail_ids: list[ComponentId], mocks: _Mocks + ) -> None: + now = datetime.now(tz=timezone.utc) + for idx, comp_id in enumerate(mocks.microgrid.steam_boiler_ids): + mocks.streamer.update_stream( + SteamBoilerDataWrapper( + comp_id, + now, + states=( + {ComponentStateCode.ERROR} + if comp_id in fail_ids + else {ComponentStateCode.READY} + ), + active_power=0.0, + active_power_inclusion_lower_bound=0.0, + active_power_inclusion_upper_bound=10000.0 * (idx + 1), + ), + ) + + def _assert_report( # pylint: disable=too-many-arguments + self, + report: SteamBoilerPoolReport | None, + *, + power: float | None, + lower: float, + upper: float, + dist_result: _power_distributing.Result | None = None, + expected_result_pred: ( + typing.Callable[[_power_distributing.Result], bool] | None + ) = None, + ) -> None: + assert report is not None and report.target_power == ( + Power.from_watts(power) if power is not None else None + ) + assert report.bounds is not None + assert report.bounds.lower == Power.from_watts(lower) + assert report.bounds.upper == Power.from_watts(upper) + if expected_result_pred is not None: + assert dist_result is not None + assert expected_result_pred(dist_result) + + async def _recv_reports_until( + self, + bounds_rx: Receiver[SteamBoilerPoolReport], + check: typing.Callable[[SteamBoilerPoolReport], bool], + ) -> SteamBoilerPoolReport | None: + """Receive reports until the given condition is met.""" + max_reports = 10 + ctr = 0 + latest_report: SteamBoilerPoolReport | None = None + while ctr < max_reports: + ctr += 1 + latest_report = await bounds_rx.receive() + if check(latest_report): + break + + return latest_report + + async def test_setting_power( # pylint: disable=too-many-statements + self, + mocks: _Mocks, + mocker: MockerFixture, + ) -> None: + """Test setting power.""" + set_power = typing.cast( + AsyncMock, + microgrid.connection_manager.get().api_client.set_component_power_active, + ) + + await self._init_steam_boilers(mocks) + steam_boiler_pool = microgrid.new_steam_boiler_pool(priority=5) + bounds_rx = steam_boiler_pool.power_status.new_receiver() + latest_report = await self._recv_reports_until( + bounds_rx, + lambda x: x.bounds is not None and x.bounds.upper.as_watts() == 100000.0, + ) + dist_results_rx = steam_boiler_pool.power_distribution_results.new_receiver() + + self._assert_report(latest_report, power=None, lower=0.0, upper=100000.0) + boiler_ids = mocks.microgrid.steam_boiler_ids + + await steam_boiler_pool.propose_power(Power.from_watts(80000.0)) + await self._recv_reports_until( + bounds_rx, + lambda x: x.target_power is not None + and x.target_power.as_watts() == 80000.0, + ) + self._assert_report( + await bounds_rx.receive(), power=80000.0, lower=0.0, upper=100000.0 + ) + await asyncio.sleep(0.0) + + # Components are set initial power + assert set_power.call_count == 4 + assert sorted(set_power.call_args_list, key=lambda x: x.args[0]) == [ + mocker.call(boiler_ids[0], 10000.0), + mocker.call(boiler_ids[1], 20000.0), + mocker.call(boiler_ids[2], 25000.0), + mocker.call(boiler_ids[3], 25000.0), + ] + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.from_watts(80000.0) + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == set(boiler_ids) + + # Throttling to a lower power should distribute it evenly again. + set_power.reset_mock() + await steam_boiler_pool.propose_power(Power.from_watts(4000.0)) + await self._recv_reports_until( + bounds_rx, + lambda x: x.target_power is not None + and x.target_power.as_watts() == 4000.0, + ) + self._assert_report( + await bounds_rx.receive(), power=4000.0, lower=0.0, upper=100000.0 + ) + await asyncio.sleep(0.0) + + assert set_power.call_count == 4 + assert sorted(set_power.call_args_list, key=lambda x: x.args[0]) == [ + mocker.call(boiler_ids[0], 1000.0), + mocker.call(boiler_ids[1], 1000.0), + mocker.call(boiler_ids[2], 1000.0), + mocker.call(boiler_ids[3], 1000.0), + ] + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.from_watts(4000.0) + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == set(boiler_ids) + + # After failing 1 boiler, bounds should go down and power shouldn't be + # distributed to that boiler. + await self._fail_steam_boilers([boiler_ids[1]], mocks) + await self._recv_reports_until( + bounds_rx, + lambda x: x.bounds is not None and x.bounds.upper.as_watts() == 80000.0, + ) + self._assert_report( + await bounds_rx.receive(), power=4000.0, lower=0.0, upper=80000.0 + ) + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.from_watts(4000.0) + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == { + boiler_ids[0], + boiler_ids[2], + boiler_ids[3], + } + + set_power.reset_mock() + await steam_boiler_pool.propose_power(Power.from_watts(70000.0)) + await self._recv_reports_until( + bounds_rx, + lambda x: x.target_power is not None + and x.target_power.as_watts() == 70000.0, + ) + self._assert_report( + await bounds_rx.receive(), power=70000.0, lower=0.0, upper=80000.0 + ) + await asyncio.sleep(0.0) + + assert set_power.call_count == 3 + assert sorted(set_power.call_args_list, key=lambda x: x.args[0]) == [ + mocker.call(boiler_ids[0], 10000.0), + mocker.call(boiler_ids[2], 30000.0), + mocker.call(boiler_ids[3], 30000.0), + ] + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.from_watts(70000.0) + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == { + boiler_ids[0], + boiler_ids[2], + boiler_ids[3], + } + + # After the failed boiler recovers, bounds should go back up and power + # should be distributed to all boilers. + await self._fail_steam_boilers([], mocks) + await self._recv_reports_until( + bounds_rx, + lambda x: x.bounds is not None and x.bounds.upper.as_watts() == 100000.0, + ) + self._assert_report( + await bounds_rx.receive(), power=70000.0, lower=0.0, upper=100000.0 + ) + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.from_watts(70000.0) + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == set(boiler_ids) + + # Proposing more than the available power clamps to the system bounds. + set_power.reset_mock() + await steam_boiler_pool.propose_power(Power.from_watts(200000.0)) + await self._recv_reports_until( + bounds_rx, + lambda x: x.target_power is not None + and x.target_power.as_watts() == 100000.0, + ) + self._assert_report( + await bounds_rx.receive(), power=100000.0, lower=0.0, upper=100000.0 + ) + await asyncio.sleep(0.0) + + assert set_power.call_count == 4 + assert sorted(set_power.call_args_list, key=lambda x: x.args[0]) == [ + mocker.call(boiler_ids[0], 10000.0), + mocker.call(boiler_ids[1], 20000.0), + mocker.call(boiler_ids[2], 30000.0), + mocker.call(boiler_ids[3], 40000.0), + ] + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.from_watts(100000.0) + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == set(boiler_ids) + + # Setting 0 power should set all boilers to 0. + set_power.reset_mock() + await steam_boiler_pool.propose_power(Power.zero()) + await self._recv_reports_until( + bounds_rx, + lambda x: x.target_power is not None and x.target_power.as_watts() == 0.0, + ) + self._assert_report( + await bounds_rx.receive(), power=0.0, lower=0.0, upper=100000.0 + ) + await asyncio.sleep(0.0) + + assert set_power.call_count == 4 + assert sorted(set_power.call_args_list, key=lambda x: x.args[0]) == [ + mocker.call(boiler_ids[0], 0.0), + mocker.call(boiler_ids[1], 0.0), + mocker.call(boiler_ids[2], 0.0), + mocker.call(boiler_ids[3], 0.0), + ] + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.zero() + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == set(boiler_ids) + + # Resetting the power should lead to default (zero) power getting set for all + # boilers. + set_power.reset_mock() + await steam_boiler_pool.propose_power(None) + report = await self._recv_reports_until( + bounds_rx, + lambda x: x.target_power is None, + ) + self._assert_report(report, power=None, lower=0.0, upper=100000.0) + await asyncio.sleep(0.0) + + assert set_power.call_count == 4 + assert sorted(set_power.call_args_list, key=lambda x: x.args[0]) == [ + mocker.call(boiler_ids[0], 0.0), + mocker.call(boiler_ids[1], 0.0), + mocker.call(boiler_ids[2], 0.0), + mocker.call(boiler_ids[3], 0.0), + ] + dist_results = await dist_results_rx.receive() + assert isinstance( + dist_results, _power_distributing.Success + ), f"Expected a success, got {dist_results}" + assert dist_results.succeeded_power == Power.zero() + assert dist_results.excess_power == Power.zero() + assert dist_results.succeeded_components == set(boiler_ids) diff --git a/tests/timeseries/mock_microgrid.py b/tests/timeseries/mock_microgrid.py index d58c6e201..248cace17 100644 --- a/tests/timeseries/mock_microgrid.py +++ b/tests/timeseries/mock_microgrid.py @@ -27,6 +27,7 @@ LiIonBattery, Meter, SolarInverter, + SteamBoiler, ) from frequenz.microgrid_component_graph import ComponentGraph from pytest_mock import MockerFixture @@ -43,6 +44,7 @@ EvChargerDataWrapper, InverterDataWrapper, MeterDataWrapper, + SteamBoilerDataWrapper, ) from .mock_resampler import MockResampler @@ -55,6 +57,7 @@ class MockMicrogrid: # pylint: disable=too-many-instance-attributes grid_id = ComponentId(1) _grid_meter_id = ComponentId(4) + steam_boiler_id_suffix = 3 chp_id_suffix = 5 evc_id_suffix = 6 meter_id_suffix = 7 @@ -144,6 +147,7 @@ def inverters(component_type: type[Inverter]) -> list[ComponentId]: self.battery_inverter_ids: list[ComponentId] = inverters(BatteryInverter) self.pv_inverter_ids: list[ComponentId] = inverters(SolarInverter) + self.steam_boiler_ids: list[ComponentId] = filter_comp(SteamBoiler) self.bat_inv_map: dict[ComponentId, ComponentId] = ( {} @@ -156,6 +160,7 @@ def inverters(component_type: type[Inverter]) -> list[ComponentId]: ) self.evc_states: dict[ComponentId, set[ComponentStateCode]] = {} + self.steam_boiler_states: dict[ComponentId, set[ComponentStateCode]] = {} self._streaming_coros: list[tuple[ComponentId, Coroutine[None, None, None]]] = ( [] @@ -212,6 +217,7 @@ async def start(self, mocker: MockerFixture | None = None) -> None: evc_ids=self.evc_ids, meter_ids=self.meter_ids, chp_ids=self.chp_ids, + steam_boiler_ids=self.steam_boiler_ids, namespaces=self._namespaces, ) @@ -350,6 +356,24 @@ def _start_ev_charger_streaming(self, evc_id: ComponentId) -> None: ) ) + def _start_steam_boiler_streaming(self, steam_boiler_id: ComponentId) -> None: + if not self._api_client_streaming: + return + self._streaming_coros.append( + ( + steam_boiler_id, + self._comp_data_send_task( + steam_boiler_id, + lambda value, ts: SteamBoilerDataWrapper( + component_id=steam_boiler_id, + timestamp=ts, + active_power=value, + states=self.steam_boiler_states[steam_boiler_id], + ), + ), + ) + ) + def add_consumer_meters(self, count: int = 1) -> None: """Add consumer meters to the mock microgrid. @@ -495,6 +519,28 @@ def add_ev_chargers(self, count: int) -> None: ComponentConnection(source=self._connect_to, destination=evc_id) ) + def add_steam_boilers(self, count: int) -> None: + """Add steam boilers to the microgrid. + + Args: + count: Number of steam boilers to add to the microgrid. + """ + for _ in range(count): + component_id = ComponentId( + self._id_increment * 10 + self.steam_boiler_id_suffix + ) + self._id_increment += 1 + self.steam_boiler_ids.append(component_id) + + self._components.add( + SteamBoiler(id=component_id, microgrid_id=_MICROGRID_ID) + ) + self.steam_boiler_states[component_id] = {ComponentStateCode.READY} + self._start_steam_boiler_streaming(component_id) + self._connections.add( + ComponentConnection(source=self._connect_to, destination=component_id) + ) + async def send_meter_data(self, values: list[float]) -> None: """Send raw meter data from the mock microgrid. diff --git a/tests/timeseries/mock_resampler.py b/tests/timeseries/mock_resampler.py index 98440646b..ee09c9834 100644 --- a/tests/timeseries/mock_resampler.py +++ b/tests/timeseries/mock_resampler.py @@ -35,6 +35,7 @@ def __init__( # pylint: disable=too-many-arguments,too-many-positional-argument pv_inverter_ids: list[ComponentId], evc_ids: list[ComponentId], chp_ids: list[ComponentId], + steam_boiler_ids: list[ComponentId], meter_ids: list[ComponentId], namespaces: int, ) -> None: @@ -80,6 +81,9 @@ def metric_senders( self._ev_power_senders = metric_senders(evc_ids, Metric.AC_ACTIVE_POWER) self._chp_power_senders = metric_senders(chp_ids, Metric.AC_ACTIVE_POWER) + self._steam_boiler_power_senders = metric_senders( + steam_boiler_ids, Metric.AC_ACTIVE_POWER + ) self._meter_power_senders = metric_senders(meter_ids, Metric.AC_ACTIVE_POWER) self._non_existing_component_sender = metric_senders( [ComponentId(NON_EXISTING_COMPONENT_ID)], Metric.AC_ACTIVE_POWER @@ -282,6 +286,13 @@ async def send_chp_power(self, values: list[float | None]) -> None: sample = self.make_sample(value) await chan.send(sample) + async def send_steam_boiler_power(self, values: list[float | None]) -> None: + """Send the given values as resampler output for steam boiler power.""" + assert len(values) == len(self._steam_boiler_power_senders) + for chan, value in zip(self._steam_boiler_power_senders, values): + sample = self.make_sample(value) + await chan.send(sample) + async def send_pv_inverter_power(self, values: list[float | None]) -> None: """Send the given values as resampler output for PV Inverter power.""" assert len(values) == len(self._pv_inverter_power_senders) diff --git a/tests/utils/component_data_wrapper.py b/tests/utils/component_data_wrapper.py index f9e1f6e77..86a559793 100644 --- a/tests/utils/component_data_wrapper.py +++ b/tests/utils/component_data_wrapper.py @@ -26,6 +26,7 @@ EVChargerData, InverterData, MeterData, + SteamBoilerData, ) # Disable these checks for the file as we need to pass a lot of data @@ -229,6 +230,78 @@ def copy_with_new_timestamp(self, new_timestamp: datetime) -> EvChargerDataWrapp return replace(self, timestamp=new_timestamp) +@dataclass +class SteamBoilerDataWrapper(SteamBoilerData): + """Wrapper for the SteamBoilerData with default arguments.""" + + def __init__( # pylint: disable=too-many-locals + self, + component_id: ComponentId, + timestamp: datetime, + active_power: float = math.nan, + active_power_per_phase: tuple[float, float, float] = ( + math.nan, + math.nan, + math.nan, + ), + current_per_phase: tuple[float, float, float] = (math.nan, math.nan, math.nan), + voltage_per_phase: tuple[float, float, float] = (math.nan, math.nan, math.nan), + active_power_inclusion_lower_bound: float = math.nan, + active_power_exclusion_lower_bound: float = math.nan, + active_power_inclusion_upper_bound: float = math.nan, + active_power_exclusion_upper_bound: float = math.nan, + reactive_power: float = math.nan, + reactive_power_per_phase: tuple[float, float, float] = ( + math.nan, + math.nan, + math.nan, + ), + frequency: float = 50.0, + states: Set[ComponentStateCode] = frozenset(), + warnings: Set[ComponentErrorCode] = frozenset(), + errors: Set[ComponentErrorCode] = frozenset(), + ) -> None: + """Initialize the SteamBoilerDataWrapper. + + This is a wrapper for the SteamBoilerData with default arguments. The + parameters are documented in the SteamBoilerData class. + """ + super().__init__( + component_id=component_id, + timestamp=timestamp, + active_power=active_power, + active_power_per_phase=active_power_per_phase, + current_per_phase=current_per_phase, + voltage_per_phase=voltage_per_phase, + active_power_inclusion_lower_bound=active_power_inclusion_lower_bound, + active_power_exclusion_lower_bound=active_power_exclusion_lower_bound, + active_power_inclusion_upper_bound=active_power_inclusion_upper_bound, + active_power_exclusion_upper_bound=active_power_exclusion_upper_bound, + reactive_power=reactive_power, + reactive_power_per_phase=reactive_power_per_phase, + frequency=frequency, + states=states, + warnings=warnings, + errors=errors, + ) + + def copy_with_new_timestamp( + self, new_timestamp: datetime + ) -> SteamBoilerDataWrapper: + """Copy the component data but insert new timestamp. + + Because the dataclass is frozen, we can't just replace the timestamp. + We have to copy it. + + Args: + new_timestamp: New timestamp. + + Returns: + Copied component data. + """ + return replace(self, timestamp=new_timestamp) + + @dataclass class MeterDataWrapper(MeterData): """Wrapper for the MeterData with default arguments.""" diff --git a/tests/utils/graph_generator.py b/tests/utils/graph_generator.py index 584f56347..596e672ca 100644 --- a/tests/utils/graph_generator.py +++ b/tests/utils/graph_generator.py @@ -25,6 +25,7 @@ LiIonBattery, Meter, SolarInverter, + SteamBoiler, UnspecifiedInverter, ) from frequenz.microgrid_component_graph import ComponentGraph @@ -36,6 +37,7 @@ class GraphGenerator: """Utilities to generate graphs from component data structures.""" SUFFIXES: dict[ComponentCategory, int] = { + ComponentCategory.STEAM_BOILER: 3, ComponentCategory.CHP: 5, ComponentCategory.EV_CHARGER: 6, ComponentCategory.METER: 7, @@ -204,6 +206,11 @@ def component( id=self.new_id()[other], microgrid_id=_MICROGRID_ID, ) + case ComponentCategory.STEAM_BOILER: + return SteamBoiler( + id=self.new_id()[other], + microgrid_id=_MICROGRID_ID, + ) case _: assert False, "Unsupported ComponentCategory"