UK "smart" energy meters

Hey UK folks.

Its time for me to renew energy contract.
Does anyone know if any of the energy providers are fitting smart meters with an open API via ethernet/wifi?
Most of them seen to have some crappy remote display, but I didn’t see anything that is open/can be integrated with yet…

TIA!

1 Like

Mine was fitted by npower and it just has a crappy remote display with is fairly useless. I wish it could be integrated into HA :frowning:

I couldn’t find a provider that provided a meter i could integrate into HA so i decided to just go out and buy my own,

is what i got and it’s been working just fine for me for the last 4 months with HA. Hope this helps. :slight_smile:

2 Likes

Thanks for the inputs @Karl_Hardy @MarkR.
I looked a bit more and I still didnt see any options other than installing via a sensor clipped round the mains supply.
Actually I used to have one of these from (IIRC) EON which had a serial out. Its in my attic I think… maybe I will dig it out and hook up to a RPi or something.

The new ‘Smart Meters’ are different from those in-house clip on monitors.
The smart meters are a replacement of the actually meters in your home.
They use radiowaves to send your usage real-time to your provider (no more reading meters!)

As a benefit you also get a handy in-house display - however, now of these are ‘open’ at the moment as the communication isn’t via traditional network infrastructure. I’m sure someone will come up with a clever way of getting the data out, but as it is pretty important data I doubt it will be easy.

I have been trying to get round this problem for some time. I came up with a way to scrape the information from my energy suppliers website. I’m with SSE I can share the code with you but it will only really work on SSE accounts. Is anyone interested in this?

1 Like

I’m with 1st utility so I would not be able to make use of it… unless they share the same backend software - something that seems to happen with banks.

I’m with SSE (both gas and electricity) and I’d love to integrate the readings with HA.

I’m away from my computer at the moment. I’ll post the code tomorrow. It will pull the data for electricity. I don’t have Gas but you should be able to adapt the code.

1 Like

Here is the code

You need to enter your email address account number and password into the python file.

create a file called weekly_energy.py add this code to it. Run it and it will create two files a json of the data direct from SSE then a json of daily and weekly totals. SSE updates the daily figure for the day before between 12pm and 7pm each day. Run the code each day to update the json file then create a sensor to read the outputted file.

#!/usr/bin/env python

import requests, lxml.html, json

s = requests.session()

login = s.get('https://my.sse.co.uk/your-account/login')
login_html = lxml.html.fromstring(login.text)
hidden_inputs = login_html.xpath(r'//form//input[@type="hidden"]')
form = {x.attrib["name"]: x.attrib["value"] for x in hidden_inputs}

account = '1234567890'
form['email'] = '[email protected]'
form['password'] = 'yourpassword'
response = s.post('https://my.sse.co.uk/your-account/login', data=form)

test = s.get('https://my.sse.co.uk/Usage/GetBarChartData?smartUsage=%7B%22accountNumber%22%3A%22' + account + '%22%2C%22granularity%22%3A%223%22%2C%22getComparisonData%22%3Afalse%2C%22isCalendarSelected%22%3Afalse%7D')

b = test.text

file = open("/home/pi/my_code/energy/weekly_energy.json", "w")
file.write(test.content)
file.close()

with open('/home/pi/my_code/energy/weekly_energy.json') as data_file:    
    data = json.load(data_file)
one = data["BarChartData"]["datasets"][0]["data"][26]
two = data["BarChartData"]["datasets"][0]["data"][25]
three = data["BarChartData"]["datasets"][0]["data"][24]
four = data["BarChartData"]["datasets"][0]["data"][23]
five = data["BarChartData"]["datasets"][0]["data"][22]
six = data["BarChartData"]["datasets"][0]["data"][21]
seven = data["BarChartData"]["datasets"][0]["data"][20]

eight = data["BarChartData"]["datasets"][0]["data"][19]
nine = data["BarChartData"]["datasets"][0]["data"][18]
ten = data["BarChartData"]["datasets"][0]["data"][17]
eleven = data["BarChartData"]["datasets"][0]["data"][16]
twelve = data["BarChartData"]["datasets"][0]["data"][15]
thirteen = data["BarChartData"]["datasets"][0]["data"][14]
fourteen = data["BarChartData"]["datasets"][0]["data"][13]
fifteen = data["BarChartData"]["datasets"][0]["data"][12]
updated = data["ListView"]["ListViewData"][0]["ReadingUsage"]

total_wk1 = one + two + three + four + five + six + seven
total_wk2 = eight + nine + ten + eleven + twelve + thirteen + fourteen

