|
11 | 11 | from frequenz.sdk.timeseries import Power, battery_pool |
12 | 12 |
|
13 | 13 |
|
| 14 | +class StatefulTester: |
| 15 | + """A stateful tester for the Matryoshka algorithm.""" |
| 16 | + |
| 17 | + def __init__( |
| 18 | + self, |
| 19 | + batteries: frozenset[int], |
| 20 | + system_bounds: battery_pool.PowerMetrics, |
| 21 | + ) -> None: |
| 22 | + """Create a new instance of the stateful tester.""" |
| 23 | + self._call_count = 0 |
| 24 | + self._batteries = batteries |
| 25 | + self._algorithm = Matryoshka() |
| 26 | + self._system_bounds = system_bounds |
| 27 | + |
| 28 | + def tgt_power( # pylint: disable=too-many-arguments |
| 29 | + self, |
| 30 | + priority: int, |
| 31 | + power: float | None, |
| 32 | + bounds: tuple[float | None, float | None], |
| 33 | + expected: float | None, |
| 34 | + must_send: bool = False, |
| 35 | + ) -> None: |
| 36 | + """Test the target power calculation.""" |
| 37 | + self._call_count += 1 |
| 38 | + tgt_power = self._algorithm.calculate_target_power( |
| 39 | + self._batteries, |
| 40 | + Proposal( |
| 41 | + battery_ids=self._batteries, |
| 42 | + source_id=f"actor-{priority}", |
| 43 | + preferred_power=None if power is None else Power.from_watts(power), |
| 44 | + bounds=timeseries.Bounds( |
| 45 | + None if bounds[0] is None else Power.from_watts(bounds[0]), |
| 46 | + None if bounds[1] is None else Power.from_watts(bounds[1]), |
| 47 | + ), |
| 48 | + priority=priority, |
| 49 | + ), |
| 50 | + self._system_bounds, |
| 51 | + must_send, |
| 52 | + ) |
| 53 | + assert tgt_power == ( |
| 54 | + Power.from_watts(expected) if expected is not None else None |
| 55 | + ) |
| 56 | + |
| 57 | + def bounds( |
| 58 | + self, |
| 59 | + priority: int, |
| 60 | + expected_power: float | None, |
| 61 | + expected_bounds: tuple[float, float], |
| 62 | + ) -> None: |
| 63 | + """Test the status report.""" |
| 64 | + report = self._algorithm.get_status( |
| 65 | + self._batteries, priority, self._system_bounds, None |
| 66 | + ) |
| 67 | + if expected_power is None: |
| 68 | + assert report.target_power is None |
| 69 | + else: |
| 70 | + assert report.target_power is not None |
| 71 | + assert report.target_power.as_watts() == expected_power |
| 72 | + assert report.inclusion_bounds is not None |
| 73 | + assert report.inclusion_bounds.lower.as_watts() == expected_bounds[0] |
| 74 | + assert report.inclusion_bounds.upper.as_watts() == expected_bounds[1] |
| 75 | + |
| 76 | + |
14 | 77 | async def test_matryoshka_algorithm() -> None: # pylint: disable=too-many-statements |
15 | 78 | """Tests for the power managing actor.""" |
16 | 79 | batteries = frozenset({2, 5}) |
|
0 commit comments