Skip to content

Commit f48d23e

Browse files
shsmsshsms
authored
Make the SDK's representation of bounds data internal (#756)
This make the `inclusion_bounds` and `exclusion_bounds` fields of the battery pool's `Report` class private, and instead introduces a `bounds` property as a single set of bounds that users can use. This PR also introduces the `Report.adjust_to_bounds` method which can be used in case more granularity is needed on what exact power values can be proposed to the battery pool.
2 parents 0e69660 + 78bb25f commit f48d23e

File tree

7 files changed

+386
-150
lines changed

7 files changed

+386
-150
lines changed

‎src/frequenz/sdk/actor/_power_managing/_base_classes.py‎

Lines changed: 81 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313

1414
from ... import timeseries
1515
from ...timeseries import Power
16+
from . import _bounds
1617

1718
if typing.TYPE_CHECKING:
1819
from ...timeseries.battery_pool import PowerMetrics
@@ -49,14 +50,20 @@ class Report:
4950
target_power: Power | None
5051
"""The currently set power for the batteries."""
5152

52-
inclusion_bounds: timeseries.Bounds[Power] | None
53+
distribution_result: power_distributing.Result | None
54+
"""The result of the last power distribution.
55+
56+
This is `None` if no power distribution has been performed yet.
57+
"""
58+
59+
_inclusion_bounds: timeseries.Bounds[Power] | None
5360
"""The available inclusion bounds for the batteries, for the actor's priority.
5461
5562
These bounds are adjusted to any restrictions placed by actors with higher
5663
priorities.
5764
"""
5865

59-
exclusion_bounds: timeseries.Bounds[Power] | None
66+
_exclusion_bounds: timeseries.Bounds[Power] | None
6067
"""The exclusion bounds for the batteries.
6168
6269
The power manager doesn't manage exclusion bounds, so these are aggregations of
@@ -66,11 +73,79 @@ class Report:
6673
priorities.
6774
"""
6875

69-
distribution_result: power_distributing.Result | None
70-
"""The result of the last power distribution.
76+
@property
77+
def bounds(self) -> timeseries.Bounds[Power] | None:
78+
"""The bounds for the batteries.
7179
72-
This is `None` if no power distribution has been performed yet.
73-
"""
80+
These bounds are adjusted to any restrictions placed by actors with higher
81+
priorities.
82+
83+
There might be exclusion zones within these bounds. If necessary, the
84+
[`adjust_to_bounds`][frequenz.sdk.timeseries.battery_pool.Report.adjust_to_bounds]
85+
method may be used to check if a desired power value fits the bounds, or to get
86+
the closest possible power values that do fit the bounds.
87+
"""
88+
return self._inclusion_bounds
89+
90+
def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
91+
"""Adjust a power value to the bounds.
92+
93+
This method can be used to adjust a desired power value to the power bounds
94+
available to the actor.
95+
96+
If the given power value falls within the usable bounds, it will be returned
97+
unchanged.
98+
99+
If it falls outside the usable bounds, the closest possible value on the
100+
corresponding side will be returned. For example, if the given power is lower
101+
than the lowest usable power, only the lowest usable power will be returned, and
102+
similarly for the highest usable power.
103+
104+
If the given power falls within an exclusion zone that's contained within the
105+
usable bounds, the closest possible power values on both sides will be returned.
106+
107+
!!! note
108+
It is completely optional to use this method to adjust power values before
109+
proposing them through the battery pool, because the battery pool will do
110+
this automatically. This method is provided for convenience, and for
111+
granular control when there are two possible power values, both of which
112+
fall within the available bounds.
113+
114+
Example:
115+
```python
116+
from frequenz.sdk import microgrid
117+
118+
power_status_rx = microgrid.battery_pool().power_status.new_receiver()
119+
power_status = await power_status_rx.receive()
120+
desired_power = Power.from_watts(1000.0)
121+
122+
match power_status.adjust_to_bounds(desired_power):
123+
case (power, _) if power == desired_power:
124+
print("Desired power is available.")
125+
case (None, power) | (power, None) if power:
126+
print(f"Closest available power is {power}.")
127+
case (lower, upper) if lower and upper:
128+
print(f"Two options {lower}, {upper} to propose to battery pool.")
129+
case (None, None):
130+
print("No available power")
131+
```
132+
133+
Args:
134+
power: The power value to adjust.
135+
136+
Returns:
137+
A tuple of the closest power values to the desired power that fall within
138+
the available bounds for the actor.
139+
"""
140+
if self._inclusion_bounds is None:
141+
return None, None
142+
143+
return _bounds.clamp_to_bounds(
144+
power,
145+
self._inclusion_bounds.lower,
146+
self._inclusion_bounds.upper,
147+
self._exclusion_bounds,
148+
)
74149

75150

76151
@dataclasses.dataclass(frozen=True, kw_only=True)

