Integrating Anglia water usage for smart meters (UK)

It’s not completely dead in the water, they’ve come back to my email yesterday where I asked them for an official api, they’ve said

“Currently there are no plans to offer an API, however this is something we have had several requests for. Once we have fully moved over to the new platform and we start looking at enhancements, this will certainly be something that is discussed, and hopefully we can provide something.”

1 Like

Ah that’s good. Yes they are currently using a third party API that’s why there isn’t any other option put out there. Thank you again for sharing!

Ive currently got a working version, once ive tested it will post final version on here.
The reason for the change is that Anglian Water have moved to a new Website developed using AngualarJS, Selenium does not play well with this and we have to incorporate more wait times for pages to load (Javascript DOM).

This is the version that works at the moment… need a few more days testing.

UPDATED!

from selenium import webdriver
from selenium.webdriver import Chrome
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from datetime import datetime as date,timedelta
from dateutil import parser
import appdaemon.plugins.hass.hassapi as hass
import time
import json
import ctypes
import datetime

#####################
# App configuration #
#####################
# Complete with your settings
anglia_username = "_YOUR_ANGLIA_WATER_EMAIL-"
anglia_password = "_YOUR_ANGLIA_WATER_PASSWORD_"
usagesensor = "sensor.water_usage_x"
# Optional: Uncomment the below line in case you want to have a switch to run the script on demand if for eg. the script doesn't run at 11am because your HA is restarting. Make sure your input_boolean matches the below:
#force_reading = "input_boolean.water_reading"

###################
# WATER USAGE APP #
###################

webpage = "https://www.anglianwater.co.uk/"

class GetWaterUsage(hass.Hass):

# Starting APP #

  def initialize(self):
    self.log("Starting Water Usage App")
    runtime  = datetime.time(11, 00, 0)
    zero  = datetime.time(00, 00, 0)
    self.run_daily(self.StartProcess, runtime)
    self.run_daily(self.EndProcess, zero)    
    self.listen_state(self.StartProcessForced,force_reading, attribute="state", new="on", duration=10 )

  #Callback Function to Start the process 
  def StartProcess(self, kwargs):
    self.log("starting process")
    usage = self.GetUsage()
    self.log("Usage: " + usage)
    try:
      self.set_state(usagesensor, state = usage)
    except Exception as e:
      self.log("Set Sensor State Error: " + str(e))

  #FORCED Callback Function to Start the process 
  def StartProcessForced(self, entity, attribute, old, new, kwargs):
    self.log("starting process")
    self.turn_off(force_reading)
    usage = self.GetUsage()
    self.log("Usage: " + usage)
    try:
      self.set_state(usagesensor, state = usage)
    except Exception as e:
      self.log("Set Sensor State Error: " + str(e))

  #Callback Function to Zero the readings 
  def EndProcess(self, kwargs):
    self.log("zero readings")
    usage = "0.0"
    self.log("Usage: " + usage)
    self.set_state(usagesensor, state = usage)
    
# Get Water Usage from Anglia Water website #

  def GetUsage(self):
    usage = "0.0"
    try:
      self.log("starting request")
      service = Service(executable_path=r'/usr/bin/chromedriver')
      chrome_options = webdriver.ChromeOptions()
      chrome_options.add_argument('--headless')
      chrome_options.add_argument('--no-sandbox')
      chrome_options.add_argument('--disable-gpu')
      chrome_options.add_argument('--disable-dev-shm-usage')
      browser = webdriver.Chrome(service=service, options=chrome_options)
      browser.set_window_size(1600, 1200)
      try:
        self.log("Request Home page")
        browser.get(webpage)
        time.sleep(5)
        self.log("Requested Home Page")
        browser.find_element(By.LINK_TEXT, "Log In").click()
        time.sleep(5)
        self.log("Request Login")
        browser.find_element(By.ID, "email").send_keys(anglia_username)
        browser.find_element(By.ID,'password').send_keys(anglia_password)
        browser.find_element(By.ID, "next").click()
        time.sleep(10)
        self.log("Login")
        browser.find_element(By.CSS_SELECTOR, ".justify-content-between:nth-child(3) > .fw-bold").click()
        time.sleep(10)
        self.log("Request Usage")
                
        ele = browser.find_elements(By.TAG_NAME, 'path')        
        #Daily
        usage = 0
        lastreadingdate = date.today().strftime("%d-%b-%Y")
        for x in ele:
          val = x.get_attribute("aria-label")
          if val is not None:
            dss = val.split()
            stringdt = parser.parse(dss[0] + '-' + dss[1]+ '-' + dss[2]).strftime("%d-%b-%Y")
            curdt = (date.today() - timedelta(days=1)).strftime("%d-%b-%Y")
            if stringdt == curdt:
              usage = dss[3]
              lastreadingdate = stringdt
        
        browser.quit()
        self.log("Browser quit")
		
      except Exception as e:
        self.log("Request Error: " + str(e))
        browser.close()
    except Exception as e:
      self.log("GetUsage Error: " + str(e))
    return usage
