feat(sensor): Added min/max/average rates to electricity rate sensors

custom_components/octopus_energy/electricity/__init__.py

@@ -1,3 +1,5 @@
+from datetime import (datetime, timedelta)
+
 from ..api_client import OctopusEnergyApiClient
 
 def __get_interval_end(item):
@@ -44,11 +46,11 @@ async def async_calculate_electricity_cost(client: OctopusEnergyApiClient, consu
     sorted_consumption_data = __sort_consumption(consumption_data)
 
     # Only calculate our consumption if our data has changed
-    if (last_calculated_timestamp == None or last_calculated_timestamp < sorted_consumption_data[-1]["interval_end"]):
+    if (last_calculated_timestamp is None or last_calculated_timestamp < sorted_consumption_data[-1]["interval_end"]):
       rates = await client.async_get_electricity_rates(tariff_code, is_smart_meter, period_from, period_to)
       standard_charge_result = await client.async_get_electricity_standing_charge(tariff_code, period_from, period_to)
 
-      if (rates != None and len(rates) > 0 and standard_charge_result != None):
+      if (rates is not None and len(rates) > 0 and standard_charge_result is not None):
         standard_charge = standard_charge_result["value_inc_vat"]
 
         charges = []
@@ -102,4 +104,48 @@ async def async_calculate_electricity_cost(client: OctopusEnergyApiClient, consu
             result["total_off_peak"] = round(rate_charges[key_two] / 100, 2)
             result["total_peak"] = round(rate_charges[key_one] / 100, 2)
 
-        return result
+        return result
+
+def get_rate_information(rates, target: datetime):
+  min_target = target.replace(hour=0, minute=0, second=0, microsecond=0)
+  max_target = min_target + timedelta(days=1)
+
+  min_rate_value = None
+  max_rate_value = None
+  total_rate_value = 0
+  total_rates = 0
+  current_rate = None
+
+  if rates is not None:
+    for period in rates:
+      if target >= period["valid_from"] and target <= period["valid_to"]:
+        current_rate = period
+
+      if period["valid_from"] >= min_target and period["valid_to"] <= max_target:
+        if min_rate_value is None or period["value_inc_vat"] < min_rate_value:
+          min_rate_value = period["value_inc_vat"]
+
+        if max_rate_value is None or period["value_inc_vat"] > max_rate_value:
+          max_rate_value = period["value_inc_vat"]
+
+        total_rate_value = total_rate_value + period["value_inc_vat"]
+        total_rates = total_rates + 1
+
+  print(total_rate_value)
+  print(total_rates)
+
+  if current_rate is not None:
+    return {
+      "rates": list(map(lambda x: {
+        "from": x["valid_from"],
+        "to":   x["valid_to"],
+        "rate": x["value_inc_vat"],
+        "is_capped": x["is_capped"]
+      }, rates)),
+      "current_rate": current_rate,
+      "min_rate_today": min_rate_value,
+      "max_rate_today": max_rate_value,
+      "average_rate_today": total_rate_value / total_rates
+    }
+
+  return None

custom_components/octopus_energy/electricity/current_rate.py

@@ -13,6 +13,8 @@
 
 from .base import (OctopusEnergyElectricitySensor)
 
+from . import (get_rate_information)
+
 _LOGGER = logging.getLogger(__name__)
 
 class OctopusEnergyElectricityCurrentRate(CoordinatorEntity, OctopusEnergyElectricitySensor):
@@ -66,36 +68,34 @@ def state(self):
     if (self._last_updated is None or self._last_updated < (now - timedelta(minutes=30)) or (now.minute % 30) == 0):
       _LOGGER.debug(f"Updating OctopusEnergyElectricityCurrentRate for '{self._mpan}/{self._serial_number}'")
 
-      current_rate = None
-      if self.coordinator.data != None:
-        rate = self.coordinator.data[self._mpan] if self._mpan in self.coordinator.data else None
-        if rate != None:
-          for period in rate:
-            if now >= period["valid_from"] and now <= period["valid_to"]:
-              current_rate = period
-              break
-
-      if current_rate != None:
-        ratesAttributes = list(map(lambda x: {
-          "from": x["valid_from"],
-          "to":   x["valid_to"],
-          "rate": x["value_inc_vat"],
-          "is_capped": x["is_capped"]
-        }, rate))
+      rate_information = get_rate_information(self.coordinator.data[self._mpan] if self._mpan in self.coordinator.data else None, now)
+
+      if rate_information is not None:
         self._attributes = {
-          "rate": current_rate,
+          "mpan": self._mpan,
+          "serial_number": self._serial_number,
           "is_export": self._is_export,
           "is_smart_meter": self._is_smart_meter,
-          "rates": ratesAttributes
+          "rates": rate_information["rates"],
+          "rate": rate_information["current_rate"],
+          "current_day_min_rate": rate_information["min_rate_today"],
+          "current_day_max_rate": rate_information["max_rate_today"],
+          "current_day_average_rate": rate_information["average_rate_today"]
         }
 
-        if self._electricity_price_cap is not None:
-          self._attributes["price_cap"] = self._electricity_price_cap
-
-        self._state = current_rate["value_inc_vat"] / 100
+        self._state = self._attributes["current_rate"]["value_inc_vat"] / 100
       else:
+        self._attributes = {
+          "mpan": self._mpan,
+          "serial_number": self._serial_number,
+          "is_export": self._is_export,
+          "is_smart_meter": self._is_smart_meter
+        }
+
         self._state = None
-        self._attributes = {}
+
+      if self._electricity_price_cap is not None:
+        self._attributes["price_cap"] = self._electricity_price_cap
 
       self._last_updated = now
 

custom_components/octopus_energy/electricity/next_rate.py

@@ -12,6 +12,7 @@
 )
 
 from .base import (OctopusEnergyElectricitySensor)
+from . import (get_rate_information)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -67,26 +68,31 @@ def state(self):
 
       target = now + timedelta(minutes=30)
 
-      next_rate = None
-      if self.coordinator.data != None:
-        rate = self.coordinator.data[self._mpan] if self._mpan in self.coordinator.data else None
-        if rate != None:
-          for period in rate:
-            if target >= period["valid_from"] and target <= period["valid_to"]:
-              next_rate = period
-              break
+      rate_information = get_rate_information(self.coordinator.data[self._mpan] if self._mpan in self.coordinator.data else None, target)
 
-      if next_rate != None:
+      if rate_information is not None:
         self._attributes = {
-          "rate": next_rate,
+          "mpan": self._mpan,
+          "serial_number": self._serial_number,
           "is_export": self._is_export,
-          "is_smart_meter": self._is_smart_meter
+          "is_smart_meter": self._is_smart_meter,
+          "rates": rate_information["rates"],
+          "rate": rate_information["current_rate"],
+          "current_day_min_rate": rate_information["min_rate_today"],
+          "current_day_max_rate": rate_information["max_rate_today"],
+          "current_day_average_rate": rate_information["average_rate_today"]
         }
 
-        self._state = next_rate["value_inc_vat"] / 100
+        self._state = self._attributes["current_rate"]["value_inc_vat"] / 100
       else:
+        self._attributes = {
+          "mpan": self._mpan,
+          "serial_number": self._serial_number,
+          "is_export": self._is_export,
+          "is_smart_meter": self._is_smart_meter,
+        }
+
         self._state = None
-        self._attributes = {}
 
       self._last_updated = now
 

custom_components/octopus_energy/electricity/previous_rate.py

@@ -12,6 +12,7 @@
 )
 
 from .base import (OctopusEnergyElectricitySensor)
