feat: Added sensor for setting home pro text (1.5 hours dev)

_docs/entities/home_pro.md

@@ -0,0 +1,14 @@
+# Home Pro
+
+To support the Home Pro device. Once configured, the following entities will retrieve data locally from your Octopus Home Pro instead of via the Octopus Energy APIs at a target rate of every 10 seconds.
+
+* [Electricity - Current Demand](./electricity.md#current-demand)
+* [Electricity - Current Total Consumption](./electricity.md#current-total-consumption)
+* [Gas - Current Total Consumption kWh](./gas.md#current-total-consumption-kwh)
+* [Gas - Current Total Consumption m3](./gas.md#current-total-consumption-m3)
+
+## Home Pro Screen
+
+`text.octopus_energy_{{ACCOUNT_ID}}_home_pro_screen`
+
+Allows you to set scrolling text for the home pro device.

_docs/setup/account.md

@@ -99,9 +99,4 @@ Once the API has been configured, you will need to set the address to the IP add
 
 ### Entities
 
-Once configured, the following entities will retrieve data locally from your Octopus Home Pro instead of via the Octopus Energy APIs at a target rate of every 10 seconds.
-
-* [Electricity - Current Demand](../entities/electricity.md#current-demand)
-* [Electricity - Current Total Consumption](../entities/electricity.md#current-total-consumption)
-* [Gas - Current Total Consumption kWh](../entities/gas.md#current-total-consumption-kwh)
-* [Gas - Current Total Consumption m3](../entities/gas.md#current-total-consumption-m3)
+See [entities](../entities/home_pro.md) for more information.

custom_components/octopus_energy/__init__.py

@@ -57,7 +57,7 @@
   REPAIR_UNIQUE_RATES_CHANGED_KEY
 )
 
-ACCOUNT_PLATFORMS = ["sensor", "binary_sensor", "number", "switch", "time", "event"]
+ACCOUNT_PLATFORMS = ["sensor", "binary_sensor", "number", "switch", "text", "time", "event"]
 TARGET_RATE_PLATFORMS = ["binary_sensor"]
 COST_TRACKER_PLATFORMS = ["sensor"]
 TARIFF_COMPARISON_PLATFORMS = ["sensor"]

custom_components/octopus_energy/api_client_home_pro/__init__.py

@@ -83,6 +83,28 @@ async def async_get_consumption(self, is_electricity: bool) -> list | None:
       _LOGGER.warning(f'Failed to connect. Timeout of {self._timeout} exceeded.')
       raise TimeoutException()
 
+  async def async_set_screen(self, value: str, animation_type: str, type: str, brightness: int, animation_interval: int):
+    """Get the latest consumption"""
+
+    try:
+      client = self._create_client_session()
+      url = f'{self._base_url}/screen'
+      headers = { "Authorization": self._api_key }
+      payload = {
+        "value": f"{value}",
+        "animationType": f"{animation_type}",
+        "type": f"{type}",
+        "brightness": brightness,
+        "animationInterval": animation_interval
+      }
+
+      async with client.post(url, json=payload, headers=headers) as response:
+        await self.__async_read_response__(response, url)
+
+    except TimeoutError:
+      _LOGGER.warning(f'Failed to connect. Timeout of {self._timeout} exceeded.')
+      raise TimeoutException()
+
   async def __async_read_response__(self, response, url):
     """Reads the response, logging any json errors"""
 
@@ -102,12 +124,13 @@ async def __async_read_response__(self, response, url):
         _LOGGER.warning(msg)
         raise RequestException(msg, [])
 
-      _LOGGER.info(f"Response {response.status} for '{url}' received")
+      _LOGGER.info(f"Response {response.status} for '{url}' receivedL {text}")
       return None
 
     data_as_json = None
     try:
-      data_as_json = json.loads(text)
+      if text is not None and text != "":
+        data_as_json = json.loads(text)
     except:
       raise Exception(f'Failed to extract response json: {url}; {text}')
 

custom_components/octopus_energy/home_pro/screen_text.py

@@ -0,0 +1,63 @@
+import logging
+
+from homeassistant.core import HomeAssistant
+
+from homeassistant.components.text import TextEntity
+
+from homeassistant.helpers.restore_state import RestoreEntity
+
+from homeassistant.helpers.entity import generate_entity_id
+
+from ..api_client_home_pro import OctopusEnergyHomeProApiClient
+
+from ..utils.attributes import dict_to_typed_dict
+
+_LOGGER = logging.getLogger(__name__)
+
+class OctopusEnergyHomeProScreenText(TextEntity, RestoreEntity):
+  """Sensor for determining the text on the home pro"""
+
+  def __init__(self, hass: HomeAssistant, account_id: str, client: OctopusEnergyHomeProApiClient):
+    """Init sensor."""
+    self._hass = hass
+    self._client = client
+    self._account_id = account_id
+    self._attr_native_value = None
+
+    self.entity_id = generate_entity_id("text.{}", self.unique_id, hass=hass)
+
+  @property
+  def unique_id(self):
+    """The id of the sensor."""
+    return f"octopus_energy_{self._account_id}_home_pro_screen"
+
+  @property
+  def name(self):
+    """Name of the sensor."""
+    return f"Home Pro Screen ({self._account_id})"
+
+  @property
+  def icon(self):
+    """Icon of the sensor."""
+    return "mdi:led-strip"
+
+  async def async_set_value(self, value: str) -> None:
+    """Update the value."""
+    self._attr_native_value = value
+    await self._client.async_set_screen(self._attr_native_value, "scroll", "text", 200, 100)
+    self.async_write_ha_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:
+      if state.state is not None:
+        self._attr_native_value = state.state
+        self._attr_state = state.state
+
+      self._attributes = dict_to_typed_dict(state.attributes)
+
+      _LOGGER.debug(f'Restored OctopusEnergyPreviousAccumulativeElectricityCostTariffOverride state: {self._attr_state}')

custom_components/octopus_energy/text.py

@@ -0,0 +1,37 @@
+import logging
+
+from homeassistant.core import HomeAssistant
+
+from .home_pro.screen_text import OctopusEnergyHomeProScreenText
+
+from .const import (
+  CONFIG_ACCOUNT_ID,
+  DATA_HOME_PRO_CLIENT,
+  DOMAIN,
+
+  CONFIG_MAIN_API_KEY
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+async def async_setup_entry(hass, entry, async_add_entities):
+  """Setup sensors based on our entry"""
+  config = dict(entry.data)
+
+  if entry.options:
+    config.update(entry.options)
+
+  if CONFIG_MAIN_API_KEY in config:
+    await async_setup_default_sensors(hass, config, async_add_entities)
+
+async def async_setup_default_sensors(hass: HomeAssistant, config, async_add_entities):
+  account_id = config[CONFIG_ACCOUNT_ID]
+
+  home_pro_client = hass.data[DOMAIN][account_id][DATA_HOME_PRO_CLIENT] if DATA_HOME_PRO_CLIENT in hass.data[DOMAIN][account_id] else None
+
+  entities = []
+
+  if home_pro_client is not None:
+    entities.append(OctopusEnergyHomeProScreenText(hass, account_id, home_pro_client))
+
+  async_add_entities(entities)

home_pro_server/oeha_server.py

@@ -68,8 +68,29 @@ def do_POST(self):
       return
 
     if self.path.startswith("/screen"):
+      screen_auth_token = os.environ['AUTH_TOKEN']
+      if screen_auth_token is None or screen_auth_token == "":
+        self.send_response(401)
+        self.send_header('Content-type', 'text/html')
+        self.end_headers()
+        output = json.dumps({
+          "error": "AUTH_TOKEN not set",
+        })
+        self.wfile.write(output.encode("utf8"))
+        return 
+
+      if app_name is None or app_name == "":
+        self.send_response(401)
+        self.send_header('Content-type', 'text/html')
+        self.end_headers()
+        output = json.dumps({
+          "error": "APPLICATION_NAME not set",
+        })
+        self.wfile.write(output.encode("utf8"))
+        return
+
       headers = {
-        "Authorization": f"Basic {os.environ['AUTH_TOKEN']}",
+        "Authorization": f"Basic {screen_auth_token}",
         "Content-Type": "application/json"
       }
 
@@ -80,6 +101,11 @@ def do_POST(self):
         # Set headers
         self.send_header("Content-type", "application/json")
         self.end_headers()
+        output = json.dumps({
+          "error": "Failed to get screen info",
+          "native_response": response.text,
+        })
+        self.wfile.write(output.encode("utf8"))
         return
 
       # Get first screen
@@ -96,6 +122,11 @@ def do_POST(self):
         # Set headers
         self.send_header("Content-type", "application/json")
         self.end_headers()
+        output = json.dumps({
+          "error": "Failed to set screen screen info",
+          "native_response": response.text,
+        })
+        self.wfile.write(output.encode("utf8"))
         return
 
       # Set response status code

home_pro_server/setup.sh

@@ -4,7 +4,7 @@ set -e
 rm -rf bottlecapdave_homeassistant_octopus_energy
 mkdir bottlecapdave_homeassistant_octopus_energy
 cd bottlecapdave_homeassistant_octopus_energy
-wget https://raw.githubusercontent.com/BottlecapDave/HomeAssistant-OctopusEnergy/develop/home_pro_server/oeha_server.py
+wget https://raw.githubusercontent.com/BottlecapDave/HomeAssistant-OctopusEnergy/main/home_pro_server/oeha_server.py
 chmod +x oeha_server.py
-wget https://raw.githubusercontent.com/BottlecapDave/HomeAssistant-OctopusEnergy/develop/home_pro_server/start_server.sh
+wget https://raw.githubusercontent.com/BottlecapDave/HomeAssistant-OctopusEnergy/main/home_pro_server/start_server.sh
 chmod +x start_server.sh

mkdocs.yml

@@ -17,6 +17,7 @@ nav:
     - Intelligent: ./entities/intelligent.md
     - Wheel Of Fortune: ./entities/wheel_of_fortune.md
     - Greenness Forecast: ./entities/greenness_forecast.md
+    - Home Pro: ./entities/home_pro.md
   - services.md
   - events.md
   - Repairs:

tests/local_integration/api_client_home_pro/test_async_get_consumption.py

@@ -51,7 +51,7 @@ async def test_when_get_consumption_is_called_then_data_is_returned(is_electrici
     assert data[0]["demand"] is None
 
   assert "total_consumption" in data[0]
-  assert data[0]["total_consumption"] >= 0
+  assert data[0]["total_consumption"] is None or data[0]["total_consumption"] >= 0
 
   assert "start" in data[0]
   assert "end" in data[0]

tests/local_integration/api_client_home_pro/test_async_set_screen.py

@@ -0,0 +1,32 @@
+import pytest
+
+from .. import get_test_context
+from custom_components.octopus_energy.api_client import AuthenticationException
+from custom_components.octopus_energy.api_client_home_pro import OctopusEnergyHomeProApiClient
+
+@pytest.mark.asyncio
+async def test_when_set_screen_is_called_and_api_key_is_invalid_then_exception_is_raised():
+  # Arrange
+  context = get_test_context()
+
+  client = OctopusEnergyHomeProApiClient(context.base_url, "invalid-api-key")
+
+  # Act
+  exception_raised = False
+  try:
+    await client.async_set_screen("hello world", "scroll", "text", 200, 100)
+  except AuthenticationException:
+    exception_raised = True
+
+  # Assert
+  assert exception_raised == True
+
+@pytest.mark.asyncio
+async def test_when_set_screen_is_called_then_successful():
+  # Arrange
+  context = get_test_context()
+
+  client = OctopusEnergyHomeProApiClient(context.base_url, context.api_key)
+
+  # Act
+  await client.async_set_screen("hello world", "scroll", "text", 200, 100)