From cb67d18dd5866e8a4fef2eb5ae578b661785c66e Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 8 Sep 2026 07:48:31 +0000 Subject: [PATCH 1/5] Return an `ExclusionOverlap` from `check_exclusion_bounds_overlap()` The function returned a `tuple[bool, bool]`, which callers matched on with one `case` per combination. mypy can't narrow tuple items across `case` clauses (python/mypy#12364), so it can't tell whether such a `match` covers every combination, and a missing case would go unnoticed. This is about to matter, as the next `frequenz-repo-config` enables mypy's `exhaustive-match` error code, which reports these two `match` statements. Return a four-valued enum instead, which mypy does check exhaustively, and which also reads better: `(True, False)` gave no hint about which of the two bounds it referred to. The two call sites in `_matryoshka.py` only cared about one combination, so they become plain `if` statements. Both matches also get an `assert_never()` case, to fail early if an unexpected value ever reaches them: type hints are not enforced at runtime. Signed-off-by: Leandro Lucarella --- .../sdk/microgrid/_power_managing/_bounds.py | 76 +++++++++++++------ .../microgrid/_power_managing/_matryoshka.py | 20 +++-- 2 files changed, 66 insertions(+), 30 deletions(-) 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: From 5078ef3e7c6db7b8b501b2d25d8cf6cc03989a8f Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 8 Sep 2026 07:48:42 +0000 Subject: [PATCH 2/5] Fix `COALESCE()` tracking a param that evaluates to `None` `COALESCE()` keeps track of which parameter it is getting samples from, and switches to another one when the tracked parameter stops producing values, so that it can unsubscribe from the parameters after it once the new one is stable. It only recognized a `Sample` without a value as "stopped producing", not a parameter evaluating to `None`. A parameter evaluates to `None`, rather than to an empty `Sample`, when it is an expression with no value at all yet, like a nested function call. In that case the samples of the parameter actually producing values were counted against the tracked one, and once the count reached the required number of stable samples, `COALESCE()` unsubscribed from every parameter after the tracked one, dropping the only producer. Treat both cases the same, and add a test covering it. The check was written as a `match` over `used_param > 0 and args[...]`, which made the missing case easy to miss. Spell the value out first and match over it, so all the cases are visible. The match also gets an `assert_never()` case, to fail early if an unexpected value ever reaches it. Signed-off-by: Leandro Lucarella --- RELEASE_NOTES.md | 2 +- .../sdk/timeseries/formulas/_functions.py | 33 +++++--- tests/timeseries/_formulas/test_functions.py | 82 +++++++++++++++++++ 3 files changed, 106 insertions(+), 11 deletions(-) create mode 100644 tests/timeseries/_formulas/test_functions.py 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/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] From 0aa21e75ed70492fe3213c145aa6b288c8aed25c Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 8 Sep 2026 08:50:38 +0000 Subject: [PATCH 3/5] Handle every power distributing result explicitly The `match` over the results only had cases for `PartialFailure` and `Success`, so `Error` and `OutOfBounds` fell through it silently. That is the intended behaviour, as no power was set at all in either case, so there is nothing to correct, and only a successful request should clear the partial failure state. Document it, so it reads as a decision rather than an oversight. There is no behaviour change. The final `assert_never()` case makes it fail early if an unexpected result ever reaches it: type hints are not enforced at runtime. Signed-off-by: Leandro Lucarella --- .../microgrid/_power_managing/_power_managing_actor.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) 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): From 753f8f89c725a7b7c5cad091f6843cdb628610aa Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 8 Sep 2026 08:50:38 +0000 Subject: [PATCH 4/5] Take the proposal bounds apart with `if`s instead of a `match` The `match` covered all four combinations of a proposal's optional lower and upper bounds, but mypy can't narrow tuple items across `case` clauses (see python/mypy#12364), so it can't prove such a `match` exhaustive no matter how many cases are written. Nested `if`s over the two bounds cover the same four combinations, mypy does check those, and the shared code between the cases no longer needs repeating. There is no behaviour change. Signed-off-by: Leandro Lucarella --- .../_power_managing/_shifting_matryoshka.py | 33 +++++++++---------- 1 file changed, 16 insertions(+), 17 deletions(-) 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 From 3ca5a653c43ff2fc69781b7a68da3a83cd6dbda5 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Tue, 8 Sep 2026 08:50:38 +0000 Subject: [PATCH 5/5] Raise on unexpected operands in the formula operators The `match` in each binary operator covers all four combinations of its two operands, plus the ones where an operand is `None`, and was followed by an unreachable `return None`. mypy can't narrow tuple items across `case` clauses (see python/mypy#12364), so it can't prove the `match` exhaustive no matter how many cases are written. Turn that trailing `return None` into the explicit last case, and raise instead of returning, so an operand of an unexpected type fails early rather than silently evaluating the whole formula to nothing: type hints are not enforced at runtime. `assert_never()` can't be used here, as mypy doesn't narrow the subject to `Never`. Signed-off-by: Leandro Lucarella --- src/frequenz/sdk/timeseries/formulas/_ast.py | 24 ++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) 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: