feat(binary-sensor): Added ability to apply offsets to target rate se…

…nsors

This means you can now do something like "turn on an hour before the cheapest rate"

custom_components/octopus_energy/binary_sensor.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 import math
 import logging
+from custom_components.octopus_energy.utils import apply_offset
 
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.util.dt import (utcnow, now, as_utc, parse_datetime)
@@ -11,6 +12,7 @@
     BinarySensorEntity,
 )
 from .const import (
+  CONFIG_TARGET_OFFSET,
   DOMAIN,
 
   CONFIG_TARGET_NAME,
@@ -90,6 +92,11 @@ def extra_state_attributes(self):
   def is_on(self):
     """The state of the sensor."""
 
+    if CONFIG_TARGET_OFFSET in self._config:
+      offset = self._config[CONFIG_TARGET_OFFSET]
+    else:
+      offset = None
+
     # Find the current rate. Rates change a maximum of once every 30 minutes.
     current_date = utcnow()
     if (current_date.minute % 30) == 0 or len(self._target_rates) == 0:
@@ -126,22 +133,28 @@ def is_on(self):
             start_time,
             end_time,
             float(self._config[CONFIG_TARGET_HOURS]),
-            all_rates
+            all_rates,
+            offset
           )
         elif (self._config[CONFIG_TARGET_TYPE] == "Intermittent"):
           self._target_rates = calculate_intermittent_times(
             now(),
             start_time,
             end_time,
             float(self._config[CONFIG_TARGET_HOURS]),
-            all_rates
+            all_rates,
+            offset
           )
         else:
           _LOGGER.error(f"Unexpected target type: {self._config[CONFIG_TARGET_TYPE]}")
 
         self._attributes["target_times"] = self._target_rates
 
-    active_result = is_target_rate_active(current_date, self._target_rates)
-    self._attributes["next_time"] = active_result["next_time"]
+    active_result = is_target_rate_active(current_date, self._target_rates, offset)
+
+    if offset != None and active_result["next_time"] != None:
+      self._attributes["next_time"] = apply_offset(active_result["next_time"], offset)
+    else:
+      self._attributes["next_time"] = active_result["next_time"]
 
     return active_result["is_active"]

custom_components/octopus_energy/config_flow.py

@@ -20,6 +20,7 @@
   CONFIG_TARGET_END_TIME,
   CONFIG_TARGET_TYPE,
   CONFIG_TARGET_MPAN,
+  CONFIG_TARGET_OFFSET,
 
   CONFIG_SMETS1,
 
@@ -29,7 +30,8 @@
 
   REGEX_TIME,
   REGEX_ENTITY_NAME,
-  REGEX_HOURS
+  REGEX_HOURS,
+  REGEX_OFFSET_PARTS,
 )
 
 from .api_client import OctopusEnergyApiClient
@@ -55,17 +57,20 @@ def validate_target_rate_sensor(data):
       errors[CONFIG_TARGET_HOURS] = "invalid_target_hours"
 
   if CONFIG_TARGET_START_TIME in data:
-    data[CONFIG_TARGET_START_TIME] = data[CONFIG_TARGET_START_TIME]
     matches = re.search(REGEX_TIME, data[CONFIG_TARGET_START_TIME])
     if matches == None:
       errors[CONFIG_TARGET_START_TIME] = "invalid_target_time"
 
   if CONFIG_TARGET_END_TIME in data:
-    data[CONFIG_TARGET_END_TIME] = data[CONFIG_TARGET_END_TIME]
-    matches = re.search(REGEX_TIME, data[CONFIG_TARGET_START_TIME])
+    matches = re.search(REGEX_TIME, data[CONFIG_TARGET_END_TIME])
     if matches == None:
       errors[CONFIG_TARGET_END_TIME] = "invalid_target_time"
 
+  if CONFIG_TARGET_OFFSET in data:
+    matches = re.search(REGEX_OFFSET_PARTS, data[CONFIG_TARGET_OFFSET])
+    if matches == None:
+      errors[CONFIG_TARGET_OFFSET] = "invalid_offset"
+
   return errors
 
 class OctopusEnergyConfigFlow(ConfigFlow, domain=DOMAIN): 
