Python Nerdery: Data Scraping for Home Assistant

One thing I want to monitor in Home Assistant isn’t supported in any traditional way. I want to know how much propane our generator has, and I only have a local provider website to work with. Enter Python to load a browser, log me in, grab the data and send it to home assistant. Add the script to cron and you are done.

import logging
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException
from webdriver_manager.chrome import ChromeDriverManager
import requests
import json

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Constants
LOGIN_URL = 'https://account.suburbanpropane.com/'
HOME_URL = 'https://account.suburbanpropane.com/user-home'
HOME_ASSISTANT_URL = 'http://192.168.1.2:8123/api/states/'
HOME_ASSISTANT_TOKEN = 'XXXXXXXXX'
ENTITY_ID_PERCENTAGE = 'sensor.propane_tank_level_percentage'
ENTITY_ID_GALLONS = 'sensor.propane_tank_level_gallons'

def setup_driver():
    chrome_options = Options()
    chrome_options.add_argument("--headless")
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-dev-shm-usage")
    
    service = Service(ChromeDriverManager().install())
    return webdriver.Chrome(service=service, options=chrome_options)

def login(driver, username, password):
    driver.get(LOGIN_URL)
    try:
        WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "email_check")))
        driver.find_element(By.ID, "email_check").send_keys(username)
        driver.find_element(By.ID, "password_check").send_keys(password)
        driver.find_element(By.ID, "cmdLogin").click()
        
        WebDriverWait(driver, 10).until(EC.url_contains("/user-home"))
        return True
    except (TimeoutException, NoSuchElementException) as e:
        logging.error(f"Login failed: {e}")
        return False

def get_propane_levels(driver):
    driver.get(HOME_URL)
    try:
        gauge_value = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.CLASS_NAME, "c3-gauge-value"))
        )
        percentage = float(gauge_value.text.strip('%'))
        
        summary_element = driver.find_element(By.ID, "tank_gauge_summary")
        gallons = float(summary_element.text.split()[3])
        
        return percentage, gallons
    except (TimeoutException, NoSuchElementException) as e:
        logging.error(f"Failed to retrieve propane levels: {e}")
        return None, None

def update_home_assistant(entity_id, state, attributes):
    headers = {
        'Authorization': f'Bearer {HOME_ASSISTANT_TOKEN}',
        'Content-Type': 'application/json',
    }
    payload = {
        'state': state,
        'attributes': attributes
    }
    try:
        response = requests.post(f"{HOME_ASSISTANT_URL}{entity_id}", headers=headers, json=payload)
        response.raise_for_status()
        return True
    except requests.RequestException as e:
        logging.error(f"Failed to update Home Assistant for {entity_id}: {e}")
        return False

def main():
    driver = setup_driver()
    try:
        if login(driver, 'user@username.com', 'yourpassword'):
            percentage, gallons = get_propane_levels(driver)
            
            if percentage is not None and gallons is not None:
                if update_home_assistant(ENTITY_ID_PERCENTAGE, percentage, {
                    'unit_of_measurement': '%',
                    'friendly_name': 'Propane Tank Level Percentage'
                }):
                    logging.info(f"Updated propane level percentage: {percentage}%")
                
                if update_home_assistant(ENTITY_ID_GALLONS, gallons, {
                    'unit_of_measurement': 'gallons',
                    'friendly_name': 'Propane Tank Level Gallons'
                }):
                    logging.info(f"Updated propane level gallons: {gallons} gallons")
            else:
                logging.error("Failed to retrieve propane levels")
        else:
            logging.error("Login failed, unable to proceed with scraping")
    except Exception as e:
        logging.error(f"An error occurred: {e}")
    finally:
        driver.quit()

if __name__ == "__main__":
    main()