Feature: ShineDiag — portabler Vor-Ort-Diagnose-Gateway (Pi 3B)
Flask-App + mobile Web UI für Diagnose vor Ort ohne HAOS/MQTT. Pi 3B: eth0 → ShineLAN-X (DHCP), wlan0 → Hotspot "ShineDiag". Browser auf http://10.0.1.1: Modell wählen, alle Sensoren auslesen, Rohdaten-Register-Dump, Export als JSON. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,110 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ShineDiag — Vor-Ort-Diagnose für Growatt-Wechselrichter via ShineLAN-X."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
|
||||
from flask import Flask, jsonify, request, send_from_directory
|
||||
|
||||
# Shared code aus dem Add-on
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../haos-addon/src"))
|
||||
from inverters import INVERTERS
|
||||
from modbus_client import ModbusReader
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
WEB_DIR = os.path.join(os.path.dirname(__file__), "web")
|
||||
SHINELANX_IP = os.environ.get("SHINELANX_IP", "10.0.0.100")
|
||||
|
||||
app = Flask(__name__, static_folder=WEB_DIR)
|
||||
|
||||
|
||||
@app.get("/")
|
||||
def index():
|
||||
return send_from_directory(WEB_DIR, "index.html")
|
||||
|
||||
|
||||
@app.get("/api/models")
|
||||
def api_models():
|
||||
return jsonify({
|
||||
k: {"id": v.id, "name": v.name, "manufacturer": v.manufacturer,
|
||||
"sensor_count": len(v.sensors)}
|
||||
for k, v in INVERTERS.items()
|
||||
})
|
||||
|
||||
|
||||
@app.post("/api/scan")
|
||||
def api_scan():
|
||||
"""Liest alle definierten Sensoren eines Geräts aus."""
|
||||
body = request.get_json(force=True) or {}
|
||||
model_id = body.get("model", "MIC_1500_TL_X")
|
||||
ip = body.get("ip", SHINELANX_IP)
|
||||
port = int(body.get("port", 502))
|
||||
slave = int(body.get("slave", 1))
|
||||
|
||||
inverter = INVERTERS.get(model_id)
|
||||
if not inverter:
|
||||
return jsonify({"error": f"Unbekanntes Modell: {model_id}"}), 400
|
||||
|
||||
reader = ModbusReader(host=ip, port=port, slave=slave, timeout=5.0)
|
||||
values = reader.read(inverter)
|
||||
reader.close()
|
||||
|
||||
if values is None:
|
||||
return jsonify({"ok": False, "error": "Keine Verbindung zum ShineLAN-X"}), 502
|
||||
|
||||
sensors_out = []
|
||||
for s in inverter.sensors:
|
||||
sensors_out.append({
|
||||
"id": s.id,
|
||||
"name": s.name,
|
||||
"value": values.get(s.id),
|
||||
"unit": s.unit,
|
||||
"device_class": s.device_class,
|
||||
"icon": s.icon,
|
||||
})
|
||||
|
||||
return jsonify({"ok": True, "model": inverter.name,
|
||||
"manufacturer": inverter.manufacturer, "sensors": sensors_out})
|
||||
|
||||
|
||||
@app.post("/api/raw")
|
||||
def api_raw():
|
||||
"""Liest einen rohen Register-Bereich aus — für Diagnose unbekannter Werte."""
|
||||
body = request.get_json(force=True) or {}
|
||||
ip = body.get("ip", SHINELANX_IP)
|
||||
port = int(body.get("port", 502))
|
||||
slave = int(body.get("slave", 1))
|
||||
start = max(0, int(body.get("start", 0)))
|
||||
count = min(125, max(1, int(body.get("count", 100))))
|
||||
|
||||
from pymodbus.client import ModbusTcpClient
|
||||
client = ModbusTcpClient(ip, port=port, timeout=5.0)
|
||||
if not client.connect():
|
||||
return jsonify({"ok": False, "error": "Verbindung fehlgeschlagen"}), 502
|
||||
|
||||
try:
|
||||
result = client.read_input_registers(start, count, slave=slave)
|
||||
if result.isError():
|
||||
return jsonify({"ok": False, "error": str(result)}), 502
|
||||
regs = []
|
||||
for i, val in enumerate(result.registers):
|
||||
regs.append({
|
||||
"addr": start + i,
|
||||
"hex": f"0x{val:04X}",
|
||||
"dec": val,
|
||||
"signed": val if val < 0x8000 else val - 0x10000,
|
||||
})
|
||||
return jsonify({"ok": True, "registers": regs})
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
port = int(os.environ.get("PORT", "80"))
|
||||
log.info("ShineDiag startet auf Port %d — ShineLAN-X: %s", port, SHINELANX_IP)
|
||||
app.run(host="0.0.0.0", port=port, threaded=True)
|
||||
Reference in New Issue
Block a user