KSTAR Hybrid-Inverter mit Python und Modbus auslesen

pip install pysolarmanv5 pyyaml

import argparse
from pysolarmanv5 import PySolarmanV5
import yaml
import struct
import datetime
import sys
import time

parser = argparse.ArgumentParser(description='Solarman Script')
parser.add_argument('--debug', type=int, choices=[0, 1], default=0, help='Debug mode (0=off, 1=on)')
args = parser.parse_args()

# Debug-Ausgabe Funktion
def debug_print(message):
    if args.debug == 1:
        print(message)

# Lese die Konfigurationsdatei ein
with open('kstar_hybrid.yaml', 'r') as config_file:
    config = yaml.safe_load(config_file)

# Verbindung zu Solarman herstellen
modbus = PySolarmanV5("192.168.2.150", 1234567890)  # IP und Seriennummer des Datenloggers anpassen

debug_print("")
datum = datetime.datetime.now()
datum_formatiert = datum.strftime("%d.%m.%Y %H:%M:%S")

blocks = []
for request in config['requests']:
    start = request['start']
    end = request['end']
    blocks.append((start, end))
       

debug_print("Register auslesen...")
register_values = {}
for start, end in blocks:
    q = end - start + 1
    
    Versuch = 1    
    while Versuch <= 3:
        try:
            debug_print(f"   .... {start} bis {end} | Versuch: {Versuch} | Anzahl: {q}")
            result = modbus.read_input_registers(register_addr=start, quantity=q)
            if result:
                for i in range(q):
                    register_values[start + i] = result[i]
            break # fertig
        except Exception as e:
            Versuch += 1
            if Versuch <= 3:
                time.sleep(1)   # 1 Sekunde warten...
            else:
                debug_print(f"   Register konnten nicht ausgelesen werden!")
                sys.exit(1)  # Beendet das Skript mit einem Fehlerstatus


datumbis = datetime.datetime.now()
datumbis_formatiert = datumbis.strftime("%d.%m.%Y %H:%M:%S")
datediff = datumbis - datum
total_seconds = round(datediff.total_seconds(),0)
debug_print(f"Register ausgelesen ({total_seconds}s)")

debug_print("")
debug_print("Parameter auswerten...")


# Funktion zum Auslesen von 16-Bit-Registern und Zusammenfassen zu einem Wert mit umgekehrter Wertigkeit
def read_and_combine_registers(register_list):
    combined_value = 0
    for register in reversed(register_list):  # Umkehrung der Reihenfolge der Register
        if register in register_values:
            combined_value <<= 16
            combined_value |= register_values[register]
    return combined_value

# Funktion zum Runden eines Werts basierend auf der scale-Einstellung
def round_value(value, scale):
    if scale == 0.1:
        return round(value, 1)
    elif scale == 0.01:
        return round(value, 2)
    elif scale == 0.001:
        return round(value, 3)
    else:
        return value

# Funktion zum Konvertieren des Werts anhand der lookup-Tabelle
def convert_with_lookup(value, lookup_table):
    for entry in lookup_table:
        if entry['key'] == value:
            return entry['value']
    return value

# Funktion zum Konvertieren von Byte-Folgen in Strings mit UTF-8-Kodierung oder ASCII-Zeichen
def convert_bytes_to_string(register_list):
    byte_list = []
    for register in register_list:
        if register in register_values:
            # Extrahiere die beiden Bytes aus dem Register und füge sie zur Liste hinzu
            byte_list.extend(register_values[register].to_bytes(2, byteorder='big'))
    
    try:
        # Versuche, die Byte-Liste als UTF-8-kodierten String zu dekodieren
        string_value = bytes(byte_list).decode('utf-8')
    except UnicodeDecodeError:
        # Im Falle eines Decoding-Fehlers gib die Bytes als ASCII-Zeichen aus
        string_value = 'unbekannt'
    
    return string_value