2 Likes

Smashing! Thanks a lot! Keep us posted, I will replace it on the main post :wink:

Hi @yatik I’m getting the following error when I try your updated script:

2024-02-12 17:00:00.013486 INFO WaterUsage: starting process
2024-02-12 17:00:00.015359 INFO WaterUsage: starting request
2024-02-12 17:00:00.548166 INFO WaterUsage: Request Home page
2024-02-12 17:00:07.569364 INFO WaterUsage: Requested Home Page
2024-02-12 17:00:07.605873 INFO WaterUsage: Request Error: Message: no such element: Unable to locate element: {“method”:“link text”,“selector”:“Log In”}
(Session info: chrome-headless-shell=121.0.6167.85)

Hopefully something simple? Thanks for all your work on this @Zeunas & @yatik

Ive updated the code and it was a typo on my part.

Ive been running it for the last 3 days, it seems to update between 8pm-10pm.

Seems to be working reliably so can say this is final version below.

from selenium import webdriver
from selenium.webdriver import Chrome
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from datetime import datetime as date,timedelta
from dateutil import parser
import appdaemon.plugins.hass.hassapi as hass
import time
import json
import ctypes
import datetime

#####################
# App configuration #
#####################
# Complete with your settings
anglia_username = "_YOUR_ANGLIA_WATER_EMAIL-"
anglia_password = "_YOUR_ANGLIA_WATER_PASSWORD_"
usagesensor = "sensor.water_usage_x"
# Optional: Uncomment the below line in case you want to have a switch to run the script on demand if for eg. the script doesn't run at 11am because your HA is restarting. Make sure your input_boolean matches the below:
#force_reading = "input_boolean.water_reading"

###################
# WATER USAGE APP #
###################

webpage = "https://www.anglianwater.co.uk/"

class GetWaterUsage(hass.Hass):

# Starting APP #

  def initialize(self):
    self.log("Starting Water Usage App")
    runtime  = datetime.time(22, 00, 0)
    zero  = datetime.time(00, 00, 0)
    self.run_daily(self.StartProcess, runtime)
    self.run_daily(self.EndProcess, zero)    
    self.listen_state(self.StartProcessForced,force_reading, attribute="state", new="on", duration=10 )

  #Callback Function to Start the process 
  def StartProcess(self, kwargs):
    self.log("starting process")
    usage = self.GetUsage()
    self.log("Usage: " + usage)
    try:
      self.set_state(usagesensor, state = usage)
    except Exception as e:
      self.log("Set Sensor State Error: " + str(e))

  #FORCED Callback Function to Start the process 
  def StartProcessForced(self, entity, attribute, old, new, kwargs):
    self.log("starting process")
    self.turn_off(force_reading)
    usage = self.GetUsage()
    self.log("Usage: " + usage)
    try:
      self.set_state(usagesensor, state = usage)
    except Exception as e:
      self.log("Set Sensor State Error: " + str(e))

  #Callback Function to Zero the readings 
  def EndProcess(self, kwargs):
    self.log("zero readings")
    usage = "0.0"
    self.log("Usage: " + usage)
    self.set_state(usagesensor, state = usage)
    
