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 }}"