111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
from pymodbus.client import ModbusTcpClient
|
|
import struct
|
|
import sys
|
|
from typing import Optional
|
|
|
|
# === Verbindungseinstellungen ===
|
|
MODBUS_IP = "192.168.1.112"
|
|
MODBUS_PORT = 502 # SetApp: 1502; LCD-Menü: 502 -> ggf. anpassen
|
|
UNIT_ID = 1 # Default laut Doku: 1
|
|
|
|
client = ModbusTcpClient(MODBUS_IP, port=MODBUS_PORT)
|
|
if not client.connect():
|
|
print("Verbindung fehlgeschlagen.")
|
|
sys.exit(1)
|
|
|
|
def read_regs(addr: int, count: int):
|
|
"""Hilfsfunktion: liest 'count' Holding-Register ab base-0 'addr'."""
|
|
rr = client.read_holding_registers(address=addr, count=count)
|
|
if rr.isError():
|
|
return None
|
|
return rr.registers
|
|
|
|
def read_string(addr: int, words: int) -> Optional[str]:
|
|
"""
|
|
SunSpec-Strings: ASCII, Big-Endian, 2 Bytes pro Register, 0x00 gepadded.
|
|
"""
|
|
regs = read_regs(addr, words)
|
|
if regs is None:
|
|
return None
|
|
b = b"".join(struct.pack(">H", r) for r in regs)
|
|
# SunSpec Strings sind meist mit \x00 und Spaces gepadded:
|
|
s = b.decode("ascii", errors="ignore").rstrip("\x00 ").strip()
|
|
return s or None
|
|
|
|
def to_int16(u16: int) -> int:
|
|
"""unsigned 16 -> signed 16"""
|
|
return struct.unpack(">h", struct.pack(">H", u16))[0]
|
|
|
|
def apply_sf(raw: int, sf: int) -> float:
|
|
return raw * (10 ** sf)
|
|
|
|
def read_scaled(value_addr: int, sf_addr: int) -> Optional[float]:
|
|
regs = read_regs(value_addr, 1)
|
|
sf = read_regs(sf_addr, 1)
|
|
if regs is None or sf is None:
|
|
return None
|
|
raw = to_int16(regs[0])
|
|
sff = to_int16(sf[0])
|
|
return apply_sf(raw, sff)
|
|
|
|
def read_u32_with_sf(value_addr: int, sf_addr: int) -> Optional[float]:
|
|
"""
|
|
Liest 32-bit Zähler (acc32, Big-Endian, 2 Register) + SF.
|
|
"""
|
|
regs = read_regs(value_addr, 2)
|
|
sf = read_regs(sf_addr, 1)
|
|
if regs is None or sf is None:
|
|
return None
|
|
# Big-Endian zusammenbauen:
|
|
u32 = (regs[0] << 16) | regs[1]
|
|
sff = to_int16(sf[0])
|
|
return apply_sf(u32, sff)
|
|
|
|
# ==== Common Block (base-0) ====
|
|
manufacturer = read_string(40004, 16) # C_Manufacturer
|
|
model = read_string(40020, 16) # C_Model
|
|
version = read_string(40044, 8) # C_Version
|
|
serial = read_string(40052, 16) # C_SerialNumber
|
|
|
|
print(f"Hersteller: {manufacturer}")
|
|
print(f"Modell: {model}")
|
|
print(f"Version: {version}")
|
|
print(f"Seriennummer: {serial}")
|
|
|
|
# ==== Inverter Block (base-0) ====
|
|
# AC Power + Scale Factor
|
|
ac_power = read_scaled(40083, 40084) # I_AC_Power, I_AC_Power_SF
|
|
if ac_power is not None:
|
|
print(f"AC Power: {ac_power} W")
|
|
else:
|
|
print("Fehler beim Lesen von AC Power")
|
|
|
|
# AC Spannung L-N Durchschnitt (falls 1ph/3ph mit N verfügbar) + SF
|
|
ac_voltage = read_scaled(40079, 40082) # I_AC_VoltageAN, I_AC_Voltage_SF
|
|
if ac_voltage is not None:
|
|
print(f"AC Spannung: {ac_voltage} V")
|
|
|
|
# AC Frequenz + SF
|
|
ac_freq = read_scaled(40085, 40086) # I_AC_Frequency, _SF
|
|
if ac_freq is not None:
|
|
print(f"Frequenz: {ac_freq} Hz")
|
|
|
|
# DC Power + SF
|
|
dc_power = read_scaled(40100, 40101) # I_DC_Power, _SF
|
|
if dc_power is not None:
|
|
print(f"DC Power: {dc_power} W")
|
|
|
|
# Lifetime Energy (AC_Energy_WH, acc32) + SF
|
|
lifetime_wh = read_u32_with_sf(40093, 40095) # I_AC_Energy_WH, _SF
|
|
if lifetime_wh is not None:
|
|
print(f"Lifetime Energy: {lifetime_wh} Wh")
|
|
|
|
# Status
|
|
status_regs = read_regs(40107, 2) # I_Status, I_Status_Vendor
|
|
if status_regs:
|
|
i_status = status_regs[0]
|
|
i_status_vendor = status_regs[1]
|
|
print(f"Status: {i_status} (Vendor: {i_status_vendor})")
|
|
|
|
client.close()
|