Meshtastic Messenger (CLI)
This commit is contained in:
JustinOros
@@ -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)
|
||||
Reference in New Issue
Block a user