Enphase Envoy Local Data - SwiftBar/xBar Menu Bar Script

enphase_average.5m.py

#! /usr/bin/env python3

# This script is explained here: <https://sixcolors.com/post/2024/10/putting-my-solar-system-in-my-menu-bar/>
# Adapted from [email protected] <https://github.com/grzegorz914/homebridge-enphase-envoy>

# <xbar.title>Enphase Solar (Averages)</bitbar.title>
# <xbar.version>v1.2</xbar.version>
# <xbar.author>Jason Snell</xbar.author>
# <xbar.author.github>jasonsnell</xbar.author.github>
# <xbar.desc>Display local Enphase solar stats.</xbar.desc>
# <swiftbar.hideAbout>true</swiftbar.hideAbout>
# <swiftbar.hideRunInTerminal>true</swiftbar.hideRunInTerminal>
# <swiftbar.hideLastUpdated>false</swiftbar.hideLastUpdated>
# <swiftbar.hideDisablePlugin>true</swiftbar.hideDisablePlugin>
# <swiftbar.hideSwiftBar>false</swiftbar.hideSwiftBar>


import json
import requests
import os
import re
from datetime import datetime, timedelta

# Define the file to store the run history for averaging purposes - I put mine in iCloud Documents 
history_file = os.path.expanduser('~/Library/Mobile Documents/com~apple~CloudDocs/Documents/enphase_history.json')

# Define the URL of your local enphase envoy device. envoy.local works in many cases.
envoy_ip = "https://envoy.local"

# Your max panel output in kilowatts on its sunniest day; this lets it estimate a % efficiency
max_KW = 6.7

# Your enphase API token - get one at https://entrez.enphaseenergy.com/entrez_tokens
# NOTE: You need to start typing your system name into the first box before it'll autocomplete.
# You'll need to get a new one every 12 months

bearer_token = "[your bearer token goes here]"

def auto_format_power(val):
    val = val / 1000 # convert from milliwats
    if -50 < val < 50:
        return f"{val:.1f} W"
    elif 50 <= val < 1000000 or -1000000 <= val <= -50:
        return f"{val / 1000:.1f} kW"
    elif val <= -1000000 or val >= 1000000:
        return f"{val / 1000000:.1f} MW"

def auto_format_energy(val):
    if -1000 < val < 1000:
        return f"{val:.1f} Wh"
    elif 1000 <= val < 1000000 or -1000000 <= val <= -1000:
        return f"{val / 1000:.3f} kWh"
    elif val <= -1000000 or val >= 1000000:
        return f"{val / 1000000:.3f} MWh"


# Set up the headers with the Bearer token
headers = {
    "Authorization": f"Bearer {bearer_token}"
}

# Disable warnings related to self-signed certificates (not recommended for production)
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)

# Check status

def fetch_solar_data():

    # Enable stream
    try:
        uri = f"{envoy_ip}/ivp/livedata/status"    
        response = requests.get(uri, headers=headers, verify=False)
        response.raise_for_status()
        envoyStatus = response.json()
        streamEnabled = envoyStatus['connection']['sc_stream']

        if streamEnabled == "disabled":
            # Live streaming is not enabled right now, so enable it and re-load envoyStatus
            uri = f"{envoy_ip}/ivp/livedata/stream"
            enableData = { "enable": 1 }
            response = requests.post(uri, headers=headers, json=enableData, verify=False)
            response.raise_for_status()
            uri = f"{envoy_ip}/ivp/livedata/status"        
            response = requests.get(uri, headers=headers, verify=False)
            response.raise_for_status()
            envoyStatus = response.json()
            # print(envoyStatus)        
            # streamEnabled = envoyStatus['connection']['sc_stream']
            # lastUpdate = envoyStatus['meters']['last_update']            

        batteryCharge = envoyStatus["meters"]["soc"]
        pvMilliwatts = envoyStatus["meters"]["pv"]["agg_p_mw"]
        storageMilliwatts = envoyStatus["meters"]["storage"]["agg_p_mw"]
        gridMilliwatts = envoyStatus["meters"]["grid"]["agg_p_mw"]
        loadMilliwatts = envoyStatus["meters"]["load"]["agg_p_mw"]
    
        # Add a timestamp
        timestamp = datetime.now().isoformat()
        
        return {
            "batteryCharge": batteryCharge,
            "pvMilliwatts": pvMilliwatts,
            "storageMilliwatts": storageMilliwatts,
            "gridMilliwatts": gridMilliwatts,
            "loadMilliwatts": loadMilliwatts,
            "timestamp": timestamp
        }
    except Exception as e:
        # print(f"Error fetching data: {e}")
        print("Error fetching data.")
        return None

