Get Dominion Energy hourly usage data, then save to SQLite. Login with Selenium+Chrome

dominion-energy-usage-sync.py

import datetime
import logging
import logging.config
import os
import pickle
import pytz
import requests
import sqlite3
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

logging.config.dictConfig(
    {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "standard": {"format": "%(asctime)s [%(levelname)s] %(message)s"},
        },
        "handlers": {
            "default": {
                "level": "INFO",
                "formatter": "standard",
                "class": "logging.StreamHandler",
            },
        },
        "loggers": {"": {"handlers": ["default"], "level": "INFO", "propagate": True}},
    }
)

logger = logging.getLogger(__name__)

LOGIN_PAGE_URL = "https://login.dominionenergy.com/CommonLogin?SelectedAppName=electric"
API_BASE_URL = "https://prodsvc-dominioncip.smartcmobile.com/Service/api/1"
WAIT_TIMEOUT = 10
TIMEZONE = "US/Eastern"  # Required to convert local time
SAVEDATA = "data.db"

options = webdriver.ChromeOptions()
options.add_argument("--headless")
options.add_argument("--window-size=1920,1080")
browser = webdriver.Chrome(options=options)


def dump_page(content, filename="error-debug.html"):
    with open(filename, "w") as f:
        f.write(content)


def dump_screenshot(content, filename):
    with open(filename, "wb") as f:
        f.write(content)


access_token = None
account_number = None


def login():
    """
    Login to Dominion Energy website. Save access token and account number to global variables.
    """

    logger.info("Logging in...")

    if os.path.exists("credential.pkl"):
        logger.info("Loading credentials from file...")
        with open("credential.pkl", "rb") as f:
            credentials = pickle.load(f)
        username = credentials["username"]
        password = credentials["password"]
    else:
        logger.info("No credentials found. Please enter your username and password.")
        username = input("Username: ")
        password = input("Password: ")
        with open("credential.pkl", "wb") as f:
            pickle.dump({"username": username, "password": password}, f)

    # Open login page
    browser.get(LOGIN_PAGE_URL)

    try:
        login_button = WebDriverWait(browser, WAIT_TIMEOUT).until(
            EC.element_to_be_clickable(
                (
                    By.XPATH,
                    "//div[@id='login-container_content']//input[@type='submit']",
                )
            )
        )
        username_input = WebDriverWait(browser, WAIT_TIMEOUT).until(
            EC.presence_of_element_located(
                (
                    By.XPATH,
                    "//div[@id='login-container_content']//input[@name='username']",
                )
            )
        )
        password_input = WebDriverWait(browser, WAIT_TIMEOUT).until(
            EC.presence_of_element_located(
                (
                    By.XPATH,
                    "//div[@id='login-container_content']//input[@name='password']",
                )
            )
        )

        username_input.send_keys(username)
        password_input.send_keys(password)
        login_button.click()

        # wait for redirect url
        WebDriverWait(browser, WAIT_TIMEOUT).until(EC.title_contains("Home"))

        # Get access token in cookies
        global access_token
        access_token = browser.get_cookie("accessToken")
        if access_token is None:
            raise Exception("Failed to login. Please check your credentials.")
        access_token = access_token["value"]
        logger.debug(f"Access token: {access_token}")
        logger.info("Logged in!")

        # Wait for account information to load
        logger.info("Getting account information...")

        WebDriverWait(browser, WAIT_TIMEOUT).until(
            EC.presence_of_element_located(
                (By.XPATH, "//div[@id='billPaymentAccountDropdown']//ul/li")
            )
        )

        # Get account number in local storage
        global account_number
        account_number = browser.execute_script(
            "return localStorage.getItem('ACCOUNT_NUMBER')"
        )
        if account_number is None:
            raise Exception("Failed to get account number.")
        logger.info(f"Account number: {account_number}")

    except Exception as e:
        dump_page(browser.page_source)
        raise e


