Python script to graph the real-time power/energy and voltage data from a Fronius inverter

froniusPV_graph.py

#! /bin/python
#
# F R O N I U S P V _ G R A P H . P Y
#
# Last Modified on Thu Apr 17 20:47:00 2025
#
# Code to graph the real-time power/energy and voltage data from a Fronius inverter
#
######################################################################################
# This code started out as AI (ChatGPT) generated code.
# Prompt to ChatGPT:
#
# write a program to graph the output voltage and power of a
# Fronius Primo Solar Inverter in real time and update the graph on
# the screen every 10 seconds.
#
# ChatGPT response was; -
# Key Points:
# 1. Mock Data: The get_inverter_data function simulates inverter data. Replace it with your
#    actual data fetching logic.
# 2. Real-Time Update: The animation.FuncAnimation updates the graph every 10 seconds.
# 3. Data Handling: The program keeps only the last 30 data points for clarity in visualization.
#
# You may need to install the matplotlib library if it’s not already available. Run the following
# command to install it:
# pip install matplotlib
######################################################################################
#
# The Mock data was replaced with some Copilot AI generated code, that actually does fetch
# data from a Fronius inverter using the solar_api specification.
#
# Seems to work OK on Win 10 using python 3.13.1
# but didn't seem stable on Ubuntu 22.04 ??
#
# 0v6 defaults to display half an hour of data,
#     NB: -f is now -g option and -E selects show Energy [Wh]

import requests
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import time
from datetime import datetime
import sys  # exit()
import getopt  # getopt()

code_info = "froniusPV_graph.py 0v6, Apr 2025"
inverter_ip = "192.168.1.100"  # IP address of Fronius Device
base_url = "/solar_api/v1/"  # Expected version of API to use
estimatedTimeToGetDataAndGraphIt = 230 # Guess time taken in mSec

options = {
    "count": int(1),
    "debug": False,
    "energy": False,
    "file": "froniusPV_data",
    "graph": "froniusPV_graph",
    "help": False,
    "points": int(181),  # default points limit on graph
    "power": False,  # default to showing Power (W) unless -E or --energy is used
    "verbose": False,
    "wait": int(10),  # interval between updates defaults to 10 sec
}


def usage():
    print(
        "Usage:\n%s [-cX][-D][-E][-fA.Z][-gA.Z][-h][-pX][-P][-v][-wX] [targetFronius]"
        % sys.argv[0]
    )
    print(" where; -\n   -cX              do count graph updates (not yet implemented)")
    print("   -D or --debug    prints out Debug information")
    print("   -E or --energy   Show Energy (Wh), not Power (W) in graph")
    print("   -fA.Z            specify file name to save data")
    print("   -gA.Z            specify file name to save last graph")
    print("   -h or --help     outputs this usage message")
    print(f"   -pX              display X points in graph [10 <= X <= 360]")
    print("   -P or --power    Show Power (W), not Energy (Wh) in graph")
    print("   -v or --verbose  prints verbose output")
    print(
        f"   -wX              wait X sec [4 <= X <= 60] instead of default ({options['wait']}) sec to update graph"
    )
    print("   targetFronius is either the name or IP address of the fronius device")
    print(" E.g.; -")
    print(f"   python {sys.argv[0]} -v fronius.local")


# Get options and arguments from the command line
def processCommandLine():
    try:
        opts, args = getopt.getopt(
            sys.argv[1:],
            "c:DEf:g:hp:Pvw:",
            [
                "",         # count
                "debug",
                "energy",
                "",         # file
                "",         # graph
                "help",
                "",         # points
                "power",
                "verbose",
                "",         # wait
            ],
        )
    except getopt.GetoptError as err:
        print(str(err))
        usage()
        sys.exit(1)  # Indicate unsuccessful to shell, if there is one.
    for o, a in opts:
        if o in ("-c", "--count"):
            options["count"] = int(a)
            if options["count"] < 1:
                options["count"] = 1
        elif o in ("-D", "--debug"):
            options["debug"] = True
        elif o in ("-E", "--energy"):
            options["energy"] = True
        elif o in ("-f", "--file"):
            options["file"] = a
        elif o in ("-g", "--graph"):
            options["graph"] = a
        elif o in ("-h", "--help"):
            options["help"] = True
        elif o in "-p":
            options["points"] = int(a)      # Get Points limit for graph
            if options["points"] < 10:
                options["points"] = 10
            elif options["points"] > 360:   # limit to 60 minutes of data at default 10 sec b/w points
                options["points"] = 361
        elif o in ("-P", "--power"):
            options["power"] = True
        elif o in ("-v", "--verbose"):
            options["verbose"] = True
        elif o in "-w":
            options["wait"] = int(a)    # Get Seconds between sample points
            if options["wait"] < 4:     # Don't allow smaller than 4 sec between fronius get calls
                options["wait"] = 4     # Fastest sampling rate with min points equals a 40 sec graph
            elif options["wait"] > 60:  # Don't allow greater than 60 sec between fronius get calls
                options["wait"] = 60    # Slowest sampling rate with max points equals a 6 hours graph
    if options["debug"]:
        options["verbose"] = True  # Debug implies verbose output
    return args


