diff --git a/mqtt-json-parse.py b/mqtt-json-parse.py index e31d0c3..bbeb27d 100644 --- a/mqtt-json-parse.py +++ b/mqtt-json-parse.py @@ -1,31 +1,58 @@ import json import time +import logging import paho.mqtt.client as mqtt from meshtastic_mqtt_json import MeshtasticMQTT -# --- KONFIGURATION --- -LOCAL_BROKER = " -CHANNEL_KEY = "" -REGION = "EU_868" -ENCRYPTED_ROOT = f"msh/{REGION}/2/e/" +# Channel Details +CHANNEL_NAME = "" # Name of the channel (e.g., "Puig") +CHANNEL_INDEX = 0 # Channel index on the hardware (usually 0) +CHANNEL_KEY = "" # Base64 encryption key +REGION = "EU_868" -# --- SETUP LOKALER PUBLISHER --- +# Path Configuration for ioBroker Monitoring +ENCRYPTED_ROOT = f"msh/{REGION}/2/e/" +SERVICE_BASE_PATH = f"service/Decryptor/{CHANNEL_NAME}" + +# --- LOGGING SETUP --- +# Format includes the channel name for better clarity in journalctl +logging.basicConfig( + level=logging.INFO, + format=f'%(asctime)s - [{CHANNEL_NAME}] - %(levelname)s - %(message)s' +) + +# --- MQTT LOGGING HANDLER --- +# Forwards local log events to an MQTT topic for remote monitoring +class MQTTHandler(logging.Handler): + def emit(self, record): + try: + # Only attempt publish if the publisher client is connected + if 'publisher' in globals() and publisher.is_connected(): + log_entry = self.format(record) + publisher.publish(f"{SERVICE_BASE_PATH}/log", log_entry) + except Exception: + pass + +# --- LOCAL PUBLISHER SETUP --- +# Used for sending decrypted JSON data and service logs to ioBroker publisher = mqtt.Client() publisher.username_pw_set(LOCAL_USER, LOCAL_PASS) -publisher.connect(LOCAL_BROKER, LOCAL_PORT) -publisher.loop_start() + +# Attach the MQTT logger after the client object is created +logging.getLogger().addHandler(MQTTHandler()) def send_to_iobroker(sender_id_dec, msg_type, payload_data): - """Baut das Meshtastic-JSON-Format nach und sendet es lokal.""" + """ + Constructs a Meshtastic-compatible JSON structure and publishes it locally. + Target: msh/REGION/2/json/CHANNEL/!senderhex + """ sender_id_hex = hex(sender_id_dec)[2:].lower().zfill(8) - # Ziel-Topic für den ioBroker JavaScript-Trigger topic = f"msh/{REGION}/2/json/{CHANNEL_NAME}/!{sender_id_hex}" full_payload = { @@ -37,48 +64,69 @@ def send_to_iobroker(sender_id_dec, msg_type, payload_data): } publisher.publish(topic, json.dumps(full_payload), qos=1) -# --- CALLBACKS --- - +# --- DECRYPTION CALLBACKS --- def on_text_message(json_data): - """Verarbeitet reine Textnachrichten.""" - # json_data["decoded"]["payload"] enthält bei Textnachrichten den String - send_to_iobroker(json_data["from"], "text", {"text": json_data["decoded"]["payload"]}) - print(f'Relayed Text: {json_data["decoded"]["payload"]}') + """Callback for incoming decrypted text messages.""" + msg_text = json_data["decoded"]["payload"] + send_to_iobroker(json_data["from"], "text", {"text": msg_text}) + logging.info(f"Relayed Text from !{hex(json_data['from'])[2:]}: {msg_text}") def on_position(json_data): - """Verarbeitet GPS-Positionen.""" - # json_data["decoded"]["payload"] enthält hier das Positions-Objekt + """Callback for incoming decrypted position updates.""" p = json_data["decoded"]["payload"] - - # Payload-Struktur für dein ioBroker-Skript aufbauen - # Meshtastic nutzt oft latitude_i (Integer) statt Float für Präzision pos_payload = { "latitude_i": p.get("latitude_i"), "longitude_i": p.get("longitude_i"), "altitude": p.get("altitude") } - send_to_iobroker(json_data["from"], "position", pos_payload) - print(f'Relayed Position Update from {json_data["from"]}') + logging.info(f"Relayed Position from !{hex(json_data['from'])[2:]}") -# --- SETUP DECRYPTOR --- +# --- DECRYPTOR INITIALIZATION --- +# Using MeshtasticMQTT to handle encrypted MQTT streams decryptor = MeshtasticMQTT() - -# Registrierung der gewünschten Callbacks decryptor.register_callback('TEXT_MESSAGE_APP', on_text_message) decryptor.register_callback('POSITION_APP', on_position) -# Authentifizierung am internen Client setzen +# Configure authentication for the internal decryptor client if hasattr(decryptor, '_client'): decryptor._client.username_pw_set(LOCAL_USER, LOCAL_PASS) -# Verbindung zum lokalen Broker herstellen -decryptor.connect( - LOCAL_BROKER, - LOCAL_PORT, - ENCRYPTED_ROOT, - CHANNEL_NAME, - LOCAL_USER, - LOCAL_PASS, - CHANNEL_KEY -) +# --- START SERVICE --- +try: + logging.info(f"Connecting to local broker at {LOCAL_BROKER}...") + publisher.connect(LOCAL_BROKER, LOCAL_PORT) + publisher.loop_start() + + logging.info(f"Starting Decryptor service for channel '{CHANNEL_NAME}'...") + decryptor.connect( + LOCAL_BROKER, LOCAL_PORT, ENCRYPTED_ROOT, + CHANNEL_NAME, LOCAL_USER, LOCAL_PASS, CHANNEL_KEY + ) + + start_time = time.time() + last_heartbeat = 0 + + # --- MAIN MAINTENANCE LOOP --- + while True: + time.sleep(1) + + # Publish health status every 60 seconds + if time.time() - last_heartbeat > 60: + status = { + "status": "online", + "channel": CHANNEL_NAME, + "uptime_seconds": int(time.time() - start_time), + "timestamp": time.strftime('%Y-%m-%d %H:%M:%S') + } + publisher.publish(f"{SERVICE_BASE_PATH}/status", json.dumps(status), retain=True) + last_heartbeat = time.time() + +except KeyboardInterrupt: + logging.info("Service stopped by user.") +except Exception as e: + logging.error(f"Critical error in main loop: {e}") +finally: + # Clean shutdown of MQTT connections + publisher.loop_stop() + publisher.disconnect()