total_wk1_not_updated = two + three + four + five + six + seven + eight
total_wk2_not_updated = nine + ten + eleven + twelve + thirteen + fourteen + fifteen

if updated == "-.--":
	data = {}
	data['yesterday'] = "Not Updated"
	data['wk1'] = total_wk1_not_updated
	data['wk2'] = total_wk2_not_updated
	json_data = json.dumps(data)

	file = open("/home/pi/my_code/energy/weekly_energy_final.json", "w")
	file.write(json_data)
	file.close()
else:
	data = {}
	data['yesterday'] = one
	data['wk1'] = total_wk1
	data['wk2'] = total_wk2
	json_data = json.dumps(data)

	file = open("/home/pi/my_code/energy/weekly_energy_final.json", "w")
	file.write(json_data)
	file.close()
2 Likes

Thanks! I will give it a try!

Hi Michael, I’m with SSE so looking forward to trying it.

Just to check, have there been any breaking changes or updates to your script since this was posted?

Thanks.

It still works for this week and last week. Daily does not update anymore. I have not got around to seeing what is causing the issue.

I am also looking for the same. Now I am using EON but I am willing to switch providers for one that has a public metwrikg API. l

I have been playing around with the code for the past few months. If you have SSE than you might find my code uselful. I never got around to actually making it into a component (perhaps someone could take it from here):

Create a file called sse.py with following contents:

import requests
import requests.utils
import lxml.html
import json
import time
import re

SESSION_EXPIRED_PAGE = 'https://my.sse.co.uk/error/session-expired'
LOGIN_PAGE = 'https://my.sse.co.uk/your-account/login'
MAIN_PAGE = 'https://my.sse.co.uk/your-products'
ACCOUNTS_PAGE = 'https://my.sse.co.uk/your-usage/smart'
DATA_PAGE = 'https://my.sse.co.uk/Usage/GetBarChartData?smartUsage='
BILLS_PAGE = 'https://my.sse.co.uk/bills-and-payments'

HOURLY = '5'
DAILY = '3'
MONTHLY = '1'


def find_accounts(data):
    reg = re.compile(r'(Electricity|Gas)\s\-\s(\d+)', re.I | re.M | re.U)
    accounts = {}
    for key in data:
        account = data[key]
        match = reg.match(account)
        if match:
            accounts[match.group(1).lower()] = match.group(2)
    return accounts


def find_products(data):
    reg = re.compile(r'(Electricity|Gas)\s\-\s(\d+)', re.I | re.M | re.U)
    products = {}
    for key in data:
        account = data[key]
        match = reg.match(account)
        if match:
            products[match.group(1).lower()] = key
    return products


def parse_accounts(text):
    html = lxml.html.fromstring(text)
    options = html.xpath(r'//select[@id="Products_Value"]/option')
    return {x.attrib["value"]: x.text for x in options}


def parse_tokens(text):
    html = lxml.html.fromstring(text)
    hidden_inputs = html.xpath(r'//form//input[@type="hidden"]')
    return {x.attrib["name"]: x.attrib["value"] for x in hidden_inputs}


def parse_bills(text):
    reg1 = r'balance is [<>\w]*(\D)?([\.\d]+)[<>\w\/]+ \(([\w\s]+)\)'
    reg2 = r'payment of [<>\w]*(\D)?([\.\d]+)[<>\w\/]+ '
    reg2 += 'on [<>\w\s\=\"]*>([\d\s\w]+)[<>\w\/]+'
    balance_reg = re.compile(reg1, re.I | re.M | re.U)
    payment_reg = re.compile(reg2, re.I | re.M | re.U)
    balance_match = re.search(balance_reg, text)
    payment_match = re.search(payment_reg, text)

    if payment_match:
        payment_amount = float(payment_match.group(2))
        payment_date = payment_match.group(3)
    else:
        payment_amount = None
        payment_date = None
    if balance_match:
        balance = float(balance_match.group(2))
        balance_type = balance_match.group(3)
    else:
        balance = None
        balance_type = None

    return {
        'payment_date': payment_date,
        'payment_amount': payment_amount,
        'balance': balance,
        'balance_type': balance_type}


def parse_date(date):
    date = time.strptime(date, '%Y-%m-%dT%H:%M:%S')
    return time.mktime(date)


