A script to fetch data from a Kostal Piko 5 inverter

collect_kostal.py

#!/usr/local/bin/python

# encoding: utf-8
"""
collect_kostal.py

Created by Christian Stade-Schuldt on 2014-07-27.
"""

import urllib2
import datetime

from lxml import html
import sqlite3


URL = 'http://192.168.1.125' # use your address
USERNAME = 'your_username'
PASSWORD = 'your_password'


def get_data():
    """reads the data from the kostal inverter"""

    password_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm()

    top_level_url = URL
    username = USERNAME
    password = PASSWORD
    password_mgr.add_password(None, top_level_url, username, password)
    handler = urllib2.HTTPBasicAuthHandler(password_mgr)

    opener = urllib2.build_opener(handler)
    opener.open(URL)

    urllib2.install_opener(opener)
    response = urllib2.urlopen(URL)
    root = html.fromstring(response.read().strip())
    data = [v.text.strip() for v in root.xpath("//td[@bgcolor='#FFFFFF']")]

    current_power = data[0]
    total_energy = data[1]
    daily_energy = data[2]
    dc_u_1 = data[3]
    ac_u_1 = data[4]
    dc_i_1 = data[5]
    ac_p_1 = data[6]
    dc_u_2 = data[7]
    ac_u_2 = data[8]
    dc_i_2 = data[9]
    ac_p_2 = data[10]
    dc_u_3 = data[11]
    ac_u_3 = data[12]
    dc_i_3 = data[13]
    ac_p_3 = data[14]

    data = [dc_u_1, dc_i_1, ac_u_1, ac_p_1, dc_u_2, dc_i_2, ac_u_2, ac_p_2, dc_u_3, dc_i_3, ac_u_3, ac_p_3,
            current_power, daily_energy, total_energy]
    return data


def save_data_to_db(data):
    """saves the data tuple to the database"""
    con = sqlite3.connect("PATH_TO_SQLITE_FILE")

    con.isolation_level = None
    cur = con.cursor()

    query = '''INSERT INTO pvdata VALUES (null, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'''

    cur.execute(query, tuple([datetime.datetime.now().isoformat()] + data))

    con.commit()
    con.close()


def is_number(s):
    """checks if input is a valid number """
    try:
        int(s)
        return True
    except ValueError:
        return False


def main():
    data = get_data()
    if is_number(data[0]):
        save_data_to_db(data)


if __name__ == '__main__':
    main()

If you update the webserver on the Piko, you can do something like:

#!/bin/bash

declare -A dxsEntries
dxsEntries["dc_input"]="33556736,gauge"
dxsEntries["ac_output"]="67109120,gauge"
dxsEntries["daily_yield"]="251658754,counter"
dxsEntries["total_yield"]="251658753,counter"
dxsEntries["operating_status"]="16780032,gauge"

ip="192.168.10.70"
url="http://${ip}/api/dxs.json?"

for k in "${!dxsEntries[@]}"; do
    url="${url}&dxsEntries=${dxsEntries[$k]%%,*}"
done
url="${url/?&/?}"

output="$(curl -s "${url}")"

for k in "${!dxsEntries[@]}"; do
    metricName="piko_${k}"
    echo "# TYPE ${metricName} ${dxsEntries[$k]##*,}"
    echo -n "${metricName} "
    echo $output | jq ".dxsEntries[] | select(.dxsId==${dxsEntries[$k]%%,*}) | .value"
done

Hi there,
I have a new version of Webserver on mi Piko.
My Problem is, that I have no idea to bring the example of klausenbusk together with the collect_kostal.py of Tafkas.
Somebody has an idea. Thanks!

I solved it by myself.... It's not perfect.... but it works:

#!/usr/local/bin/python

encoding: utf-8

"""
collect_kostal.py

Created by Christian Stade-Schuldt on 2014-07-27.
Updated by Walter Grossenbacher on 2018-12-30 for new webserver
"""

import urllib2
import datetime

from lxml import html
import sqlite3

IPAdresse = '192.168.1.200'

