|
| 1 | +# ----------------------------------------------------------------------------- |
| 2 | +# Predbat Home Battery System |
| 3 | +# Copyright Trefor Southwell 2026 - All Rights Reserved |
| 4 | +# This application maybe used for personal use only and not for commercial use |
| 5 | +# ----------------------------------------------------------------------------- |
| 6 | +# fmt off |
| 7 | +# pylint: disable=consider-using-f-string |
| 8 | +# pylint: disable=line-too-long |
| 9 | +# pylint: disable=attribute-defined-outside-init |
| 10 | +"""Tests for the region tiling geometry and the region portfolio branch selection.""" |
| 11 | + |
| 12 | +from plan import REGION_SIZE_MIN, REGION_SIZE_START, REGION_SWITCH_THRESHOLD |
| 13 | + |
| 14 | + |
| 15 | +def _fail(name, message): |
| 16 | + """Print a test failure and return True""" |
| 17 | + print("ERROR: {} - {}".format(name, message)) |
| 18 | + return True |
| 19 | + |
| 20 | + |
| 21 | +def test_region_passes_uniform(my_predbat): |
| 22 | + """Uniform tiling halves the region width each pass and tiles the record end to start""" |
| 23 | + failed = False |
| 24 | + passes = my_predbat.compute_region_passes(0, 48 * 60, stagger=False) |
| 25 | + |
| 26 | + sizes = [size for size, _ in passes] |
| 27 | + if sizes != [960, 480, 240, 120]: |
| 28 | + failed |= _fail("region_passes_uniform", "expected widths [960, 480, 240, 120] got {}".format(sizes)) |
| 29 | + |
| 30 | + counts = [len(regions) for _, regions in passes] |
| 31 | + if counts != [3, 6, 12, 24]: |
| 32 | + failed |= _fail("region_passes_uniform", "expected counts [3, 6, 12, 24] got {}".format(counts)) |
| 33 | + |
| 34 | + # Tiles must not overlap within a pass and must be exactly one region wide (bar the clipped first tile) |
| 35 | + for size, regions in passes: |
| 36 | + ordered = sorted(regions) |
| 37 | + for index in range(1, len(ordered)): |
| 38 | + if ordered[index][0] < ordered[index - 1][1]: |
| 39 | + failed |= _fail("region_passes_uniform", "width {} tiles {} and {} overlap".format(size, ordered[index - 1], ordered[index])) |
| 40 | + return failed |
| 41 | + |
| 42 | + |
| 43 | +def test_region_passes_stagger(my_predbat): |
| 44 | + """Staggered tiling advances by half a region so each boundary is interior to the next tile""" |
| 45 | + failed = False |
| 46 | + uniform = my_predbat.compute_region_passes(0, 48 * 60, stagger=False) |
| 47 | + stagger = my_predbat.compute_region_passes(0, 48 * 60, stagger=True) |
| 48 | + |
| 49 | + if [size for size, _ in uniform] != [size for size, _ in stagger]: |
| 50 | + failed |= _fail("region_passes_stagger", "stagger changed the pass widths") |
| 51 | + |
| 52 | + for (size, uniform_regions), (_, stagger_regions) in zip(uniform, stagger): |
| 53 | + if size == REGION_SIZE_MIN: |
| 54 | + # The half-region step is clamped to the minimum width, so the finest pass is identical |
| 55 | + # in both layouts. There is nothing narrower to offset into, and the fine passes refine |
| 56 | + # an already-good plan rather than discovering pairs, so the offset buys nothing there. |
| 57 | + if sorted(stagger_regions) != sorted(uniform_regions): |
| 58 | + failed |= _fail("region_passes_stagger", "the finest pass should not be staggered") |
| 59 | + continue |
| 60 | + |
| 61 | + if len(stagger_regions) <= len(uniform_regions): |
| 62 | + failed |= _fail("region_passes_stagger", "width {} produced {} tiles, expected more than uniform's {}".format(size, len(stagger_regions), len(uniform_regions))) |
| 63 | + |
| 64 | + # Every interior boundary of the uniform layout must fall strictly inside some staggered |
| 65 | + # tile - that is the whole point, so a window pair the uniform tiling splits stays together. |
| 66 | + interior = sorted({start for start, _ in uniform_regions})[1:] |
| 67 | + for boundary in interior: |
| 68 | + if not any(start < boundary < end for start, end in stagger_regions): |
| 69 | + failed |= _fail("region_passes_stagger", "width {} boundary {} is not interior to any staggered tile".format(size, boundary)) |
| 70 | + return failed |
| 71 | + |
| 72 | + |
| 73 | +def test_region_passes_bounds(my_predbat): |
| 74 | + """Regions stay inside the record and drop tiles that end before the current time""" |
| 75 | + failed = False |
| 76 | + minutes_now = 600 |
| 77 | + end_max = 600 + 36 * 60 |
| 78 | + for stagger in (False, True): |
| 79 | + for size, regions in my_predbat.compute_region_passes(minutes_now, end_max, stagger=stagger): |
| 80 | + for start, end in regions: |
| 81 | + if start < 0 or end > end_max: |
| 82 | + failed |= _fail("region_passes_bounds", "width {} tile {} escapes [0, {}]".format(size, (start, end), end_max)) |
| 83 | + if end < minutes_now: |
| 84 | + failed |= _fail("region_passes_bounds", "width {} tile {} ends before minutes_now {}".format(size, (start, end), minutes_now)) |
| 85 | + if start >= end: |
| 86 | + failed |= _fail("region_passes_bounds", "width {} tile {} is empty".format(size, (start, end))) |
| 87 | + return failed |
| 88 | + |
| 89 | + |
| 90 | +def test_region_passes_cover_record(my_predbat): |
| 91 | + """Each pass covers the whole live part of the record, so no window is left unoptimised""" |
| 92 | + failed = False |
| 93 | + minutes_now = 300 |
| 94 | + end_max = 300 + 48 * 60 |
| 95 | + for stagger in (False, True): |
| 96 | + for size, regions in my_predbat.compute_region_passes(minutes_now, end_max, stagger=stagger): |
| 97 | + covered = sorted(regions) |
| 98 | + reach = covered[0][0] |
| 99 | + if reach > minutes_now: |
| 100 | + failed |= _fail("region_passes_cover", "width {} starts at {} leaving {} uncovered".format(size, reach, minutes_now)) |
| 101 | + for start, end in covered: |
| 102 | + if start > reach: |
| 103 | + failed |= _fail("region_passes_cover", "width {} has a gap at {}".format(size, reach)) |
| 104 | + reach = max(reach, end) |
| 105 | + if reach < end_max: |
| 106 | + failed |= _fail("region_passes_cover", "width {} reaches {} not {}".format(size, reach, end_max)) |
| 107 | + return failed |
| 108 | + |
| 109 | + |
| 110 | +def test_region_passes_min_size(my_predbat): |
| 111 | + """The descent stops at the minimum region width and never emits a narrower pass""" |
| 112 | + failed = False |
| 113 | + for stagger in (False, True): |
| 114 | + passes = my_predbat.compute_region_passes(0, 48 * 60, min_region_size=240, stagger=stagger) |
| 115 | + sizes = [size for size, _ in passes] |
| 116 | + if min(sizes) < 240: |
| 117 | + failed |= _fail("region_passes_min_size", "emitted a pass narrower than the minimum: {}".format(sizes)) |
| 118 | + if sizes[0] != REGION_SIZE_START: |
| 119 | + failed |= _fail("region_passes_min_size", "first pass should be the widest ({}) got {}".format(REGION_SIZE_START, sizes[0])) |
| 120 | + return failed |
| 121 | + |
| 122 | + |
| 123 | +def test_select_region_branch(my_predbat): |
| 124 | + """Branch B is kept only when it clears the switching threshold, otherwise the incumbent wins""" |
| 125 | + failed = False |
| 126 | + cases = [ |
| 127 | + ([100.0], 0, "a lone branch is always the winner"), |
| 128 | + ([100.0, 100.0 - REGION_SWITCH_THRESHOLD - 0.01], 1, "clearing the threshold switches"), |
| 129 | + ([100.0, 100.0 - REGION_SWITCH_THRESHOLD + 0.01], 0, "a gain inside the threshold is noise and must not switch"), |
| 130 | + ([100.0, 100.0], 0, "a tie keeps the incumbent"), |
| 131 | + ([100.0, 101.0], 0, "a worse branch never wins"), |
| 132 | + ([100.0, 95.0, 90.0], 2, "the best branch clearing the threshold wins"), |
| 133 | + ([100.0, 90.0, 99.9], 1, "a marginal third branch does not displace a clear winner"), |
| 134 | + ] |
| 135 | + for metrics, expected, reason in cases: |
| 136 | + chosen = my_predbat.select_region_branch(metrics) |
| 137 | + if chosen != expected: |
| 138 | + failed |= _fail("select_region_branch", "{}: {} chose branch {} expected {}".format(reason, metrics, chosen, expected)) |
| 139 | + return failed |
| 140 | + |
| 141 | + |
| 142 | +def test_region_defaults(my_predbat): |
| 143 | + """The tiling constants match the geometry the optimiser was tuned against""" |
| 144 | + failed = False |
| 145 | + if REGION_SIZE_START != 16 * 60: |
| 146 | + failed |= _fail("region_defaults", "REGION_SIZE_START changed to {}".format(REGION_SIZE_START)) |
| 147 | + if REGION_SIZE_MIN != 120: |
| 148 | + failed |= _fail("region_defaults", "REGION_SIZE_MIN changed to {}".format(REGION_SIZE_MIN)) |
| 149 | + if REGION_SWITCH_THRESHOLD <= 0: |
| 150 | + failed |= _fail("region_defaults", "REGION_SWITCH_THRESHOLD must be positive, got {}".format(REGION_SWITCH_THRESHOLD)) |
| 151 | + return failed |
| 152 | + |
| 153 | + |
| 154 | +def run_region_portfolio_tests(my_predbat): |
| 155 | + """Run all region tiling and portfolio selection tests""" |
| 156 | + failed = False |
| 157 | + failed |= test_region_passes_uniform(my_predbat) |
| 158 | + failed |= test_region_passes_stagger(my_predbat) |
| 159 | + failed |= test_region_passes_bounds(my_predbat) |
| 160 | + failed |= test_region_passes_cover_record(my_predbat) |
| 161 | + failed |= test_region_passes_min_size(my_predbat) |
| 162 | + failed |= test_select_region_branch(my_predbat) |
| 163 | + failed |= test_region_defaults(my_predbat) |
| 164 | + if not failed: |
| 165 | + print("**** Region portfolio tests passed ****") |
| 166 | + return failed |
0 commit comments