dukeusage.py

#!/usr/bin/env python3
# Get duke energy hourly electricity usage
# before running:  pip3 install requests
# use env vars to set username, password, and lookback days
# e.g.  [email protected] DUKE_PASSWORD=hunter11 DUKE_DAYS_BACK=14 python dukeusage.py

import os
import datetime
import json
import re
import requests

sess = requests.Session()
sess.headers = {"User-Agent": "dukeusage.py/1.0.0 ([email protected])"}

def divide_time_interval(start_time, end_time, chunk_size=3600*24*7, overlap_size=0):
    """Break a datetime range into fixed size chunks (with overlap).

    Args:
      start_time: datetime of start
      end_time: datetime of end
      chunk_size: chunk size in seconds (def. 7 days)
      overlap_size: overlap in seconds (def. 0 minutes)
    Returns:
      list of (start_time, end_time) pairs that cover the range
    """
    # Avoid infinite loops
    assert chunk_size > 0
    assert chunk_size > overlap_size

    chunk_size = datetime.timedelta(0, chunk_size)
    overlap_size = datetime.timedelta(0, overlap_size)

    jobs = []
    front = start_time
    back = start_time
    while back < end_time:
        back = front + chunk_size
        back = min(back, end_time)
        jobs.append((front, back))
        front = back - overlap_size

    return jobs

def login():
    un = os.getenv("DUKE_USERNAME")
    pw = os.getenv("DUKE_PASSWORD")
    if not un or not pw:
        raise ValueError("Must specify DUKE_USERNAME and DUKE_PASSWORD env vars")

    # Log in
    sess.post(
        "https://www.duke-energy.com/form/SignIn/GetAccountValidationMessage",
        data={
            "userId":un,
            "userPassword":pw,
            "pageId":"416298df-33d4-4347-9d27-08f66d922a96",
            "deviceprofile":"mobile",
        }
    )

def find_electric_meters():
    # Find available meters
    mresp = sess.get("https://www.duke-energy.com/my-account/usage-analysis")
    m = re.search(r'(?mis)<duke-dropdown id="usageAnalysisMeter".*?items="(\[.*?\])".*?</duke-dropdown>', mresp.text)
    mjs = json.loads(m.group(1).replace("&quot;", '"'))

    meters = [x['value'] for x in mjs]

    # filter to electric only
    meters = [m for m in meters if m.startswith('ELECTRIC')]

    return meters

def get_hourly_usage(meter, week):
    # Get usage for a week
    wk = week.strftime("%m / %d / %Y")
    resp = sess.post(
        "https://www.duke-energy.com/api/UsageAnalysis/GetUsageChartData",
        json={
            "MeterNumber": meter,
            "Date": wk,
            "Graph": "hourlyEnergyUse",
            "BillingFrequency": "Week",
            "GraphText": "Hourly Energy Usage ",
            "ActiveDate": "01/01/2000",
        },
        headers={"Accept": "application/json, text/plain, */*"},
    )

    js = resp.json()

    return list(zip(
        js['graphDates'], js['meterData']['Electric'],
    ))

def main():
    login()

    meter = find_electric_meters()[0]

    days_back = int(os.getenv('DUKE_DAYS_BACK') or 7)

    today = datetime.datetime.now()
    start = today - datetime.timedelta(days_back)

    usage = []
    for _, back in divide_time_interval(start, today):
        usg = get_hourly_usage(meter, back)
        usage.extend(usg)
    usage.sort()
    print("\n".join("%s\t%s" % (a, b) for (a, b) in usage))

if __name__ == "__main__":
    main()