Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions RELEASE_NOTES.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@

<!-- Here goes the main new features and examples or instructions on how to use them -->

- 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.
Expand Down
54 changes: 52 additions & 2 deletions src/frequenz/sdk/timeseries/_base_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand All @@ -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
92 changes: 92 additions & 0 deletions tests/timeseries/test_base_types.py
Original file line number Diff line number Diff line change
@@ -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


Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would also add some tests for corner cases, like using plain ints, using an empty range, like Bounds(lower=0, upper=0), using floats, negative numbers, and trying Bounds(lower=10, upper=-10), which is actually not enforced now, not related to this PR but it could be a nice addition to validate the arguments in __post_init__().

I think also adding kw_only to the @dataclass(froze=True, kw_only=True) for both bounds classes would be a great addition too, as it would be very confusing if users use Bounds(10, 14); which is which?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I reopened this because the suggested changes were not made. Even if the extra validation is not added, at least having tests for this would be good:

I would also add some tests for corner cases, like using plain ints, using an empty range, like Bounds(lower=0, upper=0)

If you think this should not be done, it would be good to give a reason.

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