‎src/frequenz/sdk/actor/_power_managing/_bounds.py‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
# License: MIT
2+
# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
3+
4+
"""Utilities for checking and clamping bounds and power values to exclusion bounds."""
5+
6+
from ...timeseries import Bounds, Power
7+
8+
9+
def check_exclusion_bounds_overlap(
10+
lower_bound: Power,
11+
upper_bound: Power,
12+
exclusion_bounds: Bounds[Power] | None,
13+
) -> tuple[bool, bool]:
14+
"""Check if the given bounds overlap with the given exclusion bounds.
15+
16+
Example:
17+
18+
```
19+
lower upper
20+
.----- exclusion zone -----.
21+
-----|✓✓✓✓✓✓✓✓✓✓✓✓|xxxxxxxxxxxxxxx|----------|----
22+
`-- usable --'-- exclusion --´
23+
| overlap |
24+
| |
25+
lower upper
26+
bound bound
27+
(inside the exclusion zone)
28+
```
29+
30+
Resulting in `(False, True)` because only the upper bound is inside the
31+
exclusion zone.
32+
33+
Args:
34+
lower_bound: The lower bound to check.
35+
upper_bound: The upper bound to check.
36+
exclusion_bounds: The exclusion bounds to check against.
37+
38+
Returns:
39+
A tuple containing a boolean indicating if the lower bound is bounded by the
40+
exclusion bounds, and a boolean indicating if the upper bound is bounded by
41+
the exclusion bounds.
42+
"""
43+
if exclusion_bounds is None:
44+
return False, False
45+
46+
bounded_lower = False
47+
bounded_upper = False
48+
49+
if exclusion_bounds.lower < lower_bound < exclusion_bounds.upper:
50+
bounded_lower = True
51+
if exclusion_bounds.lower < upper_bound < exclusion_bounds.upper:
52+
bounded_upper = True
53+
54+
return bounded_lower, bounded_upper
55+
56+
57+
def adjust_exclusion_bounds(
58+
lower_bound: Power,
59+
upper_bound: Power,
60+
exclusion_bounds: Bounds[Power] | None,
61+
) -> tuple[Power, Power]:
62+
"""Adjust the given bounds to exclude the given exclusion bounds.
63+
64+
Args:
65+
lower_bound: The lower bound to adjust.
66+
upper_bound: The upper bound to adjust.
67+
exclusion_bounds: The exclusion bounds to adjust to.
68+
69+
Returns:
70+
The adjusted lower and upper bounds.
71+
"""
72+
if exclusion_bounds is None:
73+
return lower_bound, upper_bound
74+
75+
# If the given bounds are within the exclusion bounds, there's no room to adjust,
76+
# so return zero.
77+
#
78+
# And if the given bounds overlap with the exclusion bounds on one side, then clamp
79+
# the given bounds on that side.
80+
match check_exclusion_bounds_overlap(lower_bound, upper_bound, exclusion_bounds):
81+
case (True, True):
82+
return Power.zero(), Power.zero()
83+
case (False, True):
84+
return lower_bound, exclusion_bounds.lower
85+
case (True, False):
86+
return exclusion_bounds.upper, upper_bound
87+
return lower_bound, upper_bound
88+
89+
90+
# Just 20 lines of code in this function, but unfortunately 8 of those are return
91+
# statements, and that's too many for pylint.
92+
def clamp_to_bounds( # pylint: disable=too-many-return-statements
93+
value: Power,
94+
lower_bound: Power,
95+
upper_bound: Power,
96+
exclusion_bounds: Bounds[Power] | None,
97+
) -> tuple[Power | None, Power | None]:
98+
"""Clamp the given value to the given bounds.
99+
100+
When the given value can falls within the exclusion zone, and can be clamped to
101+
both sides, both options will be returned.
102+
103+
When the given value falls outside the usable bounds and can be clamped only to
104+
one side, only that option will be returned.
105+
106+
Args:
107+
value: The value to clamp.
108+
lower_bound: The lower bound to clamp to.
109+
upper_bound: The upper bound to clamp to.
110+
exclusion_bounds: The exclusion bounds to clamp outside of.
111+
112+
Returns:
113+
The clamped value.
114+
"""
115+
# If the given bounds are within the exclusion bounds, return zero.
116+
#
117+
# And if the given bounds overlap with the exclusion bounds on one side, and the
118+
# given power is in that overlap region, clamp it to the exclusion bounds on that
119+
# side.
120+
if exclusion_bounds is not None:
121+
match check_exclusion_bounds_overlap(
122+
lower_bound, upper_bound, exclusion_bounds
123+
):
124+
case (True, True):
125+
return None, None
126+
case (True, False):
127+
if value < exclusion_bounds.upper:
128+
return None, exclusion_bounds.upper
129+
case (False, True):
130+
if value > exclusion_bounds.lower:
131+
return exclusion_bounds.lower, None
132+
133+
# If the given value is outside the given bounds, clamp it to the closest bound.
134+
if value < lower_bound:
135+
return lower_bound, None
136+
if value > upper_bound:
137+
return None, upper_bound
138+
139+
# If the given value is within the exclusion bounds and the exclusion bounds are
140+
# within the given bounds, clamp the given value to the closest exclusion bound.
141+
if exclusion_bounds is not None:
142+
if exclusion_bounds.lower < value < exclusion_bounds.upper:
143+
return exclusion_bounds.lower, exclusion_bounds.upper
144+
145+
return value, value

0 commit comments

Comments
 (0)