Add expensive-now price braking

Author: JohanAlvedal

custom_components/pumpsteer/sensor/sensor.py

@@ -128,7 +128,11 @@ def compute_block_window(
     block_start = update_time + timedelta(minutes=block.start_offset_minutes)
     block_end = update_time + timedelta(minutes=block.end_offset_minutes)
 
-    now_utc = dt_util.as_utc(update_time)
+    now_utc = (
+        dt_util.utcnow()
+        if hasattr(dt_util, "utcnow")
+        else dt_util.as_utc(dt_util.now())
+    )
     block_start_utc = dt_util.as_utc(block_start)
     block_end_utc = dt_util.as_utc(block_end)
     in_price_block = block_start_utc <= now_utc < block_end_utc
@@ -274,6 +278,8 @@ def _compute_controls(
         current_slot_index: int,
         price_interval_minutes: int,
         config: Dict[str, Any],
+        current_price: float,
+        price_category: str,
         update_time: datetime,
     ) -> Dict[str, Any]:
         """Compute price brake and comfort push controls."""
@@ -310,9 +316,31 @@ def _compute_controls(
         desired_brake_level = price_brake["brake_level"]
         brake_blocked_reason = "allowed"
 
-        if desired_brake_level <= 0.0 and not in_price_block:
+        category_label = price_category.split(" ")[0]
+        is_very_expensive = category_label in {"very_expensive", "extreme"}
+        is_expensive = category_label == "expensive"
+        expensive_now = is_very_expensive or (
+            is_expensive and current_price > price_brake["threshold"]
+        )
+        if desired_brake_level <= 0.0 and expensive_now:
+            min_price = min(combined_prices) if combined_prices else current_price
+            max_price = max(combined_prices) if combined_prices else current_price
+            price_range = max_price - min_price
+            if price_range > 0.0:
+                price_factor = (current_price - min_price) / price_range
+            else:
+                price_factor = 0.0
+            price_factor = max(0.0, min(price_factor, 1.0))
+            aggressiveness_factor = min(
+                1.0, max(0.0, sensor_data["aggressiveness"] / 5.0)
+            )
+            min_brake_level = 0.25 + 0.25 * (
+                0.5 * (aggressiveness_factor + price_factor)
+            )
+            desired_brake_level = max(desired_brake_level, min_brake_level)
+        if desired_brake_level <= 0.0 and not in_price_block and not expensive_now:
             brake_blocked_reason = "no_price_block"
-        elif too_cold_to_brake:
+        if too_cold_to_brake:
             desired_brake_level = 0.0
             brake_blocked_reason = "too_cold"
 
@@ -767,6 +795,8 @@ async def async_update(self) -> None:
             current_slot_index,
             price_interval_minutes,
             config,
+            current_price,
+            price_category,
             update_time,
         )
 

tests/test_block_window.py

@@ -1,10 +1,12 @@
 from datetime import datetime, timedelta
 
+from homeassistant.util import dt as dt_util
+
 from custom_components.pumpsteer.price_brake import PriceBlock
 from custom_components.pumpsteer.sensor.sensor import compute_block_window
 
 
-def test_compute_block_window_active_now():
+def test_compute_block_window_active_now(monkeypatch):
     update_time = datetime(2024, 1, 1, 12, 0, 0)
     block = PriceBlock(
         start_index=0,
@@ -13,6 +15,10 @@ def test_compute_block_window_active_now():
         area=1.5,
         peak=2.5,
     )
+    if hasattr(dt_util, "utcnow"):
+        monkeypatch.setattr(dt_util, "utcnow", lambda: dt_util.as_utc(update_time))
+    else:
+        monkeypatch.setattr(dt_util, "now", lambda: update_time)
 
     block_start, block_end, in_price_block, block_state = compute_block_window(
         update_time, block

tests/test_temperature_vs_price.py

@@ -1,4 +1,5 @@
 import sys
+from datetime import datetime
 from pathlib import Path
 
 from custom_components.pumpsteer.sensor import sensor
@@ -84,6 +85,48 @@ def test_extreme_price_brake_when_neutral():
     assert fake_temp == data["outdoor_temp"]
 
 
+def test_expensive_now_braking_outside_block(monkeypatch):
+    s = create_sensor()
+    data = base_sensor_data()
+    combined_prices = [1.0, 3.28, 2.0]
+    update_time = datetime(2024, 1, 1, 12, 0, 0)
+
+    def fake_compute_price_brake(**_kwargs):
+        return {
+            "brake_level": 0.0,
+            "baseline": 0.0,
+            "threshold": 2.54,
+            "area": 0.0,
+            "amplitude": 0.0,
+            "block": None,
+        }
+
+    monkeypatch.setattr(sensor, "compute_price_brake", fake_compute_price_brake)
+
+    pi_data = s._compute_controls(
+        data,
+        combined_prices,
+        0,
+        60,
+        {},
+        3.28,
+        "expensive",
+        update_time,
+    )
+    assert pi_data["price_brake_level"] > 0.0
+    assert pi_data["brake_blocked_reason"] != "no_price_block"
+
+    fake_temp, mode = s._calculate_output_temperature(data, "expensive", 0)
+    assert mode == "neutral"
+    if (
+        mode == "neutral"
+        and pi_data["price_brake_level"] > 0.0
+        and pi_data["brake_blocked_reason"] in {"allowed", "rate_limited"}
+    ):
+        mode = "braking_by_price"
+    assert mode == "braking_by_price"
+
+
 def test_very_cheap_price_overshoots_target():
     s = create_sensor()
     data = base_sensor_data()