solaredge_exporter/solaredge_api.py

#!/usr/bin/python3

import bottle
import datetime
import logging
import requests
from collections import defaultdict
from sys import stderr

api_key = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
site_id = 4143190

inverter = '7B0E5700-E0'

endpoint = "https://monitoringapi.solaredge.com/"

query_delta = datetime.timedelta(minutes=20)
freshness_delta = datetime.timedelta(minutes=20)
cache = {}

numeric_modes = {
    'OFF': 0,
    'SLEEPING': 1,
    'STARTING': 2,
    'MPPT': 3,
    'SHUTTING_DOWN': 4,
    'FAULT': 5,
    'STANDBY': 6,
    'LOCKED': 7,
}

metric_data = {
    "dc_voltage_volts": ["gauge", "Input bus voltage"],
    "meter_energy_watthours_total": ["counter", "Power meter"],
    "temperature_celsius": ["gauge", "Temperature"],
    "inverter_mode": ["gauge", "Inverter state (0=off, 3=producing, 5=fault)"],
    "ac_phase_volts": ["gauge", "Output bus phase voltage"],
    "ac_current_amps": ["gauge", "Output bus current"],
    "ac_voltage_volts": ["gauge", "Output bus voltage"],
    'ac_frequency_hertz': ["gauge", "Output bus frequency"],
    'ac_apparent_power_watts': ["gauge", "Apparent AC power"],
    'ac_active_power_watts': ["gauge", "Active AC power"],
    'ac_reactive_power_watts': ["gauge", "Reactive AC power"],
    'ac_cos_phi': ["gauge", "AC Phase factor"] 
}

last_ts = {}

def is_fresh(date):
    return (datetime.datetime.now()) - date < freshness_delta

def numeric_mode(s):
    if s in numeric_modes:
        return numeric_modes[s]
    if s.startswith('LOCKED'):
        return 7

def time(t):
    return t.strftime("%F %T")

def ptime(s):
    return datetime.datetime.strptime(s, "%Y-%m-%d %H:%M:%S")

def get(path, params={}, **kw):
    r = requests.get(endpoint + path, params | { "api_key": api_key }, **kw)
    r.raise_for_status()
    return r.json() | { 'fetch_time': datetime.datetime.now().timestamp() }
    
def details():
    return get(f"site/{site_id}/details.json")

def inventory():
    return get(f"site/{site_id}/inventory.json")

def get_inverters():
    return [ i['SN'] for i in inventory()['Inventory']['inverters'] ]

def energy(**kw):
    return get(f"site/{site_id}/energyDetails.json",
    kw | {"timeUnit": "QUARTER_OF_AN_HOUR"})

def tech_data(**kw):
    return get(f"equipment/{site_id}/{inverter}/data.json", kw | {"timeUnit": "QUARTER_OF_AN_HOUR"})

def meters(**kw):
    return get(f"site/{site_id}/meters.json", kw | {"timeUnit": "QUARTER_OF_AN_HOUR"})

def flow(**kw):
    return get(f"site/{site_id}/currentPowerFlow.json", kw)

def latest(method, timefn, **kw):
    name = method.__name__
    if name in cache and is_fresh(timefn(cache[name])):
        logging.info("Cache for {} is fresh", method.__name__)
        return cache[name]

    logging.info("Cache for {} is stale, requesting", method.__name__)
    end = datetime.datetime.now()
    start = end - query_delta
    data = method(startTime=time(start), endTime=time(end), **kw)
    cache[name] = data
    return data

def tech_check(c):
        if c['data']['telemetries']:
            return ptime(c['data']['telemetries'][-1]['date'])
        else:
            return datetime.datetime.fromtimestamp(c['fetch_time'])


def meters_check(c):
    return ptime(c['meterEnergyDetails']['meters'][0]['values'][-1]['date'])

def collect():
    tech = latest(tech_data, tech_check)['data']['telemetries']
    ms = latest(meters, meters_check)['meterEnergyDetails']['meters']
    if len(tech):
        point = tech[-1]
        date = ptime(point['date'])
        yield (date, 'dc_voltage_volts', {}, point.get('dcVoltage'))
        yield (date, 'meter_energy_watthours_total', {'meter': 'production'}, point.get('totalEnergy'))
        yield (date, 'temperature_celsius', {}, point.get('temperature'))
        yield (date, 'inverter_mode', {}, numeric_mode(point.get('inverterMode')))
        yield (date, 'ac_phase_volts', {'phase':'1-2'}, point.get('vL1To2'))
        yield (date, 'ac_phase_volts', {'phase':'2-3'}, point.get('vL2To3'))
        yield (date, 'ac_phase_volts', {'phase':'3-1'}, point.get('vL3To1'))

        for p in ('L1Data','L2Data','L3Data'):
            phase = point[p]
            yield (date, 'ac_current_amps', {'phase': p[1:2] }, phase.get('acCurrent'))
            yield (date, 'ac_voltage_volts', {'phase': p[1:2] }, phase.get('acVoltage'))
            yield (date, 'ac_frequency_hertz', {'phase': p[1:2] }, phase.get('acFrequency'))
            yield (date, 'ac_apparent_power_watts', {'phase': p[1:2] }, phase.get('apparentPower'))
            yield (date, 'ac_active_power_watts', {'phase': p[1:2] }, phase.get('activePower'))
            yield (date, 'ac_reactive_power_watts', {'phase': p[1:2] }, phase.get('reactivePower'))
            yield (date, 'ac_cos_phi', {'phase': p[1:2] }, phase.get('cosPhi'))
    else:
        yield (datetime.datetime.now(), 'inverter_mode', {}, 1)

    for m in ms:
        point = m['values'][-1]
        date = ptime(point['date'])
        yield (date, 'meter_energy_watthours_total', {'meter': m['meterType'].lower()}, point['value'])

def format_metrics(entries, timestamps=False):
    collected = defaultdict(list)
    for entry in entries:
        collected[entry[1]].append(entry)

    for metric in collected:
        md = metric_data.get(metric)
        if md is not None:
            yield "# TYPE {} {}\n".format(metric, md[0])
            yield "# HELP {} {}\n".format(metric, md[1])

        for (time, metric, attrs, value) in collected[metric]:
            time_s = int(time.timestamp() * 1000)
            attr_s = ""
            if attrs:
                attr_s = "{" + ','.join('{}="{}"'.format(k, attrs[k]) for k in attrs ) + "}"
            if timestamps:
                yield f"{metric} {attr_s} {value} {time_s}\n"
            else:
                yield f"{metric} {attr_s} {value}\n"

@bottle.route('/')
def root():
    return "SolarEdge prometheus exporter"

@bottle.route('/metrics')
def metrics():
    return format_metrics(collect())

def main():
    bottle.run(host='localhost', port=9150)


if __name__ == '__main__':
    main()