From 222fe3a7f17b034081bcf2e63ac642c46fb63413 Mon Sep 17 00:00:00 2001 From: acidvegas Date: Fri, 3 Jan 2025 23:36:43 -0500 Subject: [PATCH] Fixed all nan values, added map report handling and unknown app processing --- setup.py | 2 +- src/meshtastic_mqtt_json/__init__.py | 6 +- src/meshtastic_mqtt_json/client.py | 164 ++++++++++++++++----------- 3 files changed, 104 insertions(+), 68 deletions(-) diff --git a/setup.py b/setup.py index 3a653a7..ec654e8 100644 --- a/setup.py +++ b/setup.py @@ -2,7 +2,7 @@ from setuptools import setup, find_packages setup( name="meshtastic-mqtt-json", - version='1.0.8', + version='1.0.9', author='acidvegas', author_email='acid.vegas@acid.vegas', description='A lightweight Python library for parsing Meshtastic MQTT messages', diff --git a/src/meshtastic_mqtt_json/__init__.py b/src/meshtastic_mqtt_json/__init__.py index 9c6cbcb..0bd9849 100644 --- a/src/meshtastic_mqtt_json/__init__.py +++ b/src/meshtastic_mqtt_json/__init__.py @@ -4,6 +4,6 @@ Meshtastic MQTT Interface - A lightweight Python library for parsing Meshtastic from .client import MeshtasticMQTT -__version__ = "1.0.8" -__author__ = "acidvegas" -__license__ = "ISC" +__version__ = '1.0.9' +__author__ = 'acidvegas' +__license__ = 'ISC' diff --git a/src/meshtastic_mqtt_json/client.py b/src/meshtastic_mqtt_json/client.py index 59205cf..e198ef9 100644 --- a/src/meshtastic_mqtt_json/client.py +++ b/src/meshtastic_mqtt_json/client.py @@ -4,6 +4,7 @@ import argparse import base64 import json +import time try: from cryptography.hazmat.backends import default_backend @@ -27,6 +28,26 @@ except ImportError: raise ImportError('missing the paho-mqtt module (pip install paho-mqtt)') +def clean_json(data) -> dict: + ''' + Clean the JSON data by replacing NaN values with null + + :param data: The JSON data to clean + ''' + # Handle protobuf messages + if hasattr(data, 'DESCRIPTOR'): + data = json.loads(MessageToJson(data)) + + # Remove empty and NaN values from the JSON data + if isinstance(data, dict): + return {k: v for k, v in ((k, clean_json(v)) for k, v in data.items()) if str(v) not in ('None', 'nan', '')} + elif isinstance(data, list): + return [v for v in (clean_json(v) for v in data) if str(v) not in ('None', 'nan', '')] + + # Return primitive types as-is + return data + + class MeshtasticMQTT(object): def __init__(self): '''Initialize the Meshtastic MQTT client''' @@ -52,6 +73,7 @@ class MeshtasticMQTT(object): # Initialize the MQTT client client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id='', clean_session=True, userdata=None) + client.connect_timeout = 10 # Set the username and password for the MQTT broker client.username_pw_set(username=username, password=password) @@ -69,11 +91,16 @@ class MeshtasticMQTT(object): raise # Set the MQTT callbacks - client.on_connect = self.event_mqtt_connect - client.on_message = self.event_mqtt_recv + client.on_connect = self.event_mqtt_connect + client.on_message = self.event_mqtt_recv + client.on_disconnect = self.event_mqtt_disconnect # Connect to the MQTT broker - client.connect(broker, port, 60) + try: + client.connect(broker, port, 60) + except Exception as e: + print(f'Error connecting to MQTT broker: {e}') + self.event_mqtt_disconnect(client, '', 1, None) # Set the subscribe topic self.subscribe_topic = f'{root}{channel}/#' @@ -88,7 +115,6 @@ class MeshtasticMQTT(object): :param mp: The message packet to decrypt ''' - try: # Extract the nonce from the packet nonce_packet_id = getattr(mp, 'id').to_bytes(8, 'little') @@ -96,22 +122,32 @@ class MeshtasticMQTT(object): nonce = nonce_packet_id + nonce_from_node # Decrypt the message - cipher = Cipher(algorithms.AES(self.key_bytes), modes.CTR(nonce), backend=default_backend()) - decryptor = cipher.decryptor() + cipher = Cipher(algorithms.AES(self.key_bytes), modes.CTR(nonce), backend=default_backend()) + decryptor = cipher.decryptor() decrypted_bytes = decryptor.update(getattr(mp, 'encrypted')) + decryptor.finalize() # Parse the decrypted message data = mesh_pb2.Data() - data.ParseFromString(decrypted_bytes) + try: + data.ParseFromString(decrypted_bytes) + except: + # Ignore this as the message does not need to be decrypted + return None + mp.decoded.CopyFrom(data) + return mp + except Exception as e: print(f'Error decrypting message: {e}') - print(mp) + print(f'Message packet details:') + print(f'- From: {getattr(mp, "from", "unknown")}') + print(f'- To: {getattr(mp, "to", "unknown")}') + print(f'- Channel: {getattr(mp, "channel", "unknown")}') + print(f'- ID: {getattr(mp, "id", "unknown")}') return None - def event_mqtt_connect(self, client, userdata, flags, rc, properties): ''' Callback for when the client receives a CONNACK response from the server. @@ -137,7 +173,7 @@ class MeshtasticMQTT(object): :param userdata: The private user data as set in Client() or user_data_set() :param msg: An instance of MQTTMessage ''' - + try: # Define the service envelope service_envelope = mqtt_pb2.ServiceEnvelope() @@ -168,160 +204,151 @@ class MeshtasticMQTT(object): return # Convert to JSON and handle NaN values in one shot - json_packet = json.loads(MessageToJson(mp)) - - # Replace all NaN values with null before any further processing - def replace_nan(obj): - ''' - Replace all NaN values with null before any further processing - - :param obj: The object to replace NaN values in - ''' - if isinstance(obj, dict): - return {k: replace_nan(v) for k, v in obj.items()} - elif isinstance(obj, list): - return [replace_nan(x) for x in obj] - elif isinstance(obj, float) and str(obj).lower() == 'nan': - return None - elif isinstance(obj, str) and obj.lower() == 'nan': - return None - - return obj - - json_packet = replace_nan(json_packet) + json_packet = clean_json(mp) + + #print(f'Raw packet: {json_packet}') # Debug print # Process the message based on its type if mp.decoded.portnum == portnums_pb2.ADMIN_APP: data = mesh_pb2.Admin() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.ATAK_FORWARDER: data = mesh_pb2.AtakForwarder() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.ATAK_PLUGIN: data = mesh_pb2.AtakPlugin() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.AUDIO_APP: data = mesh_pb2.Audio() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.DETECTION_SENSOR_APP: data = mesh_pb2.DetectionSensor() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.IP_TUNNEL_APP: data = mesh_pb2.IPTunnel() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') + + elif mp.decoded.portnum == portnums_pb2.MAP_REPORT_APP: + map_report = mesh_pb2.MapReport() + map_report.ParseFromString(mp.decoded.payload) + json_packet['decoded']['payload'] = clean_json(map_report) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.NEIGHBORINFO_APP: neighborInfo = mesh_pb2.NeighborInfo() neighborInfo.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(neighborInfo)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(neighborInfo) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.NODEINFO_APP: from_id = getattr(mp, 'from') node_info = mesh_pb2.User() node_info.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(node_info)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(node_info) + print(f'{json.dumps(json_packet)}') self.names[from_id] = node_info.long_name elif mp.decoded.portnum == portnums_pb2.PAXCOUNTER_APP: data = mesh_pb2.Paxcounter() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.POSITION_APP: position = mesh_pb2.Position() position.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(position)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(position) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.PRIVATE_APP: data = mesh_pb2.Private() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.RANGE_TEST_APP: data = mesh_pb2.RangeTest() data.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(data)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(data) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.REMOTE_HARDWARE_APP: data = mesh_pb2.RemoteHardware() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.REPLY_APP: data = mesh_pb2.Reply() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.ROUTING_APP: routing = mesh_pb2.Routing() routing.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(routing)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(routing) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.SERIAL_APP: data = mesh_pb2.Serial() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.SIMULATOR_APP: data = mesh_pb2.Simulator() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.STORE_FORWARD_APP: - print(f'{MessageToJson(mp)}') + print(f'{clean_json(mp)}') print(f'{mp.decoded.payload}') elif mp.decoded.portnum == portnums_pb2.TELEMETRY_APP: telemetry = telemetry_pb2.Telemetry() telemetry.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(telemetry)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(telemetry) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.TEXT_MESSAGE_APP: text_payload = mp.decoded.payload.decode('utf-8') json_packet['decoded']['payload'] = text_payload - print(json.dumps(json_packet)) + print(f'{json.dumps(json_packet)}') elif mp.decoded.portnum == portnums_pb2.TEXT_MESSAGE_COMPRESSED_APP: data = mesh_pb2.TextMessageCompressed() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.TRACEROUTE_APP: routeDiscovery = mesh_pb2.RouteDiscovery() routeDiscovery.ParseFromString(mp.decoded.payload) - json_packet['decoded']['payload'] = json.loads(MessageToJson(routeDiscovery)) - print(json.dumps(json_packet)) + json_packet['decoded']['payload'] = clean_json(routeDiscovery) + print(f'{json.dumps(json_packet)}') + + elif mp.decoded.portnum == portnums_pb2.UNKNOWN_APP: + print(f'{clean_json(mp)}') elif mp.decoded.portnum == portnums_pb2.WAYPOINT_APP: data = mesh_pb2.Waypoint() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') elif mp.decoded.portnum == portnums_pb2.ZPS_APP: data = mesh_pb2.Zps() data.ParseFromString(mp.decoded.payload) - print(f'{MessageToJson(data)}') + print(f'{clean_json(data)}') else: print(f'UNKNOWN: Received Portnum name: {portnum_name}') - print(f'UNKNOWN: {MessageToJson(mp)}') + print(f'UNKNOWN: {clean_json(mp)}') except Exception as e: print(f'Error processing message: {e}') @@ -329,6 +356,15 @@ class MeshtasticMQTT(object): print(f'Payload: {msg.payload}') + def event_mqtt_disconnect(self, client, userdata, rc, packet_from_broker=None, properties=None, reason_code=None): + '''Callback for when the client disconnects from the server.''' + print(f'Disconnected with result code: {rc}') + + time.sleep(5) + print('Attempting to reconnect...') + client.reconnect() + + def main(): parser = argparse.ArgumentParser(description='Meshtastic MQTT Interface') parser.add_argument('--broker', default='mqtt.meshtastic.org', help='MQTT broker address') @@ -351,4 +387,4 @@ def main(): if __name__ == '__main__': - main() \ No newline at end of file + main()