justin/Python
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: replace sys.exit with graceful cleanup, signal handling and os._exit to prevent hanging on quit.

Browse Source
This commit is contained in:
justin committed 2026-01-26 15:29:01 -07:00
1 parent a00901efdc
commit 235b3d7be9
1 file changed
+146 -126
Download Patch File Download Diff File Expand all files Collapse all files
meshtastic-messenger.py
+146 -126
View File
@@ -5,9 +5,14 @@
# Source: https://github.com/JustinOros
import sys
import os # ← added
import time
import argparse
import signal # ← added
import atexit # ← added
import threading # ← added
from datetime import datetime, timedelta
import meshtastic.serial_interface
try:
import meshtastic.ble_interface
@@ -15,7 +20,10 @@ try:
except ImportError:
BLE_AVAILABLE = False
class MeshtasticCLI:
"""CLI wrapper around the Meshtastic Python library."""
def __init__(self, debug_mode=False):
self.interface = None
self.nodes = {}
@@ -23,26 +31,61 @@ class MeshtasticCLI:
self.exit_flag = False
self.selected_node = None
self.debug_mode = debug_mode
self.current_mode = "node_selection" # Can be "node_selection" or "messaging"
self.current_mode = "node_selection" # "node_selection" or "messaging"
# ------------------------------------------------------------------------
# EXIT / CLEAN‑UP HELPERS
# ------------------------------------------------------------------------
def _close_interface(self):
"""Close the Meshtastic interface without blocking forever."""
if not self.interface:
return
try:
# Run ``close()`` in its own thread and give it a few seconds.
# If the library hangs we simply give up – the process will be
# terminated immediately after this call.
close_thread = threading.Thread(target=self.interface.close,
daemon=True)
close_thread.start()
close_thread.join(timeout=3) # ← 3 s timeout (adjustable)
except Exception as e: # pragma: no cover
if self.debug_mode:
print(f"[DEBUG] error while closing interface: {e}")
finally:
self.interface = None
def _cleanup(self):
"""Registered with ``atexit`` – guarantees the interface is closed."""
self._close_interface()
def setup_exit_handler(self):
"""Setup exit handler"""
# Exit via /quit command
pass
"""Install SIGINT / SIGTERM handlers and atexit cleanup."""
# Catch Ctrl‑C from anywhere in the program.
signal.signal(signal.SIGINT, lambda s, f: self.exit_program())
# Catch termination from the OS (e.g. `kill`).
if hasattr(signal, "SIGTERM"):
signal.signal(signal.SIGTERM, lambda s, f: self.exit_program())
# Ensure a final cleanup runs even if we call ``sys.exit`` elsewhere.
atexit.register(self._cleanup)
def exit_program(self):
"""Exit the program gracefully"""
"""Graceful termination used by `/quit` and signal handlers."""
# Prevent duplicate printing if we get called twice.
if self.exit_flag:
return
print("\n\nExiting...")
self.exit_flag = True
if self.interface:
try:
self.interface.close()
except:
pass
sys.exit(0)
self._close_interface()
# ``os._exit`` stops the interpreter immediately – it does NOT
# execute ``finally`` blocks or other atexit handlers, which is exactly
# what we want when background threads misbehave.
os._exit(0)
# ------------------------------------------------------------------------
# INTERACTION METHODS (unchanged apart from exiting via exit_program)
# ------------------------------------------------------------------------
def scan_radios(self):
"""Scan for nearby Meshtastic radios via Bluetooth"""
"""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")
@@ -64,22 +107,18 @@ class MeshtasticCLI:
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
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
(has_emoji and len(name) <= 10 and '_' in name)
)):
meshtastic_devices.append({
'name': d.name,
'address': d.address
})
return meshtastic_devices, devices
meshtastic_devices, all_devices = asyncio.run(scan())
@@ -88,13 +127,10 @@ class MeshtasticCLI:
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
@@ -102,7 +138,6 @@ class MeshtasticCLI:
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:
@@ -114,12 +149,12 @@ class MeshtasticCLI:
return None
def connect_radio(self, devices):
"""Prompt user to select and connect to a radio"""
"""Prompt user to select and connect to a radio."""
while True:
try:
choice = input("\nEnter number to connect ('r' to retry, 'q' to quit): ").strip()
if choice.lower() == 'q':
sys.exit(0)
self.exit_program() # ← changed
elif choice.lower() == 'r':
return None # Signal to refresh scan
@@ -150,7 +185,7 @@ class MeshtasticCLI:
return False
def on_receive(self, packet, interface=None):
"""Callback for received packets"""
"""Callback for received packets."""
if self.debug_mode:
print(f"\n[DEBUG] Packet received: {packet.keys() if isinstance(packet, dict) else type(packet)}")
@@ -164,7 +199,6 @@ class MeshtasticCLI:
if 'fromId' in packet and packet['fromId']:
node_name = packet['fromId']
else:
# Try to get name from node database
node_name = node_short_id
if hasattr(self.interface, 'nodesByNum') and node_id in self.interface.nodesByNum:
node_info = self.interface.nodesByNum[node_id]
@@ -183,7 +217,6 @@ class MeshtasticCLI:
}
else:
self.nodes[node_id]['last_seen'] = datetime.now()
# Update name if we got a better one
if 'fromId' in packet and packet['fromId']:
self.nodes[node_id]['name'] = packet['fromId']
@@ -195,29 +228,26 @@ class MeshtasticCLI:
# 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
decoded.get('portnum') == 1 or
'text' in decoded):
# Extract text - handle both string and bytes
# 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', '')
payload = decoded['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
if text_content:
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'
to_short = (f"!{packet.get('to', 0):08x}"
if packet.get('to') != 0xffffffff else 'Broadcast')
msg = {
'from': from_name,
@@ -228,17 +258,18 @@ class MeshtasticCLI:
'time': datetime.now()
}
self.messages.append(msg)
# Print new messages in real-time
# Print new messages in real‑time
if self.current_mode == "messaging":
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']}")
time_str = msg['time'].strftime('%Y-%m-%d %H:%M:%S')
print(f"\n[{time_str}] {msg['from']} ({msg['from_short']}) -> "
f"{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"""
"""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}
@@ -250,39 +281,34 @@ class MeshtasticCLI:
print(f"\nActive Nodes (Seen in the last hour) ({len(recent_nodes)}):")
node_list = list(recent_nodes.values())
# Sort by last seen (most recent first)
node_list.sort(key=lambda x: x['last_seen'], reverse=True)
for i, node in enumerate(node_list, 1):
time_ago = (datetime.now() - node['last_seen']).seconds // 60
# Display the name, fall back to short ID if no name
display_name = node['name'] if node['name'] and node['name'] != node['short_id'] else node['short_id']
print(f" {i}. {display_name} (seen {time_ago}m ago)")
minutes_ago = (datetime.now() - node['last_seen']).seconds // 60
displayed = node['name'] if node['name'] and node['name'] != node['short_id'] else node['short_id']
print(f" {i}. {displayed} (seen {minutes_ago}m ago)")
return node_list
def select_contact(self, node_list):
"""Prompt user to select a contact"""
"""Prompt user to select a contact."""
while True:
try:
choice = input("\nEnter number to message ('r' to refresh, 'b' for broadcast, 'q' to quit): ").strip()
if choice.lower() == 'q':
self.exit_program()
return False
self.exit_program() # ← changed
elif choice.lower() == 'b':
self.selected_node = None
print("Broadcasting to all nodes")
return True
elif choice.lower() == 'r':
return False # Signal to refresh
return False # refresh
idx = int(choice) - 1
if 0 <= idx < len(node_list):
self.selected_node = node_list[idx]
# Use the display name for messaging
display_name = self.selected_node['name'] if self.selected_node['name'] and self.selected_node['name'] != self.selected_node['short_id'] else self.selected_node['short_id']
display_name = (self.selected_node['name']
if self.selected_node['name'] and self.selected_node['name'] != self.selected_node['short_id']
else self.selected_node['short_id'])
print(f"Messaging: {display_name}")
return True
else:
@@ -291,27 +317,26 @@ class MeshtasticCLI:
print("Please enter a valid number.")
def display_messages(self):
"""Display all past messages"""
"""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']}")
for msg in self.messages[-20:]:
t = msg['time'].strftime('%Y-%m-%d %H:%M:%S')
print(f"[{t}] {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"""
"""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'])
# Use display name for output
display_name = self.selected_node['name'] if self.selected_node['name'] and self.selected_node['name'] != self.selected_node['short_id'] else self.selected_node['short_id']
display_name = (self.selected_node['name']
if self.selected_node['name'] and self.selected_node['name'] != self.selected_node['short_id']
else self.selected_node['short_id'])
to_display = f"{display_name} ({self.selected_node['short_id']})"
print(f"[{time_str}] You -> {to_display}: {text}")
else:
@@ -322,7 +347,7 @@ class MeshtasticCLI:
print(f"Failed to send message: {e}")
def message_loop(self):
"""Main messaging loop"""
"""Main messaging loop."""
print("\n=== Messaging Interface ===")
print("Type your message and press ENTER to send")
print("Type /back to return to node selection")
@@ -332,41 +357,38 @@ class MeshtasticCLI:
try:
msg = input("> ").strip()
# Check for special commands
# Special commands
if msg.lower() == '/quit':
self.exit_program()
break
self.exit_program() # ← changed (os._exit will stop everything)
elif msg.lower() == '/back':
print("\nReturning to node selection...\n")
self.current_mode = "node_selection"
return False # Signal to go back
if self.exit_flag:
break
return False # go back to node selection
if msg:
self.send_message(msg)
except KeyboardInterrupt:
print("\n\nExiting...")
break
self.exit_program() # ← changed
except EOFError:
print("\n\nExiting...")
break
self.exit_program() # ← changed
# Clean up
if self.interface:
try:
self.interface.close()
except:
pass
return True # Signal to exit
# The loop only exits when ``self.exit_flag`` is True – which only happens via
# ``exit_program`` and therefore never reaches the code below.
# Keeping it for completeness (no‑op if we already exited).
self._close_interface()
return True
# ------------------------------------------------------------------------
# MAIN PROGRAM FLOW
# ------------------------------------------------------------------------
def run(self):
"""Main program flow"""
"""Main program flow."""
self.setup_exit_handler()
# Step 1: Scan for radios with refresh option
try:
# ------------------------------------------------------------
# Step 1 – scan for radios (with refresh option)
# ------------------------------------------------------------
while True:
devices = self.scan_radios()
if not devices:
@@ -376,26 +398,28 @@ class MeshtasticCLI:
return
continue
# Step 2: Connect to radio with refresh option
# --------------------------------------------------------
# Step 2 – connect to a radio (with refresh option)
# --------------------------------------------------------
connection_result = self.connect_radio(devices)
if connection_result is None:
# User selected 'r' to refresh
if connection_result is None: # user asked for a refresh
continue
elif connection_result:
break # Successfully connected
break # successfully connected
else:
# Connection failed
print("\nConnection failed. Press Enter to try again, or 'q' to quit: ", end='')
choice = input().strip().lower()
if choice == 'q':
return
# Setup packet callback using the correct method
# ------------------------------------------------------------
# Step 3 – set up packet receive callback
# ------------------------------------------------------------
print("Setting up message listener...")
callback_set = False
try:
# Try the pub/sub method for newer meshtastic versions
# Newer Meshtastic versions use pub/sub
from pubsub import pub
def on_receive_wrapper(packet, interface=None):
@@ -409,7 +433,7 @@ class MeshtasticCLI:
print(f"Pub/sub setup failed: {e}")
if not callback_set:
# Fallback to old callback method
# Old callback API
try:
self.interface.addReceiveCallback(self.on_receive)
print("✓ Message listener active (callback)")
@@ -417,36 +441,25 @@ class MeshtasticCLI:
except AttributeError:
print("⚠ Warning: Could not set up message callback.")
# Step 3: Wait for node discovery
# ------------------------------------------------------------
# Step 4 – give the radio time to sync & pull its node database
# ------------------------------------------------------------
print("\nDiscovering nodes and syncing data...")
print("This may take 10-20 seconds...\n")
print("This may take 10‑20 seconds...\n")
# Get nodedb to populate nodes with proper names
try:
if hasattr(self.interface, 'nodesByNum'):
for node_id, node_info in self.interface.nodesByNum.items():
user_info = node_info.get('user', {})
# Get the best available name
long_name = user_info.get('longName', '')
short_name = user_info.get('shortName', '')
# Prefer long name, then short name, then fall back to ID
if long_name:
display_name = long_name
elif short_name:
display_name = short_name
else:
display_name = f"!{node_id:08x}"
# Convert lastHeard timestamp if it exists
user = node_info.get('user', {})
long_name = user.get('longName', '')
short_name = user.get('shortName', '')
display_name = long_name or short_name or f"!{node_id:08x}"
last_seen = datetime.now()
if node_info.get('lastHeard'):
try:
last_seen = datetime.fromtimestamp(node_info['lastHeard'])
except:
except Exception:
pass
self.nodes[node_id] = {
'id': node_id,
'last_seen': last_seen,
@@ -456,39 +469,42 @@ class MeshtasticCLI:
'short_name': short_name
}
print(f"Loaded {len(self.nodes)} nodes from radio memory")
if self.debug_mode:
print("\n[DEBUG] Loaded nodes:")
for node_id, node in list(self.nodes.items())[:5]: # Show first 5 for debugging
print(f" {node['short_id']}: '{node['name']}' (long: '{node.get('long_name', '')}', short: '{node.get('short_name', '')}')")
except Exception as e:
if self.debug_mode:
print(f"Note: Could not load node database: {e}")
time.sleep(10)
time.sleep(10) # give the receive thread a chance to pick up any pending msgs
# Main loop that allows switching between modes
# ------------------------------------------------------------
# Main UI loop – switch between node‑selection and messaging
# ------------------------------------------------------------
while not self.exit_flag:
if self.current_mode == "node_selection":
node_list = self.list_recent_nodes()
if node_list is None:
time.sleep(5)
continue
if self.select_contact(node_list):
self.current_mode = "messaging"
self.display_messages()
else:
continue # Refresh node list
continue # refresh node list
elif self.current_mode == "messaging":
should_exit = self.message_loop()
if should_exit:
break # Exit program
# If we get here, user typed /back, so we'll loop again in node_selection mode
break # (will never be reached – exit_program kills the process)
# If we get here the user typed /back → loop again in node_selection
finally:
# ----------------------------------------------------------------
# ALWAYS clean up the interface – even if we never called exit_program
# ----------------------------------------------------------------
self._cleanup()
# --------------------------------------------------------------------
# Entry‑point
# --------------------------------------------------------------------
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')
@@ -498,8 +514,12 @@ if __name__ == "__main__":
cli = MeshtasticCLI(debug_mode=args.debug)
cli.run()
except KeyboardInterrupt:
# If the user hits Ctrl‑C before we have installed our signal handler
# we still want a clean shutdown.
print("\n\nExiting...")
cli._cleanup()
sys.exit(0)
except Exception as e:
print(f"\nError: {e}")
sys.exit(1)