#!/usr/bin/python3 import requests #from tabulate import tabulate import json import time import os from datetime import datetime from json import JSONEncoder from urllib3.exceptions import InsecureRequestWarning class ElectricalData: def __init__(self, data): self.production = [ProductionData(prod_data) for prod_data in data.get("production", [])] self.consumption = [ConsumptionData(cons_data) for cons_data in data.get("consumption", [])] self.storage = [StorageData(storage_data) for storage_data in data.get("storage", [])] def __str__(self): return f"Type: {self.type}, Active Count: {self.active_count}, Reading Time: {self.reading_time}, W Now: {self.w_now}, WH Lifetime: {self.wh_lifetime}" def to_json(self): return { "production": [prod.to_json() for prod in self.production], "consumption": [cons.to_json() for cons in self.consumption], "storage": [storage.to_json() for storage in self.storage] } class ProductionData: def __init__(self, prod_data): self.type = prod_data.get("type", "") self.active_count = prod_data.get("activeCount", 0) self.reading_time = prod_data.get("readingTime", 0) self.w_now = prod_data.get("wNow", 0) self.wh_lifetime = prod_data.get("whLifetime", 0) self.wh_today = prod_data.get("whToday",0) def __str__(self): production_str = "\n".join(str(prod) for prod in self.production) consumption_str = "\n".join(str(cons) for cons in self.consumption) storage_str = "\n".join(str(storage) for storage in self.storage) return f"Production:\n{production_str}\n\nConsumption:\n{consumption_str}\n\nStorage:\n{storage_str}" def to_json(self): return { "type": self.type, "active_count": self.active_count, "reading_time": self.reading_time, "w_now": self.w_now, "wh_lifetime": float(self.wh_lifetime), # Convert to float explicitly "wh_today": float(self.wh_today) } class ConsumptionData: def __init__(self, cons_data): self.type = cons_data.get("type", "") self.active_count = cons_data.get("activeCount", 0) self.measurement_type = cons_data.get("measurementType", "") self.reading_time = cons_data.get("readingTime", 0) self.w_now = cons_data.get("wNow", 0) self.wh_lifetime = cons_data.get("whLifetime", 0) self.varh_lead_lifetime = cons_data.get("varhLeadLifetime", 0) self.varh_lag_lifetime = cons_data.get("varhLagLifetime", 0) self.vah_lifetime = cons_data.get("vahLifetime", 0) self.rms_current = cons_data.get("rmsCurrent", 0) self.rms_voltage = cons_data.get("rmsVoltage", 0) self.react_pwr = cons_data.get("reactPwr", 0) self.apprnt_pwr = cons_data.get("apprntPwr", 0) self.pwr_factor = cons_data.get("pwrFactor", 0) self.wh_today = cons_data.get("whToday", 0) self.wh_last_seven_days = cons_data.get("whLastSevenDays", 0) self.vah_today = cons_data.get("vahToday", 0) self.varh_lead_today = cons_data.get("varhLeadToday", 0) self.varh_lag_today = cons_data.get("varhLagToday", 0) self.lines = [ElectricalLine(line_data) for line_data in cons_data.get("lines", [])] def __str__(self): lines_str = "\n".join(str(line) for line in self.lines) return f"Type: {self.type}, Active Count: {self.active_count}, Measurement Type: {self.measurement_type}, Reading Time: {self.reading_time}, W Now: {self.w_now}, WH Lifetime: {self.wh_lifetime}, Lines: {lines_str}" def to_json(self): return { "type": self.type, "active_count": self.active_count, "measurement_type": self.measurement_type, "reading_time": self.reading_time, "w_now": self.w_now, "wh_lifetime": self.wh_lifetime, "varh_lead_lifetime": self.varh_lead_lifetime, "varh_lag_lifetime": self.varh_lag_lifetime, "vah_lifetime": self.vah_lifetime, "rms_current": self.rms_current, "rms_voltage": self.rms_voltage, "react_pwr": self.react_pwr, "apprnt_pwr": self.apprnt_pwr, "pwr_factor": self.pwr_factor, "wh_today": self.wh_today, "wh_last_seven_days": self.wh_last_seven_days, "vah_today": self.vah_today, "varh_lead_today": self.varh_lead_today, "varh_lag_today": self.varh_lag_today, "lines": [line.to_json() for line in self.lines] } class StorageData: def __init__(self, storage_data): self.type = storage_data.get("type", "") self.active_count = storage_data.get("activeCount", 0) self.reading_time = storage_data.get("readingTime", 0) self.w_now = storage_data.get("wNow", 0) self.wh_now = storage_data.get("whNow", 0) self.state = storage_data.get("state", "") def __str__(self): return f"Type: {self.type}, Active Count: {self.active_count}, Reading Time: {self.reading_time}, W Now: {self.w_now}, WH Now: {self.wh_now}, State: {self.state}" def to_json(self): return { "type": self.type, "active_count": self.active_count, "reading_time": self.reading_time, "w_now": self.w_now, "wh_now": self.wh_now, "state": self.state } class ElectricalLine: def __init__(self, line_data): self.w_now = line_data.get("wNow", 0) self.wh_lifetime = line_data.get("whLifetime", 0) self.varh_lead_lifetime = line_data.get("varhLeadLifetime", 0) self.varh_lag_lifetime = line_data.get("varhLagLifetime", 0) self.vah_lifetime = line_data.get("vahLifetime", 0) self.rms_current = line_data.get("rmsCurrent", 0) self.rms_voltage = line_data.get("rmsVoltage", 0) self.react_pwr = line_data.get("reactPwr", 0) self.apprnt_pwr = line_data.get("apprntPwr", 0) self.pwr_factor = line_data.get("pwrFactor", 0) self.wh_today = line_data.get("whToday", 0) self.wh_last_seven_days = line_data.get("whLastSevenDays", 0) self.vah_today = line_data.get("vahToday", 0) self.varh_lead_today = line_data.get("varhLeadToday", 0) self.varh_lag_today = line_data.get("varhLagToday", 0) def __str__(self): return f"W Now: {self.w_now}, WH Lifetime: {self.wh_lifetime}, Varh Lead Lifetime: {self.varh_lead_lifetime}, Varh Lag Lifetime: {self.varh_lag_lifetime}, VAh Lifetime: {self.vah_lifetime}, RMS Current: {self.rms_current}, RMS Voltage: {self.rms_voltage}, React Power: {self.react_pwr}, Apparent Power: {self.apprnt_pwr}, Power Factor: {self.pwr_factor}, WH Today: {self.wh_today}, WH Last Seven Days: {self.wh_last_seven_days}, VAh Today: {self.vah_today}, Varh Lead Today: {self.varh_lead_today}, Varh Lag Today: {self.varh_lag_today}" def to_json(self): return { "w_now": self.w_now, "wh_lifetime": self.wh_lifetime, "varh_lead_lifetime": self.varh_lead_lifetime, "varh_lag_lifetime": self.varh_lag_lifetime, "vah_lifetime": self.vah_lifetime, "rms_current": self.rms_current, "rms_voltage": self.rms_voltage, "react_pwr": self.react_pwr, "apprnt_pwr": self.apprnt_pwr, "pwr_factor": self.pwr_factor, "wh_today": self.wh_today, "wh_last_seven_days": self.wh_last_seven_days, "vah_today": self.vah_today, "varh_lead_today": self.varh_lead_today, "varh_lag_today": self.varh_lag_today } class inverter(): def __init__(self,serialnumber,lastreportdate,devtype,lastreportwatts,maxreportwatts): self.serialnumber=str(serialnumber) self.lastreportdate=int(lastreportdate) self.devtype=int(devtype) self.lastreportwatts=int(lastreportwatts) self.maxreportwatts=int(maxreportwatts) def print(self): print(self.serialnumber, self.lastreportdate, self.devtype, self.lastreportwatts, self.maxreportwatts) enphasetoken = os.environ["ENPHASE_TOKEN"] enphaseurl = os.environ["ENPHASE_URL"] infdbuser = os.environ["INFLUX_USER"] infdbpass = os.environ["INFLUX_PASSWORD"] influxuri = os.environ["INFLUX_URI"] infdbuser = os.environ["INFLUX_USER"] while True: inverterlist=[] session = requests.Session() session.verify = False session.headers.update({"Authorization":"Bearer "+enphasetoken}) details = enphaseurl + "/production.json?details=1" inverteruri= enphaseurl + "/api/v1/production/inverters" mainresponse = session.get(details,verify=False).json() inverterresponse = session.get(inverteruri,verify=False).json() #print(json.dumps(mainresponse,indent=1)) for inverterdata in inverterresponse: myinverter = inverter(inverterdata['serialNumber'],inverterdata['lastReportDate'],inverterdata['devType'],inverterdata['lastReportWatts'],inverterdata['maxReportWatts']) inverterlist.append(myinverter) #print(json.dumps(response.json(),indent=1)) #for myinverter in inverterlist: #pass # myinverter.print() print(json.dumps(mainresponse,indent=1)) maindata = ElectricalData(mainresponse) influxsession = requests.Session() influxsession.verify = False influxsession.auth = (infdbuser, infdbpass) #influxsession.post( url=influxuri, data="NetConsumption value=10") measurements = "" totalproduction=0 for myinverter in inverterlist: totalproduction+=myinverter.lastreportwatts measurements += f"Production,Panel={myinverter.serialnumber} value={myinverter.lastreportwatts}\n" measurements += f"MaxProduction,Panel={myinverter.serialnumber} value={myinverter.maxreportwatts}\n" measurements += f"TotalProduction value={totalproduction}\n" measurements += f"TodayProduction value={maindata.production[1].wh_today}\n" consumptionTypes=[ "Total", "Net" ] currentconsumption=0 for constype in consumptionTypes: if constype=='Total': consindex=0 elif constype=='Net': consindex=1 linenum=0 consdata=maindata.consumption[consindex] for line in consdata.lines: measurements += f"{constype}_Leg_w_now,Leg={str(linenum)} value={line.w_now}\n" measurements += f"{constype}_Leg_wh_lifetime,Leg={str(linenum)} value={line.wh_lifetime}\n" measurements += f"{constype}_Leg_varh_lead_lifetime,Leg={str(linenum)} value={line.varh_lead_lifetime}\n" measurements += f"{constype}_Leg_varh_lag_lifetime,Leg={str(linenum)} value={line.varh_lag_lifetime}\n" measurements += f"{constype}_Leg_vah_lifetime,Leg={str(linenum)} value={line.vah_lifetime}\n" measurements += f"{constype}_Leg_rms_current,Leg={str(linenum)} value={line.rms_current}\n" measurements += f"{constype}_Leg_rms_voltage,Leg={str(linenum)} value={line.rms_voltage}\n" measurements += f"{constype}_Leg_react_pwr,Leg={str(linenum)} value={line.react_pwr}\n" measurements += f"{constype}_Leg_apprnt_pwr,Leg={str(linenum)} value={line.apprnt_pwr}\n" measurements += f"{constype}_Leg_pwr_factor,Leg={str(linenum)} value={line.pwr_factor}\n" measurements += f"{constype}_Leg_wh_today,Leg={str(linenum)} value={line.wh_today}\n" measurements += f"{constype}_Leg_wh_last_seven_days,Leg={str(linenum)} value={line.wh_last_seven_days}\n" measurements += f"{constype}_Leg_vah_today,Leg={str(linenum)} value={line.vah_today}\n" measurements += f"{constype}_Leg_varh_lead_today,Leg={str(linenum)} value={line.varh_lead_today}\n" measurements += f"{constype}_Leg_varh_lag_today,Leg={str(linenum)} value={line.varh_lag_today}\n" linenum +=1 measurements += f"{constype}_w_now value={consdata.w_now}\n" measurements += f"{constype}_wh_lifetime value={consdata.wh_lifetime}\n" measurements += f"{constype}_varh_lead_lifetime value={consdata.varh_lead_lifetime}\n" measurements += f"{constype}_varh_lag_lifetime value={consdata.varh_lag_lifetime}\n" measurements += f"{constype}_vah_lifetime value={consdata.vah_lifetime}\n" measurements += f"{constype}_rms_current value={consdata.rms_current}\n" measurements += f"{constype}_rms_voltage value={consdata.rms_voltage}\n" measurements += f"{constype}_react_pwr value={consdata.react_pwr}\n" measurements += f"{constype}_apprnt_pwr value={consdata.apprnt_pwr}\n" measurements += f"{constype}_pwr_factor value={consdata.pwr_factor}\n" measurements += f"{constype}_vah_lifetime value={consdata.vah_lifetime}\n" measurements += f"{constype}_wh_today value={consdata.wh_today}\n" measurements += f"{constype}_wh_last_seven_days value={consdata.wh_last_seven_days}\n" measurements += f"{constype}_vah_today value={consdata.vah_today}\n" measurements += f"{constype}_varh_lead_today value={consdata.varh_lead_today}\n" measurements += f"{constype}_varh_lag_today value={consdata.varh_lag_today}\n" if consindex==1: currentconsumption=consdata.w_now if currentconsumption > 0: totalexport = 0 NetConsumption=currentconsumption elif currentconsumption < 0: totalexport = - int(currentconsumption) NetConsumption=0 else: totalexport=0 NetConsumption=0 productionconsumed = maindata.production[1].w_now - totalexport measurements+= f"TotalExport value={totalexport}\n" measurements+= f"NetConsumption value={NetConsumption}\n" measurements+= f"ProductionConsumed value={productionconsumed}\n" #print(str(productionconsumed)) influxsession.post( url=influxuri, data=measurements) time.sleep(5)