class SSEReceiver:
    ELECTRICITY = 'electricity'
    GAS = 'gas'

    session = None
    accounts = None
    products = None

    def __init__(self, email, password):
        self.email = email
        self.password = password
        self.session = self._get_session()
        if not self._check_login():
            raise Exception("Unable to create session")
        self._load_accounts()

    def get_data(self, key):
        months = self.get_months_data(key)
        current = months['current'] if 'current' in months else None
        previous = months['previous'] if 'previous' in months else None
        return {
            'bills': self.get_bills_data(key),
            'week': self.get_week_data(key),
            'current_month': current,
            'previous_month': previous}

    def get_bills_data(self, key):
        product = self._get_product(key)
        if product is None:
            return None
        return self._get_bills(product)

    def get_week_data(self, key):
        data = self._get_data(key, DAILY)
        return self._concat_days(data, 8)

    def get_months_data(self, key):
        data = self._get_data(key, MONTHLY)
        return self._slice_months(data)

    def _get_data(self, key, granularity):
        account = self._get_account(key)
        if account is None:
            return None
        url = self._get_data_url(account, granularity)
        response = self._request(url)
        return self._parse_data(response, granularity)

    def _concat_days(self, data, days):
        res = {
            'date': data[0]['date'] if data[0] else None,
            'usage': 0, 'amount': 0}
        for item in data[:days]:
            if item['usage'] is None:
                continue
            res['usage'] += item['usage']
            res['amount'] += item['amount']
        return res

    def _slice_months(self, data):
        if not data or not len(data):
            return None
        return {
            'current': data[0],
            'previous': data[1]}

    def _get_product(self, key):
        if self.products is None or key not in self.products:
            return None
        return self.products[key]

    def _get_account(self, key):
        if self.accounts is None or key not in self.accounts:
            return None
        return self.accounts[key]

    def _get_bills(self, product):
        response = self._post(BILLS_PAGE, {'Products.Value': product})
        return parse_bills(response)

    def _load_accounts(self):
        response = self._request(ACCOUNTS_PAGE)
        if response is None:
            return None
        accounts = parse_accounts(response)
        self.accounts = find_accounts(accounts)
        self.products = find_products(accounts)

    def _get_data_url(self, account, granularity):
        data = {
            'accountNumber': account,
            'granularity': granularity,
            'selectedDateTime': None,
            'getComparisonData': False,
            'isCalendarSelected': False}
        json_data = json.dumps(data, sort_keys=True)
        return DATA_PAGE + requests.utils.quote(json_data)

    def _get_session(self):
        return self._create_login_session()

    def _get_csrf_tokens(self, session):
        response = session.get(LOGIN_PAGE)
        time.sleep(0.3)
        if response.status_code != requests.codes.ok:
            return None
        return parse_tokens(response.text)

    def _login(self, session, csrf_tokens):
        form = {'email': self.email, 'password': self.password}
        form.update(csrf_tokens)
        response = session.post(LOGIN_PAGE, data=form)
        return response.status_code == requests.codes.ok

    def _create_login_session(self):
        session = requests.session()
        csrf_tokens = self._get_csrf_tokens(session)
        if csrf_tokens is None:
            return None
        login_status = self._login(session, csrf_tokens)
        if not login_status:
            return None
        return session

    def _check_login(self):
        return self._request(MAIN_PAGE) is not None

    def _request(self, url):
        if self.session is None:
            return None
        response = self.session.get(url)
        time.sleep(0.3)
        if response.url == LOGIN_PAGE or response.url == SESSION_EXPIRED_PAGE:
            return None
        return response.text

    def _post(self, url, data):
        if self.session is None:
            return None
        response = self.session.post(url, data=data)
        time.sleep(0.3)
        if response.url == LOGIN_PAGE or response.url == SESSION_EXPIRED_PAGE:
            return None
        return response.text

    def _parse_data(self, data, granularity):
        data = json.loads(data)
        if data is None:
            return None
        if 'Success' not in data or not data['Success']:
            return None
        if 'ListView' not in data:
            return None
        if 'ListViewData' not in data['ListView']:
            return None
        return self._parse_data_list(data['ListView']['ListViewData'])

    def _parse_data_list(self, list):
        result = []
        for item in list:
            if item['ReadingUsage'] == '-.--':
                result.append({
                    'date': parse_date(item['ReadingDateTime']),
                    'usage': None,
                    'amount': None,
                    'estimate': item['ReadingType'] == '(e)',
                })
            else:
                result.append({
                    'date': parse_date(item['ReadingDateTime']),
                    'usage': float(item['ReadingUsage']),
                    'amount': float(item['ReadingAmount'][1:]),
                    'estimate': item['ReadingType'] == '(e)',
                })
        return result

Create a file called sse_update.py next to it:

#!/usr/bin/env python
from sse import SSEReceiver
import json
import sys

try:
    sse = SSEReceiver(sys.argv[2], sys.argv[3])
    res = {'data': {
        'gas': sse.get_data(sse.GAS),
        'electricity': sse.get_data(sse.ELECTRICITY)}}
except Exception as er:
    res = {'error': '{0}'.format(er)}

with open(sys.argv[1], "w") as f:
    f.write(json.dumps(res))
    f.close()

In Homeassistant create a shell command:

update_sse: "python /config/python_scripts/sse/update_sse.py '/config/sse.json' 'YOUR_SSE_LOGIN' 'YOUR_SSE_PASSWORD'"

I have an automation which is calling it every two hours(it does take some time to run it):

- alias: "Update SSE"
  hide_entity: true
  trigger:
    - platform: time
      minutes: '/120'
      seconds: 00     
  action:
    - service: shell_command.update_sse

I than add a bunch of sensors in Homeassistant:

- platform: file
  name: "Power usage this month"
  file_path: "/config/sse.json"
  unit_of_measurement: "kWh"
  value_template: "{{ value_json.data.electricity.current_month.usage }}"

- platform: file
  name: "Power cost this month"
  file_path: "/config/sse.json"
  value_template: "£{{ value_json.data.electricity.current_month.amount }}"

- platform: file
  name: "Gas usage this month"
  file_path: "/config/sse.json"
  unit_of_measurement: "kWh"
  value_template: "{{ value_json.data.gas.current_month.usage }}"

- platform: file
  name: "Gas cost this month"
  file_path: "/config/sse.json"
  value_template: "£{{ value_json.data.gas.current_month.amount }}"

- platform: file
  name: "Power usage last month"
  file_path: "/config/sse.json"
  unit_of_measurement: "kWh"
  value_template: "{{ value_json.data.electricity.previous_month.usage }}"

- platform: file
  name: "Power cost last month"
  file_path: "/config/sse.json"
  value_template: "£{{ value_json.data.electricity.previous_month.amount }}"

- platform: file
  name: "Gas usage last month"
  file_path: "/config/sse.json"
  unit_of_measurement: "kWh"
  value_template: "{{ value_json.data.gas.previous_month.usage }}"

- platform: file
  name: "Gas cost last month"
  file_path: "/config/sse.json"
  value_template: "£{{ value_json.data.gas.previous_month.amount }}"
2 Likes

I know that this is an old thread but, I have got a UK smart meter installed about a year ago now and have just got around to taking a more in-depth look at how it works, I have found that the smart meter the Sottish Power installed for me was the Elster AM110R Enhanced Credit meter. and looking at the pdf i found on google (below)

It says that it uses a ZigBee Smart Energy Profile v1.1 interface for the Home Area Network (HAN).

Now I was wondering if i got a ZigBee receiver module would it be able to pick up the HAN from the meter? Dose anyone have any experience of this and then is it possible in import to HA ?

1 Like

:point_up_2:t2: I’ve come here thinking the same thing…

I had a “smart” meter install yesterday and noticed a setting for ‘Channel’ in the settings, which was set to 25 – “smells like ZigBee”, I thought – lo, and behold, it is. (Annoyingly, I now have to look at mitigating interference with my hue bulbs, and Xiaomi presence sensors **sigh**).

I’m curious if there is any way to sniff this data similar to how Zigbee2MQTT operates? This is is probably naïve thinking without looking into the actual transport mechanism (encryption, etc.), but worth a ponder…

I also have one of the smart meters but I think we are out of luck with the Zigbee stuff. My understanding is that they have dilerberarly limited it to connect only to the the dumb display they ship. I had a couple of ideas that I’ve not yet had a chance to explore. One is to point a raspi camera at the display and do some OCR on it. The second was to open it up and see if there is a management port somewhere. ( Although I will probably break it at that point! )

I constantly turn down invites of upgrading to a smart-meter because I currently have a Loop Energy monitoring kit installed (works only on analog gas meters) and is a great option if people want to integrate their gas/electric with HA.

It has the added benefit of working with Uswitch. Meaning when it’s time to renew I just go to Uswitch via the Loop account and it puts in my true usage over the last year.
Their support staff are brilliant too. I had my gas meter monitor die (battery I imagine) and they sent out a new one with a pre-paid return envelope for the dead one so they could diagnose the issue.

1 Like

Do the ‘smart’ meters still have the flashing red light, like the old ones do? I mounted a light sensor over the top of mine, and then time the intervals between the pulses. That way you can calculate how many Watts are in use and publish it to MQTT.