feat(sensors): Added functions for retrieving next/previous rates

custom_components/octopus_energy/utils/rate_information.py

@@ -0,0 +1,114 @@
+from datetime import (datetime, timedelta)
+
+def get_current_rate_information(rates, now: datetime):
+  min_target = now.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 now >= period["valid_from"] and now <= 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
+
+  if current_rate is not None:
+    return {
+      "rates": list(map(lambda x: {
+        "valid_from": x["valid_from"],
+        "valid_to":   x["valid_to"],
+        "value_inc_vat": x["value_inc_vat"],
+        "is_capped": x["is_capped"],
+        "is_intelligent_adjusted": x["is_intelligent_adjusted"] if "is_intelligent_adjusted" in x else False
+      }, 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
+
+def get_valid_from(rate):
+  return rate["valid_from"]
+
+def get_previous_rate_information(rates, now: datetime):
+  current_rate = None
+  applicable_rates = []
+
+  if rates is not None:
+    for period in reversed(rates):
+      if now >= period["valid_from"] and now <= period["valid_to"]:
+        current_rate = period
+
+      if current_rate is not None and current_rate["value_inc_vat"] != period["value_inc_vat"]:
+        if len(applicable_rates) == 0 or period["value_inc_vat"] == applicable_rates[0]["value_inc_vat"]:
+          applicable_rates.append(period)
+        else:
+          break
+
+  applicable_rates.sort(key=get_valid_from)
+
+  if len(applicable_rates) > 0:
+    return {
+      "rates": list(map(lambda x: {
+        "valid_from": x["valid_from"],
+        "valid_to":   x["valid_to"],
+        "value_inc_vat": x["value_inc_vat"],
+        "is_capped": x["is_capped"],
+        "is_intelligent_adjusted": x["is_intelligent_adjusted"] if "is_intelligent_adjusted" in x else False
+      }, applicable_rates)),
+      "previous_rate": {
+        "valid_from": applicable_rates[0]["valid_from"],
+        "valid_to": applicable_rates[-1]["valid_to"],
+        "value_inc_vat": applicable_rates[0]["value_inc_vat"],
+      }
+    }
+
+  return None
+
+def get_next_rate_information(rates, now: datetime):
+  current_rate = None
+  applicable_rates = []
+
+  if rates is not None:
+    for period in rates:
+      if now >= period["valid_from"] and now <= period["valid_to"]:
+        current_rate = period
+
+      if current_rate is not None and current_rate["value_inc_vat"] != period["value_inc_vat"]:
+        if len(applicable_rates) == 0 or period["value_inc_vat"] == applicable_rates[0]["value_inc_vat"]:
+          applicable_rates.append(period)
+        else:
+          break
+
+  if len(applicable_rates) > 0:
+    return {
+      "rates": list(map(lambda x: {
+        "valid_from": x["valid_from"],
+        "valid_to":   x["valid_to"],
+        "value_inc_vat": x["value_inc_vat"],
+        "is_capped": x["is_capped"],
+        "is_intelligent_adjusted": x["is_intelligent_adjusted"] if "is_intelligent_adjusted" in x else False
+      }, applicable_rates)),
+      "next_rate": {
+        "valid_from": applicable_rates[0]["valid_from"],
+        "valid_to": applicable_rates[-1]["valid_to"],
+        "value_inc_vat": applicable_rates[0]["value_inc_vat"],
+      }
+    }
+
+  return None

tests/unit/utils/test_get_current_rate_information.py

@@ -0,0 +1,141 @@
+from datetime import (datetime, timedelta)
+import pytest
+
+from unit import (create_rate_data)
+from custom_components.octopus_energy.utils.rate_information import get_current_rate_information
+
+@pytest.mark.asyncio
+async def test_when_target_has_no_rates_and_gmt_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")
+  now = datetime.strptime("2022-03-01T00:00:01Z", "%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_current_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_target_has_no_rates_and_bst_then_no_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-28T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-01T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  now = datetime.strptime("2022-03-02T00:00:01+01:00", "%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_current_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_target_has_rates_and_gmt_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")
+  now = datetime.strptime("2022-02-28T00: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_current_rate_information(rate_data, now)
+
+  # 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]["valid_from"] == expected_period_from
+    assert rate_information["rates"][index]["valid_to"] == expected_period_from + timedelta(minutes=30)
+
+    assert rate_information["rates"][index]["value_inc_vat"] == (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]["valid_from"] >= min_target and rate_information["rates"][index]["valid_to"] <= max_target:
+      total_rate_value = total_rate_value + rate_information["rates"][index]["value_inc_vat"]
+
+  assert "current_rate" in rate_information
+  assert rate_information["current_rate"]["valid_from"] == datetime.strptime("2022-02-28T00: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_min_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
+
+@pytest.mark.asyncio
+async def test_when_target_has_rates_and_bst_then_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-27T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-02T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  now = datetime.strptime("2022-02-28T00:12:00+01:00", "%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_current_rate_information(rate_data, now)
+
+  # 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]["valid_from"] == expected_period_from
+    assert rate_information["rates"][index]["valid_to"] == expected_period_from + timedelta(minutes=30)
+
+    assert rate_information["rates"][index]["value_inc_vat"] == (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]["valid_from"] >= min_target and rate_information["rates"][index]["valid_to"] <= max_target:
+      total_rate_value = total_rate_value + rate_information["rates"][index]["value_inc_vat"]
+
+  assert "current_rate" in rate_information
+  assert rate_information["current_rate"]["valid_from"] == datetime.strptime("2022-02-28T00:00:00+01:00", "%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_min_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

tests/unit/utils/test_get_next_rate_information.py

@@ -0,0 +1,135 @@
+from datetime import (datetime, timedelta)
+import pytest
+
+from unit import (create_rate_data)
+from custom_components.octopus_energy.utils.rate_information import get_next_rate_information
+
+@pytest.mark.asyncio
+async def test_when_target_has_no_rates_and_gmt_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")
+  now = datetime.strptime("2022-03-01T00:00:01Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  rate_data = create_rate_data(period_from, period_to, [10, 10, 20, 20, 30, 30])
+
+  # Act
+  rate_information = get_next_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_target_has_no_rates_and_bst_then_no_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-28T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-01T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  now = datetime.strptime("2022-03-02T00:00:01+01:00", "%Y-%m-%dT%H:%M:%S%z")
+
+  rate_data = create_rate_data(period_from, period_to, [10, 10, 20, 20, 30, 30])
+
+  # Act
+  rate_information = get_next_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_target_is_at_start_of_rates_and_bst_then_no_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-28T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-01T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  now = datetime.strptime("2022-02-28T00:00:01+01:00", "%Y-%m-%dT%H:%M:%S%z")
+
+  rate_data = create_rate_data(period_from, period_to, [10, 10, 20, 20, 30, 30])
+
+  # Act
+  rate_information = get_next_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_next_rate_is_identical_to_current_rate_then_no_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-28T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-01T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  now = datetime.strptime("2022-03-01T00:12:01+01:00", "%Y-%m-%dT%H:%M:%S%z")
+
+  rate_data = create_rate_data(period_from, period_to, [10])
+
+  # Act
+  rate_information = get_next_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is None
+
+@pytest.mark.asyncio
+async def test_when_target_has_rates_and_gmt_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")
+  now = datetime.strptime("2022-02-28T01:12:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  rate_data = create_rate_data(period_from, period_to, [10, 10, 20, 20, 30, 30])
+  expected_current_rate = 20
+  for rate in rate_data:
+    if now >= rate["valid_from"] and now <= rate["valid_to"]:
+      assert expected_current_rate == rate["value_inc_vat"]
+
+  # Act
+  rate_information = get_next_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is not None
+
+  assert "next_rate" in rate_information
+  assert rate_information["next_rate"]["value_inc_vat"] == 30
+  assert rate_information["next_rate"]["valid_from"] == datetime.strptime("2022-02-28T02:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["next_rate"]["valid_to"] == datetime.strptime("2022-02-28T03:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  assert "rates" in rate_information
+  assert len(rate_information["rates"]) == 2
+
+  assert rate_information["rates"][0]["value_inc_vat"] == 30
+  assert rate_information["rates"][0]["valid_from"] == datetime.strptime("2022-02-28T02:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["rates"][0]["valid_to"] == datetime.strptime("2022-02-28T02:30:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  assert rate_information["rates"][1]["value_inc_vat"] == 30
+  assert rate_information["rates"][1]["valid_from"] == datetime.strptime("2022-02-28T02:30:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["rates"][1]["valid_to"] == datetime.strptime("2022-02-28T03:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+@pytest.mark.asyncio
+async def test_when_target_has_rates_and_bst_then_rate_information_is_returned():
+  # Arrange
+  period_from = datetime.strptime("2022-02-27T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  period_to = datetime.strptime("2022-03-02T23:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  now = datetime.strptime("2022-02-28T01:12:00+01:00", "%Y-%m-%dT%H:%M:%S%z")
+
+  rate_data = create_rate_data(period_from, period_to, [10, 10, 20, 20, 30, 30])
+  expected_current_rate = 20
+  for rate in rate_data:
+    if now >= rate["valid_from"] and now <= rate["valid_to"]:
+      assert expected_current_rate == rate["value_inc_vat"]
+
+  # Act
+  rate_information = get_next_rate_information(rate_data, now)
+
+  # Assert
+  assert rate_information is not None
+
+  assert "next_rate" in rate_information
+  assert rate_information["next_rate"]["value_inc_vat"] == 30
+  assert rate_information["next_rate"]["valid_from"] == datetime.strptime("2022-02-28T01:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["next_rate"]["valid_to"] == datetime.strptime("2022-02-28T02:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  assert "rates" in rate_information
+  assert len(rate_information["rates"]) == 2
+
+  assert rate_information["rates"][0]["value_inc_vat"] == 30
+  assert rate_information["rates"][0]["valid_from"] == datetime.strptime("2022-02-28T01:00:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["rates"][0]["valid_to"] == datetime.strptime("2022-02-28T01:30:00Z", "%Y-%m-%dT%H:%M:%S%z")
+
+  assert rate_information["rates"][1]["value_inc_vat"] == 30
+  assert rate_information["rates"][1]["valid_from"] == datetime.strptime("2022-02-28T01:30:00Z", "%Y-%m-%dT%H:%M:%S%z")
+  assert rate_information["rates"][1]["valid_to"] == datetime.strptime("2022-02-28T02:00:00Z", "%Y-%m-%dT%H:%M:%S%z")