adam/contact
1
0
Fork 0
mirror of https://github.com/pdxlocations/contact.git synced 2026-08-03 15:33:49 +02:00
Code Issues Packages Projects Releases Wiki Activity

Major Refactor - Part 1 (#16)

Browse Source 
* begin refactor

* continue refactor - notifications not working

* refactor - fix notif - chanels broken

* refactor - settings broken

* working refactor

* continue refactor

* remove unused import
This commit is contained in:
pdxlocations
2024-12-19 14:00:05 -08:00
committed by GitHub
parent e5fa27d2bc
commit 6732180ca9
10 changed files with 603 additions and 604 deletions
Download Patch File Download Diff File Expand all files Collapse all files
curses-client.py
-555
View File
@@ -1,555 +0,0 @@
#!/usr/bin/env python3
'''
Curses Client for Meshtastic by http://github.com/pdxlocations
Powered by Meshtastic.org
V 0.1.8
'''
import curses
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
from pubsub import pub
import textwrap # Import the textwrap module
try:
from meshtastic.protobuf import config_pb2
from meshtastic import BROADCAST_NUM
except ImportError:
from meshtastic import config_pb2, BROADCAST_NUM
from settings import settings
import argparse
parser = argparse.ArgumentParser(
add_help=False,
epilog="If no connection arguments are specified, we attempt a serial connection and then a TCP connection to localhost.")
connOuter = parser.add_argument_group('Connection', 'Optional arguments to specify a device to connect to and how.')
conn = connOuter.add_mutually_exclusive_group()
conn.add_argument(
"--port",
"--serial",
"-s",
help="The port to connect to via serial, e.g. `/dev/ttyUSB0`.",
nargs="?",
default=None,
const=None,
)
conn.add_argument(
"--host",
"--tcp",
"-t",
help="The hostname or IP address to connect to using TCP.",
nargs="?",
default=None,
const="localhost",
)
conn.add_argument(
"--ble",
"-b",
help="The BLE device MAC address or name to connect to.",
nargs="?",
default=None,
const="any"
)
args = parser.parse_args()
# Initialize Meshtastic interface
if args.ble:
interface = meshtastic.ble_interface.BLEInterface(args.ble if args.ble != "any" else None)
elif args.host:
interface = meshtastic.tcp_interface.TCPInterface(args.host)
else:
try:
interface = meshtastic.serial_interface.SerialInterface(args.port)
except PermissionError as ex:
print("You probably need to add yourself to the `dialout` group to use a serial connection.")
if interface.devPath is None:
interface = meshtastic.tcp_interface.TCPInterface("meshtastic.local")
myinfo = interface.getMyNodeInfo()
myNodeNum = myinfo['num']
all_messages = {}
channel_list = []
selected_channel = 0
selected_node = 0
direct_message = False
packet_buffer = []
display_log = False
def get_channels():
global channel_list
node = interface.getNode('^local')
device_channels = node.channels
channel_output = []
for device_channel in device_channels:
if device_channel.role:
if device_channel.settings.name:
channel_output.append(device_channel.settings.name)
all_messages[device_channel.settings.name] = []
else:
# If channel name is blank, use the modem preset
lora_config = node.localConfig.lora
modem_preset_enum = lora_config.modem_preset
modem_preset_string = config_pb2._CONFIG_LORACONFIG_MODEMPRESET.values_by_number[modem_preset_enum].name
channel_output.append(convert_to_camel_case(modem_preset_string))
all_messages[convert_to_camel_case(modem_preset_string)] = []
channel_list = list(all_messages.keys())
def get_node_list():
node_list = []
if interface.nodes:
for node in interface.nodes.values():
node_list.append(node['num'])
return node_list
def decimal_to_hex(decimal_number):
return f"!{decimal_number:08x}"
def convert_to_camel_case(string):
words = string.split('_')
camel_case_string = ''.join(word.capitalize() for word in words)
return camel_case_string
def get_name_from_number(number, type='long'):
name = ""
for node in interface.nodes.values():
if number == node['num']:
if type == 'long':
name = node['user']['longName']
return name
elif type == 'short':
name = node['user']['shortName']
return name
else:
pass
else:
name = str(decimal_to_hex(number)) # If long name not found, use the ID as string
return name
def on_receive(packet, interface):
global all_messages, selected_channel, channel_list, packet_buffer
# update packet log
packet_buffer.append(packet)
if len(packet_buffer) > 20:
# trim buffer to 20 packets
packet_buffer = packet_buffer[-20:]
if display_log:
update_packetlog_win()
try:
if 'decoded' in packet and packet['decoded']['portnum'] == 'NODEINFO_APP':
get_node_list()
draw_node_list()
elif 'decoded' in packet and packet['decoded']['portnum'] == 'TEXT_MESSAGE_APP':
message_bytes = packet['decoded']['payload']
message_string = message_bytes.decode('utf-8')
if packet.get('channel'):
channel_number = packet['channel']
else:
channel_number = 0
if packet['to'] == myNodeNum:
if packet['from'] in channel_list:
pass
else:
channel_list.append(packet['from'])
all_messages[packet['from']] = []
draw_channel_list()
channel_number = channel_list.index(packet['from'])
if channel_list[channel_number] != channel_list[selected_channel]:
add_notification(channel_number)
# Add received message to the messages list
message_from_id = packet['from']
message_from_string = ""
for node in interface.nodes.values():
if message_from_id == node['num']:
message_from_string = node["user"]["longName"] # Get the long name using the node ID
break
else:
message_from_string = str(decimal_to_hex(message_from_id)) # If long name not found, use the ID as string
if channel_list[channel_number] in all_messages:
all_messages[channel_list[channel_number]].append((f">> {message_from_string} ", message_string))
else:
all_messages[channel_list[channel_number]] = [(f">> {message_from_string} ", message_string)]
draw_channel_list()
update_messages_window()
except KeyError as e:
print(f"Error processing packet: {e}")
def send_message(message, destination=BROADCAST_NUM, channel=0):
global all_messages, channel_list
# FIXME if sending a DM, always send on channel 0
send_on_channel = 0
if isinstance(channel_list[channel], int):
send_on_channel = 0
destination = channel_list[channel]
elif isinstance(channel_list[channel], str):
send_on_channel = channel
interface.sendText(
text=message,
destinationId=destination,
wantAck=False,
wantResponse=False,
onResponse=None,
channelIndex=send_on_channel,
)
# Add sent message to the messages dictionary
if channel_list[channel] in all_messages:
all_messages[channel_list[channel]].append((">> Sent: ", message))
else:
all_messages[channel_list[channel]] = [(">> Sent: ", message)]
update_messages_window()
messages_win.refresh()
def add_notification(channel_number):
global channel_win
_, win_width = channel_win.getmaxyx() # Get the width of the channel window
if isinstance(channel_list[channel_number], str):
channel_name = channel_list[channel_number]
elif isinstance(channel_list[channel_number], int):
channel_name = get_name_from_number(channel_list[channel_number])
# Truncate the channel name if it's too long to fit in the window
truncated_channel_name = channel_name[:win_width - 5] + '-' if len(channel_name) > win_width - 5 else channel_name
channel_win.addstr(channel_number + 1, len(str(truncated_channel_name))+1, " *", curses.color_pair(4))
channel_win.refresh()
def remove_notification(channel_number):
global channel_win
_, win_width = channel_win.getmaxyx() # Get the width of the channel window
if isinstance(channel_list[channel_number], str):
channel_name = channel_list[channel_number]
elif isinstance(channel_list[channel_number], int):
channel_name = get_name_from_number(channel_list[channel_number])
# Truncate the channel name if it's too long to fit in the window
truncated_channel_name = channel_name[:win_width - 5] + '-' if len(channel_name) > win_width - 5 else channel_name
channel_win.addstr(channel_number + 1, len(str(truncated_channel_name))+1, " ", curses.color_pair(4))
channel_win.refresh()
def update_messages_window():
global all_messages, selected_channel, messages_win
messages_win.clear()
# Calculate how many messages can fit in the window
max_messages = messages_win.getmaxyx()[0] - 2 # Subtract 2 for the top and bottom border
# Determine the starting index for displaying messages
if channel_list[selected_channel] in all_messages:
start_index = max(0, len(all_messages[channel_list[selected_channel]]) - max_messages)
else:
# Handle the case where selected_channel does not exist
start_index = 0 # Set start_index to 0 or any other appropriate value
# Display messages starting from the calculated start index
# Check if selected_channel exists in all_messages before accessing it
if channel_list[selected_channel] in all_messages:
row = 1
for _, (prefix, message) in enumerate(all_messages[channel_list[selected_channel]][start_index:], start=1):
full_message = f"{prefix}{message}"
wrapped_messages = textwrap.wrap(full_message, messages_win.getmaxyx()[1] - 2)
for wrapped_message in wrapped_messages:
messages_win.addstr(row, 1, wrapped_message, curses.color_pair(1) if prefix.startswith(">> Sent:") else curses.color_pair(2))
row += 1
messages_win.box()
messages_win.refresh()
update_packetlog_win()
def update_packetlog_win():
if display_log:
packetlog_win.clear()
packetlog_win.box()
# Get the dimensions of the packet log window
height, width = packetlog_win.getmaxyx()
# Add headers
headers = f"{'From':<20} {'To':<20} {'Port':<15} {'Payload':<{width-55}}"
packetlog_win.addstr(1, 1, headers[:width - 2],curses.A_UNDERLINE) # Truncate headers if they exceed window width
for i, packet in enumerate(reversed(packet_buffer)):
if i >= height - 3: # Skip if exceeds the window height
break
# Format each field
from_id = get_name_from_number(packet['from']).ljust(20)
to_id = (
"BROADCAST".ljust(20) if str(packet['to']) == "4294967295"
else get_name_from_number(packet['to']).ljust(20)
)
if 'decoded' in packet:
port = packet['decoded']['portnum'].ljust(15)
payload = (packet['decoded']['payload']).ljust(30)
else:
port = "NO KEY".ljust(15)
payload = "NO KEY".ljust(30)
# Combine and truncate if necessary
logString = f"{from_id} {to_id} {port} {payload}"
logString = logString[:width - 3]
# Add to the window
packetlog_win.addstr(i + 2, 1, logString)
packetlog_win.refresh()
def draw_text_field(win, text):
win.clear()
win.border()
win.addstr(1, 1, text)
def draw_channel_list():
global direct_message
# Get the dimensions of the channel window
_, win_width = channel_win.getmaxyx()
for i, (channel, message_list) in enumerate(all_messages.items()):
# Convert node number to long name if it's an integer
if isinstance(channel, int):
channel = get_name_from_number(channel, type='long')
# Truncate the channel name if it's too long to fit in the window
truncated_channel = channel[:win_width - 5] + '-' if len(channel) > win_width - 5 else channel
if selected_channel == i and not direct_message:
channel_win.addstr(i + 1, 1, truncated_channel, curses.color_pair(3))
remove_notification(selected_channel)
else:
channel_win.addstr(i + 1, 1, truncated_channel, curses.color_pair(4))
channel_win.refresh()
def draw_node_list():
global selected_node, direct_message
nodes_win.clear()
height, width = nodes_win.getmaxyx()
start_index = max(0, selected_node - (height - 3)) # Calculate starting index based on selected node and window height
for i, node in enumerate(get_node_list()[start_index:], start=1):
if i < height - 1 : # Check if there is enough space in the window
if selected_node + 1 == start_index + i and direct_message:
nodes_win.addstr(i, 1, get_name_from_number(node, "long"), curses.color_pair(3))
else:
nodes_win.addstr(i, 1, get_name_from_number(node, "long"), curses.color_pair(4))
nodes_win.box()
nodes_win.refresh()
def draw_debug(value):
function_win.addstr(1, 100, f"debug: {value} ")
function_win.refresh()
def select_channels(direction):
global selected_channel
channel_list_length = len(channel_list)
selected_channel += direction
if selected_channel < 0:
selected_channel = channel_list_length - 1
elif selected_channel >= channel_list_length:
selected_channel = 0
draw_channel_list()
update_messages_window()
def select_nodes(direction):
global selected_node
node_list_length = len(get_node_list())
selected_node += direction
if selected_node < 0:
selected_node = node_list_length - 1
elif selected_node >= node_list_length:
selected_node = 0
draw_node_list()
def main(stdscr):
global messages_win, nodes_win, channel_win, function_win, selected_node, selected_channel, direct_message, packetlog_win, display_log
stdscr.keypad(True)
# Initialize colors
curses.start_color()
curses.init_pair(1, curses.COLOR_CYAN, curses.COLOR_BLACK)
curses.init_pair(2, curses.COLOR_YELLOW, curses.COLOR_BLACK)
curses.init_pair(3, curses.COLOR_BLACK, curses.COLOR_WHITE)
curses.init_pair(4, curses.COLOR_WHITE, curses.COLOR_BLACK)
curses.init_pair(5, curses.COLOR_RED, curses.COLOR_BLACK)
# Calculate window max dimensions
height, width = stdscr.getmaxyx()
# Define window dimensions and positions
entry_win = curses.newwin(3, width, 0, 0)
channel_width = 3 * (width // 16)
nodes_width = 5 * (width // 16)
messages_width = width - channel_width - nodes_width
channel_win = curses.newwin(height - 6, channel_width, 3, 0)
messages_win = curses.newwin(height - 6, messages_width, 3, channel_width)
packetlog_win = curses.newwin(int(height / 3), messages_width, height - int(height / 3) - 3, channel_width)
nodes_win = curses.newwin(height - 6, nodes_width, 3, channel_width + messages_width)
function_win = curses.newwin(3, width, height - 3, 0)
draw_text_field(function_win, f"↑↓ = Switch Channels ← → = Channels/Nodes ENTER = Send / Select DM ` = Settings / = Display Packet Log ESC = Quit")
# Enable scrolling for messages and nodes windows
messages_win.scrollok(True)
nodes_win.scrollok(True)
channel_win.scrollok(True)
get_channels()
channel_win.refresh()
draw_channel_list()
draw_node_list()
# Draw boxes around windows
channel_win.box()
entry_win.box()
messages_win.box()
nodes_win.box()
function_win.box()
# Refresh all windows
entry_win.refresh()
messages_win.refresh()
nodes_win.refresh()
channel_win.refresh()
function_win.refresh()
input_text = ""
direct_message = False
entry_win.keypad(True)
while True:
draw_text_field(entry_win, f"Input: {input_text}")
# Get user input from entry window
entry_win.move(1, len(input_text) + 8)
char = entry_win.getch()
# draw_debug(f"Keypress: {char}")
if char == curses.KEY_UP:
if direct_message:
draw_channel_list()
select_nodes(-1)
else:
select_channels(-1)
elif char == curses.KEY_DOWN:
if direct_message:
draw_channel_list()
select_nodes(1)
else:
select_channels(1)
elif char == curses.KEY_LEFT:
if direct_message == False:
pass
else:
direct_message = False
draw_channel_list()
draw_node_list()
elif char == curses.KEY_RIGHT:
if direct_message == False:
direct_message = True
draw_channel_list()
draw_node_list()
else:
pass
# Check for Esc
elif char == 27:
break
elif char == curses.KEY_ENTER or char == 10 or char == 13:
if direct_message:
node_list = get_node_list()
if node_list[selected_node] not in channel_list:
channel_list.append(node_list[selected_node])
all_messages[node_list[selected_node]] = []
selected_channel = channel_list.index(node_list[selected_node])
selected_node = 0
direct_message = False
draw_node_list()
draw_channel_list()
update_messages_window()
else:
# Enter key pressed, send user input as message
send_message(input_text, channel=selected_channel)
# Clear entry window and reset input text
input_text = ""
entry_win.clear()
entry_win.refresh()
elif char == curses.KEY_BACKSPACE or char == 127:
input_text = input_text[:-1]
elif char == 96:
curses.curs_set(0) # Hide cursor
settings(stdscr, interface)
curses.curs_set(1) # Show cursor again
elif char == 47:
# Display packet log
if display_log is False:
display_log = True
update_messages_window()
else:
display_log = False
packetlog_win.clear()
update_messages_window()
else:
# Append typed character to input text
input_text += chr(char)
# draw_debug(char)
pub.subscribe(on_receive, 'meshtastic.receive')
if __name__ == "__main__":
curses.wrapper(main)
curses_ui.py
+321
View File
@@ -0,0 +1,321 @@
import curses
import textwrap
import globals
from utils import get_node_list, get_name_from_number, get_channels
from settings import settings
from tx_handler import send_message
def handle_notification(channel_number, add=True):
global channel_win
_, win_width = channel_win.getmaxyx() # Get the width of the channel window
# Get the channel name
if isinstance(globals.channel_list[channel_number], str): # Channels
channel_name = globals.channel_list[channel_number]
elif isinstance(globals.channel_list[channel_number], int): # DM's
channel_name = get_name_from_number(globals.channel_list[channel_number])
else:
return
# Truncate the channel name if it's too long to fit in the window
truncated_channel_name = channel_name[:win_width - 5] + '-' if len(channel_name) > win_width - 5 else channel_name
# Add or remove the notification indicator
notification = " *" if add else " "
channel_win.addstr(channel_number + 1, len(truncated_channel_name) + 1, notification, curses.color_pair(4))
channel_win.refresh()
def add_notification(channel_number):
handle_notification(channel_number, add=True)
def remove_notification(channel_number):
handle_notification(channel_number, add=False)
def update_messages_window():
global messages_win
messages_win.clear()
# Calculate how many messages can fit in the window
max_messages = messages_win.getmaxyx()[0] - 2 # Subtract 2 for the top and bottom border
# Determine the starting index for displaying messages
if globals.channel_list[globals.selected_channel] in globals.all_messages:
start_index = max(0, len(globals.all_messages[globals.channel_list[globals.selected_channel]]) - max_messages)
else:
# Handle the case where selected_channel does not exist
start_index = 0 # Set start_index to 0 or any other appropriate value
# Display messages starting from the calculated start index
# Check if selected_channel exists in all_messages before accessing it
if globals.channel_list[globals.selected_channel] in globals.all_messages:
row = 1
for _, (prefix, message) in enumerate(globals.all_messages[globals.channel_list[globals.selected_channel]][start_index:], start=1):
full_message = f"{prefix}{message}"
wrapped_messages = textwrap.wrap(full_message, messages_win.getmaxyx()[1] - 2)
for wrapped_message in wrapped_messages:
messages_win.addstr(row, 1, wrapped_message, curses.color_pair(1) if prefix.startswith(">> Sent:") else curses.color_pair(2))
row += 1
messages_win.box()
messages_win.refresh()
update_packetlog_win()
def update_packetlog_win():
if globals.display_log:
packetlog_win.clear()
packetlog_win.box()
# Get the dimensions of the packet log window
height, width = packetlog_win.getmaxyx()
columns = [10,10,15,30]
span = 0
for column in columns[:-1]:
span += column
# Add headers
headers = f"{'From':<{columns[0]}} {'To':<{columns[1]}} {'Port':<{columns[2]}} {'Payload':<{width-span}}"
packetlog_win.addstr(1, 1, headers[:width - 2],curses.A_UNDERLINE) # Truncate headers if they exceed window width
for i, packet in enumerate(reversed(globals.packet_buffer)):
if i >= height - 3: # Skip if exceeds the window height
break
# Format each field
from_id = get_name_from_number(packet['from'], 'short').ljust(columns[0])
to_id = (
"BROADCAST".ljust(columns[1]) if str(packet['to']) == "4294967295"
else get_name_from_number(packet['to'], 'short').ljust(columns[1])
)
if 'decoded' in packet:
port = packet['decoded']['portnum'].ljust(columns[2])
payload = (packet['decoded']['payload']).ljust(columns[3])
else:
port = "NO KEY".ljust(columns[2])
payload = "NO KEY".ljust(columns[3])
# Combine and truncate if necessary
logString = f"{from_id} {to_id} {port} {payload}"
logString = logString[:width - 3]
# Add to the window
packetlog_win.addstr(i + 2, 1, logString)
packetlog_win.refresh()
def draw_text_field(win, text):
win.clear()
win.border()
win.addstr(1, 1, text)
def draw_channel_list():
# Get the dimensions of the channel window
_, win_width = channel_win.getmaxyx()
for i, (channel, message_list) in enumerate(globals.all_messages.items()):
# Convert node number to long name if it's an integer
if isinstance(channel, int):
channel = get_name_from_number(channel, type='long')
# Truncate the channel name if it's too long to fit in the window
truncated_channel = channel[:win_width - 5] + '-' if len(channel) > win_width - 5 else channel
if globals.selected_channel == i and not globals.direct_message:
channel_win.addstr(i + 1, 1, truncated_channel, curses.color_pair(3))
remove_notification(globals.selected_channel)
else:
channel_win.addstr(i + 1, 1, truncated_channel, curses.color_pair(4))
channel_win.refresh()
def draw_node_list():
nodes_win.clear()
height, width = nodes_win.getmaxyx()
start_index = max(0, globals.selected_node - (height - 3)) # Calculate starting index based on selected node and window height
for i, node in enumerate(get_node_list()[start_index:], start=1):
if i < height - 1 : # Check if there is enough space in the window
if globals.selected_node + 1 == start_index + i and globals.direct_message:
nodes_win.addstr(i, 1, get_name_from_number(node, "long"), curses.color_pair(3))
else:
nodes_win.addstr(i, 1, get_name_from_number(node, "long"), curses.color_pair(4))
nodes_win.box()
nodes_win.refresh()
def draw_debug(value):
function_win.addstr(1, 100, f"debug: {value} ")
function_win.refresh()
def select_channels(direction):
channel_list_length = len(globals.channel_list)
globals.selected_channel += direction
if globals.selected_channel < 0:
globals.selected_channel = channel_list_length - 1
elif globals.selected_channel >= channel_list_length:
globals.selected_channel = 0
draw_channel_list()
update_messages_window()
def select_nodes(direction):
node_list_length = len(get_node_list())
globals.selected_node += direction
if globals.selected_node < 0:
globals.selected_node = node_list_length - 1
elif globals.selected_node >= node_list_length:
globals.selected_node = 0
draw_node_list()
def main_ui(stdscr):
global messages_win, nodes_win, channel_win, function_win, packetlog_win
stdscr.keypad(True)
get_channels()
# Initialize colors
curses.start_color()
curses.init_pair(1, curses.COLOR_CYAN, curses.COLOR_BLACK)
curses.init_pair(2, curses.COLOR_YELLOW, curses.COLOR_BLACK)
curses.init_pair(3, curses.COLOR_BLACK, curses.COLOR_WHITE)
curses.init_pair(4, curses.COLOR_WHITE, curses.COLOR_BLACK)
curses.init_pair(5, curses.COLOR_RED, curses.COLOR_BLACK)
# Calculate window max dimensions
height, width = stdscr.getmaxyx()
# Define window dimensions and positions
entry_win = curses.newwin(3, width, 0, 0)
channel_width = 3 * (width // 16)
nodes_width = 5 * (width // 16)
messages_width = width - channel_width - nodes_width
channel_win = curses.newwin(height - 6, channel_width, 3, 0)
messages_win = curses.newwin(height - 6, messages_width, 3, channel_width)
packetlog_win = curses.newwin(int(height / 3), messages_width, height - int(height / 3) - 3, channel_width)
nodes_win = curses.newwin(height - 6, nodes_width, 3, channel_width + messages_width)
function_win = curses.newwin(3, width, height - 3, 0)
draw_text_field(function_win, f"↑↓ = Switch Channels ← → = Channels/Nodes ENTER = Send / Select DM ` = Settings / = Toggle Log ESC = Quit")
# Enable scrolling for messages and nodes windows
messages_win.scrollok(True)
nodes_win.scrollok(True)
channel_win.scrollok(True)
channel_win.refresh()
draw_channel_list()
draw_node_list()
# Draw boxes around windows
channel_win.box()
entry_win.box()
messages_win.box()
nodes_win.box()
function_win.box()
# Refresh all windows
entry_win.refresh()
messages_win.refresh()
nodes_win.refresh()
channel_win.refresh()
function_win.refresh()
input_text = ""
globals.direct_message = False
entry_win.keypad(True)
while True:
draw_text_field(entry_win, f"Input: {input_text}")
# Get user input from entry window
entry_win.move(1, len(input_text) + 8)
char = entry_win.getch()
# draw_debug(f"Keypress: {char}")
if char == curses.KEY_UP:
if globals.direct_message:
draw_channel_list()
select_nodes(-1)
else:
select_channels(-1)
elif char == curses.KEY_DOWN:
if globals.direct_message:
draw_channel_list()
select_nodes(1)
else:
select_channels(1)
elif char == curses.KEY_LEFT:
if globals.direct_message == False:
pass
else:
globals.direct_message = False
draw_channel_list()
draw_node_list()
elif char == curses.KEY_RIGHT:
if globals.direct_message == False:
globals.direct_message = True
draw_channel_list()
draw_node_list()
else:
pass
# Check for Esc
elif char == 27:
break
elif char == curses.KEY_ENTER or char == 10 or char == 13:
if globals.direct_message:
node_list = get_node_list()
if node_list[globals.selected_node] not in globals.channel_list:
globals.channel_list.append(node_list[globals.selected_node])
globals.all_messages[node_list[globals.selected_node]] = []
globals.selected_channel = globals.channel_list.index(node_list[globals.selected_node])
globals.selected_node = 0
globals.direct_message = False
draw_node_list()
draw_channel_list()
update_messages_window()
else:
# Enter key pressed, send user input as message
send_message(input_text, channel=globals.selected_channel)
update_messages_window()
messages_win.refresh()
# Clear entry window and reset input text
input_text = ""
entry_win.clear()
entry_win.refresh()
elif char == curses.KEY_BACKSPACE or char == 127:
input_text = input_text[:-1]
elif char == 96:
curses.curs_set(0) # Hide cursor
settings(stdscr, globals.interface)
curses.curs_set(1) # Show cursor again
elif char == 47:
# Display packet log
if globals.display_log is False:
globals.display_log = True
update_messages_window()
else:
display_log = False
packetlog_win.clear()
update_messages_window()
else:
# Append typed character to input text
input_text += chr(char)
globals.py
+9
View File
@@ -0,0 +1,9 @@
all_messages = {}
channel_list = []
packet_buffer = []
myNodeNum = 0
selected_channel = 0
selected_node = 0
direct_message = False
interface = None
display_log = False
interfaces.py
+17
View File
@@ -0,0 +1,17 @@
import meshtastic.serial_interface
import meshtastic.tcp_interface
import meshtastic.ble_interface
import globals
def initialize_interface(args):
if args.ble:
return meshtastic.ble_interface.BLEInterface(args.ble if args.ble != "any" else None)
elif args.host:
return meshtastic.tcp_interface.TCPInterface(args.host)
else:
try:
return meshtastic.serial_interface.SerialInterface(args.port)
except PermissionError as ex:
print("You probably need to add yourself to the `dialout` group to use a serial connection.")
if globals.interface.devPath is None:
return meshtastic.tcp_interface.TCPInterface("meshtastic.local")
main.py
+25
View File
@@ -0,0 +1,25 @@
#!/usr/bin/env python3
'''
Curses Client for Meshtastic by http://github.com/pdxlocations
Powered by Meshtastic.org
V 0.2.0
'''
import curses
from pubsub import pub
from parsers import setup_parser
from interfaces import initialize_interface
from rx_handler import on_receive
from curses_ui import main_ui
from utils import get_channels
import globals
if __name__ == "__main__":
parser = setup_parser()
args = parser.parse_args()
globals.interface = initialize_interface(args)
globals.channel_list = get_channels()
pub.subscribe(on_receive, 'meshtastic.receive')
curses.wrapper(main_ui)
parsers.py
+37
View File
@@ -0,0 +1,37 @@
import argparse
def setup_parser():
parser = argparse.ArgumentParser(
add_help=True,
epilog="If no connection arguments are specified, we attempt a serial connection and then a TCP connection to localhost.")
connOuter = parser.add_argument_group('Connection', 'Optional arguments to specify a device to connect to and how.')
conn = connOuter.add_mutually_exclusive_group()
conn.add_argument(
"--port",
"--serial",
"-s",
help="The port to connect to via serial, e.g. `/dev/ttyUSB0`.",
nargs="?",
default=None,
const=None,
)
conn.add_argument(
"--host",
"--tcp",
"-t",
help="The hostname or IP address to connect to using TCP.",
nargs="?",
default=None,
const="localhost",
)
conn.add_argument(
"--ble",
"-b",
help="The BLE device MAC address or name to connect to.",
nargs="?",
default=None,
const="any"
)
return parser
rx_handler.py
+61
View File
@@ -0,0 +1,61 @@
from meshtastic import BROADCAST_NUM
from utils import get_node_list, decimal_to_hex, get_nodeNum
import globals
from curses_ui import update_packetlog_win, draw_node_list, update_messages_window, draw_channel_list, add_notification
def on_receive(packet):
# update packet log
globals.packet_buffer.append(packet)
if len(globals.packet_buffer) > 20:
# trim buffer to 20 packets
globals.packet_buffer = globals.packet_buffer[-20:]
if globals.display_log:
update_packetlog_win()
try:
if 'decoded' in packet and packet['decoded']['portnum'] == 'NODEINFO_APP':
get_node_list()
draw_node_list()
elif 'decoded' in packet and packet['decoded']['portnum'] == 'TEXT_MESSAGE_APP':
message_bytes = packet['decoded']['payload']
message_string = message_bytes.decode('utf-8')
if packet.get('channel'):
channel_number = packet['channel']
else:
channel_number = 0
myNodeNum = get_nodeNum()
if packet['to'] == myNodeNum:
if packet['from'] in globals.channel_list:
pass
else:
globals.channel_list.append(packet['from'])
globals.all_messages[packet['from']] = []
draw_channel_list()
channel_number = globals.channel_list.index(packet['from'])
if globals.channel_list[channel_number] != globals.channel_list[globals.selected_channel]:
add_notification(channel_number)
# Add received message to the messages list
message_from_id = packet['from']
message_from_string = ""
for node in globals.interface.nodes.values():
if message_from_id == node['num']:
message_from_string = node["user"]["longName"] # Get the long name using the node ID
break
else:
message_from_string = str(decimal_to_hex(message_from_id)) # If long name not found, use the ID as string
if globals.channel_list[channel_number] in globals.all_messages:
globals.all_messages[globals.channel_list[channel_number]].append((f">> {message_from_string} ", message_string))
else:
globals.all_messages[globals.channel_list[channel_number]] = [(f">> {message_from_string} ", message_string)]
draw_channel_list()
update_messages_window()
except KeyError as e:
print(f"Error processing packet: {e}")
settings.py
+47 -49
View File
@@ -1,11 +1,9 @@
import curses
import meshtastic.serial_interface, meshtastic.tcp_interface
import ipaddress
try:
from meshtastic.protobuf import config_pb2, module_config_pb2, mesh_pb2, channel_pb2
except ImportError:
from meshtastic import config_pb2, module_config_pb2, mesh_pb2, channel_pb2
import meshtastic.serial_interface, meshtastic.tcp_interface
from meshtastic.protobuf import config_pb2, module_config_pb2, mesh_pb2, channel_pb2
import globals
def display_enum_menu(stdscr, enum_values, menu_item):
menu_height = len(enum_values) + 2
@@ -336,7 +334,7 @@ def display_bool_menu(stdscr, setting_value):
return display_enum_menu(stdscr, bool_options, setting_value)
def generate_menu_from_protobuf(message_instance, interface):
def generate_menu_from_protobuf(message_instance):
if not hasattr(message_instance, "DESCRIPTOR"):
return # This is not a protobuf message instance, exit
menu = {}
@@ -346,12 +344,12 @@ def generate_menu_from_protobuf(message_instance, interface):
field_descriptor = message_instance.DESCRIPTOR.fields_by_name[field_name]
if field_descriptor is not None:
nested_message_instance = getattr(message_instance, field_name)
menu[field_name] = generate_menu_from_protobuf(nested_message_instance, interface)
menu[field_name] = generate_menu_from_protobuf(nested_message_instance)
return menu
def change_setting(stdscr, interface, menu_path):
node = interface.localNode
def change_setting(stdscr, menu_path):
node = globals.interface.localNode
field_descriptor = None
setting_value = 0
@@ -363,7 +361,7 @@ def change_setting(stdscr, interface, menu_path):
# Determine the level of nesting based on the length of menu_path
if menu_path[1] == "User Settings":
n = interface.getMyNodeInfo()
n = globals.interface.getMyNodeInfo()
setting_string = n['user'].get(snake_to_camel(menu_path[2]), 0)
@@ -381,11 +379,11 @@ def change_setting(stdscr, interface, menu_path):
if menu_path[2] in ["long_name", "short_name"]:
if menu_path[2] == "short_name" and len(setting_value) > 4:
setting_value = setting_value[:4]
settings_set_owner(interface, long_name=setting_value if menu_path[2] == "long_name" else None,
settings_set_owner(long_name=setting_value if menu_path[2] == "long_name" else None,
short_name=setting_value if menu_path[2] == "short_name" else None)
elif menu_path[2] == "is_licensed":
ln = n['user']['longName']
settings_set_owner(interface, long_name=ln, is_licensed=setting_value)
settings_set_owner(long_name=ln, is_licensed=setting_value)
stdscr.clear()
stdscr.border()
@@ -456,7 +454,7 @@ def change_setting(stdscr, interface, menu_path):
# formatted_text = f"{menu_path[2]}.{menu_path[3]} = {setting_value}"
# menu_header(stdscr,formatted_text,2)
ourNode = interface.localNode
ourNode = globals.interface.localNode
# Convert "true" to 1, "false" to 0, leave other values as they are
if setting_value == "True" or setting_value == "1":
@@ -496,14 +494,14 @@ def snake_to_camel(snake_str):
return components[0] + ''.join(x.title() for x in components[1:])
def display_values(stdscr, interface, key_list, menu_path):
node = interface.localNode
def display_values(stdscr, key_list, menu_path):
node = globals.interface.localNode
user_settings = ["long_name", "short_name", "is_licensed"]
for i, key in enumerate(key_list):
if len(menu_path) == 2:
if menu_path[1] == 'User Settings':
n = interface.getMyNodeInfo()
n = globals.interface.getMyNodeInfo()
try:
setting = n['user'][snake_to_camel(key_list[i])]
except:
@@ -537,7 +535,7 @@ def menu_header(window, text, start_y=1):
window.addstr(start_y, start_x, formatted_text)
window.refresh()
def nested_menu(stdscr, menu, interface):
def nested_menu(stdscr, menu):
menu_item = 0
current_menu = menu
prev_menu = []
@@ -565,7 +563,7 @@ def nested_menu(stdscr, menu, interface):
stdscr.addstr(i+3, 1, key)
# Display current values
display_values(stdscr, interface, key_list, menu_path)
display_values(stdscr, key_list, menu_path)
char = stdscr.getch()
@@ -584,10 +582,10 @@ def nested_menu(stdscr, menu, interface):
elif char == curses.KEY_RIGHT:
# if selected_key == "Region":
# settings_region(interface)
# settings_region()
# break
if selected_key == "Channels":
channels_editor(interface, stdscr)
channels_editor(stdscr)
elif selected_key not in ["Reboot", "Reset NodeDB", "Shutdown", "Factory Reset"]:
menu_path.append(selected_key)
@@ -614,15 +612,15 @@ def nested_menu(stdscr, menu, interface):
elif char == ord('\n'):
if selected_key == "Channels":
channels_editor(interface, stdscr)
channels_editor(stdscr)
if selected_key == "Reboot":
settings_reboot(interface)
settings_reboot()
elif selected_key == "Reset NodeDB":
settings_reset_nodedb(interface)
settings_reset_nodedb()
elif selected_key == "Shutdown":
settings_shutdown(interface)
settings_shutdown()
elif selected_key == "Factory Reset":
settings_factory_reset(interface)
settings_factory_reset()
elif selected_value is not None:
stdscr.refresh()
@@ -643,10 +641,10 @@ def nested_menu(stdscr, menu, interface):
if last_menu_level == True:
if not isinstance(current_menu.get(next_key), dict):
change_setting(stdscr, interface, menu_path)
change_setting(stdscr, menu_path)
def settings(stdscr, interface):
def settings(stdscr):
popup_height = 22
popup_width = 60
popup_win = None
@@ -667,19 +665,19 @@ def settings(stdscr, interface):
user = mesh_pb2.User()
user_settings = ["long_name", "short_name", "is_licensed"]
user_config = generate_menu_from_protobuf(user, interface)
user_config = generate_menu_from_protobuf(user)
user_config = {key: value for key, value in user_config.items() if key in user_settings}
channel = channel_pb2.ChannelSettings()
channel_config = generate_menu_from_protobuf(channel, interface)
channel_config = generate_menu_from_protobuf(channel)
channel_config = [channel_config.copy() for i in range(8)]
radio = config_pb2.Config()
radio_config = generate_menu_from_protobuf(radio, interface)
radio_config = generate_menu_from_protobuf(radio)
module = module_config_pb2.ModuleConfig()
module_config = generate_menu_from_protobuf(module, interface)
module_config = generate_menu_from_protobuf(module)
# Add top-level menu items
top_level_menu = {
@@ -694,41 +692,41 @@ def settings(stdscr, interface):
}
# Call nested_menu function to display and handle the nested menu
nested_menu(popup_win, top_level_menu, interface)
nested_menu(popup_win, top_level_menu)
# Close the popup window
popup_win.clear()
popup_win.refresh()
# def settings_region(interface):
# selected_option, do_set = set_region(interface)
# def settings_region():
# selected_option, do_set = set_region()
# if do_set:
# ourNode = interface.localNode
# setattr(ourNode.localConfig.lora, "region", selected_option)
# ourNode.writeConfig("lora")
def settings_reboot(interface):
interface.localNode.reboot()
def settings_reboot():
globals.interface.localNode.reboot()
def settings_reset_nodedb(interface):
interface.localNode.resetNodeDb()
def settings_reset_nodedb():
globals.interface.localNode.resetNodeDb()
def settings_shutdown(interface):
interface.localNode.shutdown()
def settings_shutdown():
globals.interface.localNode.shutdown()
def settings_factory_reset(interface):
interface.localNode.factoryReset()
def settings_factory_reset():
globals.interface.localNode.factoryReset()
def settings_set_owner(interface, long_name=None, short_name=None, is_licensed=False):
def settings_set_owner(long_name=None, short_name=None, is_licensed=False):
if is_licensed == 'True':
is_licensed = True
elif is_licensed == 'False':
is_licensed = False
interface.localNode.setOwner(long_name, short_name, is_licensed)
globals.interface.localNode.setOwner(long_name, short_name, is_licensed)
def channels_editor(interface, stdscr):
def channels_editor(stdscr):
# Define the list of channels
channels = [f"{i}" for i in range(8)]
@@ -742,12 +740,12 @@ def channels_editor(interface, stdscr):
# Fetch and print roles for each channel
for index, channel_index in enumerate(channels):
channel = interface.localNode.getChannelByChannelIndex(index)
channel = globals.interface.localNode.getChannelByChannelIndex(index)
role = "DISABLED" if channel.role == 0 else "PRIMARY" if channel.role == 1 else "SECONDARY"
channel_settings = channel.settings
channel_name = channel_settings.name
if not channel_name and role != "DISABLED":
config = interface.localNode.localConfig
config = globals.interface.localNode.localConfig
channel_name_int = config.lora.modem_preset
channel_name = config_pb2.Config.LoRaConfig.ModemPreset.Name(channel_name_int)
@@ -788,7 +786,7 @@ def channels_editor(interface, stdscr):
if __name__ == "__main__":
interface = meshtastic.serial_interface.SerialInterface()
globals.interface = meshtastic.serial_interface.SerialInterface()
# radio = config_pb2.Config()
# module = module_config_pb2.ModuleConfig()
@@ -798,6 +796,6 @@ if __name__ == "__main__":
def main(stdscr):
stdscr.keypad(True)
while True:
settings(stdscr, interface)
settings(stdscr)
curses.wrapper(main)
tx_handler.py
+25
View File
@@ -0,0 +1,25 @@
from meshtastic import BROADCAST_NUM
import globals
def send_message(message, destination=BROADCAST_NUM, channel=0):
send_on_channel = 0
if isinstance(globals.channel_list[channel], int):
send_on_channel = 0
destination = globals.channel_list[channel]
elif isinstance(globals.channel_list[channel], str):
send_on_channel = channel
globals.interface.sendText(
text=message,
destinationId=destination,
wantAck=False,
wantResponse=False,
onResponse=None,
channelIndex=send_on_channel,
)
# Add sent message to the messages dictionary
if globals.channel_list[channel] in globals.all_messages:
globals.all_messages[globals.channel_list[channel]].append((">> Sent: ", message))
else:
globals.all_messages[globals.channel_list[channel]] = [(">> Sent: ", message)]
utils.py
+61
View File
@@ -0,0 +1,61 @@
import globals
from meshtastic.protobuf import config_pb2
def get_nodeNum():
myinfo = globals.interface.getMyNodeInfo()
myNodeNum = myinfo['num']
return myNodeNum
def get_channels():
node = globals.interface.getNode('^local')
device_channels = node.channels
channel_output = []
for device_channel in device_channels:
if device_channel.role:
if device_channel.settings.name:
channel_output.append(device_channel.settings.name)
globals.all_messages[device_channel.settings.name] = []
else:
# If channel name is blank, use the modem preset
lora_config = node.localConfig.lora
modem_preset_enum = lora_config.modem_preset
modem_preset_string = config_pb2._CONFIG_LORACONFIG_MODEMPRESET.values_by_number[modem_preset_enum].name
channel_output.append(convert_to_camel_case(modem_preset_string))
globals.all_messages[convert_to_camel_case(modem_preset_string)] = []
return list(globals.all_messages.keys())
def get_node_list():
node_list = []
if globals.interface.nodes:
for node in globals.interface.nodes.values():
node_list.append(node['num'])
return node_list
def decimal_to_hex(decimal_number):
return f"!{decimal_number:08x}"
def convert_to_camel_case(string):
words = string.split('_')
camel_case_string = ''.join(word.capitalize() for word in words)
return camel_case_string
def get_name_from_number(number, type='long'):
name = ""
for node in globals.interface.nodes.values():
if number == node['num']:
if type == 'long':
name = node['user']['longName']
return name
elif type == 'short':
name = node['user']['shortName']
return name
else:
pass
else:
name = str(decimal_to_hex(number)) # If long name not found, use the ID as string
return name