# Get Water Usage from Anglia Water website #

  def GetUsage(self):
    usage = "0.0"
    try:
      self.log("starting request")
      service = Service(executable_path=r'/usr/bin/chromedriver')
      chrome_options = webdriver.ChromeOptions()
      chrome_options.add_argument('--headless')
      chrome_options.add_argument('--no-sandbox')
      chrome_options.add_argument('--disable-gpu')
      chrome_options.add_argument('--disable-dev-shm-usage')
      browser = webdriver.Chrome(service=service, options=chrome_options)
      browser.set_window_size(1600, 1200)
      try:
        self.log("Request Home page")
        browser.get(webpage)
        time.sleep(5)
        self.log("Requested Home Page")
        browser.find_element(By.LINK_TEXT, "Log In").click()
        time.sleep(8)
        self.log("Request Login")
        browser.find_element(By.ID, "email").send_keys(anglia_username)
        browser.find_element(By.ID,'password').send_keys(anglia_password)
        browser.find_element(By.ID, "next").click()
        time.sleep(10)
        self.log("Login")
        browser.find_element(By.CSS_SELECTOR, ".justify-content-between:nth-child(3) > .fw-bold").click()
        time.sleep(10)
        self.log("Request Usage")
                
        ele = browser.find_elements(By.TAG_NAME, 'path')        
        #Daily
        usage = 0
        lastreadingdate = date.today().strftime("%d-%b-%Y")
        for x in ele:
          val = x.get_attribute("aria-label")
          if val is not None:
            dss = val.split()
            stringdt = parser.parse(dss[0] + '-' + dss[1]+ '-' + dss[2]).strftime("%d-%b-%Y")
            curdt = (date.today() - timedelta(days=1)).strftime("%d-%b-%Y")
            if stringdt == curdt:
              usage = dss[3]
              lastreadingdate = stringdt
        
        browser.quit()
        self.log("Browser quit")
		
      except Exception as e:
        self.log("Request Error: " + str(e))
        browser.close()
    except Exception as e:
      self.log("GetUsage Error: " + str(e))
    return usage
4 Likes

Thanks a lot! I have updated my post to revert to your script. Kudos for you!

THanks @yatik I can confirm that it now pulls in yesterday’s usage, great work!


Thanks for all this folks, really appreciate the work. Getting this in the log for starting the app.


If uncomment the force reading at the top of the script i get this

I have just updated the script but i’m getting the following error:

2024-02-17 20:47:05.212075 INFO WaterUsage: Request Login
2024-02-17 20:47:05.226017 INFO WaterUsage: Request Error: Message: no such element: Unable to locate element: {“method”:“css selector”,“selector”:“[id=“email”]”}
(Session info: chrome-headless-shell=121.0.6167.184); For documentation on this error, please visit: Understanding Common Errors | Selenium

2024-02-17 20:47:05.294652 INFO WaterUsage: Usage: 0.0

Any ideas?

If you are running on a raspberry pi / something slower / busy Home Assistant, then increase the time.sleep(x) → to something like 10-15 seconds.

Try logging into Anglian Water website, make sure there is a reading for yesterday (daily usage).
Also, you can change the execution time but Anglian water only seem to post updates for yesterday between 8-11pm.

Thank you @yatik. I’m using running in Promox on an i7 so resources are not an issue. which time.sleep do you recommend adjusting?

Increase them all by 2-3 seconds, it could be other factors… also check internet connection as all the script does is scrape the Website for data.
The other reason (weekend just gone 17-18 Feb), Anglian water were doing an upgrade.

Hi, so i can login and see details on the web site.
Readings are from 19th Feb to yesterday.
I was obviously a bit early for adding the integration.
When trying to add to the energy dashboard, i get the No Matching Statistics Found.
And un commenting the force_reading i get the image first posted about input_boolean.water_reading not found in namespace default.

If i comment it out again i get the nameError: name force_reading is not defined. If i comment the FORCED Callback section out, i have to comment out the self.listen_state as well.



