feat: Added support for setting Home Pro screen (3 hours dev time)

You will need to reinstall the API on the Home Pro device to support this - https://bottlecapdave.github.io/HomeAssistant-OctopusEnergy/setup/account/#prerequisites

_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. If the text is greater than 3 characters, then it will scroll on the device, otherwise it will be statically displayed.

_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,29 @@ 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 = {
+        # API doesn't support none or empty string as a valid value
+        "value": f"{value}" if value is not None and value != "" else " ",
+        "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 +125,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,67 @@
+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
+    animation_type = "static"
+    if value is not None and len(value) > 3:
+      animation_type = "scroll"
+
+    await self._client.async_set_screen(self._attr_native_value, animation_type, "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

@@ -6,6 +6,8 @@
 
 han_host = os.getenv("HAN_API_HOST")
 auth_token = os.getenv("SERVER_AUTH_TOKEN")
+screen_host = os.getenv('SCREEN_API_HOST')
+app_name = os.getenv('APPLICATION_NAME')
 
 class RequestHandler(BaseHTTPRequestHandler):
   # Handle GET requests
@@ -20,13 +22,13 @@ def do_GET(self):
       self.end_headers()
       return
 
-    if self.path.startswith("/get_meter_consumption"):
+    if self.path.startswith(f"/get_meter_consumption"):
       # Set response status code
       self.send_response(200)
       # Set headers
       self.send_header("Content-type", "application/json")
       self.end_headers()
-      response = request_get_response("get_meter_consumption", params)
+      response = request_get_response(f"{han_host}/get_meter_consumption", params)
       if response.ok:
         self.wfile.write(to_response(response, "meter_consump"))
         return
@@ -37,7 +39,7 @@ def do_GET(self):
       # Set headers
       self.send_header("Content-type", "application/json")
       self.end_headers()
-      response = request_get_response("get_meter_status", params)
+      response = request_get_response(f"{han_host}/get_meter_status", params)
       if response.ok:
         self.wfile.write(to_response(response, "meter_status"))
         return
@@ -48,18 +50,97 @@ def do_GET(self):
       # Set headers
       self.send_header("Content-type", "application/json")
       self.end_headers()
-      response = request_get_response("get_meter_info", params)
+      response = request_get_response(f"{han_host}/get_meter_info", params)
       if response.ok:
         self.wfile.write(to_response(response, "meter_info"))
         return
     else:
       self.send_response(404)
       self.send_header("Content-type", "text/plain")
       self.end_headers()
+
+  def do_POST(self):
+    auth_header = self.headers.get('Authorization')
+    if auth_header == None or auth_header != auth_token:
+      self.send_response(401)
+      self.send_header('Content-type', 'text/html')
+      self.end_headers()
+      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 {screen_auth_token}",
+        "Content-Type": "application/json"
+      }
+
+      response = requests.get(f"{screen_host}/api/v1/screen/application/{app_name}", headers=headers)
+      if not response.ok:
+        print("Failed to get screen info")
+        self.send_response(400)
+        # 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
+      first_screen = response.json()[0]
+      screen_id = first_screen['_id']
+
+      screen_url = f"{screen_host}/api/v1/screen/{screen_id}"
+      print(f"Calling API (PATCH): {screen_url}")
+      data = (self.rfile.read(int(self.headers['content-length']))).decode('utf-8')
+      response = requests.patch(screen_url, headers=headers, data=data)
+      print(f"Response: {response.text}")
+      if not response.ok:
+        self.send_response(400)
+        # 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
+      self.send_response(200)
+      # Set headers
+      self.send_header("Content-type", "application/json")
+      self.end_headers()
+    else:
+      self.send_response(404)
+      self.send_header("Content-type", "text/plain")
+      self.end_headers()
 
-def request_get_response(api, params):
-  print(f"Calling API: {api}")
-  url = f"{han_host}/" + f"{api}"
+def request_get_response(url, params):
+  print(f"Calling API (GET): {url}")
   response = requests.get(url, json=params)
   print(f"Response: {response.text}")
   return response

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,43 @@
+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)
+
+# @pytest.mark.asyncio
+# async def test_when_set_screen_is_called_with_empty_value_then_successful():
+#   # Arrange
+#   context = get_test_context()
+
+#   client = OctopusEnergyHomeProApiClient(context.base_url, context.api_key)
+
+#   # Act
+#   await client.async_set_screen("", "scroll", "text", 200, 100)
+#   await client.async_set_screen(None, "scroll", "text", 200, 100)