dukeusage.py

#!/usr/bin/python
# Get Duke Energy electricity usage information. Needs lxml and requests. Supply info in envvars
import os
import json
import re
import datetime

import requests
import lxml.etree

sess = requests.Session()
username = os.getenv('DUKE_USERNAME')
password = os.getenv('DUKE_PASSWORD')
account_id = os.getenv('DUKE_ACCOUNTID')

def unix2dt(unx):
    return datetime.datetime(1970, 1, 1, 0, 0, 0) + datetime.timedelta(0, unx)

def login():
    resp = sess.post(
        url='https://p-scapi.duke-energy.com/form/Login/GetAccountValidationMessage',
        headers={
            'User-Agent': 'Duke Energy Usage fetcher (Python3, %s)' % username,
            'Origin': 'https://www.duke-energy.com',
        },
        data={
            'userId': username,
            'userPassword': password,
        },
        timeout=31,
    )
    #print(resp.text)

def get_usage():
    resp = sess.post(
        url='https://p-auth.duke-energy.com/form/PlanRate/GetEnergyUsage',
        headers={
            'User-Agent': 'Duke Energy Usage fetcher (Python3, %s)' % username,
            'Origin': 'https://p-auth.duke-energy.com',
            'ID': 'test',
        },
        json={
            'request': json.dumps({
                "SrcAcctId": account_id,
                "SrcAcctId2": "",
                "SrcSysCd": "ISU",
                "ServiceType": "ELECTRIC",
            })
        },
        timeout=31,
    )
    return resp.content

def parse_usage(byts):
    # remove namespaces
    byts = re.sub(b'xmlns:[^ >]*', b'', byts)
    byts = re.sub(b'<[a-z0-9]*:', b'<', byts)
    byts = re.sub(b'</[a-z0-9]*:', b'</', byts)

    # parse as xml
    root = lxml.etree.fromstring(byts)

    # find interval and compute scaling factor
    # e.g. kWh per 15min => 4x scaling to get kW used
    interval = int(root.xpath('//secondsPerInterval')[0].text)
    scale = 3600.0 / interval

    # compute the power usage
    out = []
    for elm in root.xpath('//IntervalReading'):
        tim = unix2dt(int(elm.xpath('timePeriod/start')[0].text)).isoformat()
        val = float(elm.xpath('value')[0].text)
        val *= scale
        out.append((tim, val))
    return out

def main():
    assert username
    assert password
    assert account_id
    login()
    byts = get_usage()
    with open('/tmp/dukeusage.xml', 'wb') as out:
        out.write(byts)
    usg = parse_usage(byts)
    print("\n".join("\t".join(str(s) for s in x) for x in usg))

if __name__ == "__main__":
    main()