retr0
391e615893
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>
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
#!/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)
|