def get_hourly_usage(date: str) -> list:
    """
    Get hourly usage data for a specific date.
    :param date: Date in format `YYYY-MM-DD`, e.g. "2024-02-20"
    :return: List of hourly usage data, e.g.
            [{
                "accountNumber": "225999013117",
                "meterNumber": "000000000468943103",
                "consumption": "0.86",
                "tier": "",
                "uom": "",
                "readDate": "2/20/2024 12:00:00 AM",
                "unitGenerated": "0",
                "netUnit": "0",
                "demandKW": "0"
            }, ...]
    """

    logger.info(f"Getting hourly usage for {date}...")
    url = f"{API_BASE_URL}/Usage/UsageData"
    params = {
        "accountNumber": account_number,
        "EndDate": date,
        "ActionCode": 4,
    }
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
    }

    # Send with requests
    response = requests.get(url, params=params, headers=headers)
    response.raise_for_status()

    # Parse response
    data = response.json()
    logger.debug(data["status"])
    logger.info(f"Obtained {len(data['data']['electricUsages'])} hourly usage data.")

    return data["data"]["electricUsages"]


def date_unix_timestamp(local_time_str: str, tz_str: str = TIMEZONE) -> float:
    """
    Convert freedom time string to unix timestamp.
    :param local_time_str: Local time string in format `MM/DD/YYYY hh:mm:ss AM/PM`, e.g. "2/20/2024 12:00:00 AM"
    :param tz_str: Timezone string, e.g. "US/Eastern"
    :return: Unix timestamp, e.g. 1700000000.0
    """

    dt_naive = datetime.datetime.strptime(local_time_str, "%m/%d/%Y %I:%M:%S %p")
    dt_localized = pytz.timezone(tz_str).localize(dt_naive)

    return dt_localized.timestamp()


def format_hourly_usage(hourly_usage: list) -> list:
    """
    Format hourly usage data.
    :param hourly_usage: List of hourly usage data
    :return: List of formatted hourly usage data, e.g.
            [{
                "time": 1700000000.0,
                "account_number": "225999013117",
                "meter_number": "000000000468943103",
                "consumption": 0.86,
                "tier": None,
                "uom": None,
                "unit_generated": 0,
                "net_unit": 0,
                "demand_kw": 0
            }, ...]
    """

    def process_usage(usage):
        try:
            return {
                "time": date_unix_timestamp(usage["readDate"]),
                "account_number": usage["accountNumber"],
                "meter_number": usage["meterNumber"],
                "consumption": float(usage["consumption"]),
                "tier": usage["tier"] or None,
                "uom": usage["uom"] or None,
                "unit_generated": float(usage["unitGenerated"]),
                "net_unit": float(usage["netUnit"]),
                "demand_kw": float(usage["demandKW"]),
            }
        except Exception:
            # Skip this usage if any error occurs.
            logger.warning(f"Failed to process usage: {usage}")
            return None

    return [
        item
        for item in (process_usage(usage) for usage in hourly_usage)
        if item is not None
    ]


def persist_hourly_usage(hourly_usage: list):
    """
    Persist hourly usage data to database.
    :param hourly_usage: List of formatted hourly usage data
    """

    logger.info(f"Persisting hourly usage data to {SAVEDATA}...")

    conn = sqlite3.connect(SAVEDATA)
    cursor = conn.cursor()

    # Create table if not exists
    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS hourly_usage (
            time REAL PRIMARY KEY,
            account_number TEXT,
            meter_number TEXT,
            consumption REAL,
            tier TEXT,
            uom TEXT,
            unit_generated REAL,
            net_unit REAL,
            demand_kw REAL
        )
        """
    )
    conn.commit()

    # Insert records, or replace if already exists
    cursor.executemany(
        """
        INSERT OR REPLACE INTO hourly_usage (time, account_number, meter_number, consumption, tier, uom, unit_generated, net_unit, demand_kw)
        VALUES (:time, :account_number, :meter_number, :consumption, :tier, :uom, :unit_generated, :net_unit, :demand_kw)
        """,
        hourly_usage,
    )
    conn.commit()


if __name__ == "__main__":
    login()
    browser.quit() # Browser is for login only, move on to use good old requests

    # Get hourly usage data for yesterday (since today's data may not be available yet)
    yesterday = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
    hourly_usage = get_hourly_usage(yesterday)
    formatted_hourly_usage = format_hourly_usage(hourly_usage)
    persist_hourly_usage(formatted_hourly_usage)
    logger.info("Done!")

requirements.txt

selenium
requests
pytz
sqlite3