# Get Fronius inverter data. Data is expected in JSON format.
def get_inverter_data():
    status_code = 254
    power = -1
    powerUnit = "W"  # Default to Energy
    day_power = -1
    day_powerUnit = "Wh"  # Default to Energy
    voltage = -1
    voltageUnit = "V"
    if options["debug"]:
        print(f"Debug: get_inverter_data(): status_code is {status_code}")
    try:
        response = requests.get(
            f"http://{inverter_ip}{base_url}GetInverterRealtimeData.cgi?Scope=Device&DeviceId=1&DataCollection=CommonInverterData",
            timeout=5,
            verify=True,
        )
        response.raise_for_status()
    except requests.exceptions.HTTPError as errh:
        print("HTTP Error")
        print(errh.args[0])
        raise SystemExit(errh)
    except requests.exceptions.Timeout as errrt:
        print("Connection Time-out Error")
        raise SystemExit(errrt)
    except requests.exceptions.ConnectionError as conerr:
        print("Connection Error")
        raise SystemExit(conerr)
    except KeyboardInterrupt:
        raise
    except requests.exceptions.RequestException as e:
        raise SystemExit(e)

    # See if we got a response with valid JSON
    try:
        data = response.json()
    except requests.exceptions.JSONDecodeError as jsonerr:
        print("\nError: JSON Decode Error in response from Network Device.\n")
        print(f"Warning: No Data available : {jsonerr}.")
    else:
        if options["debug"]:
            print(data)         # output the json response data if in debug mode
        if "Head" in data and "Status" in data["Head"]:
            status_code = data["Head"]["Status"]["Code"]
        if status_code != 0:
            if "Head" in data and "Status" in data["Head"]:
                status_reason = data["Head"]["Status"]["Reason"]
            print(
                f"Warning: {status_reason}: Returned Data is not trustworthy (status code: {status_code})."
            )
        if "Body" in data and "Data" in data["Body"]:
            if "PAC" in data["Body"]["Data"]:
                power = data["Body"]["Data"]["PAC"]["Value"]
                powerUnit = data["Body"]["Data"]["PAC"]["Unit"]
            else:
                print("Warning: No Power data available.")
            if "DAY_ENERGY" in data["Body"]["Data"]:
                day_power = data["Body"]["Data"]["DAY_ENERGY"]["Value"]
                day_powerUnit = data["Body"]["Data"]["DAY_ENERGY"]["Unit"]
            else:
                print("Warning: No Day Energy data available.")
            if "UAC" in data["Body"]["Data"]:
                voltage = data["Body"]["Data"]["UAC"]["Value"]
                voltageUnit = data["Body"]["Data"]["UAC"]["Unit"]
            else:
                print("Warning: No AC Voltage data available.")
            current_time = datetime.now().strftime("%H:%M:%S")
            print(
                f"{current_time}: Day Energy = {day_power} {day_powerUnit}, AC Power = {power} {powerUnit}, AC Voltage = {voltage} {voltageUnit}"
            )
        else:
            print("Warning: No Data available.")
    return status_code, voltage, power, day_power


# Start up and get command line configs if there are any
args = processCommandLine()
if options["debug"]:
    print()
if options["debug"] or options["verbose"]:
    print(f"{code_info}")
if options["debug"]:
    print("\nGraph data from a networked Fronius Inverter")
    print(
        '\nDebug: "%s" Python script running on system type "%s"'
        % (sys.argv[0], sys.platform)
    )
    print(f"Debug: Argument List {sys.argv}")
    print(f"Debug: Python version {sys.version}")
    print(f"Debug: {options['points']} Graph Points, {options['wait']} sec b/w Points")

# If help is requested then print usage() and exit
if options["help"]:
    usage()
    sys.exit(1)

# Attempt 1st contact with Fronius device
if len(args) > 0:
    inverter_ip = args[
        0
    ]  # First non-switch is assumed to be an IP address or Network name
