#!/usr/bin/env python3 """Receive Meshtastic mesh text messages over MQTT and expose them in Home Assistant. Flow: 1. Subscribe to a Meshtastic gateway's MQTT topic (public server or local broker). 2. Decrypt channel packets (AES-CTR) using the channel PSK. 3. Publish the latest text message and parsed Temperature/Humidity values to Home Assistant via MQTT Discovery (so entities appear automatically). """ import base64 import json import os import re import sys import time from datetime import datetime, timezone import paho.mqtt.client as mqtt import requests from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from meshtastic.protobuf import mesh_pb2, mqtt_pb2, portnums_pb2 def log(msg): print(msg, file=sys.stdout, flush=True) # --- Add-on options (written by the Supervisor) ------------------------------ with open("/data/options.json", encoding="utf-8") as fh: OPTS = json.load(fh) SRC_BROKER = OPTS["source_broker"] SRC_PORT = int(OPTS.get("source_port", 1883)) SRC_USER = OPTS.get("source_username", "") or "" SRC_PASS = OPTS.get("source_password", "") or "" SRC_TLS = bool(OPTS.get("source_tls", False)) REGION = OPTS.get("region", "+") or "+" CHANNEL_NAME = OPTS["channel_name"] CHANNEL_PSK = OPTS.get("channel_psk", "AQ==") MATCH_PREFIX = OPTS.get("match_prefix", "") or "" DEVICE_NAME = OPTS.get("device_name", "Meshtastic Mesh") DISCOVERY_PREFIX = OPTS.get("discovery_prefix", "homeassistant") SENSORS = OPTS.get("sensors", []) DEVICE_UID = re.sub(r"[^a-z0-9_]", "_", DEVICE_NAME.lower()) or "meshtastic_mesh" BASE_TOPIC = f"meshtastic_receiver/{DEVICE_UID}" AVAIL_TOPIC = f"{BASE_TOPIC}/status" # --- PSK handling ------------------------------------------------------------ # "AQ==" (byte 0x01) is Meshtastic's shorthand for this well-known default key. _DEFAULT_KEY = base64.b64decode("1PG7OiApB1nwvP+rz05pAQ==") def expand_psk(psk_b64): raw = base64.b64decode(psk_b64) return _DEFAULT_KEY if raw == b"\x01" else raw CHANNEL_KEY = expand_psk(CHANNEL_PSK) # Matches "T1:20.8", "H2:54.4", "P:1013" etc. (unit suffix is ignored here). PAIR_RE = re.compile(r"([A-Za-z][A-Za-z0-9_]*)\s*:\s*(-?\d+(?:\.\d+)?)") # Built-in per-node diagnostic sensors derived from packet metadata. DIAGNOSTICS = [ { "key": "last_seen", "name": "Last Seen", "device_class": "timestamp", "icon": "mdi:clock-outline", }, { "key": "rssi", "name": "RSSI", "device_class": "signal_strength", "unit": "dBm", "state_class": "measurement", }, { "key": "snr", "name": "SNR", "unit": "dB", "state_class": "measurement", "icon": "mdi:signal", }, { "key": "hops_away", "name": "Hops Away", "state_class": "measurement", "icon": "mdi:transit-connection-variant", }, ] # Set once the HA broker client is connected. ha_client = None # --- Decryption -------------------------------------------------------------- def decrypt_packet(encrypted, key, from_id, packet_id): """AES-CTR decrypt. Nonce = packetId(uint64 LE) | fromNode(uint32 LE) | 0*4.""" try: nonce = ( packet_id.to_bytes(4, "little") + b"\x00" * 4 + from_id.to_bytes(4, "little") + b"\x00" * 4 ) cipher = Cipher( algorithms.AES(key), modes.CTR(nonce), backend=default_backend() ) d = cipher.decryptor() return d.update(encrypted) + d.finalize() except Exception as exc: # noqa: BLE001 log(f"[warn] decryption error: {exc}") return None # --- Home Assistant MQTT broker resolution (via Supervisor service) ---------- def resolve_ha_mqtt(): token = os.environ.get("SUPERVISOR_TOKEN") if not token: return None try: r = requests.get( "http://supervisor/services/mqtt", headers={"Authorization": f"Bearer {token}"}, timeout=10, ) if r.status_code == 200: d = r.json()["data"] return { "host": d["host"], "port": int(d["port"]), "username": d.get("username") or None, "password": d.get("password") or None, "ssl": bool(d.get("ssl", False)), } log(f"[warn] supervisor mqtt service returned HTTP {r.status_code}") except Exception as exc: # noqa: BLE001 log(f"[warn] could not resolve HA MQTT via supervisor: {exc}") return None # --- MQTT Discovery ---------------------------------------------------------- def device_block(): return { "identifiers": [DEVICE_UID], "name": DEVICE_NAME, "manufacturer": "Meshtastic", "model": "MQTT Mesh Receiver", } def publish_discovery(): # Latest text message (with from/to/time as attributes). cfg = { "name": "Last Message", "unique_id": f"{DEVICE_UID}_last_message", "state_topic": f"{BASE_TOPIC}/last_message", "json_attributes_topic": f"{BASE_TOPIC}/last_message/attributes", "icon": "mdi:message-text", "availability_topic": AVAIL_TOPIC, "device": device_block(), } ha_client.publish( f"{DISCOVERY_PREFIX}/sensor/{DEVICE_UID}/last_message/config", json.dumps(cfg), retain=True, ) for s in SENSORS: label = s["label"] cfg = { "name": s.get("name", label), "unique_id": f"{DEVICE_UID}_{label}", "state_topic": f"{BASE_TOPIC}/{label}", "availability_topic": AVAIL_TOPIC, "device": device_block(), } if s.get("unit"): cfg["unit_of_measurement"] = s["unit"] if s.get("device_class"): cfg["device_class"] = s["device_class"] cfg["state_class"] = "measurement" ha_client.publish( f"{DISCOVERY_PREFIX}/sensor/{DEVICE_UID}/{label}/config", json.dumps(cfg), retain=True, ) for d in DIAGNOSTICS: key = d["key"] cfg = { "name": d["name"], "unique_id": f"{DEVICE_UID}_{key}", "state_topic": f"{BASE_TOPIC}/{key}", "availability_topic": AVAIL_TOPIC, "entity_category": "diagnostic", "device": device_block(), } if d.get("unit"): cfg["unit_of_measurement"] = d["unit"] if d.get("device_class"): cfg["device_class"] = d["device_class"] if d.get("state_class"): cfg["state_class"] = d["state_class"] if d.get("icon"): cfg["icon"] = d["icon"] ha_client.publish( f"{DISCOVERY_PREFIX}/sensor/{DEVICE_UID}/{key}/config", json.dumps(cfg), retain=True, ) log( f"[ha] published discovery for last_message + {len(SENSORS)} sensors " f"+ {len(DIAGNOSTICS)} diagnostics" ) # --- Message handling -------------------------------------------------------- def publish_metadata(mp): """Publish link-quality diagnostics from a received MeshPacket.""" # Last seen: prefer the gateway's rx_time, fall back to now. ts = ( datetime.fromtimestamp(mp.rx_time, tz=timezone.utc) if mp.rx_time else datetime.now(timezone.utc) ) ha_client.publish(f"{BASE_TOPIC}/last_seen", ts.isoformat(), retain=True) # RSSI/SNR are 0 for packets the gateway originated itself — only publish # genuine over-the-air measurements. if mp.rx_rssi != 0: ha_client.publish(f"{BASE_TOPIC}/rssi", mp.rx_rssi, retain=True) ha_client.publish(f"{BASE_TOPIC}/snr", round(mp.rx_snr, 2), retain=True) # hop_start is the original TTL; hops taken = hop_start - hop_limit. # hop_start == 0 means older firmware that didn't report it. if mp.hop_start: ha_client.publish( f"{BASE_TOPIC}/hops_away", mp.hop_start - mp.hop_limit, retain=True ) def handle_text(text, from_id, to_id): # HA sensor states are limited to 255 chars. ha_client.publish(f"{BASE_TOPIC}/last_message", text[:255], retain=True) ha_client.publish( f"{BASE_TOPIC}/last_message/attributes", json.dumps( { "from": f"!{from_id:08x}", "to": f"!{to_id:08x}", "received": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "full_text": text, } ), retain=True, ) if MATCH_PREFIX and MATCH_PREFIX not in text: return known = {s["label"] for s in SENSORS} published = [] for label, value in PAIR_RE.findall(text): if label in known: ha_client.publish(f"{BASE_TOPIC}/{label}", value, retain=True) published.append(f"{label}={value}") if published: log(f"[parse] {' '.join(published)}") # --- Source MQTT callbacks --------------------------------------------------- def on_src_connect(client, userdata, flags, reason_code, properties=None): if reason_code == 0: topic = f"msh/{REGION}/2/e/{CHANNEL_NAME}/#" client.subscribe(topic) log(f"[source] connected; subscribed to {topic}") else: log(f"[source] connection refused: reason_code={reason_code}") def on_src_message(client, userdata, msg): try: env = mqtt_pb2.ServiceEnvelope() env.ParseFromString(msg.payload) except Exception as exc: # noqa: BLE001 log(f"[warn] could not parse ServiceEnvelope: {exc}") return mp = env.packet from_id = getattr(mp, "from") # Link diagnostics + last-seen update for every packet on the channel. publish_metadata(mp) if mp.HasField("decoded"): data = mp.decoded elif mp.HasField("encrypted"): plain = decrypt_packet(mp.encrypted, CHANNEL_KEY, from_id, mp.id) if plain is None: return data = mesh_pb2.Data() try: data.ParseFromString(plain) except Exception: # noqa: BLE001 return # wrong PSK or not for this channel else: return if data.portnum == portnums_pb2.PortNum.TEXT_MESSAGE_APP: text = data.payload.decode("utf-8", errors="replace") log(f"[rx] !{from_id:08x}: {text}") handle_text(text, from_id, mp.to) # --- Entry point ------------------------------------------------------------- def main(): global ha_client ha = resolve_ha_mqtt() if ha is None: log( "[fatal] No MQTT broker available. Install and start the Mosquitto " "broker add-on, then restart this add-on." ) sys.exit(1) log(f"[ha] using MQTT broker {ha['host']}:{ha['port']}") ha_client = mqtt.Client( mqtt.CallbackAPIVersion.VERSION2, client_id="meshtastic_receiver" ) if ha["username"]: ha_client.username_pw_set(ha["username"], ha["password"]) if ha["ssl"]: ha_client.tls_set() ha_client.will_set(AVAIL_TOPIC, "offline", retain=True) def on_ha_connect(client, userdata, flags, reason_code, properties=None): if reason_code == 0: log("[ha] connected") client.publish(AVAIL_TOPIC, "online", retain=True) publish_discovery() else: log(f"[ha] connection refused: reason_code={reason_code}") ha_client.on_connect = on_ha_connect ha_client.connect(ha["host"], ha["port"], keepalive=60) ha_client.loop_start() src = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) if SRC_USER: src.username_pw_set(SRC_USER, SRC_PASS) if SRC_TLS: src.tls_set() src.on_connect = on_src_connect src.on_message = on_src_message log(f"[source] connecting to {SRC_BROKER}:{SRC_PORT} …") src.connect(SRC_BROKER, SRC_PORT, keepalive=60) try: src.loop_forever() except KeyboardInterrupt: pass finally: ha_client.publish(AVAIL_TOPIC, "offline", retain=True) ha_client.loop_stop() if __name__ == "__main__": main()