meshtastic-metrics-exporter/main.py

import logging
import os

import paho.mqtt.client as mqtt
import redis
from dotenv import load_dotenv
from meshtastic.mesh_pb2 import MeshPacket
from meshtastic.mqtt_pb2 import ServiceEnvelope
from paho.mqtt.enums import CallbackAPIVersion
from prometheus_client import CollectorRegistry, start_http_server

from exporter.processors import MessageProcessor


def handle_connect(client, userdata, flags, reason_code, properties):
    print(f"Connected with result code {reason_code}")
    client.subscribe(os.getenv('MQTT_TOPIC', 'msh/israel/#'))


def handle_message(client, userdata, message):
    print(f"Received message on topic '{message.topic}'")
    envelope = ServiceEnvelope()
    envelope.ParseFromString(message.payload)

    packet: MeshPacket = envelope.packet
    if redis_client.set(str(packet.id), 1, nx=True, ex=os.getenv('redis_expiration', 60), get=True) is not None:
        logging.debug(f"Packet {packet.id} already processed")
        return

    # Process the packet
    processor.process(packet)


if __name__ == "__main__":
    load_dotenv()
    # Create Redis client
    try:
        redis_client = redis.Redis(
            host=os.getenv('REDIS_HOST'),
            port=int(os.getenv('REDIS_PORT')),
            db=int(os.getenv('REDIS_DB', 0)),
            password=os.getenv('REDIS_PASSWORD', None),
        )
    except Exception as e:
        logging.error(f"Failed to connect to Redis: {e}")
        exit(1)

    # Configure Prometheus exporter
    registry = CollectorRegistry()
    start_http_server(int(os.getenv('PROMETHEUS_COLLECTOR_PORT', 8000)), registry=registry)

    # Create an MQTT client
    mqtt_client = mqtt.Client(
        callback_api_version=CallbackAPIVersion.VERSION2,
        protocol=mqtt.MQTTv5
    )

    mqtt_client.on_connect = handle_connect
    mqtt_client.on_message = handle_message

    if os.getenv('MQTT_IS_TLS', 'false') == 'true':
        tls_context = mqtt.ssl.create_default_context()
        mqtt_client.tls_set_context(tls_context)

    if os.getenv('MQTT_USERNAME', None) and os.getenv('MQTT_PASSWORD', None):
        mqtt_client.username_pw_set(os.getenv('MQTT_USERNAME'), os.getenv('MQTT_PASSWORD'))

    try:
        mqtt_client.connect(
            os.getenv('MQTT_HOST'),
            int(os.getenv('MQTT_PORT')),
            keepalive=int(os.getenv('MQTT_KEEPALIVE', 60)),

        )
    except Exception as e:
        logging.error(f"Failed to connect to MQTT broker: {e}")
        exit(1)
    # Configure the Processor and the Exporter
    processor = MessageProcessor(registry, redis_client)

    mqtt_client.loop_forever()