json dumps output for better inegstion

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name="meshtastic-mqtt-json",
-    version='1.0.5',
+    version='1.0.10',
     author='acidvegas',
     author_email='acid.vegas@acid.vegas',
     description='A lightweight Python library for parsing Meshtastic MQTT messages',

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.5"
+__version__ = '1.0.10'
-__author__  = "acidvegas"
+__author__  = 'acidvegas'
-__license__ = "ISC" 
+__license__ = 'ISC'

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_connect    = self.event_mqtt_connect
-		client.on_message = self.event_mqtt_recv
+		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())
+			cipher = Cipher(algorithms.AES(self.key_bytes), modes.CTR(nonce), backend=default_backend())
-			decryptor       = cipher.decryptor()
+			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,163 +204,167 @@ class MeshtasticMQTT(object):
 				return
 
 			# Convert to JSON and handle NaN values in one shot
-			json_packet = json.loads(MessageToJson(mp, including_default_value_fields=True, 
+			json_packet = clean_json(mp)
-												 preserving_proto_field_name=True,
+			
-												 float_precision=10))
+			#print(f'Raw packet: {json_packet}')  # Debug print
-
-			# 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)
 
 			# 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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.ATAK_FORWARDER:
 				data = mesh_pb2.AtakForwarder()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.ATAK_PLUGIN:
 				data = mesh_pb2.AtakPlugin()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.AUDIO_APP:
 				data = mesh_pb2.Audio()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.DETECTION_SENSOR_APP:
 				data = mesh_pb2.DetectionSensor()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.IP_TUNNEL_APP:
 				data = mesh_pb2.IPTunnel()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
+
+			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))
+				json_packet['decoded']['payload'] = clean_json(neighborInfo)
-				print(json.dumps(json_packet))
+				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))
+				json_packet['decoded']['payload'] = clean_json(node_info)
-				print(json.dumps(json_packet))
+				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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			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))
+				json_packet['decoded']['payload'] = clean_json(position)
-				print(json.dumps(json_packet))
+				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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			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))
+				json_packet['decoded']['payload'] = clean_json(data)
-				print(json.dumps(json_packet))
+				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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.REPLY_APP:
 				data = mesh_pb2.Reply()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			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))
+				json_packet['decoded']['payload'] = clean_json(routing)
-				print(json.dumps(json_packet))
+				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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.SIMULATOR_APP:
 				data = mesh_pb2.Simulator()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			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)
-				telemetry_json = json.loads(MessageToJson(telemetry))
+				json_packet['decoded']['payload'] = clean_json(telemetry)
-				json_packet['decoded']['payload'] = replace_nan(telemetry_json)
+				print(f'{json.dumps(json_packet)}')
-				print(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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.TRACEROUTE_APP:
-					routeDiscovery = mesh_pb2.RouteDiscovery()
+				routeDiscovery = mesh_pb2.RouteDiscovery()
-					routeDiscovery.ParseFromString(mp.decoded.payload)
+				routeDiscovery.ParseFromString(mp.decoded.payload)
-					json_packet['decoded']['payload'] = json.loads(MessageToJson(routeDiscovery))
+				json_packet['decoded']['payload'] = clean_json(routeDiscovery)
-					print(json.dumps(json_packet))
+				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)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			elif mp.decoded.portnum == portnums_pb2.ZPS_APP:
 				data = mesh_pb2.Zps()
 				data.ParseFromString(mp.decoded.payload)
-				print(f'{MessageToJson(data)}')
+				msg = json.dumps(clean_json(data))
+				print(f'{msg}')
 
 			else:
 				print(f'UNKNOWN: Received Portnum name: {portnum_name}')
-				print(f'UNKNOWN: {MessageToJson(mp)}')
+				msg = json.dumps(clean_json(mp))
+				print(f'UNKNOWN: {msg}')
 
 		except Exception as e:
 			print(f'Error processing message: {e}')
@@ -332,6 +372,21 @@ 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}')
+		while True:
+			print('Attempting to reconnect...')
+			try:
+				client.reconnect()
+			except Exception as e:
+				print(f'Error reconnecting to MQTT broker: {e}')
+				time.sleep(5)
+			else:
+				print('Reconnected to MQTT broker')
+				break
+
+
 def main():
     parser = argparse.ArgumentParser(description='Meshtastic MQTT Interface')
     parser.add_argument('--broker', default='mqtt.meshtastic.org', help='MQTT broker address')
@@ -354,4 +409,4 @@ def main():
 
 
 if __name__ == '__main__':
-    main() 
+    main()