# Iteriere über die Konfigurationsparameter
export = f"Export.Datum={datum_formatiert}|"
for parameter in config['parameters']:
    if 'group' in parameter and 'items' in parameter:
        group = parameter['group']
        items = parameter['items']

        for item in items:            
            if 'name' in item and 'registers' in item and 'scale' in item:
                name = item['name']
                registers = item['registers']
                scale = item['scale']

                export_value = ""
                if 'rule' in item and item['rule'] == 5:
                    # Wenn rule auf 5 gesetzt ist, handelt es sich um eine Byte-Folge, die einen String darstellt
                    string_value = convert_bytes_to_string(registers)
                    export_value = string_value
                else:
                    # Andernfalls handelt es sich um einen numerischen Wert
                    # Lese die Registerwerte aus und kombiniere sie zu einem Wert mit umgekehrter Wertigkeit
                    combined_value = read_and_combine_registers(registers)

                    # Berücksichtige den Scale-Faktor und runde den Wert entsprechend
                    scaled_value = round_value(combined_value * scale, scale)

                    # Wenn eine lookup-Tabelle definiert ist, konvertiere den Wert entsprechend
                    if 'lookup' in item:
                        converted_value = convert_with_lookup(scaled_value, item['lookup'])
                    else:
                        converted_value = scaled_value
                    export_value = converted_value
                
                s = f"{group}.{name}={export_value}" 
                export += s + "|"
                
                debug_print(f"   {s}")


export += f"Export.Enddatum={datumbis_formatiert}|"

debug_print("Parameter ausgewertet")

if args.debug == 0:
    print(export)

# KSTAR Hybrid Inverter
# Register laut KSTAR „MODBUS RS485 Communication Protocol V2.5“
#
# Input-Register:  3000–3660
# Holding-Register: 3200–3237
# Pro Modbus-Request max. 125 Register → Anfragen aufteilen.

requests:
  - start: 3000
    end: 3124
    mb_functioncode: 0x04
  - start: 3125
    end: 3199
    mb_functioncode: 0x04
  - start: 3200
    end: 3217
    mb_functioncode: 0x03   # Holding-Register (Systeminfos)
  - start: 3228
    end: 3249
    mb_functioncode: 0x04
  - start: 3250
    end: 3301
    mb_functioncode: 0x04

parameters:
  - group: solar
    items:
      - name: "PV1 Voltage"
        scale: 0.1
        rule: 1
        registers: [3000]
      - name: "PV1 Power"
        scale: 1
        rule: 2
        registers: [3024]
      - name: "Daily Production"
        scale: 0.1
        rule: 1
        registers: [3036]

  - group: Battery
    items:
      - name: "Battery SoC"
        scale: 0.1
        rule: 1
        registers: [3066]

  - group: Power Grid
    items:
      - name: "Daily Energy Feed-In"
        scale: 0.1
        rule: 1
        registers: [3116]

# … weitere Gruppen (Verbrauch, Phasen, Inverter-Status) analog definiert.
# rule: 1 = 16-Bit-Wert, 2 = vorzeichenbehaftet, 3 = 32-Bit (2 Register),
#       5 = String aus Byte-Folge, 6 = Bitfeld

Export.Datum=12.07.2026 10:15:00|solar.PV1 Voltage=312.4|solar.PV1 Power=1842|Battery.Battery SoC=78.5|Power Grid.Daily Energy Feed-In=12.3|Export.Enddatum=12.07.2026 10:15:04|

#!/bin/bash

cd /home/ubuntu/python

# Führe das Python-Skript aus und speichere die Ausgabe
python3 export.py > /var/www/html/export_neu.html

# Überprüfe, ob die Datei nicht leer ist
if [ -s /var/www/html/export_neu.html ]; then
    cp /var/www/html/export_neu.html /var/www/html/export.html
    echo "Die Datei export.html wurde erfolgreich aktualisiert."
else
    echo "Das Python-Skript hat kein Ergebnis zurückgegeben."
fi