diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index f355a5375..1d989834d 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -15,6 +15,8 @@ +- Classes Bounds and SystemBounds now work with the `in` operator + ## Bug Fixes - Fixed a typing issue that occurs in some cases when composing formulas with constants. diff --git a/src/frequenz/sdk/timeseries/_base_types.py b/src/frequenz/sdk/timeseries/_base_types.py index c5e9ddd6b..f6732fb27 100644 --- a/src/frequenz/sdk/timeseries/_base_types.py +++ b/src/frequenz/sdk/timeseries/_base_types.py @@ -8,7 +8,7 @@ from collections.abc import Callable, Iterator from dataclasses import dataclass from datetime import datetime, timezone -from typing import Generic, Self, TypeVar, overload +from typing import Any, Generic, Protocol, Self, TypeVar, cast, overload from ._quantities import Power, QuantityT @@ -134,7 +134,22 @@ def map( ) -_T = TypeVar("_T") +class Comparable(Protocol): + """A protocol that requires the implementation of comparison methods. + + This protocol is used to ensure that types can be compared using + the less than or equal to (`<=`) and greater than or equal to (`>=`) + operators. + """ + + def __le__(self, other: Any, /) -> bool: + """Return whether this instance is less than or equal to `other`.""" + + def __ge__(self, other: Any, /) -> bool: + """Return whether this instance is greater than or equal to `other`.""" + + +_T = TypeVar("_T", bound=Comparable | None) @dataclass(frozen=True) @@ -147,6 +162,25 @@ class Bounds(Generic[_T]): upper: _T """Upper bound.""" + def __contains__(self, item: _T) -> bool: + """ + Check if the value is within the range of the container. + + Args: + item: The value to check. + + Returns: + bool: True if value is within the range, otherwise False. + """ + if self.lower is None and self.upper is None: + return True + if self.lower is None: + return item <= self.upper + if self.upper is None: + return self.lower <= item + + return cast(Comparable, self.lower) <= item <= cast(Comparable, self.upper) + @dataclass(frozen=True, kw_only=True) class SystemBounds: @@ -171,3 +205,19 @@ class SystemBounds: This is the range within which power requests are NOT allowed by the pool. If present, they will be a subset of the inclusion bounds. """ + + def __contains__(self, item: Power) -> bool: + """ + Check if the value is within the range of the container. + + Args: + item: The value to check. + + Returns: + bool: True if value is within the range, otherwise False. + """ + if not self.inclusion_bounds or item not in self.inclusion_bounds: + return False + if self.exclusion_bounds and item in self.exclusion_bounds: + return False + return True diff --git a/tests/timeseries/test_base_types.py b/tests/timeseries/test_base_types.py new file mode 100644 index 000000000..37ebee078 --- /dev/null +++ b/tests/timeseries/test_base_types.py @@ -0,0 +1,92 @@ +# License: MIT +# Copyright © 2024 Frequenz Energy-as-a-Service GmbH + +"""Tests for timeseries base types.""" + + +from datetime import datetime + +from frequenz.sdk.timeseries._base_types import Bounds, SystemBounds +from frequenz.sdk.timeseries._quantities import Power + + +def test_bounds_contains() -> None: + """Tests with complete bounds.""" + bounds = Bounds(lower=Power.from_watts(10), upper=Power.from_watts(100)) + assert Power.from_watts(50) in bounds # within + assert Power.from_watts(10) in bounds # at lower + assert Power.from_watts(100) in bounds # at upper + assert Power.from_watts(9) not in bounds # below lower + assert Power.from_watts(101) not in bounds # above upper + + +def test_bounds_contains_no_lower() -> None: + """Tests without lower bound.""" + bounds_no_lower = Bounds(lower=None, upper=Power.from_watts(100)) + assert Power.from_watts(50) in bounds_no_lower # within upper + assert Power.from_watts(100) in bounds_no_lower # at upper + assert Power.from_watts(101) not in bounds_no_lower # above upper + + +def test_bounds_contains_no_upper() -> None: + """Tests without upper bound.""" + bounds_no_upper = Bounds(lower=Power.from_watts(10), upper=None) + assert Power.from_watts(50) in bounds_no_upper # within lower + assert Power.from_watts(10) in bounds_no_upper # at lower + assert Power.from_watts(9) not in bounds_no_upper # below lower + + +def test_bounds_contains_no_bounds() -> None: + """Tests with no bounds.""" + bounds_no_bounds: Bounds[Power | None] = Bounds(lower=None, upper=None) + assert Power.from_watts(50) in bounds_no_bounds # any value within bounds + + +INCLUSION_BOUND = Bounds(lower=Power.from_watts(10), upper=Power.from_watts(100)) +EXCLUSION_BOUND = Bounds(lower=Power.from_watts(40), upper=Power.from_watts(50)) + + +def test_system_bounds_contains() -> None: + """Tests with complete system bounds.""" + system_bounds = SystemBounds( + timestamp=datetime.now(), + inclusion_bounds=INCLUSION_BOUND, + exclusion_bounds=EXCLUSION_BOUND, + ) + + assert Power.from_watts(30) in system_bounds # within inclusion, not in exclusion + assert Power.from_watts(45) not in system_bounds # within inclusion and exclusion + assert Power.from_watts(110) not in system_bounds # outside inclusion + + +def test_system_bounds_contains_no_exclusion() -> None: + """Tests with no exclusion bounds.""" + system_bounds_no_exclusion = SystemBounds( + timestamp=datetime.now(), + inclusion_bounds=INCLUSION_BOUND, + exclusion_bounds=None, + ) + assert Power.from_watts(30) in system_bounds_no_exclusion # within inclusion + assert Power.from_watts(110) not in system_bounds_no_exclusion # outside inclusion + + +def test_system_bounds_contains_no_inclusion() -> None: + """Tests with no inclusion bounds.""" + system_bounds_no_inclusion = SystemBounds( + timestamp=datetime.now(), + inclusion_bounds=None, + exclusion_bounds=EXCLUSION_BOUND, + ) + assert Power.from_watts(30) not in system_bounds_no_inclusion # outside exclusion + assert Power.from_watts(45) not in system_bounds_no_inclusion # within exclusion + + +def test_system_bounds_contains_no_bounds() -> None: + """Tests with no bounds.""" + system_bounds_no_bounds = SystemBounds( + timestamp=datetime.now(), + inclusion_bounds=None, + exclusion_bounds=None, + ) + assert Power.from_watts(30) not in system_bounds_no_bounds # any value outside + assert Power.from_watts(110) not in system_bounds_no_bounds # any value outside