feat: Added sensors for available wheel of fortune spins

custom_components/octopus_energy/api_client/__init__.py

@@ -12,6 +12,7 @@
 from .intelligent_settings import IntelligentSettings
 from .intelligent_dispatches import IntelligentDispatchItem, IntelligentDispatches
 from .saving_sessions import JoinSavingSessionResponse, SavingSession, SavingSessionsResponse
+from .wheel_of_fortune import WheelOfFortuneSpinsResponse
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -261,6 +262,17 @@
 	}}
 }}'''
 
+wheel_of_fortune_query = '''query {{
+  wheelOfFortuneSpins(accountNumber: "{account_id}") {{
+    electricity {{
+      remainingSpinsThisMonth
+    }}
+    gas {{
+      remainingSpinsThisMonth
+    }}
+  }}
+}}'''
+
 def get_valid_from(rate):
   return rate["valid_from"]
 
@@ -979,6 +991,31 @@ async def async_get_intelligent_device(self, account_id: str):
           _LOGGER.error("Failed to retrieve intelligent device")
 
     return None
+
+  async def async_get_wheel_of_fortune_spins(self, account_id: str) -> WheelOfFortuneSpinsResponse:
+    """Get the user's wheel of fortune spins"""
+    await self.async_refresh_token()
+
+    async with aiohttp.ClientSession(timeout=self.timeout) as client:
+      url = f'{self._base_url}/v1/graphql/'
+      payload = { "query": wheel_of_fortune_query.format(account_id=account_id) }
+      headers = { "Authorization": f"JWT {self._graphql_token}" }
+      async with client.post(url, json=payload, headers=headers) as response:
+        response_body = await self.__async_read_response__(response, url)
+        _LOGGER.debug(f'async_get_wheel_of_fortune_spins: {response_body}')
+
+        if (response_body is not None and "data" in response_body and
+            "wheelOfFortuneSpins" in response_body["data"]):
+
+          spins = response_body["data"]["wheelOfFortuneSpins"]
+          return WheelOfFortuneSpinsResponse(
+            int(spins["electricity"]["remainingSpinsThisMonth"]) if "electricity" in spins and "remainingSpinsThisMonth" in spins["electricity"] else 0,
+            int(spins["gas"]["remainingSpinsThisMonth"]) if "gas" in spins and "remainingSpinsThisMonth" in spins["gas"] else 0
+          )
+        else:
+          _LOGGER.error("Failed to retrieve wheel of fortune spins")
+
+    return None
 
   def __get_interval_end(self, item):
     return item["interval_end"]

custom_components/octopus_energy/api_client/wheel_of_fortune.py

@@ -0,0 +1,11 @@
+class WheelOfFortuneSpinsResponse:
+  electricity: int
+  gas: int
+
+  def __init__(
+    self,
+    electricity: int,
+    gas: int
+  ):
+    self.electricity = electricity
+    self.gas = gas

custom_components/octopus_energy/const.py

@@ -47,10 +47,9 @@
 DATA_INTELLIGENT_DISPATCHES_COORDINATOR = "INTELLIGENT_DISPATCHES_COORDINATOR"
 DATA_INTELLIGENT_SETTINGS = "INTELLIGENT_SETTINGS"
 DATA_INTELLIGENT_SETTINGS_COORDINATOR = "INTELLIGENT_SETTINGS_COORDINATOR"
-
 DATA_ELECTRICITY_STANDING_CHARGE_KEY = "ELECTRICITY_STANDING_CHARGES_{}_{}"
-
 DATA_GAS_STANDING_CHARGE_KEY = "GAS_STANDING_CHARGES_{}_{}"
+DATA_WHEEL_OF_FORTUNE_SPINS = "WHEEL_OF_FORTUNE_SPINS"
 
 STORAGE_COMPLETED_DISPATCHES_NAME = "octopus_energy.{}-completed-intelligent-dispatches.json"
 

custom_components/octopus_energy/coordinators/wheel_of_fortune.py

@@ -0,0 +1,75 @@
+import logging
+from datetime import datetime, timedelta
+
+from homeassistant.util.dt import (now)
+from homeassistant.helpers.update_coordinator import (
+  DataUpdateCoordinator
+)
+
+from ..const import (
+  COORDINATOR_REFRESH_IN_SECONDS,
+  DOMAIN,
+  DATA_CLIENT,
+  DATA_ACCOUNT_ID,
+  DATA_WHEEL_OF_FORTUNE_SPINS,
+)
+
+from ..api_client import OctopusEnergyApiClient
+from ..api_client.saving_sessions import SavingSession
+from ..api_client.wheel_of_fortune import WheelOfFortuneSpinsResponse
+
+_LOGGER = logging.getLogger(__name__)
+
+class WheelOfFortuneSpinsCoordinatorResult:
+  last_retrieved: datetime
+  spins: WheelOfFortuneSpinsResponse
+
+  def __init__(self, last_retrieved: datetime, spins: WheelOfFortuneSpinsResponse):
+    self.last_retrieved = last_retrieved
+    self.spins = spins
+
+async def async_refresh_wheel_of_fortune_spins(
+    current: datetime,
+    client: OctopusEnergyApiClient,
+    account_id: str,
+    existing_result: WheelOfFortuneSpinsCoordinatorResult
+) -> WheelOfFortuneSpinsCoordinatorResult:
+  if existing_result is None or current.minute % 30 == 0:
+    try:
+      result = await client.async_get_wheel_of_fortune_spins(account_id)
+
+      return WheelOfFortuneSpinsCoordinatorResult(current, result)
+    except:
+      _LOGGER.debug('Failed to retrieve wheel of fortune spins')
+
+  return existing_result
+
+async def async_setup_wheel_of_fortune_spins_coordinator(hass, account_id: str):
+  async def async_update_data():
+    """Fetch data from API endpoint."""
+    current = now()
+    client: OctopusEnergyApiClient = hass.data[DOMAIN][DATA_CLIENT]
+
+    hass.data[DOMAIN][DATA_WHEEL_OF_FORTUNE_SPINS] = await async_refresh_wheel_of_fortune_spins(
+      current,
+      client,
+      account_id,
+      hass.data[DOMAIN][DATA_WHEEL_OF_FORTUNE_SPINS] if DATA_WHEEL_OF_FORTUNE_SPINS in hass.data[DOMAIN] else None
+    )
+
+    return hass.data[DOMAIN][DATA_WHEEL_OF_FORTUNE_SPINS]
+
+  coordinator = DataUpdateCoordinator(
+    hass,
+    _LOGGER,
+    name=f"{account_id}_wheel_of_fortune_spins",
+    update_method=async_update_data,
+    # Because of how we're using the data, we'll update every minute, but we will only actually retrieve
+    # data every 30 minutes
+    update_interval=timedelta(seconds=COORDINATOR_REFRESH_IN_SECONDS),
+    always_update=True
+  )
+
+  await coordinator.async_config_entry_first_refresh()
+
+  return coordinator

custom_components/octopus_energy/sensor.py

@@ -1,3 +1,4 @@
+from custom_components.octopus_energy.coordinators.wheel_of_fortune import async_setup_wheel_of_fortune_spins_coordinator
 import voluptuous as vol
 import logging
 
@@ -35,6 +36,8 @@
 from .gas.current_accumulative_cost import OctopusEnergyCurrentAccumulativeGasCost
 from .gas.standing_charge import OctopusEnergyGasCurrentStandingCharge
 from .gas.previous_accumulative_cost_override import OctopusEnergyPreviousAccumulativeGasCostOverride
+from .wheel_of_fortune.electricity_spins import OctopusEnergyWheelOfFortuneElectricitySpins
+from .wheel_of_fortune.gas_spins import OctopusEnergyWheelOfFortuneGasSpins
 
 from .coordinators.current_consumption import async_create_current_consumption_coordinator
 from .coordinators.gas_rates import async_setup_gas_rates_coordinator
@@ -101,8 +104,15 @@ async def async_setup_default_sensors(hass: HomeAssistant, entry, async_add_enti
   await saving_session_coordinator.async_config_entry_first_refresh()
 
   account_info = hass.data[DOMAIN][DATA_ACCOUNT]
+  account_id = account_info["id"]
+
+  wheel_of_fortune_coordinator = await async_setup_wheel_of_fortune_spins_coordinator(hass, account_id)
 
-  entities = [OctopusEnergyOctoplusPoints(hass, client, account_info["id"])]
+  entities = [
+    OctopusEnergyOctoplusPoints(hass, client, account_id),
+    OctopusEnergyWheelOfFortuneElectricitySpins(hass, wheel_of_fortune_coordinator, account_id),
+    OctopusEnergyWheelOfFortuneGasSpins(hass, wheel_of_fortune_coordinator, account_id)
+  ]
 
   now = utcnow()
 
@@ -177,7 +187,7 @@ async def async_setup_default_sensors(hass: HomeAssistant, entry, async_add_enti
                 severity=ir.IssueSeverity.ERROR,
                 learn_more_url="https://github.com/BottlecapDave/HomeAssistant-OctopusEnergy/blob/develop/_docs/repairs/octopus_mini_not_valid.md",
                 translation_key="octopus_mini_not_valid",
-                translation_placeholders={ "type": "electricity", "account_id": account_info["id"], "mpan_mprn": mpan, "serial_number": serial_number },
+                translation_placeholders={ "type": "electricity", "account_id": account_id, "mpan_mprn": mpan, "serial_number": serial_number },
               )
       else:
         for meter in point["meters"]:
@@ -254,7 +264,7 @@ async def async_setup_default_sensors(hass: HomeAssistant, entry, async_add_enti
                 severity=ir.IssueSeverity.ERROR,
                 learn_more_url="https://github.com/BottlecapDave/HomeAssistant-OctopusEnergy/blob/develop/_docs/repairs/octopus_mini_not_valid.md",
                 translation_key="octopus_mini_not_valid",
-                translation_placeholders={ "type": "gas", "account_id": account_info["id"], "mpan_mprn": mprn, "serial_number": serial_number },
+                translation_placeholders={ "type": "gas", "account_id": account_id, "mpan_mprn": mprn, "serial_number": serial_number },
               )
       else:
         for meter in point["meters"]:

custom_components/octopus_energy/wheel_of_fortune/electricity_spins.py

@@ -0,0 +1,79 @@
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import generate_entity_id
+from homeassistant.helpers.update_coordinator import (
+  CoordinatorEntity,
+)
+from homeassistant.components.sensor import (
+  RestoreSensor,
+  SensorStateClass
+)
+from ..utils import account_id_to_unique_key
+from ..coordinators.wheel_of_fortune import WheelOfFortuneSpinsCoordinatorResult
+
+_LOGGER = logging.getLogger(__name__)
+
+class OctopusEnergyWheelOfFortuneElectricitySpins(CoordinatorEntity, RestoreSensor):
+  """Sensor for current wheel of fortune spins for electricity"""
+
+  def __init__(self, hass: HomeAssistant, coordinator, account_id: str):
+    """Init sensor."""
+    CoordinatorEntity.__init__(self, coordinator)
+
+    self._account_id = account_id
+    self._state = None
+    self._attributes = {
+      "last_evaluated": None
+    }
+    self._last_evaluated = None
+
+    self.entity_id = generate_entity_id("sensor.{}", self.unique_id, hass=hass)
+
+  @property
+  def unique_id(self):
+    """The id of the sensor."""
+    return f"octopus_energy_{account_id_to_unique_key(self._account_id)}_wheel_of_fortune_spins_electricity"
+
+  @property
+  def name(self):
+    """Name of the sensor."""
+    return f"Octopus Energy {self._account_id} Wheel Of Fortune Spins Electricity"
+
+  @property
+  def icon(self):
+    """Icon of the sensor."""
+    return "mdi:star-circle"
+
+  @property
+  def extra_state_attributes(self):
+    """Attributes of the sensor."""
+    return self._attributes
+
+  @property
+  def state_class(self):
+    """The state class of sensor"""
+    return SensorStateClass.TOTAL
+
+  @property
+  def state(self):
+    result: WheelOfFortuneSpinsCoordinatorResult = self.coordinator.data if self.coordinator is not None and self.coordinator.data is not None else None
+    if result is not None:
+      self._state = result.spins.electricity
+      self._attributes["last_evaluated"] = result.last_retrieved
+
+    return self._state
+
+  async def async_added_to_hass(self):
+    """Call when entity about to be added to hass."""
+    # If not None, we got an initial value.
+    await super().async_added_to_hass()
+    state = await self.async_get_last_state()
+
+    if state is not None and self._state is None:
+      self._state = state.state
+      self._attributes = {}
+      for x in state.attributes.keys():
+        self._attributes[x] = state.attributes[x]
+
+      _LOGGER.debug(f'Restored OctopusEnergyWheelOfFortuneElectricitySpins state: {self._state}')

custom_components/octopus_energy/wheel_of_fortune/gas_spins.py

@@ -0,0 +1,79 @@
+import logging
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import generate_entity_id
+from homeassistant.helpers.update_coordinator import (
+  CoordinatorEntity,
+)
+from homeassistant.components.sensor import (
+  RestoreSensor,
+  SensorStateClass
+)
+from ..utils import account_id_to_unique_key
+from ..coordinators.wheel_of_fortune import WheelOfFortuneSpinsCoordinatorResult
+
+_LOGGER = logging.getLogger(__name__)
+
+class OctopusEnergyWheelOfFortuneGasSpins(CoordinatorEntity, RestoreSensor):
+  """Sensor for current wheel of fortune spins for gas"""
+
+  def __init__(self, hass: HomeAssistant, coordinator, account_id: str):
+    """Init sensor."""
+    CoordinatorEntity.__init__(self, coordinator)
+
+    self._account_id = account_id
+    self._state = None
+    self._attributes = {
+      "last_evaluated": None
+    }
+    self._last_evaluated = None
+
+    self.entity_id = generate_entity_id("sensor.{}", self.unique_id, hass=hass)
+
+  @property
+  def unique_id(self):
+    """The id of the sensor."""
+    return f"octopus_energy_{account_id_to_unique_key(self._account_id)}_wheel_of_fortune_spins_gas"
+
+  @property
+  def name(self):
+    """Name of the sensor."""
+    return f"Octopus Energy {self._account_id} Wheel Of Fortune Spins Gas"
+
+  @property
+  def icon(self):
+    """Icon of the sensor."""
+    return "mdi:star-circle"
+
+  @property
+  def extra_state_attributes(self):
+    """Attributes of the sensor."""
+    return self._attributes
+
+  @property
+  def state_class(self):
+    """The state class of sensor"""
+    return SensorStateClass.TOTAL
+
+  @property
+  def state(self):
+    result: WheelOfFortuneSpinsCoordinatorResult = self.coordinator.data if self.coordinator is not None and self.coordinator.data is not None else None
+    if result is not None:
+      self._state = result.spins.gas
+      self._attributes["last_evaluated"] = result.last_retrieved
+
+    return self._state
+
+  async def async_added_to_hass(self):
+    """Call when entity about to be added to hass."""
+    # If not None, we got an initial value.
+    await super().async_added_to_hass()
+    state = await self.async_get_last_state()
+
+    if state is not None and self._state is None:
+      self._state = state.state
+      self._attributes = {}
+      for x in state.attributes.keys():
+        self._attributes[x] = state.attributes[x]
+
+      _LOGGER.debug(f'Restored OctopusEnergyWheelOfFortuneGasSpins state: {self._state}')

tests/integration/api_client/test_get_wheel_of_fortune_spins.py

@@ -0,0 +1,21 @@
+import pytest
+
+from integration import get_test_context
+from custom_components.octopus_energy.api_client import OctopusEnergyApiClient
+
+@pytest.mark.asyncio
+async def test_when_get_wheel_of_fortune_spins_called_then_spins_are_returned():
+    # Arrange
+    context = get_test_context()
+
+    client = OctopusEnergyApiClient(context["api_key"])
+    account_id = context["account_id"]
+
+    # Act
+    result = await client.async_get_wheel_of_fortune_spins(account_id)
+
+    # Assert
+    assert result is not None
+
+    assert result.electricity >= 0
+    assert result.gas >= 0