@@ -115,6 +120,7 @@ async def async_setup_target_rate_schema(self):
       ),
       vol.Optional(CONFIG_TARGET_START_TIME): str,
       vol.Optional(CONFIG_TARGET_END_TIME): str,
+      vol.Optional(CONFIG_TARGET_OFFSET): str,
     })
 
   async def async_step_target_rate(self, user_input):
@@ -188,6 +194,10 @@ async def __async_setup_target_rate_schema(self, config, errors):
 
     if (CONFIG_TARGET_MPAN not in config):
       config[CONFIG_TARGET_MPAN] = meters[0]
+
+    offset = None
+    if (CONFIG_TARGET_OFFSET in config):
+      offset = config[CONFIG_TARGET_OFFSET]
 
     return self.async_show_form(
       step_id="target_rate",
@@ -198,6 +208,7 @@ async def __async_setup_target_rate_schema(self, config, errors):
         ),
         vol.Optional(CONFIG_TARGET_START_TIME, default=config[CONFIG_TARGET_START_TIME]): str,
         vol.Optional(CONFIG_TARGET_END_TIME, default=config[CONFIG_TARGET_END_TIME]): str,
+        vol.Optional(CONFIG_TARGET_OFFSET, default=offset): str,
       }),
       errors=errors
     )

custom_components/octopus_energy/const.py

@@ -12,6 +12,7 @@
 CONFIG_TARGET_START_TIME = "Start time"
 CONFIG_TARGET_END_TIME = "End time"
 CONFIG_TARGET_MPAN = "MPAN"
+CONFIG_TARGET_OFFSET = "offset"
 
 DATA_CONFIG = "CONFIG"
 DATA_ELECTRICITY_RATES_COORDINATOR = "ELECTRICITY_RATES_COORDINATOR"
@@ -20,10 +21,11 @@
 DATA_GAS_TARIFF_CODE = "GAS_TARIFF_CODE"
 DATA_ACCOUNT_ID = "ACCOUNT_ID"
 
-REGEX_HOURS = "^[0-9]+(\.[0-9]+)*$"
+REGEX_HOURS = "^[0-9]+(\\.[0-9]+)*$"
 REGEX_TIME = "^([0-1]?[0-9]|2[0-3]):[0-5][0-9]$"
 REGEX_ENTITY_NAME = "^[a-z0-9_]+$"
 REGEX_TARIFF_PARTS = "^([A-Z])-([0-9A-Z]+)-([A-Z0-9-]+)-([A-Z])$"