try:
    response = requests.get(
        f"http://{inverter_ip}/solar_api/GetAPIVersion.cgi",
        timeout=4,
        verify=True,
    )
except requests.exceptions.Timeout as errrt:
    print(
        f"\nError: Connection Time-out Error (Probably no Network Device at {inverter_ip}?)\n"
    )
    raise SystemExit(errrt)
except requests.exceptions.ConnectionError as conerr:
    print(
        f"\nError: Connection Error (Probably no HTTP enabled Network Device at {inverter_ip}?)\n"
    )
    raise SystemExit(conerr)
except requests.exceptions.RequestException as e:
    raise SystemExit(e)

# Determine API version if possible
if response.status_code == 404:  # Possibly Version 0 API ?
    base_url = "/solar_api/"
    print(
        f"Warning: Fronius device did not responded to API version request; Defaulting to: {base_url}"
    )
else:
    # See if we got a response with valid JSON
    try:
        data = response.json()
    except requests.exceptions.JSONDecodeError as jsonerr:
        print(
            "\nError: JSON Decode Error in response from Network Device (Probably not a Fronius?)\n"
        )
        raise SystemExit(jsonerr)

    if "BaseURL" in data:
        base_url = data["BaseURL"]
        if options["verbose"]:
            print(f"Fronius device responded to API version request with: {base_url}")
    else:
        print(
            f"Warning: Fronius device did not responded to API version request; Defaulting to: {base_url}"
        )
    if options["debug"]:
        print(data)

# Initialize lists to store time, voltage, and power values.
timestamps = []
voltages = []
powers = []
day_powers = []
timeRelativeToMostRecentReading = []

# Create a figure and two subplots for voltage and power.
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))

# Configure the voltage subplot.
ax1.set_title("Output Voltage")
ax1.set_xlabel("Time relative to last reading [Sec]")
ax1.set_ylabel("Voltage [V]")
ax1.grid(True)

# Configure the Power/Energy subplot.
if options["energy"]:
    ax2.set_title("Day Energy")
    ax2.set_ylabel("Energy [Wh]")
else:
    ax2.set_title("Output Power")
    ax2.set_ylabel("Power [W]")
ax2.set_xlabel("Time relative to last reading [Sec]")
ax2.grid(True)

# Update function for real-time graphing.
def update(frame):
    # Get current time and inverter data.
    current_time = time.time()
    result_code, voltage, power, day_power = get_inverter_data()

    # Append the data to the lists if it is valid data.
    if result_code == 0:
        timestamps.append(current_time)
        voltages.append(voltage)
        powers.append(power)
        day_powers.append(day_power)

    # Limit the lists to the last -p X (defaults to 30 if not specified) points for clarity.
    if len(timestamps) > options["points"]:
        timestamps.pop(0)
        voltages.pop(0)
        powers.pop(0)
        day_powers.pop(0)

    # Calculate X Axis values relative to most recent reading
    lastIndex = len(timestamps) - 1
    timeRelativeToMostRecentReading = [t - timestamps[lastIndex] for t in timestamps]

    # Clear and replot the voltage subplot.
    ax1.clear()
    ax1.plot(
        timeRelativeToMostRecentReading,
        voltages,
        label="Voltage [V]",
        color="blue",
    )
    ax1.set_title("Output Voltage [V]")
    ax1.set_xlabel("Time relative to last reading [Sec]")
    ax1.set_ylabel("Voltage [V]")
    ax1.grid(True)

    # Clear and replot the power subplot.
    ax2.clear()
    if options["energy"]:
        ax2.plot(
            timeRelativeToMostRecentReading,
            day_powers,
            label="Energy [Wh]",
            color="red",
        )
        ax2.set_title("Day Energy [Wh]")
        ax2.set_ylabel("Energy [Wh]")
    else:
        ax2.plot(
            timeRelativeToMostRecentReading,
            powers,
            label="Power [W]",
            color="red",
        )
        ax2.set_title("Output Power [W]")
        ax2.set_ylabel("Power [W]")
    ax2.set_xlabel("Time relative to last reading [Sec]")
    ax2.grid(True)


# Make sure we trap keyboard ^C and exit relatively cleanly.
try:
    # Set up animation to update periodically (defaults to 10 seconds).
    ani = animation.FuncAnimation(
        fig, update, interval=options["wait"] * 1000 - estimatedTimeToGetDataAndGraphIt, save_count=1
    )

    # Display the graph.
    plt.tight_layout()
    plt.show()

except KeyboardInterrupt:
    plt.savefig(f"{options['graph']}.png")
    print(f"Program terminated and last graph saved as \"{options['graph']}.png\".")