#!/usr/bin/env python
 # Meshtastic Serial Interface - Developed by Acidvegas in Python (https://git.acid.vegas)
 
 import argparse
 import logging
 import os
 import time
 
 try:
 	import meshtastic
 except ImportError:
 	raise ImportError('meshtastic library not found (pip install meshtastic)')
 
 try:
 	from pubsub import pub
 except ImportError:
 	raise ImportError('pubsub library not found (pip install pypubsub)')
 
 
 # Initialize logging
 logging.basicConfig(level=logging.INFO, format='%(asctime)s | %(levelname)9s | %(funcName)s | %(message)s', datefmt='%Y-%m-%d %I:%M:%S')
 
 
 def now():
 	'''Returns the current date and time in a formatted string'''
 
 	return time.strftime('%Y-%m-%d %H:%M:%S')
 
 
 class MeshtasticClient(object):
 	def __init__(self, option: str, value: str):
 		self.interface = None
 		self.me        = {}
 		self.nodes     = {}
 
 		self.interface_option = option
 		self.interface_value  = value
 
 		if self.interface_option == 'serial':
 			from meshtastic.serial_interface import SerialInterface as MeshInterface
 			from meshtastic.util import findPorts
 		elif self.interface_option == 'tcp':
 			from meshtastic.tcp_interface import TCPInterface as MeshInterface
 
 
 	def connect(self, option: str, value: str):
 		'''
   		Connect to the Meshtastic interface
     
     	:param option: The interface option to connect to
 		:param value:  The value of the interface option
      	'''
   
 		while True:
 			try:
 				if option == 'serial':
 					if devices := findPorts():
 						if not os.path.exists(args.serial) or not args.serial in devices:
 							raise Exception(f'Invalid serial port specified: {args.serial} (Available: {devices})')
 					else:
 						raise Exception('No serial devices found')
 					self.interface = SerialInterface(value)
 		
 				elif option == 'tcp':
 					self.interface = TCPInterface(value)
 
 				else:
 					raise SystemExit('Invalid interface option')
 
 			except Exception as e:
 				logging.error(f'Failed to connect to the radio: {e}')
 				logging.error('Retrying in 15 seconds...')
 				time.sleep(15)
 
 			else:
 				self.me = self.interface.getMyNodeInfo()
 				break
 
 
 	def send(self, message: str):
 		'''
 		Send a message to the Meshtastic interface
 
 		:param message: The message to send
 		'''
 
 		if len(message) > 255:
 			logging.warning('Message exceeds 255 characters')
 			message = message[:255]
 
 		self.interface.sendText(message)
 
 		logging.info(f'Sent broadcast message: {message}')
 
 
 	def listen(self):
 		'''Create the Meshtastic callback subscriptions'''
 
 		pub.subscribe(self.event_connect,    'meshtastic.connection.established')
 		pub.subscribe(self.event_data,       'meshtastic.receive.data.portnum')
 		pub.subscribe(self.event_disconnect, 'meshtastic.connection.lost')
 		pub.subscribe(self.event_node,       'meshtastic.node')
 		pub.subscribe(self.event_position,   'meshtastic.receive.position')
 		pub.subscribe(self.event_text,       'meshtastic.receive.text')
 		pub.subscribe(self.event_user,       'meshtastic.receive.user')
 
 		logging.debug('Listening for Meshtastic events...')
 		
 		
 	def event_connect(self, interface, topic=pub.AUTO_TOPIC):
 		'''
 		Callback function for connection established
 
 		:param interface: Meshtastic interface
 		:param topic:     PubSub topic
 		'''
 
 		logging.info(f'Connected to the {self.me["user"]["longName"]} radio on {self.me["user"]["hwModel"]} hardware')
 		logging.info(f'Found a total of {len(self.nodes):,} nodes')
 
 
 	def event_data(self, packet: dict, interface):
 		'''
 		Callback function for data updates
 
 		:param packet: Data information
 		:param interface: Meshtastic interface
 		'''
 
 		logging.info(f'Data update: {data}')
 
 
 	def event_disconnect(self, interface, topic=pub.AUTO_TOPIC):
 		'''
 		Callback function for connection lost
 
 		:param interface: Meshtastic interface
 		:param topic:     PubSub topic
 		'''
 
 		logging.warning('Lost connection to radio!')
 
 		time.sleep(10)
 
 		 # TODO: Consider storing the interface option and value in a class variable since we don't want to reference the args object inside the class
 		self.connect('serial' if args.serial else 'tcp', args.serial if args.serial else args.tcp)
 
 	
 	def event_node(self, node):
 		'''
 		Callback function for node updates
 
 		:param node: Node information
 		'''
 
 		# Node ID Formula = f'!{hex(node_num)[2:]}'
 
 		self.nodes[node['num']] = node
 
 		logging.info(f'Node found: {node["user"]["id"]} - {node["user"]["shortName"].ljust(4)} - {node["user"]["longName"]}')
 		print(node)
 
 
 	def event_position(self, packet: dict, interface):
 		'''
 		Callback function for position updates
 
 		:param packet: Position information
 		:param interface: Meshtastic interface
 		'''
 
 		sender    = packet['from']
 		msg       = packet['decoded']['payload'].hex()
 		id        = self.nodes[sender]['user']['id']       if sender in self.nodes else '!unk   '
 		name      = self.nodes[sender]['user']['longName'] if sender in self.nodes else 'UNK'
 		longitude = packet['decoded']['position']['longitudeI'] / 1e7
 		latitude  = packet['decoded']['position']['latitudeI'] / 1e7
 		altitude  = packet['decoded']['position']['altitude']
 		snr	      = packet['rxSnr']
 		rssi	  = packet['rxRssi']
 
 		logging.info(f'{id} - {name}: {longitude}, {latitude}, {altitude}m (SNR: {snr}, RSSI: {rssi}) - {msg}')
 
 
 	def event_text(self, packet: dict, interface):
 		'''
 		Callback function for received packets
 
 		:param packet: Packet received
 		'''
 
 		sender = packet['from']
 		to     = packet['to'] 
 		msg    = packet['decoded']['payload'].decode('utf-8')
 		id     = self.nodes[sender]['user']['id']       if sender in self.nodes else '!unk   '
 		name   = self.nodes[sender]['user']['longName'] if sender in self.nodes else 'UNK'
 		target = self.nodes[to]['user']['longName']     if to     in self.nodes else 'UNK'
 
 		logging.info(f'{id} {name} -> {target}: {msg}')
 		print(packet)
 
 
 	def event_user(self, packet: dict, interface):
 		'''
 		Callback function for user updates
 
 		:param user: User information
 		'''
 
 		'''
 		{
 			'from' : 862341900,
 			'to'   : 4294967295,
 			'decoded' : {
 				'portnum'      : 'NODEINFO_APP',
 				'payload'      : b'\n\t!33664b0c\x12\x08HELLDIVE\x1a\x04H3LL"\x06d\xe83fK\x0c(+8\x03',
 				'wantResponse' : True,
 				'user' : {
 					'id'        : '!33664b0c',
 					'longName'  : 'HELLDIVE',
 					'shortName' : 'H3LL',
 					'macaddr'   : 'ZOgzZksM',
 					'hwModel'   : 'HELTEC_V3',
 					'role'      : 'ROUTER_CLIENT',
 					'raw'       : 'rm this'
 				}
 			},
 			'id'       : 1612906268,
 			'rxTime'   : 1714279638,
 			'rxSnr'    : 6.25,
 			'hopLimit' : 3,
 			'rxRssi'   : -38,
 			'hopStart' : 3,
 			'raw'      : 'rm this'
 		}
 		'''
 
 		# Not sure what to do with this yet...
 		pass
 
 
 
 if __name__ == '__main__':
 	parser = argparse.ArgumentParser(description='Meshtastic Interfacing Tool')
 	parser.add_argument('--serial', help='Use serial interface') # Typically /dev/ttyUSB0 or /dev/ttyACM0
 	parser.add_argument('--tcp',    help='Use TCP interface')    # Can be an IP address or hostname (meshtastic.local)
 	args = parser.parse_args()
  
 	# Ensure one interface is specified
 	if (not args.serial and not args.tcp) or (args.serial and args.tcp):
 		raise SystemExit('Must specify either --serial or --tcp interface')
 	
 	# Initialize the Meshtastic client
 	mesh = MeshtasticClient()
 
 	# Listen for Meshtastic events
 	mesh.listen()
 
 	# Connect to the Meshtastic interface
 	mesh.connect('serial' if args.serial else 'tcp', args.serial if args.serial else args.tcp)
 
 	# Keep-alive loop
 	try:
 		while True:
 			time.sleep(60)
 	except KeyboardInterrupt:
 		try:
 			mesh.interface.close()
 		except:
 			pass
 	finally:
 		logging.info('Connection to radio lost')
 
 '''
 Notes:
 		conf = self.interface.localNode.localConfig
 		ok = interface.getNode('^local')
 		print(ok.channels)
 '''

© 2023 acidvegas, inc • generated with stagit