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()