+from . import (get_rate_information)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -67,26 +68,31 @@ def state(self):
 
       target = now - timedelta(minutes=30)
 
-      previous_rate = None
-      if self.coordinator.data != None:
-        rate = self.coordinator.data[self._mpan] if self._mpan in self.coordinator.data else None
-        if rate != None:
-          for period in rate:
-            if target >= period["valid_from"] and target <= period["valid_to"]:
-              previous_rate = period
-              break
+      rate_information = get_rate_information(self.coordinator.data[self._mpan] if self._mpan in self.coordinator.data else None, target)
 
-      if previous_rate != None:
+      if rate_information is not None:
         self._attributes = {
-          "rate": previous_rate,
+          "mpan": self._mpan,
+          "serial_number": self._serial_number,
           "is_export": self._is_export,
-          "is_smart_meter": self._is_smart_meter
+          "is_smart_meter": self._is_smart_meter,
+          "rates": rate_information["rates"],
+          "rate": rate_information["current_rate"],
+          "current_day_min_rate": rate_information["min_rate_today"],
+          "current_day_max_rate": rate_information["max_rate_today"],
+          "current_day_average_rate": rate_information["average_rate_today"]
         }
 
-        self._state = previous_rate["value_inc_vat"] / 100
+        self._state = self._attributes["current_rate"]["value_inc_vat"] / 100
       else:
