diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 412c6f25e..567779eff 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -16,4 +16,4 @@ ## Bug Fixes - +* `COALESCE()` no longer unsubscribes from the parameter that is producing values when the parameter it was tracking evaluates to nothing at all. This could only happen when a parameter is an expression that has no value yet, like a nested function call, in which case the samples of the producing parameter were counted against the tracked one. diff --git a/src/frequenz/sdk/microgrid/_power_managing/_bounds.py b/src/frequenz/sdk/microgrid/_power_managing/_bounds.py index 267aae9cb..0c271813f 100644 --- a/src/frequenz/sdk/microgrid/_power_managing/_bounds.py +++ b/src/frequenz/sdk/microgrid/_power_managing/_bounds.py @@ -3,16 +3,40 @@ """Utilities for checking and clamping bounds and power values to exclusion bounds.""" +import enum +from typing import assert_never + from frequenz.quantities import Power from ...timeseries import Bounds +# This used to be a tuple[bool, bool], but mypy can't check that a match over a tuple +# covers every combination (see python/mypy#12364), so callers could leave a case out +# without anyone noticing. It also reads better, as (True, False) gives no hint about +# which of the two bounds it refers to. +@enum.unique +class ExclusionOverlap(enum.Enum): + """Which bounds of a pair fall inside an exclusion zone.""" + + NONE = enum.auto() + """Neither bound is inside the exclusion zone.""" + + LOWER = enum.auto() + """Only the lower bound is inside the exclusion zone.""" + + UPPER = enum.auto() + """Only the upper bound is inside the exclusion zone.""" + + BOTH = enum.auto() + """Both bounds are inside the exclusion zone.""" + + def check_exclusion_bounds_overlap( lower_bound: Power, upper_bound: Power, exclusion_bounds: Bounds[Power] | None, -) -> tuple[bool, bool]: +) -> ExclusionOverlap: """Check if the given bounds overlap with the given exclusion bounds. Example: @@ -29,8 +53,8 @@ def check_exclusion_bounds_overlap( (inside the exclusion zone) ``` - Resulting in `(False, True)` because only the upper bound is inside the - exclusion zone. + Resulting in `ExclusionOverlap.UPPER` because only the upper bound is inside + the exclusion zone. Args: lower_bound: The lower bound to check. @@ -38,22 +62,21 @@ def check_exclusion_bounds_overlap( exclusion_bounds: The exclusion bounds to check against. Returns: - A tuple containing a boolean indicating if the lower bound is bounded by the - exclusion bounds, and a boolean indicating if the upper bound is bounded by - the exclusion bounds. + Which of the given bounds are inside the exclusion bounds. """ if exclusion_bounds is None: - return False, False - - bounded_lower = False - bounded_upper = False + return ExclusionOverlap.NONE - if exclusion_bounds.lower < lower_bound < exclusion_bounds.upper: - bounded_lower = True - if exclusion_bounds.lower < upper_bound < exclusion_bounds.upper: - bounded_upper = True + bounded_lower = exclusion_bounds.lower < lower_bound < exclusion_bounds.upper + bounded_upper = exclusion_bounds.lower < upper_bound < exclusion_bounds.upper - return bounded_lower, bounded_upper + if bounded_lower and bounded_upper: + return ExclusionOverlap.BOTH + if bounded_lower: + return ExclusionOverlap.LOWER + if bounded_upper: + return ExclusionOverlap.UPPER + return ExclusionOverlap.NONE def adjust_exclusion_bounds( @@ -80,13 +103,16 @@ def adjust_exclusion_bounds( # And if the given bounds overlap with the exclusion bounds on one side, then clamp # the given bounds on that side. match check_exclusion_bounds_overlap(lower_bound, upper_bound, exclusion_bounds): - case (True, True): + case ExclusionOverlap.BOTH: return Power.zero(), Power.zero() - case (False, True): + case ExclusionOverlap.UPPER: return lower_bound, exclusion_bounds.lower - case (True, False): + case ExclusionOverlap.LOWER: return exclusion_bounds.upper, upper_bound - return lower_bound, upper_bound + case ExclusionOverlap.NONE: + return lower_bound, upper_bound + case unexpected: + assert_never(unexpected) # Just 20 lines of code in this function, but unfortunately 8 of those are return @@ -123,14 +149,20 @@ def clamp_to_bounds( # pylint: disable=too-many-return-statements match check_exclusion_bounds_overlap( lower_bound, upper_bound, exclusion_bounds ): - case (True, True): + case ExclusionOverlap.BOTH: return None, None - case (True, False): + case ExclusionOverlap.LOWER: if value < exclusion_bounds.upper: return None, exclusion_bounds.upper - case (False, True): + case ExclusionOverlap.UPPER: if value > exclusion_bounds.lower: return exclusion_bounds.lower, None + case ExclusionOverlap.NONE: + # The bounds don't overlap the exclusion zone, so the value only needs + # the generic clamping done below. + pass + case unexpected: + assert_never(unexpected) # If the given value is outside the given bounds, clamp it to the closest bound. if value < lower_bound: diff --git a/src/frequenz/sdk/microgrid/_power_managing/_matryoshka.py b/src/frequenz/sdk/microgrid/_power_managing/_matryoshka.py index 441413b76..1407227c8 100644 --- a/src/frequenz/sdk/microgrid/_power_managing/_matryoshka.py +++ b/src/frequenz/sdk/microgrid/_power_managing/_matryoshka.py @@ -111,11 +111,13 @@ def _calc_target_power( # If the bounds from the current proposal are fully within the exclusion # bounds, then don't use them to narrow the bounds further. This allows # subsequent proposals to not be blocked by the current proposal. - match _bounds.check_exclusion_bounds_overlap( - proposal_lower, proposal_upper, exclusion_bounds + if ( + _bounds.check_exclusion_bounds_overlap( + proposal_lower, proposal_upper, exclusion_bounds + ) + is _bounds.ExclusionOverlap.BOTH ): - case (True, True): - continue + continue lower_bound = max(lower_bound, proposal_lower) upper_bound = min(upper_bound, proposal_upper) lower_bound, upper_bound = _bounds.adjust_exclusion_bounds( @@ -275,11 +277,13 @@ def get_status( break proposal_lower = next_proposal.bounds.lower or lower_bound proposal_upper = next_proposal.bounds.upper or upper_bound - match _bounds.check_exclusion_bounds_overlap( - proposal_lower, proposal_upper, exclusion_bounds + if ( + _bounds.check_exclusion_bounds_overlap( + proposal_lower, proposal_upper, exclusion_bounds + ) + is _bounds.ExclusionOverlap.BOTH ): - case (True, True): - continue + continue calc_lower_bound = max(lower_bound, proposal_lower) calc_upper_bound = min(upper_bound, proposal_upper) if calc_lower_bound <= calc_upper_bound: 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..798f89a60 100644 --- a/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py +++ b/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py @@ -197,7 +197,7 @@ async def _send_updated_target_power( ) ) - @override + @override # pylint: disable-next=too-many-branches async def _run(self) -> None: """Run the power managing actor.""" last_result_partial_failure = False @@ -261,6 +261,14 @@ async def _run(self) -> None: ) case _power_distributing.Success(): last_result_partial_failure = False + case ( + _power_distributing.Error() | _power_distributing.OutOfBounds() + ): + # No power was set at all, so there is nothing to correct here. + # Only a successful request clears the partial failure state. + pass + case unexpected: + assert_never(unexpected) await self._send_reports(frozenset(result.request.component_ids)) elif selected_from(selected, drop_old_proposals_timer): diff --git a/src/frequenz/sdk/microgrid/_power_managing/_shifting_matryoshka.py b/src/frequenz/sdk/microgrid/_power_managing/_shifting_matryoshka.py index 03281c795..dfaa15075 100644 --- a/src/frequenz/sdk/microgrid/_power_managing/_shifting_matryoshka.py +++ b/src/frequenz/sdk/microgrid/_power_managing/_shifting_matryoshka.py @@ -122,25 +122,24 @@ def _calc_targets( # pylint: disable=too-many-branches,too-many-statements if upper_bound < lower_bound: break - match (next_proposal.bounds.lower, next_proposal.bounds.upper): - case (None, None): + # Bounds the proposal leaves open are taken from the currently available + # bounds, unless the bound the proposal does set is already past them, in + # which case the proposal collapses to that single value. + proposal_lower = next_proposal.bounds.lower + proposal_upper = next_proposal.bounds.upper + if proposal_lower is None: + if proposal_upper is None: proposal_lower = lower_bound proposal_upper = upper_bound - case (Power(), None): - proposal_lower = next_proposal.bounds.lower - if proposal_lower > upper_bound: - proposal_upper = proposal_lower - else: - proposal_upper = upper_bound - case (None, Power()): - proposal_upper = next_proposal.bounds.upper - if proposal_upper < lower_bound: - proposal_lower = proposal_upper - else: - proposal_lower = lower_bound - case (Power(), Power()): - proposal_lower = next_proposal.bounds.lower - proposal_upper = next_proposal.bounds.upper + elif proposal_upper < lower_bound: + proposal_lower = proposal_upper + else: + proposal_lower = lower_bound + elif proposal_upper is None: + if proposal_lower > upper_bound: + proposal_upper = proposal_lower + else: + proposal_upper = upper_bound proposal_power = next_proposal.preferred_power diff --git a/src/frequenz/sdk/timeseries/formulas/_ast.py b/src/frequenz/sdk/timeseries/formulas/_ast.py index 3cc8d2e67..6340b7793 100644 --- a/src/frequenz/sdk/timeseries/formulas/_ast.py +++ b/src/frequenz/sdk/timeseries/formulas/_ast.py @@ -177,7 +177,11 @@ async def evaluate(self) -> Sample[QuantityT] | QuantityT | None: ) case (None, _) | (_, None): return None - return None + case unexpected: + # The cases above are exhaustive, but mypy can't narrow tuple patterns + # (see python/mypy#12364), so it neither sees that nor accepts + # assert_never() here. + raise AssertionError(f"Unexpected operands: {unexpected!r}") @override def format(self, wrap: bool = False) -> str: @@ -254,7 +258,11 @@ async def evaluate(self) -> Sample[QuantityT] | QuantityT | None: ) case (None, _) | (_, None): return None - return None + case unexpected: + # The cases above are exhaustive, but mypy can't narrow tuple patterns + # (see python/mypy#12364), so it neither sees that nor accepts + # assert_never() here. + raise AssertionError(f"Unexpected operands: {unexpected!r}") @override def format(self, wrap: bool = False) -> str: @@ -331,7 +339,11 @@ async def evaluate(self) -> Sample[QuantityT] | QuantityT | None: ) case (None, _) | (_, None): return None - return None + case unexpected: + # The cases above are exhaustive, but mypy can't narrow tuple patterns + # (see python/mypy#12364), so it neither sees that nor accepts + # assert_never() here. + raise AssertionError(f"Unexpected operands: {unexpected!r}") @override def format(self, wrap: bool = False) -> str: @@ -406,7 +418,11 @@ async def evaluate(self) -> Sample[QuantityT] | QuantityT | None: ) case (None, _) | (_, None): return None - return None + case unexpected: + # The cases above are exhaustive, but mypy can't narrow tuple patterns + # (see python/mypy#12364), so it neither sees that nor accepts + # assert_never() here. + raise AssertionError(f"Unexpected operands: {unexpected!r}") @override def format(self, wrap: bool = False) -> str: diff --git a/src/frequenz/sdk/timeseries/formulas/_functions.py b/src/frequenz/sdk/timeseries/formulas/_functions.py index e494ce081..b0e9e0b3e 100644 --- a/src/frequenz/sdk/timeseries/formulas/_functions.py +++ b/src/frequenz/sdk/timeseries/formulas/_functions.py @@ -10,7 +10,7 @@ import logging from dataclasses import dataclass, field from datetime import datetime -from typing import Generic +from typing import Generic, assert_never from frequenz.quantities import Quantity from typing_extensions import override @@ -117,7 +117,7 @@ def name(self) -> str: """Return the name of the function.""" return "COALESCE" - @override + @override # pylint: disable-next=too-many-branches async def __call__(self) -> Sample[QuantityT] | QuantityT | None: """Return the first non-None argument.""" ts: datetime | None = None @@ -132,16 +132,29 @@ async def __call__(self) -> Sample[QuantityT] | QuantityT | None: match arg: case Sample(timestamp, value): if value is not None: - # Keep track of which parameter we are getting samples from. - # this slightly convoluted check ensures that we unsubscribe - # from the last parameter if any earlier one produces at least - # REQUIRED_CONSECUTIVE_STABLE_SAMPLES samples, regardless of - # intermittent non-None values received from other params. - match self.used_param > 0 and args[self.used_param - 1]: - case False | Sample(value=None): + # Keeps track of which parameter we are getting samples + # from. This slightly convoluted check ensures that we + # unsubscribe from the last parameter if any earlier + # one produces at least + # `REQUIRED_CONSECUTIVE_STABLE_SAMPLES` samples, + # regardless of intermittent non-None values received + # from other params. + used_arg = ( + args[self.used_param - 1] if self.used_param > 0 else None + ) + match used_arg: + case None | Sample(value=None): + # We are not tracking a parameter yet, or the + # one we track stopped producing values, so + # track this one instead. self.used_param = param self.num_samples = 0 - + case Sample() | Quantity(): + # The tracked parameter is still producing + # values, so keep counting samples for it. + pass + case unexpected: + assert_never(unexpected) self.num_samples += 1 if ( diff --git a/tests/timeseries/_formulas/test_functions.py b/tests/timeseries/_formulas/test_functions.py new file mode 100644 index 000000000..289c82cd2 --- /dev/null +++ b/tests/timeseries/_formulas/test_functions.py @@ -0,0 +1,82 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for the formula functions.""" + +from dataclasses import dataclass +from datetime import datetime, timezone + +from frequenz.quantities import Quantity +from typing_extensions import override + +from frequenz.sdk.timeseries import Sample +from frequenz.sdk.timeseries.formulas._base_ast_node import AstNode +from frequenz.sdk.timeseries.formulas._functions import Coalesce + + +@dataclass(kw_only=True) +class _ScriptedNode(AstNode[Quantity]): + """An AST node evaluating to a predefined sequence of values.""" + + values: list[Sample[Quantity] | Quantity | None] + """The values to return, one per evaluation, `None` once exhausted.""" + + subscribed: bool = False + """Whether this node is currently subscribed.""" + + @override + async def evaluate(self) -> Sample[Quantity] | Quantity | None: + """Return the next scripted value.""" + return self.values.pop(0) if self.values else None + + @override + def format(self, wrap: bool = False) -> str: + """Return a string representation of this node.""" + return "scripted" + + @override + async def subscribe(self) -> None: + """Mark this node as subscribed.""" + self.subscribed = True + + @override + async def unsubscribe(self) -> None: + """Mark this node as unsubscribed.""" + self.subscribed = False + + +class TestCoalesce: + """Tests for the `COALESCE()` function.""" + + async def test_param_evaluating_to_none(self) -> None: + """Test a param evaluating to `None` stops being the tracked one. + + A param evaluates to `None`, rather than to a `Sample` with no value, when it + is an expression that has no value at all yet, like a nested function call. + Such a param must stop being the tracked one, or the samples produced by + another param are counted against it, and the coalesce ends up unsubscribing + from the param actually producing values. + """ + timestamp = datetime.now(timezone.utc) + params = [ + _ScriptedNode(values=[None] * 6), + _ScriptedNode(values=[Sample(timestamp, Quantity(1.0))] + [None] * 4), + _ScriptedNode(values=[Sample(timestamp, Quantity(2.0))] * 3), + ] + coalesce = Coalesce(params=list(params)) + + # Params are subscribed to one by one, as the previous ones fail to produce a + # value: #1 from the start, #2 and #3 after each call returning `None`. + assert await coalesce() is None + assert [param.subscribed for param in params] == [True, True, False] + assert await coalesce() == Sample(timestamp, Quantity(1.0)) + assert await coalesce() is None + assert [param.subscribed for param in params] == [True, True, True] + + # From here on only #3 produces values, while #2, the tracked param, evaluates + # to `None`. So #3 must become the tracked param, instead of accumulating + # stable samples for #2, which would unsubscribe from #3 once it reached the + # required number of samples. + for _ in range(3): + assert await coalesce() == Sample(timestamp, Quantity(2.0)) + assert [param.subscribed for param in params] == [True, True, True]