diff --git a/curses-client.py b/curses-client.py deleted file mode 100644 index 9bb0d85..0000000 --- a/curses-client.py +++ /dev/null @@ -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) - diff --git a/curses_ui.py b/curses_ui.py new file mode 100644 index 0000000..a621955 --- /dev/null +++ b/curses_ui.py @@ -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) + diff --git a/globals.py b/globals.py new file mode 100644 index 0000000..900ae5d --- /dev/null +++ b/globals.py @@ -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 \ No newline at end of file diff --git a/interfaces.py b/interfaces.py new file mode 100644 index 0000000..4111d8c --- /dev/null +++ b/interfaces.py @@ -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") \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..b25bacf --- /dev/null +++ b/main.py @@ -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) \ No newline at end of file diff --git a/parsers.py b/parsers.py new file mode 100644 index 0000000..391dc0e --- /dev/null +++ b/parsers.py @@ -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 \ No newline at end of file diff --git a/rx_handler.py b/rx_handler.py new file mode 100644 index 0000000..db94371 --- /dev/null +++ b/rx_handler.py @@ -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}") + diff --git a/settings.py b/settings.py index 51c9d4e..b9efb88 100644 --- a/settings.py +++ b/settings.py @@ -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) \ No newline at end of file diff --git a/tx_handler.py b/tx_handler.py new file mode 100644 index 0000000..7a45ec6 --- /dev/null +++ b/tx_handler.py @@ -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)] \ No newline at end of file diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..fa1ba7a --- /dev/null +++ b/utils.py @@ -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 + \ No newline at end of file