From 5677bae9dec9e9be5188a8ae2135e1c10dd1e0e1 Mon Sep 17 00:00:00 2001 From: Eduard Iten Date: Thu, 2 Jul 2026 13:47:22 +0200 Subject: [PATCH] =?UTF-8?q?Gateway=20hinzugef=C3=BCgt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gateway/.gitignore | 220 ++++++++++++++++++++++++++++++ gateway/config.yaml | 13 ++ gateway/config.yaml.example | 13 ++ gateway/gateway/__init__.py | 0 gateway/gateway/controller.py | 219 +++++++++++++++++++++++++++++ gateway/gateway/mqtt_client.py | 121 ++++++++++++++++ gateway/gateway/mqtt_discovery.py | 158 +++++++++++++++++++++ gateway/gateway/udp_listener.py | 140 +++++++++++++++++++ gateway/main.py | 175 ++++++++++++++++++++++++ gateway/pooltemp.service | 12 ++ gateway/requirements.txt | 2 + 11 files changed, 1073 insertions(+) create mode 100644 gateway/.gitignore create mode 100644 gateway/config.yaml create mode 100644 gateway/config.yaml.example create mode 100644 gateway/gateway/__init__.py create mode 100644 gateway/gateway/controller.py create mode 100644 gateway/gateway/mqtt_client.py create mode 100644 gateway/gateway/mqtt_discovery.py create mode 100644 gateway/gateway/udp_listener.py create mode 100644 gateway/main.py create mode 100644 gateway/pooltemp.service create mode 100644 gateway/requirements.txt diff --git a/gateway/.gitignore b/gateway/.gitignore new file mode 100644 index 0000000..b3ec7d5 --- /dev/null +++ b/gateway/.gitignore @@ -0,0 +1,220 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[codz] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py.cover +*.lcov +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +# Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +# poetry.lock +# poetry.toml + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. +# https://pdm-project.org/en/latest/usage/project/#working-with-version-control +# pdm.lock +# pdm.toml +.pdm-python +.pdm-build/ + +# pixi +# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. +# pixi.lock +# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one +# in the .venv directory. It is recommended not to include this directory in version control. +.pixi/* +!.pixi/config.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule* +celerybeat.pid + +# Redis +*.rdb +*.aof +*.pid + +# RabbitMQ +mnesia/ +rabbitmq/ +rabbitmq-data/ + +# ActiveMQ +activemq-data/ + +# SageMath parsed files +*.sage.py + +# Environments +.env +.envrc +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +# .idea/ + +# Abstra +# Abstra is an AI-powered process automation framework. +# Ignore directories containing user credentials, local state, and settings. +# Learn more at https://abstra.io/docs +.abstra/ + +# Visual Studio Code +# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore +# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore +# and can be added to the global gitignore or merged into this file. However, if you prefer, +# you could uncomment the following to ignore the entire vscode folder +# .vscode/ +# Temporary file for partial code execution +tempCodeRunnerFile.py + +# Ruff stuff: +.ruff_cache/ + +# PyPI configuration file +.pypirc + +# Marimo +marimo/_static/ +marimo/_lsp/ +__marimo__/ + +# Streamlit +.streamlit/secrets.toml diff --git a/gateway/config.yaml b/gateway/config.yaml new file mode 100644 index 0000000..7bbea66 --- /dev/null +++ b/gateway/config.yaml @@ -0,0 +1,13 @@ +mqtt: + broker: "10.0.10.110" + port: 1883 + user: "simulator" + password: "simulator" + base_topic: "homeassistant" + sensor_topic: "pool_temp" + +gateway: + udp_in_port: 6969 + udp_out_port: 6969 + log_level: "DEBUG" + debounce_timeout: 1 diff --git a/gateway/config.yaml.example b/gateway/config.yaml.example new file mode 100644 index 0000000..cbf47fe --- /dev/null +++ b/gateway/config.yaml.example @@ -0,0 +1,13 @@ +mqtt: + broker: "ip of your broker" + port: 1883 + user: "username" + password: "password" + base_topic: "homeassistant" + sensor_topic: "pool_temp" + +gateway: + udp_in_port: 6969 + udp_out_port: 6969 + log_level: "DEBUG" + debounce_timeout: 30 diff --git a/gateway/gateway/__init__.py b/gateway/gateway/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/gateway/gateway/controller.py b/gateway/gateway/controller.py new file mode 100644 index 0000000..5bfce15 --- /dev/null +++ b/gateway/gateway/controller.py @@ -0,0 +1,219 @@ +import logging +import threading +import struct + +class Controller: + def __init__(self, config): + self.config = config + self.logger = logging.getLogger("Controller") + + self.nodes = {} + self.downlink_seq = 0 # Globaler Zähler für ausgehende UDP-Befehle + + self.udp_port = self.config.get("gateway", {}).get("udp_port", 6969) + self.debounce_timeout = self.config.get("gateway", {}).get("debounce_timeout", 15) + + self.udp_listener = None + self.mqtt_client = None + + def start(self): + self.logger.info("Initialisiere Sub-Module...") + + from .mqtt_client import MQTTClient + self.mqtt_client = MQTTClient(self, self.config) + self.mqtt_client.start() + + from .udp_listener import UDPListener + self.udp_listener = UDPListener(self, self.udp_port) + self.udp_listener.start() + + self.logger.info("Controller aktiv und bereit.") + + def _ensure_node_exists(self, uuid): + if uuid not in self.nodes: + self.nodes[uuid] = { + "lost_frame_sync": False, + "local_lost_counter": 0, + "global_lost_counter": 0, + "last_seq": None, + "ip": None, + "fallback_timer": None, + # Speicher für die aktuellen Slider-Sollwerte (Standardwerte als Fallback) + "config_shadow": { + "meas_interval": 5, # 5 Minuten + "temp_delta": 500, # 500 m°C (0.5°C) + "max_interval": 60, # 60 Minuten + "batt_interval": 24 # 24h + }, + "debounce_timer": None, + "alive_timer": None # --- NEU: Hält den Lebenszeichen-Timer --- + } + t = threading.Timer(2.0, self.mqtt_client.trigger_fallback_sync, args=[uuid]) + self.nodes[uuid]["fallback_timer"] = t + t.start() + + def _kick_alive_timer(self, uuid): + """Setzt den Lebenszeichen-Timer zurück und berechnet das Timeout dynamisch.""" + node = self.nodes[uuid] + + # Laufenden Timer stoppen, falls vorhanden + if node["alive_timer"] is not None: + node["alive_timer"].cancel() + + shadow = node["config_shadow"] + + # Deine Format-Formel: max_interval + 2 * meas_interval + # Da du aktuell in Sekunden testest, rechnen wir hier direkt 1:1 + timeout_seconds = (shadow["max_interval"] + (2 * shadow["meas_interval"])) * 60 + + # Sicherheits-Untergrenze (falls Slider auf 0 stehen, damit der Loop nicht durchdreht) + if timeout_seconds < 5: + timeout_seconds = 10 + + # Timer starten. Wenn er abläuft, wird das Node offline gemeldet + node["alive_timer"] = threading.Timer(timeout_seconds, self._node_timeout_triggered, args=[uuid]) + node["alive_timer"].start() + self.logger.debug(f"[{uuid}] Lebenszeichen registriert. Timeout neu gesetzt auf {timeout_seconds}s") + + def _node_timeout_triggered(self, uuid): + """Wird aufgerufen, wenn vom Node zu lange kein UDP-Paket mehr kam.""" + self.logger.warning(f"[{uuid}] LEBENSZEICHEN-TIMEOUT! Seit maximalem Intervall nichts gehört.") + if self.mqtt_client: + topic = f"{self.mqtt_client.sensor_topic}/{uuid}/availability" + self.mqtt_client.client.publish(topic, "offline", retain=True) + + def process_udp_packet(self, msg_type, seq_counter, value, uuid, ip_address): + self._ensure_node_exists(uuid) + node = self.nodes[uuid] + + # Online melden, wenn paket eingetroffen ist + if self.mqtt_client and self.mqtt_client.is_connected: + topic = f"{self.mqtt_client.sensor_topic}/{uuid}/availability" + self.mqtt_client.client.publish(topic, "online", retain=True) + + # Jedes eintreffende Paket (egal ob Temp oder Batt) setzt den Timer zurück! + self._kick_alive_timer(uuid) + + # Falls das Node vorher offline war, bringen wir es jetzt sofort wieder online + if node["lost_frame_sync"]: + self.mqtt_client.publish_sensor_value(uuid, "availability", "online", retain=True) + + if node["ip"] != ip_address: + node["ip"] = ip_address + if node["lost_frame_sync"]: + self.mqtt_client.publish_sensor_value(uuid, "ipv6", ip_address) + + # Paketverlust berechnen + if node["last_seq"] is not None: + raw_diff = (seq_counter - node["last_seq"]) & 0xFFFF + if raw_diff & 0x8000: + signed_diff = raw_diff - 0x10000 + else: + signed_diff = raw_diff + + if signed_diff > 1: + lost_now = signed_diff - 1 + self.logger.warning(f"[{uuid}] Paketverlust erkannt! {lost_now} Paket(e) ausgelassen.") + if node["lost_frame_sync"]: + node["global_lost_counter"] += lost_now + self.mqtt_client.publish_sensor_value(uuid, "packet_loss", node["global_lost_counter"], retain=True) + else: + node["local_lost_counter"] += lost_now + elif signed_diff < 0: + self.logger.info(f"[{uuid}] nRF52840 Reboot erkannt (Seq von {node['last_seq']} auf {seq_counter}). Setze Zähler zurück.") + + node["last_seq"] = seq_counter + + if node["lost_frame_sync"]: + if msg_type == "temperature": + temp_c = value / 1000.0 + self.logger.info(f"[{uuid}] -> MQTT: Temperatur {temp_c}°C") + self.mqtt_client.publish_sensor_value(uuid, "temperature", temp_c) + elif msg_type == "battery": + batt_v = value / 1000.0 + self.logger.info(f"[{uuid}] -> MQTT: Batteriespannung {batt_v}V") + self.mqtt_client.publish_sensor_value(uuid, "battery", batt_v) + batt_percent = 100 if batt_v > 3.25 else (10 if batt_v > 3.2 else 0) + self.mqtt_client.publish_sensor_value(uuid, "battery_level", batt_percent) + elif msg_type == "config": + meas_interval, temp_delta, max_interval, batt_interval = value + self.logger.info(f"[{uuid}] -> MQTT: Konfigurationspaket empfangen (meas_interval={meas_interval}, temp_delta={temp_delta}, max_interval={max_interval}, batt_interval={batt_interval})") + + # Update das lokale Shadow, damit die Timeout-Berechnung sich den echten Firmware-Zuständen anpasst! + node["config_shadow"]["meas_interval"] = meas_interval + node["config_shadow"]["max_interval"] = max_interval + + self.mqtt_client.publish_sensor_value(uuid, "meas_interval", meas_interval) + self.mqtt_client.publish_sensor_value(uuid, "temp_delta", temp_delta / 100.0) + self.mqtt_client.publish_sensor_value(uuid, "max_interval", max_interval) + self.mqtt_client.publish_sensor_value(uuid, "batt_interval", batt_interval / 60.0) + else: + self.logger.info(f"[{uuid}] Wert geparkt ({msg_type}, warte auf MQTT Sync)...") + + def handle_mqtt_sync(self, uuid, retained_lost_packets): + if uuid not in self.nodes: + self._ensure_node_exists(uuid) + + node = self.nodes[uuid] + + if not node["lost_frame_sync"]: + if node["fallback_timer"] is not None: + node["fallback_timer"].cancel() + + node["global_lost_counter"] = retained_lost_packets + node["local_lost_counter"] + node["lost_frame_sync"] = True + self.logger.info(f"[{uuid}] Sync abgeschlossen. Globaler Verlust-Zähler: {node['global_lost_counter']}") + + self.mqtt_client.publish_sensor_value(uuid, "packet_loss", node["global_lost_counter"], retain=True) + if node["ip"]: + self.mqtt_client.publish_sensor_value(uuid, "ipv6", node["ip"]) + + def handle_param_change_from_ha(self, uuid, param_name, float_value): + self._ensure_node_exists(uuid) + node = self.nodes[uuid] + + if param_name == "meas_interval": + node["config_shadow"]["meas_interval"] = int(float_value) + elif param_name == "temp_delta": + node["config_shadow"]["temp_delta"] = int(round(float_value * 1000)) + elif param_name == "max_interval": + node["config_shadow"]["max_interval"] = int(float_value) + elif param_name == "batt_interval": + node["config_shadow"]["batt_interval"] = int(float_value) + + self.logger.debug(f"[{uuid}] Slider-Änderung empfangen: {param_name} = {float_value}. Shadow aktualisiert: {node['config_shadow']}") + + if node["debounce_timer"] is not None: + node["debounce_timer"].cancel() + + node["debounce_timer"] = threading.Timer(self.debounce_timeout, self._send_config_downlink, args=[uuid]) + node["debounce_timer"].start() + + def _send_config_downlink(self, uuid): + node = self.nodes[uuid] + if not node["ip"]: + self.logger.error(f"Kann Config nicht an {uuid} senden: Keine IPv6-Adresse bekannt!") + return + + shadow = node["config_shadow"] + + fmt = "!BBHHHHH" + payload = struct.pack( + fmt, + 1, + 0x10, + self.downlink_seq, + shadow["meas_interval"], + shadow["temp_delta"], + shadow["max_interval"], + shadow["batt_interval"] + ) + + self.downlink_seq = (self.downlink_seq + 1) & 0xFFFF + self.logger.info(f"[{uuid}] Sendete neue Config an [{node['ip']}]: " + f"Meas: {shadow['meas_interval']}m, Delta: {shadow['temp_delta']}m°C, " + f"Max: {shadow['max_interval']}m, Batt: {shadow['batt_interval']}m") + + # Nach dem Senden der Config passen wir das lokale Timeout-Fenster sofort prophylaktisch an + self._kick_alive_timer(uuid) + self.udp_listener.send_cmd(node["ip"], payload) \ No newline at end of file diff --git a/gateway/gateway/mqtt_client.py b/gateway/gateway/mqtt_client.py new file mode 100644 index 0000000..884d3e6 --- /dev/null +++ b/gateway/gateway/mqtt_client.py @@ -0,0 +1,121 @@ +import logging +import json +from paho.mqtt import client as mqtt +from .mqtt_discovery import MQTTDiscovery +import time + +class MQTTClient: + def __init__(self, controller, config): + self.controller = controller + self.config = config + self.logger = logging.getLogger("MQTTClient") + + mqtt_conf = self.config.get("mqtt", {}) + self.broker = mqtt_conf.get("broker", "localhost") + self.port = mqtt_conf.get("port", 1883) + + username = mqtt_conf.get("user") + password = mqtt_conf.get("password") + + self.base_topic = mqtt_conf.get("base_topic", "homeassistant") + self.sensor_topic = mqtt_conf.get("sensor_topic", "pool_temp") + + # Wichtig: sensor_topic hier übergeben! + self.discovery = MQTTDiscovery(base_topic=self.base_topic, sensor_topic=self.sensor_topic) + self.client = mqtt.Client() + + if username and password: + self.client.username_pw_set(username, password) + + self.client.on_connect = self._on_connect + self.client.on_message = self._on_message + + self.sync_topics = set() + self.is_connected = False + + def start(self): + try: + self.logger.info(f"Verbinde mit MQTT-Broker {self.broker}:{self.port}...") + testament_topic = f"{self.sensor_topic}/28833e1363721b06/availability" + self.client.will_set(testament_topic, payload="offline", qos=1, retain=True) + self.client.connect(self.broker, self.port, 60) + self.client.loop_start() + except Exception as e: + self.logger.error(f"MQTT-Verbindungsfehler: {e}") + + def _on_connect(self, client, userdata, flags, rc): + if rc == 0: + self.logger.info("Erfolgreich mit MQTT-Broker verbunden.") + self.is_connected = True + + # Korrigiert: Nutze das dynamische sensor_topic für den Paketverlust-Sync! + sync_topic = f"{self.sensor_topic}/+/packet_loss/state" + self.client.subscribe(sync_topic) + self.logger.debug(f"Sync-Topic abonniert: {sync_topic}") + + # Abonnieren der Slider-Befehle aus HA (z.B. pool_temp/+/+/set) + cmd_topic = f"{self.sensor_topic}/+/+/set" + self.client.subscribe(cmd_topic) + self.logger.debug(f"Command-Topic abonniert: {cmd_topic}") + else: + self.logger.error(f"Verbindung abgelehnt mit Code: {rc}") + + def _on_message(self, client, userdata, msg): + try: + topic_parts = msg.topic.split('/') + # Aufbau: [sensor_topic]/[uuid]/[sub_topic]/[state_or_set] + if len(topic_parts) != 4: + return + + uuid = topic_parts[1] + sub_topic = topic_parts[2] + action = topic_parts[3] + + # 1. Sync-Logik beim Gateway-Start + if sub_topic == "packet_loss" and action == "state": + if uuid not in self.sync_topics: + payload_str = msg.payload.decode().strip() + retained_loss = int(payload_str) if payload_str else 0 + + self.logger.info(f"Retain-Wert für {uuid} gefunden: {retained_loss} verlorene Pakete.") + self.sync_topics.add(uuid) + self.controller.handle_mqtt_sync(uuid, retained_loss) + self.publish_discovery(uuid) + + # 2. Slider-Änderungen aus Home Assistant abfangen + elif action == "set": + value_str = msg.payload.decode().strip() + self.logger.info(f"Slider-Änderung von HA erhalten: {sub_topic} -> {value_str} für {uuid}") + + # Bestätigung direkt an den State-Kanal zurücksenden, damit der Slider nicht zurückspringt + self.client.publish(f"{self.sensor_topic}/{uuid}/{sub_topic}/state", value_str, retain=True) + + self.controller.handle_param_change_from_ha(uuid, sub_topic, float(value_str)) + + except Exception as e: + self.logger.error(f"Fehler beim Verarbeiten der MQTT-Nachricht: {e}") + + def trigger_fallback_sync(self, uuid): + if uuid not in self.sync_topics: + self.logger.info(f"Kein Retain-Wert für {uuid} empfangen. Starte initial bei 0.") + self.sync_topics.add(uuid) + self.controller.handle_mqtt_sync(uuid, 0) + self.publish_discovery(uuid) + + def publish_discovery(self, uuid): + configs = self.discovery.get_configs(uuid) + for topic, payload in configs.items(): + self.client.publish(topic, json.dumps(payload), retain=True) + self.logger.debug(f"Home Assistant Discovery gesendet: {topic}, Payload: {payload}") + self.logger.info(f"Home Assistant Discovery für Node {uuid} gesendet.") + + def publish_sensor_value(self, uuid, sensor_type, value, retain=False): + if not self.is_connected: + return + topic = f"{self.sensor_topic}/{uuid}/{sensor_type}/state" + self.client.publish(topic, str(value), retain=retain) + self.logger.debug(f"Sensorwert veröffentlicht: {topic} -> {value} (retain={retain})") + + def stop(self): + self.client.loop_stop() + self.client.disconnect() \ No newline at end of file diff --git a/gateway/gateway/mqtt_discovery.py b/gateway/gateway/mqtt_discovery.py new file mode 100644 index 0000000..46b17bf --- /dev/null +++ b/gateway/gateway/mqtt_discovery.py @@ -0,0 +1,158 @@ +import json + +class MQTTDiscovery: + def __init__(self, base_topic="homeassistant", sensor_topic="pool_temp"): + self.base_topic = base_topic + self.sensor_topic = sensor_topic + + def get_device_info(self, uuid): + return { + "identifiers": [uuid], + "name": f"Pool-Temperatur-Sensor ({uuid[-4:]})", + "model": "nRF52840 Temperature Sensor", + "manufacturer": "Iten Engineering", + "sw_version": "Zephyr v4.3.99 / NCS v3.3.0" + } + + def get_configs(self, uuid): + device_info = self.get_device_info(uuid) + configs = {} + + # --- SENSOREN (Ausgabe) --- + + # 1. Temperatur-Sensor + configs[f"{self.base_topic}/sensor/{uuid}/temperature/config"] = { + "device_class": "temperature", + "state_class": "measurement", + "name": "Temperatur", + "state_topic": f"{self.sensor_topic}/{uuid}/temperature/state", + "unit_of_measurement": "°C", + "value_template": "{{ value }}", + "unique_id": f"{uuid}_temperature", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + +# 2. Batterie-Sensor (Analog für Diagnose) + configs[f"{self.base_topic}/sensor/{uuid}/battery/config"] = { + "device_class": "voltage", + "state_class": "measurement", + "entity_category": "diagnostic", + "name": "Batteriespannung", + "state_topic": f"{self.sensor_topic}/{uuid}/battery/state", + "unit_of_measurement": "V", + "value_template": "{{ value }}", + "unique_id": f"{uuid}_battery", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + configs[f"{self.base_topic}/sensor/{uuid}/battery_level/config"] = { + "device_class": "battery", + "name": "Batterieladung", + "state_topic": f"{self.sensor_topic}/{uuid}/battery_level/state", + "unit_of_measurement": "%", + "value_template": "{{ value }}", + "unique_id": f"{uuid}_battery_level", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + # 3. Packet Loss Zähler + configs[f"{self.base_topic}/sensor/{uuid}/packet_loss/config"] = { + "state_class": "total_increasing", + "entity_category": "diagnostic", + "name": "Paketverlust Gesamt", + "state_topic": f"{self.sensor_topic}/{uuid}/packet_loss/state", + "unit_of_measurement": "Pakete", + "value_template": "{{ value }}", + "unique_id": f"{uuid}_packet_loss", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + # 4. IPv6-Adresse + configs[f"{self.base_topic}/sensor/{uuid}/ipv6_address/config"] = { + "entity_category": "diagnostic", + "name": "IPv6 Adresse", + "state_topic": f"{self.sensor_topic}/{uuid}/ipv6/state", + "value_template": "{{ value }}", + "unique_id": f"{uuid}_ipv6", + "icon": "mdi:ip-network", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + # --- SLIDER / CONTROLS (Eingabe) --- + + # 1. Messintervall (1-60 Min, Raster 1) + configs[f"{self.base_topic}/number/{uuid}/meas_interval/config"] = { + "name": "Messintervall [min]", + "state_topic": f"{self.sensor_topic}/{uuid}/meas_interval/state", + "command_topic": f"{self.sensor_topic}/{uuid}/meas_interval/set", + "min": 1, "max": 60, "step": 1, + "unit_of_measurement": "min", + "unique_id": f"{uuid}_meas_interval", + "icon": "mdi:timer", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + # 2. Sende-Delta (0.1 - 2.0 °C, Raster 0.1) + configs[f"{self.base_topic}/number/{uuid}/temp_delta/config"] = { + "name": "Temperatur Delta-T [°C]", + "state_topic": f"{self.sensor_topic}/{uuid}/temp_delta/state", + "command_topic": f"{self.sensor_topic}/{uuid}/temp_delta/set", + "min": 0.1, "max": 2.0, "step": 0.05, + "unit_of_measurement": "°C", + "unique_id": f"{uuid}_temp_delta", + "icon": "mdi:thermometer", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + # 3. Maximales Sendeintervall Temp (5-120 Min, Raster 5) + configs[f"{self.base_topic}/number/{uuid}/max_interval/config"] = { + "name": "Max Sendeintervall Temp", + "state_topic": f"{self.sensor_topic}/{uuid}/max_interval/state", + "command_topic": f"{self.sensor_topic}/{uuid}/max_interval/set", + "min": 5, "max": 120, "step": 5, + "unit_of_measurement": "min", + "unique_id": f"{uuid}_max_interval", + "icon": "mdi:timer", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + # 4. Sendeintervall Batterie (2-72 Std, Raster 2) + configs[f"{self.base_topic}/number/{uuid}/batt_interval/config"] = { + "name": "Sendeintervall Batterie [h]", + "state_topic": f"{self.sensor_topic}/{uuid}/batt_interval/state", + "command_topic": f"{self.sensor_topic}/{uuid}/batt_interval/set", + "min": 2, "max": 72, "step": 2, + "unit_of_measurement": "h", + "unique_id": f"{uuid}_batt_interval", + "icon": "mdi:timer", + "device": device_info, + "availability_topic": f"{self.sensor_topic}/{uuid}/availability", + "payload_available": "online", + "payload_not_available": "offline" + } + + return configs \ No newline at end of file diff --git a/gateway/gateway/udp_listener.py b/gateway/gateway/udp_listener.py new file mode 100644 index 0000000..e2a7a69 --- /dev/null +++ b/gateway/gateway/udp_listener.py @@ -0,0 +1,140 @@ +import logging +import socket +import struct +import threading + +class UDPListener: + def __init__(self, controller, port=6969): + self.controller = controller + self.port = port + self.logger = logging.getLogger("UDPListener") + + # Socket für IPv6 (Thread nutzt IPv6) vorbereiten + self.sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) + + # Erlaubt das schnelle Wiederbinden des Ports nach einem Neustart + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + + self.is_running = False + self._thread = None + + def start(self): + """Startet den UDP-Empfangs-Thread.""" + try: + # Bindet auf alle verfügbaren IPv6-Schnittstellen ("::") + self.sock.bind(("::", self.port)) + self.is_running = True + + self._thread = threading.Thread(target=self._listen_loop, daemon=True) + self._thread.start() + self.logger.info(f"UDP-Listener erfolgreich auf Port {self.port} gestartet (IPv6).") + except Exception as e: + self.logger.error(f"Fehler beim Starten des UDP-Sockets: {e}") + raise + + def _listen_loop(self): + """Endlosschleife, die auf eingehende UDP-Pakete blockiert und typbasiert parst.""" + # Definition der Formate nach deiner Firmware-Struktur: + # Header (12 Bytes): 1B Version, 1B Type, 8s Device-ID, 1H Sequence Counter + # Payload Temp: 1i (4 Bytes Signed Int) -> Gesamtlänge 16 + # Payload Batt: 1H (2 Bytes Unsigned Short) -> Gesamtlänge 14 + + FORMAT_TEMP = "!BB8sHi" + FORMAT_BATT = "!BB8sHH" + FORMAT_CONFIG = "!BB8sHHHHH" + + SIZE_TEMP = struct.calcsize(FORMAT_TEMP) # 16 Bytes + SIZE_BATT = struct.calcsize(FORMAT_BATT) # 14 Bytes + SIZE_CONFIG = struct.calcsize(FORMAT_CONFIG) # 12 Bytes + + + + while self.is_running: + try: + data, addr = self.sock.recvfrom(1024) + + if len(data) < 12: + self.logger.warning(f"Paket von {addr[0]} zu kurz für einen Header ({len(data)} Bytes)") + continue + + # Wir schauen direkt in das zweite Byte (Index 1) für den Typ + payload_type = data[1] + + if payload_type == 0x00: # PAYLOAD_TYPE_TEMP + if len(data) != SIZE_TEMP: + self.logger.warning(f"Ungültige Größe für Temp-Paket: {len(data)} statt {SIZE_TEMP}") + continue + + version, p_type, device_id_raw, seq, temp_raw = struct.unpack(FORMAT_TEMP, data) + + if version != 1: + continue + + self.controller.process_udp_packet( + msg_type="temperature", + seq_counter=seq, + value=temp_raw, # In Milligrad + uuid=device_id_raw.hex(), + ip_address=addr[0] # Die IPv6-Adresse des nRF52 + ) + + elif payload_type == 0x01: # PAYLOAD_TYPE_BATTERY + if len(data) != SIZE_BATT: + self.logger.warning(f"Ungültige Größe für Batterie-Paket: {len(data)} statt {SIZE_BATT}") + continue + + version, p_type, device_id_raw, seq, batt_raw = struct.unpack(FORMAT_BATT, data) + + if version != 1: + continue + + self.controller.process_udp_packet( + msg_type="battery", + seq_counter=seq, + value=batt_raw, # In Millivolt + uuid=device_id_raw.hex(), + ip_address=addr[0] # Die IPv6-Adresse des nRF52 + ) + + elif payload_type == 0x10: # PAYLOAD_TYPE_CONFIG + if len(data) != SIZE_CONFIG: + self.logger.warning(f"Ungültige Größe für Config-Paket: {len(data)} statt {SIZE_CONFIG}") + continue + + version, p_type, device_id_raw, seq, meas_interval, temp_delta, max_interval, batt_interval = struct.unpack(FORMAT_CONFIG, data) + + if version != 1: + continue + + self.controller.process_udp_packet( + msg_type="config", + seq_counter=seq, + value=(meas_interval, temp_delta, max_interval, batt_interval), + uuid=device_id_raw.hex(), + ip_address=addr[0] # Die IPv6-Adresse des nRF52 + ) + + else: + self.logger.warning(f"Unbekannter Payload-Typ 0x{payload_type:02x} von {addr[0]}") + + except Exception as e: + if self.is_running: + self.logger.error(f"Fehler im UDP-Empfangs-Loop: {e}") + + def send_cmd(self, ip_address, payload): + """ + Sendet ein UDP-Paket zurück an den nRF52. + Wird später vom Controller aufgerufen, wenn MQTT-Befehle reinkommen. + """ + try: + # sendto() blockiert nicht nennenswert, kann direkt aufgerufen werden + self.sock.sendto(payload, (ip_address, self.port)) + self.logger.debug(f"Befehl an [{ip_address}]:{self.port} gesendet.") + except Exception as e: + self.logger.error(f"Fehler beim Senden an [{ip_address}]: {e}") + + def stop(self): + """Stoppt den Listener und schließt den Socket.""" + self.is_running = False + self.sock.close() + self.logger.info("UDP-Listener gestoppt.") \ No newline at end of file diff --git a/gateway/main.py b/gateway/main.py new file mode 100644 index 0000000..d0fcca7 --- /dev/null +++ b/gateway/main.py @@ -0,0 +1,175 @@ +import os +import sys +import logging +import yaml +import argparse + +# Pfad-Setup, damit das 'gateway' Paket sauber importiert werden kann +sys.path.append(os.path.dirname(os.path.abspath(__file__))) + +def parse_arguments(): + """Definiert und liest die Kommandozeilenparameter aus.""" + parser = argparse.ArgumentParser(description="nRF52 UDP zu MQTT Gateway für den Pool-Sensor") + + # Config-Pfad + parser.add_argument("-c", "--config", help="Pfad zur Konfigurationsdatei (config.yaml)") + + # MQTT Parameter + parser.add_argument("--mqtt-broker", help="Überschreibt den MQTT Broker (IP/Hostname)") + parser.add_argument("--mqtt-port", type=int, help="Überschreibt den MQTT Broker Port") + parser.add_argument("--mqtt-user", help="Überschreibt den MQTT Benutzernamen") + parser.add_argument("--mqtt-password", help="Überschreibt das MQTT Passwort") + parser.add_argument("--mqtt-base-topic", help="Überschreibt das MQTT Base Topic (Discovery)") + parser.add_argument("--mqtt-sensor-topic", help="Überschreibt das MQTT Sensor Topic") + + # Gateway Parameter + parser.add_argument("--udp-in-port", type=int, help="Überschreibt den UDP Empfangs-Port") + parser.add_argument("--udp-out-port", type=int, help="Überschreibt den UDP Sende-Port") + parser.add_argument("--log-level", choices=["DEBUG", "INFO", "WARNING", "ERROR"], help="Überschreibt das Logging Level") + parser.add_argument("--debounce-timeout", type=int, help="Überschreibt das Debounce Timeout in Sekunden") + + return parser.parse_args() + +def load_and_merge_config(args): + """Sucht die Config, lädt sie und mergt sie mit den CLI-Parametern.""" + config_path = None + + # 1. Config-Pfad ermitteln + if args.config: + config_path = args.config + else: + # Fallback 1: Aktuelles Arbeitsverzeichnis (CWD) + cwd_path = os.path.join(os.getcwd(), "config.yaml") + # Fallback 2: Verzeichnis, in dem dieses Skript liegt + script_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml") + + if os.path.exists(cwd_path): + config_path = cwd_path + elif os.path.exists(script_path): + config_path = script_path + else: + config_path = cwd_path # Nur für die Fehlermeldung, falls nichts existiert + + # 2. YAML laden + config = {} + if os.path.exists(config_path): + with open(config_path, "r", encoding="utf-8") as f: + try: + config = yaml.safe_load(f) or {} + print(f"Lade Konfiguration von: {config_path}") + except yaml.YAMLError as e: + print(f"Fehler beim Parsen der YAML-Datei: {e}") + sys.exit(1) + else: + if args.config: + print(f"Fehler: Explizit angegebene Konfigurationsdatei '{args.config}' nicht gefunden!") + sys.exit(1) + else: + print("Warnung: Keine config.yaml gefunden! Nutze ausschließlich Kommandozeilenparameter und Defaults.") + + # Grundstruktur sicherstellen, falls die YAML leer war oder nicht existiert + if "mqtt" not in config: config["mqtt"] = {} + if "gateway" not in config: config["gateway"] = {} + + # 3. Mit CLI-Parametern überschreiben (falls angegeben) + if args.mqtt_broker: config["mqtt"]["broker"] = args.mqtt_broker + if args.mqtt_port: config["mqtt"]["port"] = args.mqtt_port + if args.mqtt_user: config["mqtt"]["user"] = args.mqtt_user + if args.mqtt_password: config["mqtt"]["password"] = args.mqtt_password + if args.mqtt_base_topic: config["mqtt"]["base_topic"] = args.mqtt_base_topic + if args.mqtt_sensor_topic: config["mqtt"]["sensor_topic"] = args.mqtt_sensor_topic + + if args.udp_in_port: config["gateway"]["udp_in_port"] = args.udp_in_port + if args.udp_out_port: config["gateway"]["udp_out_port"] = args.udp_out_port + if args.log_level: config["gateway"]["log_level"] = args.log_level + if args.debounce_timeout: config["gateway"]["debounce_timeout"] = args.debounce_timeout + + # 4. Fallback Defaults setzen, falls weder in YAML noch CLI vorhanden + config["mqtt"].setdefault("broker", "127.0.0.1") + config["mqtt"].setdefault("port", 1883) + config["mqtt"].setdefault("base_topic", "homeassistant") + config["mqtt"].setdefault("sensor_topic", "pool_temp") + config["gateway"].setdefault("udp_in_port", 6969) + config["gateway"].setdefault("udp_out_port", 6969) + config["gateway"].setdefault("log_level", "INFO") + config["gateway"].setdefault("debounce_timeout", 5) + + return config + +def setup_logging(level_str): + """Initialisiert das globale Logging, optimiert für Terminal (Farbe) oder systemd.""" + level = getattr(logging, level_str.upper(), logging.INFO) + + if not sys.stdout.isatty(): + log_format = "[%(levelname)s] %(name)s: %(message)s" + formatter = logging.Formatter(log_format) + else: + log_format = "%(asctime)s [%(levelname)s] %(name)s: %(message)s" + date_format = "%Y-%m-%d %H:%M:%S" + + class ColoredFormatter(logging.Formatter): + GREY = "\033[90m" + YELLOW = "\033[33m" + RED = "\033[31m" + RESET = "\033[0m" + + COLORS = { + logging.DEBUG: GREY, + logging.WARNING: YELLOW, + logging.ERROR: RED, + logging.CRITICAL: RED + } + + def format(self, record): + color = self.COLORS.get(record.levelno, "") + msg = super().format(record) + if color: + return f"{color}{msg}{self.RESET}" + return msg + + formatter = ColoredFormatter(log_format, datefmt=date_format) + + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(formatter) + + root_logger = logging.getLogger() + root_logger.setLevel(level) + # Verhindere doppelte Logs bei wiederholtem Aufruf + if not root_logger.handlers: + root_logger.addHandler(handler) + +def main(): + # 1. Argumente parsen + args = parse_arguments() + + # 2. Konfiguration laden & mergen + config = load_and_merge_config(args) + + # 3. Logging mit dem (möglicherweise überschriebenen) Level initialisieren + log_level = config["gateway"]["log_level"] + setup_logging(log_level) + + logger = logging.getLogger("Main") + logger.info(f"MQTT-Broker: {config['mqtt']['broker']}:{config['mqtt']['port']}") + logger.info(f"UDP Listen Port: {config['gateway']['udp_in_port']}") + logger.info("Starte Temperatursensor-Gateway...") + + try: + from gateway.controller import Controller + controller = Controller(config) + controller.start() + + logger.info("Gateway erfolgreich initialisiert. Warte auf Daten...") + + import time + while True: + time.sleep(1) + + except KeyboardInterrupt: + logger.info("Gateway wird durch Benutzer beendet...") + except Exception as e: + logger.exception(f"Unerwarteter Fehler im Hauptprogramm: {e}") + sys.exit(1) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/gateway/pooltemp.service b/gateway/pooltemp.service new file mode 100644 index 0000000..0697fcd --- /dev/null +++ b/gateway/pooltemp.service @@ -0,0 +1,12 @@ +[Unit] +Description=Pool Thermometer Python Gateway +After=network.target + +[Service] +WorkingDirectory=/opt/pooltemp/gateway +ExecStart=/opt/pooltemp/gateway/.venv/bin/python /opt/pooltemp/gateway/main.py +Restart=always +RestartSec=5s + +[Install] +WantedBy=multi-user.target \ No newline at end of file diff --git a/gateway/requirements.txt b/gateway/requirements.txt new file mode 100644 index 0000000..ca70c95 --- /dev/null +++ b/gateway/requirements.txt @@ -0,0 +1,2 @@ +pyyaml +paho-mqtt