+        self._attributes = {
+          "mpan": self._mpan,
+          "serial_number": self._serial_number,
+          "is_export": self._is_export,
+          "is_smart_meter": self._is_smart_meter,
+        }
+
         self._state = None
-        self._attributes = {}
 
       self._last_updated = now
 

tests/unit/__init__.py

@@ -40,7 +40,8 @@ def create_rate_data(period_from, period_to, expected_rates: list):
     rates.append({
       "valid_from": current_valid_from,
       "valid_to": current_valid_to,
-      "value_inc_vat": expected_rates[rate_index]
+      "value_inc_vat": expected_rates[rate_index],
+      "is_capped": False
     })
 
     current_valid_from = current_valid_to

tests/unit/electricity/test_get_rate_information.py

@@ -0,0 +1,73 @@
+from datetime import (datetime, timedelta)
+import pytest
+
+from unit import (create_rate_data)
+from custom_components.octopus_energy.electricity import get_rate_information
+
+@pytest.mark.asyncio
+async def test_when_target_has_no_rates_then_no_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-28T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-01T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  target = datetime.strptime("2022-03-01T10:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  expected_min_price = 10
+  expected_max_price = 30
+
+  rate_data = create_rate_data(period_from, period_to, [expected_min_price, 20, expected_max_price])
+
+  # Act
+  rate_information = get_rate_information(rate_data, target)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_target_has_rates_then_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-27T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-02T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  target = datetime.strptime("2022-02-28T10:12:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  expected_min_price = 10
+  expected_max_price = 30
+
+  rate_data = create_rate_data(period_from, period_to, [expected_min_price, 20, expected_max_price])
+
+  # Act
+  rate_information = get_rate_information(rate_data, target)
+
+  # Assert
+  assert rate_information is not None
+
+  assert "rates" in rate_information
+  expected_period_from = period_from
+
+  total_rate_value = 0
+  min_target = datetime.strptime("2022-02-28T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  max_target = min_target + timedelta(days=1)
+
+  for index in range(len(rate_data)):
+    assert rate_information["rates"][index]["from"] == expected_period_from
+    assert rate_information["rates"][index]["to"] == expected_period_from + timedelta(minutes=30)
+
+    assert rate_information["rates"][index]["rate"] == (expected_min_price if index % 3 == 0 else expected_max_price if index % 3 == 2 else 20)
+    assert rate_information["rates"][index]["is_capped"] == False
+    expected_period_from = expected_period_from + timedelta(minutes=30)
+
+    if rate_information["rates"][index]["from"] >= min_target and rate_information["rates"][index]["to"] <= max_target:
+      total_rate_value = total_rate_value + rate_information["rates"][index]["rate"]
+
+  assert "current_rate" in rate_information
+  assert rate_information["current_rate"]["valid_from"] == datetime.strptime("2022-02-28T10:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["current_rate"]["valid_to"] == rate_information["current_rate"]["valid_from"] + timedelta(minutes=30)
+
+  assert rate_information["current_rate"]["value_inc_vat"] == expected_max_price
+  assert rate_information["current_rate"]["is_capped"] == False
+
+  assert "min_rate_today" in rate_information
+  assert rate_information["min_rate_today"] == expected_min_price
+
+  assert "max_rate_today" in rate_information
+  assert rate_information["max_rate_today"] == expected_max_price
+
+  assert "average_rate_today" in rate_information
+  assert rate_information["average_rate_today"] == total_rate_value / 48