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)