diff --git a/meshtastic-messenger.py b/meshtastic-messenger.py new file mode 100644 index 0000000..c17c5af --- /dev/null +++ b/meshtastic-messenger.py @@ -0,0 +1,421 @@ +#!/usr/bin/env python3 +# Description: Meshtastic Bluetooth Messenger (CLI) - Connect to and message via Meshtastic radios +# Usage: python3 meshtastic-messenger.py +# Author: Justin Oros +# Source: https://github.com/JustinOros + +import sys +import time +import argparse +from datetime import datetime, timedelta +import meshtastic.serial_interface +try: + import meshtastic.ble_interface + BLE_AVAILABLE = True +except ImportError: + BLE_AVAILABLE = False + +class MeshtasticCLI: + def __init__(self, debug_mode=False): + self.interface = None + self.nodes = {} + self.messages = [] + self.exit_flag = False + self.selected_node = None + self.debug_mode = debug_mode + + def setup_exit_handler(self): + """Setup exit handler""" + # Exit via /quit command + pass + + def exit_program(self): + """Exit the program gracefully""" + print("\n\nExiting...") + self.exit_flag = True + if self.interface: + try: + self.interface.close() + except: + pass + sys.exit(0) + + def scan_radios(self): + """Scan for nearby Meshtastic radios via Bluetooth""" + print("Scanning for Meshtastic radios via Bluetooth (10 seconds)...") + print("Make sure your radio's Bluetooth is enabled!\n") + + try: + import asyncio + from bleak import BleakScanner + + async def scan(): + print("Scanning all Bluetooth devices...") + devices = await BleakScanner.discover(timeout=10.0) + + if self.debug_mode: + print(f"\nFound {len(devices)} total Bluetooth devices") + print("\nAll devices (for debugging):") + for d in devices: + print(f" - {d.name or 'Unknown'} ({d.address})") + + # Filter for Meshtastic devices (very permissive) + meshtastic_devices = [] + for d in devices: + name = d.name or "" + # Match Meshtastic patterns including emoji prefixes + # Common Meshtastic emojis: 📡 🏠 🚀 🌐 📻 and others + has_emoji = any(ord(c) > 127 for c in name) # Simple emoji detection + + if (name and ( + 'meshtastic' in name.lower() or + name.startswith('!') or + 'mesh' in name.lower() or + name.startswith('Meshtastic') or + (has_emoji and len(name) <= 10 and '_' in name) # Emoji_XXXX pattern + )): + meshtastic_devices.append({ + 'name': d.name, + 'address': d.address + }) + + return meshtastic_devices, devices + + meshtastic_devices, all_devices = asyncio.run(scan()) + + if not meshtastic_devices: + print("\nNo Meshtastic radios auto-detected.") + print("Would you like to manually select a device? (y/n): ", end='') + choice = input().strip().lower() + + if choice == 'y': + # Let user pick from all devices + print("\nAll available devices:") + for i, d in enumerate(all_devices, 1): + print(f" {i}. {d.name or 'Unknown'} - {d.address}") + + return [{'name': d.name or 'Unknown', 'address': d.address} + for d in all_devices] + return None + + print(f"\nFound {len(meshtastic_devices)} Meshtastic radio(s):") + for i, device in enumerate(meshtastic_devices, 1): + print(f" {i}. {device['name']} - {device['address']}") + + return meshtastic_devices + + except Exception as e: + print(f"Bluetooth scan error: {e}") + print("\nTroubleshooting tips:") + print(" - Ensure Bluetooth is enabled on your computer") + print(" - Make sure your Meshtastic radio's Bluetooth is on") + print(" - Try running with sudo on Linux/Mac") + return None + + def connect_radio(self, devices): + """Prompt user to select and connect to a radio""" + while True: + try: + choice = input("\nEnter number to connect (or 'q' to quit): ").strip() + if choice.lower() == 'q': + sys.exit(0) + + idx = int(choice) - 1 + if 0 <= idx < len(devices): + device = devices[idx] + print(f"\nConnecting to {device['name']}...") + + # Try BLE interface + if BLE_AVAILABLE: + try: + self.interface = meshtastic.ble_interface.BLEInterface(device['address']) + time.sleep(3) + print("Connected via BLE!") + return True + except Exception as e: + print(f"BLE connection failed: {e}") + return False + else: + print("BLE interface not available. Install bleak: pip install bleak") + return False + else: + print("Invalid selection. Try again.") + except ValueError: + print("Please enter a valid number.") + except Exception as e: + print(f"Connection failed: {e}") + return False + + def on_receive(self, packet, interface=None): + """Callback for received packets""" + if self.debug_mode: + print(f"\n[DEBUG] Packet received: {packet.keys() if isinstance(packet, dict) else type(packet)}") + + try: + # Store node info + if 'from' in packet: + node_id = packet['from'] + node_short_id = f"!{node_id:08x}" + node_long_name = packet.get('fromId', node_short_id) + + if node_id not in self.nodes: + self.nodes[node_id] = { + 'id': node_id, + 'last_seen': datetime.now(), + 'name': node_long_name, + 'short_id': node_short_id + } + else: + self.nodes[node_id]['last_seen'] = datetime.now() + if 'fromId' in packet: + self.nodes[node_id]['name'] = packet['fromId'] + + # Store and display messages + if 'decoded' in packet: + decoded = packet['decoded'] + if self.debug_mode: + print(f"[DEBUG] Decoded packet: portnum={decoded.get('portnum')}, has text={('text' in decoded)}") + + # Check if it's a text message + if (decoded.get('portnum') == 'TEXT_MESSAGE_APP' or + decoded.get('portnum') == 1 or # TEXT_MESSAGE_APP enum value + 'text' in decoded): + + # Extract text - handle both string and bytes + text_content = decoded.get('text', '') + + # If text is bytes, decode it + if isinstance(text_content, bytes): + text_content = text_content.decode('utf-8', errors='ignore') + + # If still no text, try payload + if not text_content and 'payload' in decoded: + payload = decoded.get('payload', '') + if isinstance(payload, bytes): + text_content = payload.decode('utf-8', errors='ignore') + elif isinstance(payload, dict): + text_content = payload.get('text', '') + + if text_content: # Only store if there's actual text + from_name = packet.get('fromId', 'Unknown') + from_short = f"!{packet.get('from', 0):08x}" + to_name = packet.get('toId', 'Broadcast') + to_short = f"!{packet.get('to', 0):08x}" if packet.get('to') != 0xffffffff else 'Broadcast' + + msg = { + 'from': from_name, + 'from_short': from_short, + 'to': to_name, + 'to_short': to_short, + 'text': text_content, + 'time': datetime.now() + } + self.messages.append(msg) + # Print new messages in real-time + time_str = msg['time'].strftime('%Y-%m-%d %H:%M:%S') + print(f"\n[{time_str}] {msg['from']} ({msg['from_short']}) -> {msg['to']} ({msg['to_short']}): {msg['text']}") + print("> ", end='', flush=True) + except Exception as e: + if self.debug_mode: + print(f"[DEBUG] Error processing packet: {e}") + + def list_recent_nodes(self): + """List nodes seen in the last hour""" + one_hour_ago = datetime.now() - timedelta(hours=1) + recent_nodes = {k: v for k, v in self.nodes.items() + if v['last_seen'] > one_hour_ago} + + if not recent_nodes: + print("\nNo nodes detected in the last hour.") + print("Waiting for node activity...") + return None + + print(f"\nNodes active in the last hour ({len(recent_nodes)}):") + node_list = list(recent_nodes.values()) + for i, node in enumerate(node_list, 1): + time_ago = (datetime.now() - node['last_seen']).seconds // 60 + print(f" {i}. {node['name']} (seen {time_ago}m ago)") + + return node_list + + def select_contact(self, node_list): + """Prompt user to select a contact""" + while True: + try: + choice = input("\nEnter number to message (or 'b' for broadcast, 'r' to refresh): ").strip() + + if choice.lower() == 'b': + self.selected_node = None + print("Broadcasting to all nodes") + return True + elif choice.lower() == 'r': + return False # Signal to refresh + + idx = int(choice) - 1 + if 0 <= idx < len(node_list): + self.selected_node = node_list[idx] + print(f"Messaging: {self.selected_node['name']}") + return True + else: + print("Invalid selection. Try again.") + except ValueError: + print("Please enter a valid number.") + + def display_messages(self): + """Display all past messages""" + if not self.messages: + print("\nNo messages yet.") + return + + print(f"\n--- Message History ({len(self.messages)} messages) ---") + for msg in self.messages[-20:]: # Show last 20 messages + time_str = msg['time'].strftime('%Y-%m-%d %H:%M:%S') + print(f"[{time_str}] {msg['from']} ({msg['from_short']}) -> {msg['to']} ({msg['to_short']}): {msg['text']}") + print("--- End of messages ---\n") + + def send_message(self, text): + """Send a message""" + try: + time_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + if self.selected_node: + # Direct message + self.interface.sendText(text, destinationId=self.selected_node['id']) + short_id = self.selected_node.get('short_id', f"!{self.selected_node['id']:08x}") + to_display = f"{self.selected_node['name']} ({short_id})" + print(f"[{time_str}] You -> {to_display}: {text}") + else: + # Broadcast + self.interface.sendText(text) + print(f"[{time_str}] You -> Broadcast: {text}") + except Exception as e: + print(f"Failed to send message: {e}") + + def message_loop(self): + """Main messaging loop""" + print("\n=== Messaging Interface ===") + print("Type your message and press ENTER to send") + print("Type /quit to exit\n") + + while not self.exit_flag: + try: + msg = input("> ").strip() + + # Check for quit command + if msg.lower() == '/quit': + self.exit_program() + break + + if self.exit_flag: + break + + if msg: + self.send_message(msg) + + except KeyboardInterrupt: + print("\n\nExiting...") + break + except EOFError: + print("\n\nExiting...") + break + + # Clean up + if self.interface: + try: + self.interface.close() + except: + pass + + def run(self): + """Main program flow""" + self.setup_exit_handler() + + # Step 1: Scan for radios + devices = self.scan_radios() + if not devices: + return + + # Step 2: Connect to radio + if not self.connect_radio(devices): + return + + # Setup packet callback using the correct method + print("Setting up message listener...") + callback_set = False + + try: + # Try the pub/sub method for newer meshtastic versions + from pubsub import pub + + def on_receive_wrapper(packet, interface=None): + self.on_receive(packet, interface) + + pub.subscribe(on_receive_wrapper, "meshtastic.receive") + print("✓ Message listener active (pub/sub)") + callback_set = True + except Exception as e: + if self.debug_mode: + print(f"Pub/sub setup failed: {e}") + + if not callback_set: + # Fallback to old callback method + try: + self.interface.addReceiveCallback(self.on_receive) + print("✓ Message listener active (callback)") + callback_set = True + except AttributeError: + print("⚠ Warning: Could not set up message callback.") + + # Step 3: Wait for node discovery + print("\nDiscovering nodes and syncing data...") + print("This may take 10-20 seconds...\n") + + # Get nodedb to populate nodes + try: + if hasattr(self.interface, 'nodesByNum'): + for node_id, node_info in self.interface.nodesByNum.items(): + self.nodes[node_id] = { + 'id': node_id, + 'last_seen': datetime.fromtimestamp(node_info.get('lastHeard', 0)) if node_info.get('lastHeard') else datetime.now(), + 'name': node_info.get('user', {}).get('longName', f"!{node_id:08x}"), + 'short_id': f"!{node_id:08x}" + } + print(f"Loaded {len(self.nodes)} nodes from radio memory") + except Exception as e: + if self.debug_mode: + print(f"Note: Could not load node database: {e}") + + time.sleep(10) + + # Step 4: Select contact + while not self.exit_flag: + node_list = self.list_recent_nodes() + if node_list is None: + time.sleep(5) + continue + + if self.select_contact(node_list): + break + + # Step 5: Display message history + self.display_messages() + + # Step 6: Enter messaging loop + self.message_loop() + +if __name__ == "__main__": + # Parse command line arguments + parser = argparse.ArgumentParser(description='Meshtastic Bluetooth CLI Messenger') + parser.add_argument('-d', '--debug', action='store_true', + help='Enable debug mode to see packet information') + args = parser.parse_args() + + try: + cli = MeshtasticCLI(debug_mode=args.debug) + cli.run() + except KeyboardInterrupt: + print("\n\nExiting...") + sys.exit(0) + except Exception as e: + print(f"\nError: {e}") + sys.exit(1)