+REGEX_OFFSET_PARTS = "^(-)?([0-1]?[0-9]|2[0-3]):([0-5][0-9]):([0-5][0-9])$"
 
 DATA_SCHEMA_ACCOUNT = vol.Schema({
   vol.Required(CONFIG_MAIN_API_KEY): str,

custom_components/octopus_energy/target_sensor_utils.py

@@ -1,8 +1,9 @@
 from datetime import timedelta
 import math
 from homeassistant.util.dt import (as_utc, parse_datetime)
+from .utils import (apply_offset)
 
-def __get_applicable_rates(current_date, target_start_time, target_end_time, rates):
+def __get_applicable_rates(current_date, target_start_time, target_end_time, rates, target_start_offset):
   if target_end_time != None:
     # Get the target end for today. If this is in the past, then look at tomorrow
     target_end = parse_datetime(current_date.strftime(f"%Y-%m-%dT{target_end_time}:00%z"))
@@ -28,6 +29,10 @@ def __get_applicable_rates(current_date, target_start_time, target_end_time, rat
   if (target_start < current_date):
     target_start = current_date
 
+  # Apply our offset so we make sure our target turns on within the specified timeframe
+  if (target_start_offset != None):
+    target_start = apply_offset(target_start, target_start_offset, True)
+
   # Convert our target start/end timestamps to UTC as this is what our rates are in
   target_start = as_utc(target_start)
   if target_end is not None:
@@ -48,8 +53,8 @@ def __get_rate(rate):
 def __get_valid_to(rate):
   return rate["valid_to"]
 
-def calculate_continuous_times(current_date, target_start_time, target_end_time, target_hours, rates):
-  applicable_rates = __get_applicable_rates(current_date, target_start_time, target_end_time, rates)
+def calculate_continuous_times(current_date, target_start_time, target_end_time, target_hours, rates, target_start_offset = None):
+  applicable_rates = __get_applicable_rates(current_date, target_start_time, target_end_time, rates, target_start_offset)
   applicable_rates_count = len(applicable_rates)
   total_required_rates = math.ceil(target_hours * 2)
 
@@ -81,8 +86,8 @@ def calculate_continuous_times(current_date, target_start_time, target_end_time,
 
   return []
 
-def calculate_intermittent_times(current_date, target_start_time, target_end_time, target_hours, rates):
-  applicable_rates = __get_applicable_rates(current_date, target_start_time, target_end_time, rates)
+def calculate_intermittent_times(current_date, target_start_time, target_end_time, target_hours, rates, target_start_offset = None):
+  applicable_rates = __get_applicable_rates(current_date, target_start_time, target_end_time, rates, target_start_offset)
   total_required_rates = math.ceil(target_hours * 2)
 
   applicable_rates.sort(key=__get_rate)
@@ -95,7 +100,7 @@ def calculate_intermittent_times(current_date, target_start_time, target_end_tim
   applicable_rates.sort(key=__get_valid_to)
   return applicable_rates
 
-def is_target_rate_active(current_date, applicable_rates):
+def is_target_rate_active(current_date, applicable_rates, offset = None):
   is_active = False
   next_time = None
   total_applicable_rates = len(applicable_rates)
@@ -105,7 +110,14 @@ def is_target_rate_active(current_date, applicable_rates):
       next_time = applicable_rates[0]["valid_from"]
 
     for index, rate in enumerate(applicable_rates):
-      if current_date >= rate["valid_from"] and current_date <= rate["valid_to"]:
+      if (offset != None):
+        valid_from = apply_offset(rate["valid_from"], offset)
+        valid_to = apply_offset(rate["valid_to"], offset)
+      else:
+        valid_from = rate["valid_from"]
+        valid_to = rate["valid_to"]
+
+      if current_date >= valid_from and current_date <= valid_to:
         is_active = True
 
         next_index = index + 1

custom_components/octopus_energy/translations/en.json

@@ -20,15 +20,17 @@
           "Type": "The type of target you're after",
           "MPAN": "The MPAN number of the meter to apply the target to",
           "Start time": "The minimum time to start the device",
-          "End time": "The maximum time to stop the device"
+          "End time": "The maximum time to stop the device",
+          "offset": "The offset to apply to the scheduled block to be considered active"
         }
       }
     },
     "error": {
       "account_not_found": "Account information was not found",
       "invalid_target_hours": "Target hours must be in half hour increments.",
       "invalid_target_name": "Name must only include lower case alpha characters and underscore (e.g. my_target)",
-      "invalid_target_time": "Must be in the format HH:MM"
+      "invalid_target_time": "Must be in the format HH:MM",
+      "invalid_offset": "Offset must be in the form of HH:MM:SS with an optional negative symbol"
     },
     "abort": {
       "not_supported": "Configuration for target rates is not supported at the moment."
@@ -51,11 +53,15 @@
           "Hours": "The hours you require.",
           "MPAN": "The MPAN number of the meter to apply the target to",
           "Start time": "The minimum time to start the device",
-          "End time": "The maximum time to stop the device"
+          "End time": "The maximum time to stop the device",
+          "offset": "The offset to apply to the scheduled block to be considered active"
         }
       }
     },
     "error": {
+      "invalid_target_hours": "Target hours must be in half hour increments.",
+      "invalid_target_time": "Must be in the format HH:MM",
+      "invalid_offset": "Offset must be in the form of HH:MM:SS with an optional negative symbol"
     },
     "abort": {
       "not_supported": "Configuration for target rates is not supported at the moment."

custom_components/octopus_energy/utils.py

@@ -1,9 +1,11 @@
+from datetime import timedelta
 from homeassistant.util.dt import (as_utc, parse_datetime)
 
 import re
 
 from .const import (
-  REGEX_TARIFF_PARTS
+  REGEX_TARIFF_PARTS,
+  REGEX_OFFSET_PARTS,
 )
 
 def get_tariff_parts(tariff_code):
@@ -46,4 +48,20 @@ def get_active_tariff_code(utcnow, agreements):
   if latest_agreement != None:
     return latest_agreement["tariff_code"]
 
-  return None
+  return None
+
+def apply_offset(date_time, offset, inverse = False):
+  matches = re.search(REGEX_OFFSET_PARTS, offset)
+  if matches == None:
+    raise Exception(f'Unable to extract offset: {offset}')
+
+  symbol = matches[1]
+  hours = float(matches[2])
+  minutes = float(matches[3])
+  seconds = float(matches[4])
+
+  if ((symbol == "-" and inverse == False) or (symbol != "-" and inverse == True)):
+    return date_time - timedelta(hours=hours, minutes=minutes, seconds=seconds)
+
+  return date_time + timedelta(hours=hours, minutes=minutes, seconds=seconds)
+

tests/unit/test_calculate_continuous_times.py

@@ -85,4 +85,47 @@ async def test_when_current_time_has_not_enough_time_left_then_no_continuous_tim
 
   # Assert
   assert result != None
-  assert len(result) == 0
+  assert len(result) == 0
+
+@pytest.mark.asyncio
+async def test_when_offset_set_then_next_continuous_times_returned_have_offset_applied():
+  # Arrange
+  period_from = datetime.strptime("2022-02-09T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-02-11T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  expected_rates = [0.1, 0.2, 0.3, 0.2, 0.2, 0.1]
+  current_date = datetime.strptime("2022-02-09T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  target_start_time = "11:00"
+  target_end_time = "18:00"
+  offset = "-01:00:00"
+
+  expected_first_valid_from = datetime.strptime("2022-02-09T14:30:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  # Restrict our time block
+  target_hours = 1
+
+  rates = create_rate_data(
+    period_from,
+    period_to,
+    expected_rates
+  )
+
+  # Act
+  result = calculate_continuous_times(
+    current_date,
+    target_start_time,
+    target_end_time,
+    target_hours,
+    rates,
+    offset
+  )
+
+  # Assert
+  assert result != None
+  assert len(result) == 2
+  assert result[0]["valid_from"] == expected_first_valid_from
+  assert result[0]["valid_to"] == expected_first_valid_from + timedelta(minutes=30)
+  assert result[0]["value_inc_vat"] == 0.1
+
+  assert result[1]["valid_from"] == expected_first_valid_from + timedelta(minutes=30)
+  assert result[1]["valid_to"] == expected_first_valid_from + timedelta(hours=1)
+  assert result[1]["value_inc_vat"] == 0.1

tests/unit/test_calculate_intermittent_times.py

@@ -85,4 +85,45 @@ async def test_when_current_time_has_not_enough_time_left_then_no_intermittent_t
 
   # Assert
   assert result != None
-  assert len(result) == 0
+  assert len(result) == 0
+
+@pytest.mark.asyncio
+async def test_when_offset_set_then_next_continuous_times_returned_have_offset_applied():
+  # Arrange
+  period_from = datetime.strptime("2022-02-09T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-02-11T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  expected_rates = [0.1, 0.2, 0.3]
+  current_date = datetime.strptime("2022-02-09T00:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  target_start_time = "11:00"
+  target_end_time = "18:00"
+  offset = "-01:00:00"
+
+  # Restrict our time block
+  target_hours = 1
+
+  rates = create_rate_data(
+    period_from,
+    period_to,
+    expected_rates
+  )
+
+  # Act
+  result = calculate_intermittent_times(
+    current_date,
+    target_start_time,
+    target_end_time,
+    target_hours,
+    rates,
+    offset
+  )
+
+  # Assert
+  assert result != None
+  assert len(result) == 2
+  assert result[0]["valid_from"] == datetime.strptime("2022-02-09T12:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert result[0]["valid_to"] == datetime.strptime("2022-02-09T12:00:00Z", "%Y-%m-%dT%H:%M:%S%z") + timedelta(minutes=30)
+  assert result[0]["value_inc_vat"] == 0.1
+
+  assert result[1]["valid_from"] == datetime.strptime("2022-02-09T13:30:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert result[1]["valid_to"] == datetime.strptime("2022-02-09T13:30:00Z", "%Y-%m-%dT%H:%M:%S%z") + timedelta(minutes=30)
+  assert result[1]["value_inc_vat"] == 0.1