# Function to load history from file
def load_history():
    if os.path.exists(history_file):
        with open(history_file, 'r') as file:
            return json.load(file)
    return []

# Function to save updated history to file
def save_history(history):
    with open(history_file, 'w') as file:
        json.dump(history, file)

# Function to update the history with new data
def update_history(new_data):
    history = load_history()
    
    # Append new data
    history.append(new_data)
    
    # Keep only the last 5 entries
    if len(history) > 5:
        history = history[-5:]
    
    # Save updated history
    save_history(history)

    return history

# Function to filter data from the last hour
def filter_on_time(history):
    time_ago = datetime.now() - timedelta(minutes=20)
    
    # Filter the history to include only the data from the last hour
    recent_history = [entry for entry in history if datetime.fromisoformat(entry["timestamp"]) > time_ago]
    
    return recent_history

# Function to calculate the average of the last hour's data
def calculate_averages(history):
    avg_data = {}
    if not history:
        return avg_data

    keys = ["batteryCharge", "pvMilliwatts", "storageMilliwatts", "gridMilliwatts", "loadMilliwatts"]
    for key in keys:
        avg_data[key] = sum(entry[key] for entry in history) / len(history)
    
    return avg_data


# Fetch new data
new_data = fetch_solar_data()

if new_data:
    # Update history with the new data
    history = update_history(new_data)
    
    # Filter history to include only data from the last hour
    recent_history = filter_on_time(history)
    
    # Calculate averages for data within the last hour
    if recent_history:
        avg_data = calculate_averages(recent_history)

        # Display the averaged data
        pvAverage = avg_data.get('pvMilliwatts', 0)
        batteryAverage = round(avg_data.get('batteryCharge', 0))
        storageAverage = avg_data.get('storageMilliwatts', 0)
        gridAverage = avg_data.get('gridMilliwatts', 0)
        loadAverage = avg_data.get('loadMilliwatts', 0)
        
        # figure out sun icon
        
        if pvAverage > 0:
            genPercent = round(pvAverage/(max_KW*1000)) # what percent of max generation is being generated
            if genPercent > 60:
                sunIcon = ":sun.max.circle:"
            elif genPercent > 30:
                sunIcon = ":cloud.sun.circle:"
            else:
                sunIcon = ":sun.horizon.circle:"
        
        # figure out battery icon
        
        if batteryAverage > 90:
            batteryIcon = ":battery.100percent:"
        elif batteryAverage > 70:
            batteryIcon = ":battery.75percent:"
        elif batteryAverage > 40:
            batteryIcon = ":battery.50percent:"
        elif batteryAverage > 20:
            batteryIcon = ":battery.25percent:"
        else:
            batteryIcon = ":battery.0percent:"

        if -100000 < gridAverage < 100000:
            gridStatus = (":arrow.left.arrow.right:") # no grid activity worth mentioning
        elif gridAverage < 0:
            gridStatus = (":arrowshape.left.circle.fill: " + auto_format_power(abs(gridAverage))) # grid export
        else:
            gridStatus = (":arrowshape.right.circle.fill: " + auto_format_power(gridAverage)) # grid import
        
        
        # this is where the menu bar presentation happens

        # if sun, show sun icon and grid status
        if pvAverage > 0:
            print (":sun.max.fill: " + auto_format_power(pvAverage) + ' ', end='')

        # if grid activity, show it
        if abs(gridAverage) > 100000:
            print (gridStatus + ' ', end='')

        # show home power usage
        print(":bolt.fill: " + auto_format_power(loadAverage) + ' ', end='')

        # if the battery is actively charging or discharging, show battery status icon
        if abs(storageAverage) > 100000:
            print (" " + batteryIcon, end='')

        # end of menu bar presentation
        print("\n---")
        
        # battery status line output
        print (batteryIcon + " " + str(batteryAverage), end='')
        if storageAverage < -100000:
             print("% Charging at " + auto_format_power(abs(storageAverage)))
        elif storageAverage > 100000:
            print ("% Discharging at " + auto_format_power(storageAverage))
        else:
            print ("%")
        
        if pvAverage > 0:
            genPercent = round(pvAverage/67000) # what percent of 6.7kW max
            print (":sun.max.fill: Generating " + auto_format_power(pvAverage) + " (" + str(genPercent) + "%)")
        
        if gridAverage < -100000:
            print(":arrowshape.left.circle.fill: Exporting " + auto_format_power(abs(gridAverage))) # grid export
        elif gridAverage > 100000:
            print(":arrowshape.right.circle.fill: Importing " + auto_format_power(gridAverage)) # grid import
        
        print (":bolt.fill: Using " + auto_format_power(loadAverage)) # how much being used
        
    else:
        print("No data from the last hour available.")