Merge pull request #190 from JohanAlvedal/codex/implement-advanced-braking-logic

Apply braking relative to active/next price block and expose metadata

Author: JohanAlvedal

custom_components/pumpsteer/price_brake.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 import math
 from statistics import median
-from typing import List, Optional
+from typing import List, Optional, Tuple, Union
 
 
 @dataclass
@@ -190,6 +190,35 @@ def filter_upcoming_blocks(
     return [block for block in blocks if block.end_offset_minutes > now_offset_minutes]
 
 
+def resolve_active_and_next_blocks(
+    blocks: Optional[Union[PriceBlock, List[PriceBlock]]],
+    now_offset_minutes: float,
+) -> Tuple[Optional[PriceBlock], Optional[PriceBlock], List[PriceBlock]]:
+    """Resolve active and next blocks from a block or list of blocks."""
+    if blocks is None:
+        return None, None, []
+    block_list = blocks if isinstance(blocks, list) else [blocks]
+    active_block = next(
+        (
+            block
+            for block in block_list
+            if block.start_offset_minutes <= now_offset_minutes < block.end_offset_minutes
+        ),
+        None,
+    )
+    upcoming_blocks = [
+        block
+        for block in block_list
+        if block.start_offset_minutes > now_offset_minutes
+    ]
+    next_block = (
+        min(upcoming_blocks, key=lambda block: block.start_offset_minutes)
+        if upcoming_blocks
+        else None
+    )
+    return active_block, next_block, block_list
+
+
 def compute_brake_level(
     block: Optional[PriceBlock],
     amplitude: float,
@@ -252,23 +281,10 @@ def compute_price_brake(
     )
     upcoming_blocks = filter_upcoming_blocks(blocks, now_offset_minutes)
     selected_blocks = select_price_blocks(upcoming_blocks, max_blocks=2)
-    active_block = None
-    for block in selected_blocks:
-        if block.start_offset_minutes <= now_offset_minutes < block.end_offset_minutes:
-            active_block = block
-            break
-    upcoming_block = None
-    if active_block is None:
-        upcoming_candidates = [
-            block
-            for block in selected_blocks
-            if block.start_offset_minutes > now_offset_minutes
-        ]
-        if upcoming_candidates:
-            upcoming_block = min(
-                upcoming_candidates, key=lambda block: block.start_offset_minutes
-            )
-    primary_block = active_block or upcoming_block
+    active_block, next_block, _ = resolve_active_and_next_blocks(
+        upcoming_blocks, now_offset_minutes
+    )
+    primary_block = active_block or next_block
     area = primary_block.area if primary_block else 0.0
     safe_area_scale = area_scale if area_scale > 0 else 1.0
     amplitude = clamp(area / safe_area_scale, 0.0, 1.0) if primary_block else 0.0
@@ -279,6 +295,11 @@ def compute_price_brake(
         post_release_minutes,
         now_offset_minutes,
     )
+    blocks_for_status = list(selected_blocks)
+    for candidate in (active_block, next_block):
+        if candidate and candidate not in blocks_for_status:
+            blocks_for_status.append(candidate)
+    blocks_for_status.sort(key=lambda block: block.start_index)
 
     return {
         "baseline": baseline,
@@ -287,5 +308,7 @@ def compute_price_brake(
         "amplitude": amplitude,
         "brake_level": brake_level,
         "block": primary_block,
-        "blocks": selected_blocks,
+        "blocks": blocks_for_status,
+        "active_block": active_block,
+        "next_block": next_block,
     }

custom_components/pumpsteer/sensor/sensor.py

@@ -342,6 +342,7 @@ def _compute_controls(
             area_scale=PRICE_BLOCK_AREA_SCALE,
             now_offset_minutes=0.0,
         )
+        next_block = price_brake.get("next_block")
 
         (
             block_start,
@@ -370,7 +371,16 @@ def _compute_controls(
                 desired_brake_level, base_min + 0.20 * scale
             )
             brake_blocked_reason = "expensive_now"
-        if desired_brake_level <= 0.0 and not in_price_block and not expensive_now:
+        pre_window_active = False
+        if next_block is not None:
+            pre_start = next_block.start_offset_minutes - PRICE_BRAKE_PRE_MINUTES
+            pre_window_active = pre_start <= 0 < next_block.start_offset_minutes
+        if (
+            desired_brake_level <= 0.0
+            and not in_price_block
+            and not expensive_now
+            and not pre_window_active
+        ):
             brake_blocked_reason = "no_price_block"
         if too_cold_to_brake:
             desired_brake_level = 0.0