Hallo zusammen,
ich habe mir in der Bucht einen Sensor für meinen Wasserzähler gekauft "Komplettsystem Reed-Kontakt ESPHome für Wasserzähler wie ZENNER MNK", welcher über WLAN auf einer Webseite (SSE Streaming) den Zählerstand für ESPHome zur Verfügung stellt. (soll keine Werbung sein, bin nicht beteiligt ;-) )
Hierfür habe ich - zugegeben, mit Unterstützung von Copilot - ein Plugin für SmartHomeNG erstellt.
Ich stelle das hier mal ein. Evtl. kann jemand damit was anfangen. Ich habe nichts anderes gefunden. Vielleicht gibt es auch elegantere Lösungen, aber es funktioniert.
Wenn nicht, einfach ignorieren
.
/plugins/esphome_sse/__init__.py:
/plugins/esphome_sse/plugin.yaml:
Eintrag /etc/plugin.yaml:
item:
ich habe mir in der Bucht einen Sensor für meinen Wasserzähler gekauft "Komplettsystem Reed-Kontakt ESPHome für Wasserzähler wie ZENNER MNK", welcher über WLAN auf einer Webseite (SSE Streaming) den Zählerstand für ESPHome zur Verfügung stellt. (soll keine Werbung sein, bin nicht beteiligt ;-) )
Hierfür habe ich - zugegeben, mit Unterstützung von Copilot - ein Plugin für SmartHomeNG erstellt.
Ich stelle das hier mal ein. Evtl. kann jemand damit was anfangen. Ich habe nichts anderes gefunden. Vielleicht gibt es auch elegantere Lösungen, aber es funktioniert.
Wenn nicht, einfach ignorieren
./plugins/esphome_sse/__init__.py:
Code:
import threading
import requests
import json
import time
from lib.model.smartplugin import SmartPlugin
class ESPHomeSSE(SmartPlugin):
PLUGIN_VERSION = "1.1.0"
def __init__(self, sh, *args, **kwargs):
super().__init__()
self._sh = sh
self._host = self.get_parameter_value('host')
self._port = self.get_parameter_value('port')
self._events_path = self.get_parameter_value('events_path')
self._base_url = f"http://{self._host}:{self._port}{self._events_path}"
self.logger.info(f"ESPHomeSSE Plugin initialisiert, URL: {self._base_url}")
self._items_by_id = {}
self._alive = False
self._thread = None
def run(self):
self._alive = True
self._thread = threading.Thread(target=self._sse_loop, name="ESPHomeSSE", daemon=True)
self._thread.start()
self.logger.info("ESPHomeSSE Plugin gestartet")
def stop(self):
self._alive = False
self.logger.info("ESPHomeSSE Plugin gestoppt")
def parse_item(self, item):
esphome_id = item.conf.get('esphome_id')
if esphome_id:
self.logger.info(f"Item {item.id()} mit esphome_id={esphome_id} registriert")
self._items_by_id[esphome_id] = item
return None
def _sse_loop(self):
self.logger.warning("ESPHome SSE Thread gestartet")
while self._alive:
try:
self.logger.debug(f"Verbinde zu {self._base_url}")
with requests.get(
self._base_url,
headers={"Accept": "text/event-stream"},
stream=True,
timeout=60
) as r:
if r.status_code != 200:
self.logger.warning(f"SSE HTTP Status {r.status_code}, warte 5s")
time.sleep(5)
continue
current_event = None
for raw_line in r.iter_lines(decode_unicode=True):
if not self._alive:
break
if not raw_line:
continue
# Beispiel:
# event: state
if raw_line.startswith("event:"):
current_event = raw_line.split(":", 1)[1].strip()
continue
# Beispiel:
# data: {"id":"sensor-water", ...}
if raw_line.startswith("data:"):
data_str = raw_line[5:].strip()
# leere data-Zeilen ignorieren
if not data_str:
continue
# Nur "state"-Events verarbeiten
if current_event != "state":
continue
try:
payload = json.loads(data_str)
except Exception as e:
self.logger.warning(f"Konnte JSON nicht parsen: {data_str} ({e})")
continue
sensor_id = payload.get("id")
value = payload.get("value")
if sensor_id in self._items_by_id:
item = self._items_by_id[sensor_id]
try:
fvalue = float(value)
except Exception:
self.logger.warning(f"Value nicht float-konvertierbar: {value}")
continue
item(fvalue, self.get_shortname())
self.logger.info(f"Item {item.id()} von {sensor_id} aktualisiert: {fvalue}")
except Exception as e:
self.logger.warning(f"SSE Fehler: {e}, neuer Versuch in 5s")
time.sleep(5)
Code:
plugin:
plugin_name: esphome_sse
class_name: ESPHomeSSE
description: "ESPHome SSE Plugin fuer Wasserzaehler"
author: "ooUrmeloo"
version: "1.0"
sh_minversion: "1.9"
multi_instance: False
parameters:
host:
type: str
description: "IP oder Hostname des ESPHome Geraets"
default: "192.168.178.167"
port:
type: int
description: "Port des ESPHome HTTP Servers"
default: 80
events_path:
type: str
description: "Pfad fuer SSE Events"
default: "/events"
item_attributes:
esphome_id:
type: str
description: "ID des ESPHome Sensors (z.B. sensor-water)"
Code:
esphome_sse:
plugin_name: esphome_sse
class_name: ESPHomeSSE
plugin_enabled: true
host: 192.168.xxx.xxx
port: 80
events_path: /events
Code:
wasserzaehler:
zaehler:
type: num
esphome_id: sensor-water
Kommentar