#Leistungswerte
ID_DCEingangGesamt = '33556736' # in W
ID_Ausgangsleistung = '67109120' # in W
ID_Eigenverbrauch = '83888128' # in W
#Status
ID_Status = '16780032' # 0:Off, 1:Leerlauf, 2:Anfahren, 3:Einspeisen MPP, 4:Abgeregelt, 5:Einspeisen
#Statistik - Tag
ID_Ertrag_d = '251658754' # in Wh
ID_Hausverbrauch_d = '251659010' # in Wh
ID_Eigenverbrauch_d = '251659266' # in Wh
ID_Eigenverbrauchsquote_d = '251659278' # in %
ID_Autarkiegrad_d = '251659279' # in %
#Statistik - Gesamt
ID_Ertrag_G = '251658753' # in kWh
ID_Hausverbrauch_G = '251659009' # in kWh
ID_Eigenverbrauch_G = '251659265' # in kWh
ID_Eigenverbrauchsquote_G = '251659280' # in %
ID_Autarkiegrad_G = '251659281' # in %
ID_Betriebszeit = '251658496' # in h
#Momentanwerte - PV Genertor
ID_DC1Spannung = '33555202' # in V
ID_DC1Strom = '33555201' # in A
ID_DC1Leistung = '33555203' # in W
ID_DC2Spannung = '33555458' # in V
ID_DC2Strom = '33555457' # in A
ID_DC2Leistung = '33555459' # in W
ID_DC3Spannung = '33555714' # in V
ID_DC3Strom = '33555713' # in A
ID_DC3Leistung = '33555715' # in W
#Momentanwerte Haus
ID_HausverbrauchSolar = '83886336' # in W
ID_HausverbrauchBatterie = '83886592' # in W
ID_HausverbrauchNetz = '83886848' # in W
ID_HausverbrauchPhase1 = '83887106' # in W
ID_HausverbrauchPhase2 = '83887362' # in W
ID_HausverbrauchPhase3 = '83887618' # in W
#Netz Netzparameter
ID_NetzAusgangLeistung = '67109120' # in W
ID_NetzFrequenz = '67110400' # in Hz
ID_NetzCosPhi = '67110656'
#Netz Phase 1
ID_P1Spannung = '67109378' # in V
ID_P1Strom = '67109377' # in A
ID_P1Leistung = '67109379' # in W
#Netz Phase 2
ID_P2Spannung = '67109634' # in V
ID_P2Strom = '67109633' # in A
ID_P2Leistung = '67109635' # in W
#Netz Phase 3
ID_P3Spannung = '67109890' # in V
ID_P3Strom = '67109889' # in A
ID_P3Leistung = '67109891' # in W

def read_data(ID_Code):
"""reads the data out from the PiKo Inverter were ID_Code is the ID Code"""

URL = ('http://'+IPAdresse+'/api/dxs.json?dxsEntries='+ID_Code)
response_open = urllib2.urlopen(URL)
response = response_open.read()
FILTER_STR = 'value'
result = response.split('value":') [1]
result = result.split('}],') [0]
return(result)

def get_data():
"""prepares the data for processing"""

current_power = read_data(ID_Ausgangsleistung)
total_energy = read_data(ID_Ertrag_G)
daily_energy = read_data(ID_Ertrag_d)
dc_u_1 = read_data(ID_DC1Spannung)
ac_u_1 = read_data(ID_P1Spannung)
dc_i_1 = read_data(ID_DC1Strom)
ac_p_1 = read_data(ID_P1Leistung)
dc_u_2 = read_data(ID_DC2Spannung)
ac_u_2 = read_data(ID_P2Spannung)
dc_i_2 = read_data(ID_DC2Strom)
ac_p_2 = read_data(ID_P2Leistung)
dc_u_3 = read_data(ID_DC3Spannung)
ac_u_3 = read_data(ID_P3Spannung)
dc_i_3 = read_data(ID_DC1Strom)
ac_p_3 = read_data(ID_P3Leistung)

data = [dc_u_1, dc_i_1, ac_u_1, ac_p_1, dc_u_2, dc_i_2, ac_u_2, ac_p_2, dc_u_3, dc_i_3, ac_u_3, ac_p_3,
        current_power, daily_energy, total_energy]

return data

def save_data_to_db(data):
"""saves the data tuple to the database"""
con = sqlite3.connect("/home/pi/solarpi/dev.db")

con.isolation_level = None
cur = con.cursor()

query = '''INSERT INTO pvdata VALUES (null, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'''

cur.execute(query, tuple([datetime.datetime.now().isoformat()] + data))

con.commit()
con.close()

def is_number(s):
"""checks if input is a valid number """
try:
int(s)
return True
except ValueError:
return False

def main():
data = get_data()
if is_number(data[0]):
save_data_to_db(data)

if name == 'main':
main()

II had to fix read_data(ID_Code):

def read_data(ID_Code):
"""reads the data out from the PiKo Inverter were ID_Code is the ID Code"""

URL = ('http://'+IPAdresse+'/api/dxs.json?dxsEntries='+ID_Code)
response_open = urllib2.urlopen(URL)
response = response_open.read()
FILTER_STR = 'value'
result = response.split('value":') [1]
result = result.split('}],') [0]
result_float = float(result)
return(result_float)

@wagro ... Thank you ... your script is very helpful.
But could you say me, where I can find the codes like (ID_DC1Spannung = '33555202' # in V)
I need "Externer Stromsensor Phase 1-3 & Ladeszustand Batterie"
Thank you