Screenshot 2024-02-26 at 14.24.55
but

This is what i get trying to add the water to Energy dashboard

also getting this in the logs

2024-02-26 19:02:00.023189 INFO WaterUsage: starting process
2024-02-26 19:02:00.032045 INFO WaterUsage: starting request
2024-02-26 19:02:06.735279 INFO WaterUsage: Request Home page
2024-02-26 19:02:51.346501 INFO WaterUsage: Requested Home Page
2024-02-26 19:03:13.742273 INFO WaterUsage: Request Login
2024-02-26 19:03:26.887964 INFO WaterUsage: Login
2024-02-26 19:03:50.002040 INFO WaterUsage: Request Usage
2024-02-26 19:03:51.046865 INFO WaterUsage: Request Error: list index out of range
2024-02-26 19:03:51.424048 INFO WaterUsage: Usage: 411
2024-02-26 19:03:51.762079 WARNING AppDaemon: Excessive time spent in callback 'StartProcess() in WaterUsage', Thread 'thread.thread-1' - now complete after 111.759528 seconds (limit=80.0)

Ok, i appear to have it working.
When manually changing the time it runs to daytime for setting up.
curdt = (date.today() - timedelta(days=2)).strftime("%d-%b-%Y")
Also, i noticed when the sensor is created it is not always to sensor.water_usage_x as per the configuration file.
In my case it created the template (viewable under entities) as sensor.water_usage_3 which you need to put in the python file and helper. I was looking for _x and couldnt work out why it wasnt there.
The only issue i have now is statistics, but i guess i have to wait a few days or something.


I keep getting this message in the logs, anybody know how to get rid of the warning?

2024-03-05 21:56:00.274822 INFO AppDaemon: Loading app WaterUsage using class GetWaterUsage from module GetWaterUsage
2024-03-05 21:56:00.276621 INFO AppDaemon: Calling initialize() for WaterUsage
2024-03-05 21:56:00.346341 INFO WaterUsage: Starting Water Usage App
2024-03-05 21:56:00.350462 WARNING WaterUsage: WaterUsage: Entity input_boolean.water_reading not found in namespace default
2024-03-05 21:56:00.352155 INFO AppDaemon: App initialization complete

I have removed my old settings as could not get this working. Then I have re set it up with the new code as updated.
My logs keep showing the same error.

2024-03-06 17:55:56.886765 INFO AppDaemon: Calling initialize() for WaterUsage
2024-03-06 17:55:56.893018 INFO WaterUsage: Starting Water Usage App
2024-03-06 17:55:56.908367 WARNING WaterUsage: ------------------------------------------------------------
2024-03-06 17:55:56.909133 WARNING WaterUsage: Unexpected error running initialize() for WaterUsage
2024-03-06 17:55:56.909548 WARNING WaterUsage: ------------------------------------------------------------
2024-03-06 17:55:56.912497 WARNING WaterUsage: Traceback (most recent call last):
  File "/usr/lib/python3.11/site-packages/appdaemon/app_management.py", line 162, in initialize_app
    await utils.run_in_executor(self, init)
  File "/usr/lib/python3.11/site-packages/appdaemon/utils.py", line 304, in run_in_executor
    response = future.result()
               ^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/config/apps/GetWaterUsage.py", line 40, in initialize
    self.listen_state(self.StartProcessForced,force_reading, attribute="state", new="on", duration=10 )
                                              ^^^^^^^^^^^^^
NameError: name 'force_reading' is not defined

2024-03-06 17:55:56.917026 WARNING WaterUsage: ------------------------------------------------------------

configuration.yaml has this under template:

  - sensor:
      - name: "Water usage"
        unique_id: water_usage_x
        icon: mdi:water
        state: "0.0"

apps.yaml

hello_world:
  module: hello
  class: HelloWorld
WaterUsage:
  module: GetWaterUsage
  class: GetWaterUsage

getwaterusage.py has my details that I can log in to anglian water with and “usage sensor=sensor.water_usage_x”


Any help with this would be appreciated as I have been going around in circles.