Compare commits

...

14 Commits

Author SHA1 Message Date
pdxlocations
d0ada7eb5b add utf-8 read (#166) 2025-04-13 15:17:36 -07:00
pdxlocations
8d9bbac0be Merge pull request #165 from pdxlocations/type-annotations 2025-04-13 15:02:54 -07:00
Ben Lipsey
613eeb4fab working changes 2025-04-13 14:58:31 -07:00
Ben Lipsey
f7b2645dcb working changes 2025-04-13 14:49:00 -07:00
Ben Lipsey
bc5a5951d4 current state 2025-04-13 14:18:52 -07:00
Ben Lipsey
d7eec6de6e current state 2025-04-12 21:53:27 -07:00
Ben Lipsey
8779297424 current state 2025-04-12 21:19:44 -07:00
Ben Lipsey
ccc1399644 current state 2025-04-12 09:46:00 -07:00
Ben Lipsey
f52034e61f current state 2025-04-11 22:22:34 -07:00
pdxlocations
cdd1d89062 fix typo 2025-04-08 21:17:01 -07:00
pdxlocations
c3ff85a646 bump version 2025-04-08 21:14:58 -07:00
pdxlocations
9b8cf19a0c Fix localhost fallback and allow tcp port (#164)
* update interfaces.py

* fix logging

* update

* update

* fix devpath

* fix returns

* changes
2025-04-08 21:05:32 -07:00
pdxlocations
f2e671da7f Update README.md 2025-04-08 18:10:41 -07:00
pdxlocations
3fc1293db1 Update README.md 2025-04-08 17:58:15 -07:00
22 changed files with 554 additions and 387 deletions

View File

@@ -1,8 +1,12 @@
## Contact - A Console UI for Meshtastic
### (Formerly Curses Client for Meshtastic)
#### Powered by Meshtastic.org
### Install with:
```bash
pip install contact
```
This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores.
@@ -11,7 +15,7 @@ This Python curses client for Meshtastic is a terminal-based client designed to
<br><br>
The settings dialogue can be accessed within the client or may be run standalone to configure your node by launching `contact --settings` or `contact -c`
<img width="573" alt="Contact - Settings Dialogue" src="https://github.com/user-attachments/assets/dbe1287b-5558-407c-84b8-2a1bc913dec8" />
<img width="696" alt="Screenshot 2025-04-08 at 6 10 06PM" src="https://github.com/user-attachments/assets/3d5e3964-f009-4772-bd6e-91b907c65a3b" />
## Message Persistence

View File

@@ -1,114 +1,133 @@
#!/usr/bin/env python3
'''
"""
Contact - A Console UI for Meshtastic by http://github.com/pdxlocations
Powered by Meshtastic.org
Meshtastic® is a registered trademark of Meshtastic LLC. Meshtastic software components are released under various licenses, see GitHub for details. No warranty is provided - use at your own risk.
'''
Meshtastic® is a registered trademark of Meshtastic LLC.
Meshtastic software components are released under various licenses—see GitHub for details.
No warranty is provided. Use at your own risk.
"""
# Standard library
import contextlib
import curses
import os
from pubsub import pub
import sys
import io
import logging
import os
import subprocess
import traceback
import sys
import threading
import traceback
from contact.utilities.db_handler import init_nodedb, load_messages_from_db
# Third-party
from pubsub import pub
# Local application
import contact.globals as globals
import contact.ui.default_config as config
from contact.message_handlers.rx_handler import on_receive
from contact.settings import set_region
from contact.ui.contact_ui import main_ui
from contact.ui.colors import setup_colors
from contact.ui.contact_ui import main_ui
from contact.ui.splash import draw_splash
import contact.ui.default_config as config
from contact.utilities.arg_parser import setup_parser
from contact.utilities.interfaces import initialize_interface
from contact.utilities.db_handler import init_nodedb, load_messages_from_db
from contact.utilities.input_handlers import get_list_input
from contact.utilities.utils import get_channels, get_node_list, get_nodeNum
import contact.globals as globals
from contact.utilities.interfaces import initialize_interface
from contact.utilities.utils import get_channels, get_nodeNum, get_node_list
# ------------------------------------------------------------------------------
# Environment & Logging Setup
# ------------------------------------------------------------------------------
# Set ncurses compatibility settings
os.environ["NCURSES_NO_UTF8_ACS"] = "1"
os.environ["LANG"] = "C.UTF-8"
os.environ.setdefault("TERM", "xterm-256color")
if os.environ.get("COLORTERM") == "gnome-terminal":
os.environ["TERM"] = "xterm-256color"
# Configure logging
# Run `tail -f client.log` in another terminal to view live
logging.basicConfig(
filename=config.log_file_path,
level=logging.INFO, # DEBUG, INFO, WARNING, ERROR, CRITICAL)
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
globals.lock = threading.Lock()
def main(stdscr):
# ------------------------------------------------------------------------------
# Main Program Logic
# ------------------------------------------------------------------------------
def initialize_globals(args) -> None:
"""Initializes interface and shared globals."""
globals.interface = initialize_interface(args)
# Prompt for region if unset
if globals.interface.localNode.localConfig.lora.region == 0:
confirmation = get_list_input("Your region is UNSET. Set it now?", "Yes", ["Yes", "No"])
if confirmation == "Yes":
set_region(globals.interface)
globals.interface.close()
globals.interface = initialize_interface(args)
globals.myNodeNum = get_nodeNum()
globals.channel_list = get_channels()
globals.node_list = get_node_list()
pub.subscribe(on_receive, 'meshtastic.receive')
init_nodedb()
load_messages_from_db()
def main(stdscr: curses.window) -> None:
"""Main entry point for the curses UI."""
output_capture = io.StringIO()
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):
setup_colors()
draw_splash(stdscr)
parser = setup_parser()
args = parser.parse_args()
# Check if --settings was passed and run settings.py as a subprocess
args = setup_parser().parse_args()
if getattr(args, 'settings', False):
subprocess.run([sys.executable, "-m", "contact.settings"], check=True)
return
logging.info("Initializing interface %s", args)
logging.info("Initializing interface...")
with globals.lock:
globals.interface = initialize_interface(args)
if globals.interface.localNode.localConfig.lora.region == 0:
confirmation = get_list_input("Your region is UNSET. Set it now?", "Yes", ["Yes", "No"])
if confirmation == "Yes":
set_region(globals.interface)
globals.interface.close()
globals.interface = initialize_interface(args)
logging.info("Interface initialized")
globals.myNodeNum = get_nodeNum()
globals.channel_list = get_channels()
globals.node_list = get_node_list()
pub.subscribe(on_receive, 'meshtastic.receive')
init_nodedb()
load_messages_from_db()
initialize_globals(args)
logging.info("Starting main UI")
main_ui(stdscr)
except Exception as e:
console_output = output_capture.getvalue()
logging.error("An error occurred: %s", e)
logging.error("Uncaught exception: %s", e)
logging.error("Traceback: %s", traceback.format_exc())
logging.error("Console output before crash:\n%s", console_output)
raise # Re-raise only unexpected errors
logging.error("Console output:\n%s", console_output)
raise
def start():
log_file = config.log_file_path
log_f = open(log_file, "a", buffering=1) # Enable line-buffering for immediate log writes
sys.stdout = log_f
sys.stderr = log_f
def start() -> None:
"""Launch curses wrapper and redirect logs to file."""
with open(config.log_file_path, "a", buffering=1) as log_f:
sys.stdout = log_f
sys.stderr = log_f
with contextlib.redirect_stdout(log_f), contextlib.redirect_stderr(log_f):
try:
curses.wrapper(main)
except KeyboardInterrupt:
logging.info("User exited with Ctrl+C")
sys.exit(0)
except Exception as e:
logging.error("Fatal error: %s", e)
logging.error("Traceback: %s", traceback.format_exc())
sys.exit(1)
with contextlib.redirect_stderr(log_f), contextlib.redirect_stdout(log_f):
try:
curses.wrapper(main)
except KeyboardInterrupt:
logging.info("User exited with Ctrl+C or Ctrl+X") # Clean exit logging
sys.exit(0) # Ensure a clean exit
except Exception as e:
logging.error("Fatal error in curses wrapper: %s", e)
logging.error("Traceback: %s", traceback.format_exc())
sys.exit(1) # Exit with an error code
if __name__ == "__main__":
start()

View File

@@ -1,17 +1,34 @@
import logging
import time
from contact.utilities.utils import refresh_node_list
from datetime import datetime
from contact.ui.contact_ui import draw_packetlog_win, draw_node_list, draw_messages_window, draw_channel_list, add_notification
from contact.utilities.db_handler import save_message_to_db, maybe_store_nodeinfo_in_db, get_name_from_database, update_node_info_in_db
from typing import Any
from contact.utilities.utils import refresh_node_list
from contact.ui.contact_ui import (
draw_packetlog_win,
draw_node_list,
draw_messages_window,
draw_channel_list,
add_notification,
)
from contact.utilities.db_handler import (
save_message_to_db,
maybe_store_nodeinfo_in_db,
get_name_from_database,
update_node_info_in_db,
)
import contact.ui.default_config as config
import contact.globals as globals
from datetime import datetime
def on_receive(packet, interface):
def on_receive(packet: dict[str, Any], interface: Any) -> None:
"""
Handles an incoming packet from a Meshtastic interface.
Args:
packet: The received Meshtastic packet as a dictionary.
interface: The Meshtastic interface instance that received the packet.
"""
with globals.lock:
# Update packet log
globals.packet_buffer.append(packet)

View File

@@ -1,17 +1,29 @@
from datetime import datetime
from typing import Any
import google.protobuf.json_format
from meshtastic import BROADCAST_NUM
from meshtastic.protobuf import mesh_pb2, portnums_pb2
from contact.utilities.db_handler import save_message_to_db, update_ack_nak, get_name_from_database, is_chat_archived, update_node_info_in_db
from contact.utilities.db_handler import (
save_message_to_db,
update_ack_nak,
get_name_from_database,
is_chat_archived,
update_node_info_in_db,
)
import contact.ui.default_config as config
import contact.globals as globals
ack_naks = {}
ack_naks: dict[str, dict[str, Any]] = {} # requestId -> {channel, messageIndex, timestamp}
# Note "onAckNak" has special meaning to the API, thus the nonstandard naming convention
# See https://github.com/meshtastic/python/blob/master/meshtastic/mesh_interface.py#L462
def onAckNak(packet):
def onAckNak(packet: dict[str, Any]) -> None:
"""
Handles incoming ACK/NAK response packets.
"""
from contact.ui.contact_ui import draw_messages_window
request = packet['decoded']['requestId']
if(request not in ack_naks):
@@ -41,8 +53,10 @@ def onAckNak(packet):
if globals.channel_list[channel_number] == globals.channel_list[globals.selected_channel]:
draw_messages_window()
def on_response_traceroute(packet):
"""on response for trace route"""
def on_response_traceroute(packet: dict[str, Any]) -> None:
"""
Handle traceroute response packets and render the route visually in the UI.
"""
from contact.ui.contact_ui import draw_channel_list, draw_messages_window, add_notification
refresh_channels = False
@@ -118,7 +132,10 @@ def on_response_traceroute(packet):
save_message_to_db(globals.channel_list[channel_number], packet['from'], msg_str)
def send_message(message, destination=BROADCAST_NUM, channel=0):
def send_message(message: str, destination: int = BROADCAST_NUM, channel: int = 0) -> None:
"""
Sends a chat message using the selected channel.
"""
myid = globals.myNodeNum
send_on_channel = 0
channel_id = globals.channel_list[channel]
@@ -168,7 +185,10 @@ def send_message(message, destination=BROADCAST_NUM, channel=0):
ack_naks[sent_message_data.id] = {'channel': channel_id, 'messageIndex': len(globals.all_messages[channel_id]) - 1, 'timestamp': timestamp}
def send_traceroute():
def send_traceroute() -> None:
"""
Sends a RouteDiscovery protobuf to the selected node.
"""
r = mesh_pb2.RouteDiscovery()
globals.interface.sendData(
r,

View File

@@ -14,7 +14,7 @@ from contact.utilities.arg_parser import setup_parser
from contact.utilities.interfaces import initialize_interface
def main(stdscr):
def main(stdscr: curses.window) -> None:
output_capture = io.StringIO()
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):

View File

@@ -12,7 +12,7 @@ COLOR_MAP = {
"white": curses.COLOR_WHITE
}
def setup_colors(reinit=False):
def setup_colors(reinit: bool = False) -> None:
"""
Initialize curses color pairs based on the COLOR_CONFIG.
"""
@@ -29,7 +29,7 @@ def setup_colors(reinit=False):
print()
def get_color(category, bold=False, reverse=False, underline=False):
def get_color(category: str, bold: bool = False, reverse: bool = False, underline: bool = False) -> int:
"""
Retrieve a curses color pair with optional attributes.
"""

View File

@@ -13,7 +13,7 @@ import contact.ui.default_config as config
import contact.ui.dialog
import contact.globals as globals
def handle_resize(stdscr, firstrun):
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, function_win, packetlog_win, entry_win
# Calculate window max dimensions
@@ -90,7 +90,7 @@ def handle_resize(stdscr, firstrun):
pass
def main_ui(stdscr):
def main_ui(stdscr: curses.window) -> None:
global input_text
input_text = ""
stdscr.keypad(True)
@@ -377,7 +377,7 @@ def main_ui(stdscr):
def draw_channel_list():
def draw_channel_list() -> None:
channel_pad.erase()
win_height, win_width = channel_win.getmaxyx()
start_index = max(0, globals.selected_channel - (win_height - 3)) # Leave room for borders
@@ -418,7 +418,7 @@ def draw_channel_list():
refresh_pad(0)
def draw_messages_window(scroll_to_bottom = False):
def draw_messages_window(scroll_to_bottom: bool = False) -> None:
"""Update the messages window based on the selected channel and scroll position."""
messages_pad.erase()
@@ -461,7 +461,7 @@ def draw_messages_window(scroll_to_bottom = False):
draw_packetlog_win()
def draw_node_list():
def draw_node_list() -> None:
global nodes_pad
# This didn't work, for some reason an error is thown on startup, so we just create the pad every time
@@ -501,7 +501,7 @@ def draw_node_list():
curses.curs_set(1)
entry_win.refresh()
def select_channel(idx):
def select_channel(idx: int) -> None:
old_selected_channel = globals.selected_channel
globals.selected_channel = max(0, min(idx, len(globals.channel_list) - 1))
draw_messages_window(True)
@@ -515,7 +515,7 @@ def select_channel(idx):
highlight_line(True, 0, globals.selected_channel)
refresh_pad(0)
def scroll_channels(direction):
def scroll_channels(direction: int) -> None:
new_selected_channel = globals.selected_channel + direction
if new_selected_channel < 0:
@@ -525,7 +525,7 @@ def scroll_channels(direction):
select_channel(new_selected_channel)
def scroll_messages(direction):
def scroll_messages(direction: int) -> None:
globals.selected_message += direction
msg_line_count = messages_pad.getmaxyx()[0]
@@ -533,7 +533,7 @@ def scroll_messages(direction):
refresh_pad(1)
def select_node(idx):
def select_node(idx: int) -> None:
old_selected_node = globals.selected_node
globals.selected_node = max(0, min(idx, len(globals.node_list) - 1))
@@ -543,7 +543,7 @@ def select_node(idx):
draw_function_win()
def scroll_nodes(direction):
def scroll_nodes(direction: int) -> None:
new_selected_node = globals.selected_node + direction
if new_selected_node < 0:
@@ -553,7 +553,7 @@ def scroll_nodes(direction):
select_node(new_selected_node)
def draw_packetlog_win():
def draw_packetlog_win() -> None:
columns = [10,10,15,30]
span = 0
@@ -602,7 +602,7 @@ def draw_packetlog_win():
curses.curs_set(1)
entry_win.refresh()
def search(win):
def search(win: int) -> None:
start_idx = globals.selected_node
select_func = select_node
@@ -645,7 +645,7 @@ def search(win):
entry_win.erase()
def draw_node_details():
def draw_node_details() -> None:
node = None
try:
node = globals.interface.nodesByNum[globals.node_list[globals.selected_node]]
@@ -693,7 +693,7 @@ def draw_node_details():
draw_centered_text_field(function_win, nodestr, 0, get_color("commands"))
def draw_help():
def draw_help() -> None:
cmds = ["↑→↓← = Select", " ENTER = Send", " ` = Settings", " ^P = Packet Log", " ESC = Quit", " ^t = Traceroute", " ^d = Archive Chat", " ^f = Favorite", " ^g = Ignore"]
function_str = ""
for s in cmds:
@@ -702,19 +702,18 @@ def draw_help():
draw_centered_text_field(function_win, function_str, 0, get_color("commands"))
def draw_function_win():
def draw_function_win() -> None:
if(globals.current_window == 2):
draw_node_details()
else:
draw_help()
def get_msg_window_lines():
def get_msg_window_lines() -> None:
packetlog_height = packetlog_win.getmaxyx()[0] - 1 if globals.display_log else 0
return messages_win.getmaxyx()[0] - 2 - packetlog_height
def refresh_pad(window):
# global messages_pad, nodes_pad, channel_pad
def refresh_pad(window: int) -> None:
win_height = channel_win.getmaxyx()[0]
if(window == 1):
@@ -746,7 +745,7 @@ def refresh_pad(window):
box.getbegyx()[0] + 1, box.getbegyx()[1] + 1,
box.getbegyx()[0] + lines, box.getbegyx()[1] + box.getmaxyx()[1] - 2)
def highlight_line(highlight, window, line):
def highlight_line(highlight: bool, window: int, line: int) -> None:
pad = nodes_pad
color = get_color("node_list")
@@ -767,25 +766,25 @@ def highlight_line(highlight, window, line):
pad.chgat(line, 1, select_len, color | curses.A_REVERSE if highlight else color)
def add_notification(channel_number):
def add_notification(channel_number: int) -> None:
if channel_number not in globals.notifications:
globals.notifications.append(channel_number)
def remove_notification(channel_number):
def remove_notification(channel_number: int) -> None:
if channel_number in globals.notifications:
globals.notifications.remove(channel_number)
def draw_text_field(win, text, color):
def draw_text_field(win: curses.window, text: str, color: int) -> None:
win.border()
win.addstr(1, 1, text, color)
def draw_centered_text_field(win, text, y_offset, color):
def draw_centered_text_field(win: curses.window, text: str, y_offset: int, color: int) -> None:
height, width = win.getmaxyx()
x = (width - len(text)) // 2
y = (height // 2) + y_offset
win.addstr(y, x, text, color)
win.refresh()
def draw_debug(value):
def draw_debug(value: str | int) -> None:
function_win.addstr(1, 1, f"debug: {value} ")
function_win.refresh()

View File

@@ -15,8 +15,7 @@ from contact.utilities.control_utils import parse_ini_file, transform_menu_path
from contact.ui.user_config import json_editor
from contact.ui.ui_state import MenuState
state = MenuState()
menu_state = MenuState()
# Constants
width = 80
@@ -38,11 +37,14 @@ config_folder = os.path.join(locals_dir, "node-configs")
# Load translations
field_mapping, help_text = parse_ini_file(translation_file)
# Aliases
Segment = tuple[str, str, bool, bool]
WrappedLine = list[Segment]
def display_menu(state):
def display_menu(menu_state: MenuState) -> tuple[object, object]: # curses.window or pad types
min_help_window_height = 6
num_items = len(state.current_menu) + (1 if state.show_save_option else 0)
num_items = len(menu_state.current_menu) + (1 if menu_state.show_save_option else 0)
# Determine the available height for the menu
max_menu_height = curses.LINES
@@ -62,18 +64,18 @@ def display_menu(state):
menu_win.border()
menu_win.keypad(True)
menu_pad = curses.newpad(len(state.current_menu) + 1, width - 8)
menu_pad = curses.newpad(len(menu_state.current_menu) + 1, width - 8)
menu_pad.bkgd(get_color("background"))
header = " > ".join(word.title() for word in state.menu_path)
header = " > ".join(word.title() for word in menu_state.menu_path)
if len(header) > width - 4:
header = header[:width - 7] + "..."
menu_win.addstr(1, 2, header, get_color("settings_breadcrumbs", bold=True))
transformed_path = transform_menu_path(state.menu_path)
transformed_path = transform_menu_path(menu_state.menu_path)
for idx, option in enumerate(state.current_menu):
field_info = state.current_menu[option]
for idx, option in enumerate(menu_state.current_menu):
field_info = menu_state.current_menu[option]
current_value = field_info[1] if isinstance(field_info, tuple) else ""
full_key = '.'.join(transformed_path + [option])
display_name = field_mapping.get(full_key, option)
@@ -82,46 +84,64 @@ def display_menu(state):
display_value = f"{current_value}"[:width // 2 - 4]
try:
color = get_color("settings_sensitive" if option in sensitive_settings else "settings_default", reverse=(idx == state.selected_index))
color = get_color("settings_sensitive" if option in sensitive_settings else "settings_default", reverse=(idx == menu_state.selected_index))
menu_pad.addstr(idx, 0, f"{display_option:<{width // 2 - 2}} {display_value}".ljust(width - 8), color)
except curses.error:
pass
if state.show_save_option:
if menu_state.show_save_option:
save_position = menu_height - 2
menu_win.addstr(save_position, (width - len(save_option)) // 2, save_option, get_color("settings_save", reverse=(state.selected_index == len(state.current_menu))))
menu_win.addstr(save_position, (width - len(save_option)) // 2, save_option, get_color("settings_save", reverse=(menu_state.selected_index == len(menu_state.current_menu))))
# Draw help window with dynamically updated max_help_lines
draw_help_window(start_y, start_x, menu_height, max_help_lines, transformed_path, state)
draw_help_window(start_y, start_x, menu_height, max_help_lines, transformed_path, menu_state)
menu_win.refresh()
menu_pad.refresh(
state.start_index[-1], 0,
menu_state.start_index[-1], 0,
menu_win.getbegyx()[0] + 3, menu_win.getbegyx()[1] + 4,
menu_win.getbegyx()[0] + 3 + menu_win.getmaxyx()[0] - 5 - (2 if state.show_save_option else 0),
menu_win.getbegyx()[0] + 3 + menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0),
menu_win.getbegyx()[1] + menu_win.getmaxyx()[1] - 4
)
max_index = num_items + (1 if state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if state.show_save_option else 0)
max_index = num_items + (1 if menu_state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0)
draw_arrows(menu_win, visible_height, max_index, state)
draw_arrows(menu_win, visible_height, max_index, menu_state)
return menu_win, menu_pad
def draw_help_window(menu_start_y, menu_start_x, menu_height, max_help_lines, transformed_path, state):
def draw_help_window(
menu_start_y: int,
menu_start_x: int,
menu_height: int,
max_help_lines: int,
transformed_path: list[str],
menu_state: MenuState
) -> None:
global help_win
if 'help_win' not in globals():
help_win = None # Initialize if it does not exist
selected_option = list(state.current_menu.keys())[state.selected_index] if state.current_menu else None
selected_option = list(menu_state.current_menu.keys())[menu_state.selected_index] if menu_state.current_menu else None
help_y = menu_start_y + menu_height
help_win = update_help_window(help_win, help_text, transformed_path, selected_option, max_help_lines, width, help_y, menu_start_x)
def update_help_window(help_win, help_text, transformed_path, selected_option, max_help_lines, width, help_y, help_x):
def update_help_window(
help_win: object, # curses window or None
help_text: dict[str, str],
transformed_path: list[str],
selected_option: str | None,
max_help_lines: int,
width: int,
help_y: int,
help_x: int
) -> object: # returns a curses window
"""Handles rendering the help window consistently."""
wrapped_help = get_wrapped_help_text(help_text, transformed_path, selected_option, width, max_help_lines)
@@ -159,7 +179,13 @@ def update_help_window(help_win, help_text, transformed_path, selected_option, m
return help_win
def get_wrapped_help_text(help_text, transformed_path, selected_option, width, max_lines):
def get_wrapped_help_text(
help_text: dict[str, str],
transformed_path: list[str],
selected_option: str | None,
width: int,
max_lines: int
) -> list[WrappedLine]:
"""Fetches and formats help text for display, ensuring it fits within the allowed lines."""
full_help_key = '.'.join(transformed_path + [selected_option]) if selected_option else None
@@ -178,7 +204,7 @@ def get_wrapped_help_text(help_text, transformed_path, selected_option, width, m
r'\\033\[4m(.*?)\\033\[0m': ('settings_default', False, True) # Underline
}
def extract_ansi_segments(text):
def extract_ansi_segments(text: str) -> list[Segment]:
"""Extracts and replaces ANSI color codes, ensuring spaces are preserved."""
matches = []
last_pos = 0
@@ -208,7 +234,7 @@ def get_wrapped_help_text(help_text, transformed_path, selected_option, width, m
return matches
def wrap_ansi_text(segments, wrap_width):
def wrap_ansi_text(segments: list[Segment], wrap_width: int) -> list[WrappedLine]:
"""Wraps text while preserving ANSI formatting and spaces."""
wrapped_lines = []
line_buffer = []
@@ -251,116 +277,131 @@ def get_wrapped_help_text(help_text, transformed_path, selected_option, width, m
return wrapped_help
def move_highlight(old_idx, options, menu_win, menu_pad, help_win, help_text, max_help_lines, state):
if old_idx == state.selected_index: # No-op
def move_highlight(
old_idx: int,
options: list[str],
menu_win: object,
menu_pad: object,
help_win: object,
help_text: dict[str, str],
max_help_lines: int,
menu_state: MenuState
) -> None:
if old_idx == menu_state.selected_index: # No-op
return
max_index = len(options) + (1 if state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if state.show_save_option else 0)
max_index = len(options) + (1 if menu_state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0)
# Adjust state.start_index only when moving out of visible range
if state.selected_index == max_index and state.show_save_option:
# Adjust menu_state.start_index only when moving out of visible range
if menu_state.selected_index == max_index and menu_state.show_save_option:
pass
elif state.selected_index < state.start_index[-1]: # Moving above the visible area
state.start_index[-1] = state.selected_index
elif state.selected_index >= state.start_index[-1] + visible_height: # Moving below the visible area
state.start_index[-1] = state.selected_index - visible_height
elif menu_state.selected_index < menu_state.start_index[-1]: # Moving above the visible area
menu_state.start_index[-1] = menu_state.selected_index
elif menu_state.selected_index >= menu_state.start_index[-1] + visible_height: # Moving below the visible area
menu_state.start_index[-1] = menu_state.selected_index - visible_height
pass
# Ensure state.start_index is within bounds
state.start_index[-1] = max(0, min(state.start_index[-1], max_index - visible_height + 1))
# Ensure menu_state.start_index is within bounds
menu_state.start_index[-1] = max(0, min(menu_state.start_index[-1], max_index - visible_height + 1))
# Clear old selection
if state.show_save_option and old_idx == max_index:
if menu_state.show_save_option and old_idx == max_index:
menu_win.chgat(menu_win.getmaxyx()[0] - 2, (width - len(save_option)) // 2, len(save_option), get_color("settings_save"))
else:
menu_pad.chgat(old_idx, 0, menu_pad.getmaxyx()[1], get_color("settings_sensitive") if options[old_idx] in sensitive_settings else get_color("settings_default"))
# Highlight new selection
if state.show_save_option and state.selected_index == max_index:
if menu_state.show_save_option and menu_state.selected_index == max_index:
menu_win.chgat(menu_win.getmaxyx()[0] - 2, (width - len(save_option)) // 2, len(save_option), get_color("settings_save", reverse=True))
else:
menu_pad.chgat(state.selected_index, 0, menu_pad.getmaxyx()[1], get_color("settings_sensitive", reverse=True) if options[state.selected_index] in sensitive_settings else get_color("settings_default", reverse=True))
menu_pad.chgat(menu_state.selected_index, 0, menu_pad.getmaxyx()[1], get_color("settings_sensitive", reverse=True) if options[menu_state.selected_index] in sensitive_settings else get_color("settings_default", reverse=True))
menu_win.refresh()
# Refresh pad only if scrolling is needed
menu_pad.refresh(state.start_index[-1], 0,
menu_pad.refresh(menu_state.start_index[-1], 0,
menu_win.getbegyx()[0] + 3, menu_win.getbegyx()[1] + 4,
menu_win.getbegyx()[0] + 3 + visible_height,
menu_win.getbegyx()[1] + menu_win.getmaxyx()[1] - 4)
# Update help window
transformed_path = transform_menu_path(state.menu_path)
selected_option = options[state.selected_index] if state.selected_index < len(options) else None
transformed_path = transform_menu_path(menu_state.menu_path)
selected_option = options[menu_state.selected_index] if menu_state.selected_index < len(options) else None
help_y = menu_win.getbegyx()[0] + menu_win.getmaxyx()[0]
help_win = update_help_window(help_win, help_text, transformed_path, selected_option, max_help_lines, width, help_y, menu_win.getbegyx()[1])
draw_arrows(menu_win, visible_height, max_index, state)
draw_arrows(menu_win, visible_height, max_index, menu_state)
def draw_arrows(win, visible_height, max_index, state):
def draw_arrows(
win: object,
visible_height: int,
max_index: int,
menu_state: MenuState
) -> None:
# vh = visible_height + (1 if show_save_option else 0)
mi = max_index - (2 if state.show_save_option else 0)
mi = max_index - (2 if menu_state.show_save_option else 0)
if visible_height < mi:
if state.start_index[-1] > 0:
if menu_state.start_index[-1] > 0:
win.addstr(3, 2, "", get_color("settings_default"))
else:
win.addstr(3, 2, " ", get_color("settings_default"))
if mi - state.start_index[-1] >= visible_height + (0 if state.show_save_option else 1) :
if mi - menu_state.start_index[-1] >= visible_height + (0 if menu_state.show_save_option else 1) :
win.addstr(visible_height + 3, 2, "", get_color("settings_default"))
else:
win.addstr(visible_height + 3, 2, " ", get_color("settings_default"))
def settings_menu(stdscr, interface):
def settings_menu(stdscr: object, interface: object) -> None:
curses.update_lines_cols()
menu = generate_menu_from_protobuf(interface)
state.current_menu = menu["Main Menu"]
state.menu_path = ["Main Menu"]
menu_state.current_menu = menu["Main Menu"]
menu_state.menu_path = ["Main Menu"]
modified_settings = {}
need_redraw = True
state.show_save_option = False
menu_state.show_save_option = False
while True:
if(need_redraw):
options = list(state.current_menu.keys())
options = list(menu_state.current_menu.keys())
state.show_save_option = (
len(state.menu_path) > 2 and ("Radio Settings" in state.menu_path or "Module Settings" in state.menu_path)
menu_state.show_save_option = (
len(menu_state.menu_path) > 2 and ("Radio Settings" in menu_state.menu_path or "Module Settings" in menu_state.menu_path)
) or (
len(state.menu_path) == 2 and "User Settings" in state.menu_path
len(menu_state.menu_path) == 2 and "User Settings" in menu_state.menu_path
) or (
len(state.menu_path) == 3 and "Channels" in state.menu_path
len(menu_state.menu_path) == 3 and "Channels" in menu_state.menu_path
)
# Display the menu
menu_win, menu_pad = display_menu(state)
menu_win, menu_pad = display_menu(menu_state)
need_redraw = False
# Capture user input
key = menu_win.getch()
max_index = len(options) + (1 if state.show_save_option else 0) - 1
max_index = len(options) + (1 if menu_state.show_save_option else 0) - 1
# max_help_lines = 4
if key == curses.KEY_UP:
old_selected_index = state.selected_index
state.selected_index = max_index if state.selected_index == 0 else state.selected_index - 1
move_highlight(old_selected_index, options, menu_win, menu_pad, help_win, help_text, max_help_lines, state)
old_selected_index = menu_state.selected_index
menu_state.selected_index = max_index if menu_state.selected_index == 0 else menu_state.selected_index - 1
move_highlight(old_selected_index, options, menu_win, menu_pad, help_win, help_text, max_help_lines, menu_state)
elif key == curses.KEY_DOWN:
old_selected_index = state.selected_index
state.selected_index = 0 if state.selected_index == max_index else state.selected_index + 1
move_highlight(old_selected_index, options, menu_win, menu_pad, help_win, help_text, max_help_lines, state)
old_selected_index = menu_state.selected_index
menu_state.selected_index = 0 if menu_state.selected_index == max_index else menu_state.selected_index + 1
move_highlight(old_selected_index, options, menu_win, menu_pad, help_win, help_text, max_help_lines, menu_state)
elif key == curses.KEY_RESIZE:
need_redraw = True
@@ -372,36 +413,36 @@ def settings_menu(stdscr, interface):
menu_win.refresh()
help_win.refresh()
elif key == ord("\t") and state.show_save_option:
old_selected_index = state.selected_index
state.selected_index = max_index
move_highlight(old_selected_index, options, menu_win, menu_pad, help_win, help_text, max_help_lines, state)
elif key == ord("\t") and menu_state.show_save_option:
old_selected_index = menu_state.selected_index
menu_state.selected_index = max_index
move_highlight(old_selected_index, options, menu_win, menu_pad, help_win, help_text, max_help_lines, menu_state)
elif key == curses.KEY_RIGHT or key == ord('\n'):
need_redraw = True
state.start_index.append(0)
menu_state.start_index.append(0)
menu_win.erase()
help_win.erase()
# draw_help_window(menu_win.getbegyx()[0], menu_win.getbegyx()[1], menu_win.getmaxyx()[0], max_help_lines, state.current_menu, selected_index, transform_menu_path(state.menu_path))
# draw_help_window(menu_win.getbegyx()[0], menu_win.getbegyx()[1], menu_win.getmaxyx()[0], max_help_lines, menu_state.current_menu, selected_index, transform_menu_path(menu_state.menu_path))
menu_win.refresh()
help_win.refresh()
if state.show_save_option and state.selected_index == len(options):
save_changes(interface, modified_settings, state)
if menu_state.show_save_option and menu_state.selected_index == len(options):
save_changes(interface, modified_settings, menu_state)
modified_settings.clear()
logging.info("Changes Saved")
if len(state.menu_path) > 1:
state.menu_path.pop()
state.current_menu = menu["Main Menu"]
for step in state.menu_path[1:]:
state.current_menu = state.current_menu.get(step, {})
state.selected_index = 0
if len(menu_state.menu_path) > 1:
menu_state.menu_path.pop()
menu_state.current_menu = menu["Main Menu"]
for step in menu_state.menu_path[1:]:
menu_state.current_menu = menu_state.current_menu.get(step, {})
menu_state.selected_index = 0
continue
selected_option = options[state.selected_index]
selected_option = options[menu_state.selected_index]
if selected_option == "Exit":
break
@@ -410,7 +451,7 @@ def settings_menu(stdscr, interface):
filename = get_text_input("Enter a filename for the config file")
if not filename:
logging.info("Export aborted: No filename provided.")
state.start_index.pop()
menu_state.start_index.pop()
continue # Go back to the menu
if not filename.lower().endswith(".yaml"):
filename += ".yaml"
@@ -423,14 +464,14 @@ def settings_menu(stdscr, interface):
overwrite = get_list_input(f"{filename} already exists. Overwrite?", None, ["Yes", "No"])
if overwrite == "No":
logging.info("Export cancelled: User chose not to overwrite.")
state.start_index.pop()
menu_state.start_index.pop()
continue # Return to menu
os.makedirs(os.path.dirname(yaml_file_path), exist_ok=True)
with open(yaml_file_path, "w", encoding="utf-8") as file:
file.write(config_text)
logging.info(f"Config file saved to {yaml_file_path}")
dialog(stdscr, "Config File Saved:", yaml_file_path)
state.start_index.pop()
menu_state.start_index.pop()
continue
except PermissionError:
logging.error(f"Permission denied: Unable to write to {yaml_file_path}")
@@ -438,7 +479,7 @@ def settings_menu(stdscr, interface):
logging.error(f"OS error while saving config: {e}")
except Exception as e:
logging.error(f"Unexpected error: {e}")
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "Load Config File":
@@ -461,7 +502,7 @@ def settings_menu(stdscr, interface):
overwrite = get_list_input(f"Are you sure you want to load {filename}?", None, ["Yes", "No"])
if overwrite == "Yes":
config_import(interface, file_path)
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "Config URL":
@@ -473,7 +514,7 @@ def settings_menu(stdscr, interface):
if overwrite == "Yes":
interface.localNode.setURL(new_value)
logging.info(f"New Config URL sent to node")
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "Reboot":
@@ -481,7 +522,7 @@ def settings_menu(stdscr, interface):
if confirmation == "Yes":
interface.localNode.reboot()
logging.info(f"Node Reboot Requested by menu")
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "Reset Node DB":
@@ -489,7 +530,7 @@ def settings_menu(stdscr, interface):
if confirmation == "Yes":
interface.localNode.resetNodeDb()
logging.info(f"Node DB Reset Requested by menu")
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "Shutdown":
@@ -497,7 +538,7 @@ def settings_menu(stdscr, interface):
if confirmation == "Yes":
interface.localNode.shutdown()
logging.info(f"Node Shutdown Requested by menu")
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "Factory Reset":
@@ -505,28 +546,28 @@ def settings_menu(stdscr, interface):
if confirmation == "Yes":
interface.localNode.factoryReset()
logging.info(f"Factory Reset Requested by menu")
state.start_index.pop()
menu_state.start_index.pop()
continue
elif selected_option == "App Settings":
menu_win.clear()
menu_win.refresh()
state.menu_path.append("App Settings")
state.menu_index.append(state.selected_index)
json_editor(stdscr, state) # Open the App Settings menu
state.current_menu = menu["Main Menu"]
state.menu_path = ["Main Menu"]
state.start_index.pop()
state.selected_index = 4
menu_state.menu_path.append("App Settings")
menu_state.menu_index.append(menu_state.selected_index)
json_editor(stdscr, menu_state) # Open the App Settings menu
menu_state.current_menu = menu["Main Menu"]
menu_state.menu_path = ["Main Menu"]
menu_state.start_index.pop()
menu_state.selected_index = 4
continue
# need_redraw = True
field_info = state.current_menu.get(selected_option)
field_info = menu_state.current_menu.get(selected_option)
if isinstance(field_info, tuple):
field, current_value = field_info
# Transform the menu path to get the full key
transformed_path = transform_menu_path(state.menu_path)
transformed_path = transform_menu_path(menu_state.menu_path)
full_key = '.'.join(transformed_path + [selected_option])
# Fetch human-readable name from field_mapping
@@ -536,33 +577,33 @@ def settings_menu(stdscr, interface):
if selected_option in ['longName', 'shortName']:
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = current_value if new_value is None else new_value
state.current_menu[selected_option] = (field, new_value)
menu_state.current_menu[selected_option] = (field, new_value)
elif selected_option == 'isLicensed':
new_value = get_list_input(f"{human_readable_name} is currently: {current_value}", str(current_value), ["True", "False"])
new_value = new_value == "True"
state.current_menu[selected_option] = (field, new_value)
menu_state.current_menu[selected_option] = (field, new_value)
for option, (field, value) in state.current_menu.items():
for option, (field, value) in menu_state.current_menu.items():
modified_settings[option] = value
state.start_index.pop()
menu_state.start_index.pop()
elif selected_option in ['latitude', 'longitude', 'altitude']:
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = current_value if new_value is None else new_value
state.current_menu[selected_option] = (field, new_value)
menu_state.current_menu[selected_option] = (field, new_value)
for option in ['latitude', 'longitude', 'altitude']:
if option in state.current_menu:
modified_settings[option] = state.current_menu[option][1]
if option in menu_state.current_menu:
modified_settings[option] = menu_state.current_menu[option][1]
state.start_index.pop()
menu_state.start_index.pop()
elif selected_option == "admin_key":
new_values = get_admin_key_input(current_value)
new_value = current_value if new_values is None else [base64.b64decode(key) for key in new_values]
state.start_index.pop()
menu_state.start_index.pop()
elif field.type == 8: # Handle boolean type
new_value = get_list_input(human_readable_name, str(current_value), ["True", "False"])
@@ -570,39 +611,39 @@ def settings_menu(stdscr, interface):
pass # Leave it as-is
else:
new_value = new_value == "True" or new_value is True
state.start_index.pop()
menu_state.start_index.pop()
elif field.label == field.LABEL_REPEATED: # Handle repeated field - Not currently used
new_value = get_repeated_input(current_value)
new_value = current_value if new_value is None else new_value.split(", ")
state.start_index.pop()
menu_state.start_index.pop()
elif field.enum_type: # Enum field
enum_options = {v.name: v.number for v in field.enum_type.values}
new_value_name = get_list_input(human_readable_name, current_value, list(enum_options.keys()))
new_value = enum_options.get(new_value_name, current_value)
state.start_index.pop()
menu_state.start_index.pop()
elif field.type == 7: # Field type 7 corresponds to FIXED32
new_value = get_fixed32_input(current_value)
state.start_index.pop()
menu_state.start_index.pop()
elif field.type == 13: # Field type 13 corresponds to UINT32
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = current_value if new_value is None else int(new_value)
state.start_index.pop()
menu_state.start_index.pop()
elif field.type == 2: # Field type 13 corresponds to INT64
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = current_value if new_value is None else float(new_value)
state.start_index.pop()
menu_state.start_index.pop()
else: # Handle other field types
new_value = get_text_input(f"{human_readable_name} is currently: {current_value}")
new_value = current_value if new_value is None else new_value
state.start_index.pop()
menu_state.start_index.pop()
for key in state.menu_path[3:]: # Skip "Main Menu"
for key in menu_state.menu_path[3:]: # Skip "Main Menu"
modified_settings = modified_settings.setdefault(key, {})
# Add the new value to the appropriate level
@@ -613,12 +654,12 @@ def settings_menu(stdscr, interface):
enum_value_descriptor = field.enum_type.values_by_number.get(new_value)
new_value = enum_value_descriptor.name if enum_value_descriptor else new_value
state.current_menu[selected_option] = (field, new_value)
menu_state.current_menu[selected_option] = (field, new_value)
else:
state.current_menu = state.current_menu[selected_option]
state.menu_path.append(selected_option)
state.menu_index.append(state.selected_index)
state.selected_index = 0
menu_state.current_menu = menu_state.current_menu[selected_option]
menu_state.menu_path.append(selected_option)
menu_state.menu_index.append(menu_state.selected_index)
menu_state.selected_index = 0
elif key == curses.KEY_LEFT:
@@ -628,29 +669,29 @@ def settings_menu(stdscr, interface):
help_win.erase()
# max_help_lines = 4
# draw_help_window(menu_win.getbegyx()[0], menu_win.getbegyx()[1], menu_win.getmaxyx()[0], max_help_lines, state.current_menu, selected_index, transform_menu_path(state.menu_path))
# draw_help_window(menu_win.getbegyx()[0], menu_win.getbegyx()[1], menu_win.getmaxyx()[0], max_help_lines, menu_state.current_menu, selected_index, transform_menu_path(menu_state.menu_path))
menu_win.refresh()
help_win.refresh()
if len(state.menu_path) < 2:
if len(menu_state.menu_path) < 2:
modified_settings.clear()
# Navigate back to the previous menu
if len(state.menu_path) > 1:
state.menu_path.pop()
state.current_menu = menu["Main Menu"]
for step in state.menu_path[1:]:
state.current_menu = state.current_menu.get(step, {})
state.selected_index = state.menu_index.pop()
state.start_index.pop()
if len(menu_state.menu_path) > 1:
menu_state.menu_path.pop()
menu_state.current_menu = menu["Main Menu"]
for step in menu_state.menu_path[1:]:
menu_state.current_menu = menu_state.current_menu.get(step, {})
menu_state.selected_index = menu_state.menu_index.pop()
menu_state.start_index.pop()
elif key == 27: # Escape key
menu_win.erase()
menu_win.refresh()
break
def set_region(interface):
def set_region(interface: object) -> None:
node = interface.getNode('^local')
device_config = node.localConfig
lora_descriptor = device_config.lora.DESCRIPTOR

View File

@@ -1,5 +1,5 @@
import logging
import json
import logging
import os
# Get the parent directory of the script
@@ -11,11 +11,11 @@ json_file_path = os.path.join(parent_dir, "config.json")
log_file_path = os.path.join(parent_dir, "client.log")
db_file_path = os.path.join(parent_dir, "client.db")
def format_json_single_line_arrays(data, indent=4):
def format_json_single_line_arrays(data: dict[str, object], indent: int = 4) -> str:
"""
Formats JSON with arrays on a single line while keeping other elements properly indented.
"""
def format_value(value, current_indent):
def format_value(value: object, current_indent: int) -> str:
if isinstance(value, dict):
items = []
for key, val in value.items():
@@ -31,7 +31,7 @@ def format_json_single_line_arrays(data, indent=4):
return format_value(data, indent)
# Recursive function to check and update nested dictionaries
def update_dict(default, actual):
def update_dict(default: dict[str, object], actual: dict[str, object]) -> bool:
updated = False
for key, value in default.items():
if key not in actual:
@@ -42,7 +42,7 @@ def update_dict(default, actual):
updated = update_dict(value, actual[key]) or updated
return updated
def initialize_config():
def initialize_config() -> dict[str, object]:
COLOR_CONFIG_DARK = {
"default": ["white", "black"],
"background": [" ", "black"],
@@ -161,7 +161,7 @@ def initialize_config():
return loaded_config
def assign_config_variables(loaded_config):
def assign_config_variables(loaded_config: dict[str, object]) -> None:
# Assign values to local variables
global db_file_path, log_file_path, message_prefix, sent_message_prefix

View File

@@ -1,7 +1,7 @@
import curses
from contact.ui.colors import get_color
def dialog(stdscr, title, message):
def dialog(stdscr: curses.window, title: str, message: str) -> None:
height, width = stdscr.getmaxyx()
# Calculate dialog dimensions

View File

@@ -1,19 +1,27 @@
from collections import OrderedDict
from meshtastic.protobuf import config_pb2, module_config_pb2, channel_pb2
import logging
import base64
import logging
import os
from collections import OrderedDict
from typing import Any
from google.protobuf.message import Message
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
locals_dir = os.path.dirname(os.path.abspath(__file__))
translation_file = os.path.join(locals_dir, "localisations", "en.ini")
def encode_if_bytes(value):
def encode_if_bytes(value: Any) -> str:
"""Encode byte values to base64 string."""
if isinstance(value, bytes):
return base64.b64encode(value).decode('utf-8')
return value
def extract_fields(message_instance, current_config=None):
def extract_fields(
message_instance: Message,
current_config: Message | dict[str, Any] | None = None
) -> dict[str, Any]:
if isinstance(current_config, dict): # Handle dictionaries
return {key: (None, encode_if_bytes(current_config.get(key, "Not Set"))) for key in current_config}
@@ -47,7 +55,10 @@ def extract_fields(message_instance, current_config=None):
menu[field.name] = (field, encode_if_bytes(current_value))
return menu
def generate_menu_from_protobuf(interface):
def generate_menu_from_protobuf(interface: object) -> dict[str, Any]:
"""
Builds the full settings menu structure from the protobuf definitions.
"""
menu_structure = {"Main Menu": {}}
# Add User Settings

View File

@@ -1,7 +1,8 @@
import curses
from contact.ui.colors import get_color
def draw_splash(stdscr):
def draw_splash(stdscr: object) -> None:
"""Draw the splash screen with a logo and connecting message."""
curses.curs_set(0)
stdscr.clear()

View File

@@ -1,8 +1,10 @@
from typing import Any
class MenuState:
def __init__(self):
self.menu_index = [] # Row we left the previous menus
self.start_index = [0] # Row to start the menu if it doesn't all fit
self.selected_index = 0 # Selected Row
self.current_menu = {} # Contents of the current menu
self.menu_path = [] # Menu Path
self.show_save_option = False
self.menu_index: list[int]= [] # Row we left the previous menus
self.start_index: list[int] = [0] # Row to start the menu if it doesn't all fit
self.selected_index: int = 0 # Selected Row
self.current_menu: dict[str, Any] | list[Any] | str | int = {} # Contents of the current menu
self.menu_path: list[str] = [] # Menu Path
self.show_save_option: bool = False # Display 'Save'

View File

@@ -1,6 +1,7 @@
import os
import json
import curses
from typing import Any
from contact.ui.colors import get_color, setup_colors, COLOR_MAP
from contact.ui.default_config import format_json_single_line_arrays, loaded_config
from contact.utilities.input_handlers import get_list_input
@@ -9,8 +10,7 @@ width = 80
save_option = "Save Changes"
sensitive_settings = []
def edit_color_pair(key, current_value):
def edit_color_pair(key: str, current_value: list[str]) -> list[str]:
"""
Allows the user to select a foreground and background color for a key.
"""
@@ -20,7 +20,7 @@ def edit_color_pair(key, current_value):
return [fg_color, bg_color]
def edit_value(key, current_value, state):
def edit_value(key: str, current_value: str) -> str:
height = 10
input_width = width - 16 # Allow space for "New Value: "
@@ -96,17 +96,17 @@ def edit_value(key, current_value, state):
return user_input if user_input else current_value
def display_menu(state):
def display_menu(menu_state: Any) -> tuple[Any, Any, list[str]]:
"""
Render the configuration menu with a Save button directly added to the window.
"""
num_items = len(state.current_menu) + (1 if state.show_save_option else 0)
num_items = len(menu_state.current_menu) + (1 if menu_state.show_save_option else 0)
# Determine menu items based on the type of current_menu
if isinstance(state.current_menu, dict):
options = list(state.current_menu.keys())
elif isinstance(state.current_menu, list):
options = [f"[{i}]" for i in range(len(state.current_menu))]
if isinstance(menu_state.current_menu, dict):
options = list(menu_state.current_menu.keys())
elif isinstance(menu_state.current_menu, list):
options = [f"[{i}]" for i in range(len(menu_state.current_menu))]
else:
options = [] # Fallback in case of unexpected data types
@@ -130,110 +130,122 @@ def display_menu(state):
menu_pad.bkgd(get_color("background"))
# Display the menu path
header = " > ".join(state.menu_path)
header = " > ".join(menu_state.menu_path)
if len(header) > width - 4:
header = header[:width - 7] + "..."
menu_win.addstr(1, 2, header, get_color("settings_breadcrumbs", bold=True))
# Populate the pad with menu options
for idx, key in enumerate(options):
value = state.current_menu[key] if isinstance(state.current_menu, dict) else state.current_menu[int(key.strip("[]"))]
value = menu_state.current_menu[key] if isinstance(menu_state.current_menu, dict) else menu_state.current_menu[int(key.strip("[]"))]
display_key = f"{key}"[:width // 2 - 2]
display_value = (
f"{value}"[:width // 2 - 8]
)
color = get_color("settings_default", reverse=(idx == state.selected_index))
color = get_color("settings_default", reverse=(idx == menu_state.selected_index))
menu_pad.addstr(idx, 0, f"{display_key:<{width // 2 - 2}} {display_value}".ljust(width - 8), color)
# Add Save button to the main window
if state.show_save_option:
if menu_state.show_save_option:
save_position = menu_height - 2
menu_win.addstr(save_position, (width - len(save_option)) // 2, save_option, get_color("settings_save", reverse=(state.selected_index == len(state.current_menu))))
menu_win.addstr(save_position, (width - len(save_option)) // 2, save_option, get_color("settings_save", reverse=(menu_state.selected_index == len(menu_state.current_menu))))
menu_win.refresh()
menu_pad.refresh(
state.start_index[-1], 0,
menu_state.start_index[-1], 0,
menu_win.getbegyx()[0] + 3, menu_win.getbegyx()[1] + 4,
menu_win.getbegyx()[0] + 3 + menu_win.getmaxyx()[0] - 5 - (2 if state.show_save_option else 0),
menu_win.getbegyx()[0] + 3 + menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0),
menu_win.getbegyx()[1] + menu_win.getmaxyx()[1] - 4
)
max_index = num_items + (1 if state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if state.show_save_option else 0)
max_index = num_items + (1 if menu_state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0)
draw_arrows(menu_win, visible_height, max_index, state)
draw_arrows(menu_win, visible_height, max_index, menu_state)
return menu_win, menu_pad, options
def move_highlight(old_idx, options, menu_win, menu_pad, state):
if old_idx == state.selected_index: # No-op
def move_highlight(
old_idx: int,
options: list[str],
menu_win: curses.window,
menu_pad: curses.window,
menu_state: Any
) -> None:
if old_idx == menu_state.selected_index: # No-op
return
max_index = len(options) + (1 if state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if state.show_save_option else 0)
max_index = len(options) + (1 if menu_state.show_save_option else 0) - 1
visible_height = menu_win.getmaxyx()[0] - 5 - (2 if menu_state.show_save_option else 0)
# Adjust state.start_index only when moving out of visible range
if state.selected_index == max_index and state.show_save_option:
# Adjust menu_state.start_index only when moving out of visible range
if menu_state.selected_index == max_index and menu_state.show_save_option:
pass
elif state.selected_index < state.start_index[-1]: # Moving above the visible area
state.start_index[-1] = state.selected_index
elif state.selected_index >= state.start_index[-1] + visible_height: # Moving below the visible area
state.start_index[-1] = state.selected_index - visible_height
elif menu_state.selected_index < menu_state.start_index[-1]: # Moving above the visible area
menu_state.start_index[-1] = menu_state.selected_index
elif menu_state.selected_index >= menu_state.start_index[-1] + visible_height: # Moving below the visible area
menu_state.start_index[-1] = menu_state.selected_index - visible_height
pass
# Ensure state.start_index is within bounds
state.start_index[-1] = max(0, min(state.start_index[-1], max_index - visible_height + 1))
# Ensure menu_state.start_index is within bounds
menu_state.start_index[-1] = max(0, min(menu_state.start_index[-1], max_index - visible_height + 1))
# Clear old selection
if state.show_save_option and old_idx == max_index:
if menu_state.show_save_option and old_idx == max_index:
menu_win.chgat(menu_win.getmaxyx()[0] - 2, (width - len(save_option)) // 2, len(save_option), get_color("settings_save"))
else:
menu_pad.chgat(old_idx, 0, menu_pad.getmaxyx()[1], get_color("settings_sensitive") if options[old_idx] in sensitive_settings else get_color("settings_default"))
# Highlight new selection
if state.show_save_option and state.selected_index == max_index:
if menu_state.show_save_option and menu_state.selected_index == max_index:
menu_win.chgat(menu_win.getmaxyx()[0] - 2, (width - len(save_option)) // 2, len(save_option), get_color("settings_save", reverse=True))
else:
menu_pad.chgat(state.selected_index, 0, menu_pad.getmaxyx()[1], get_color("settings_sensitive", reverse=True) if options[state.selected_index] in sensitive_settings else get_color("settings_default", reverse=True))
menu_pad.chgat(menu_state.selected_index, 0, menu_pad.getmaxyx()[1], get_color("settings_sensitive", reverse=True) if options[menu_state.selected_index] in sensitive_settings else get_color("settings_default", reverse=True))
menu_win.refresh()
# Refresh pad only if scrolling is needed
menu_pad.refresh(state.start_index[-1], 0,
menu_pad.refresh(menu_state.start_index[-1], 0,
menu_win.getbegyx()[0] + 3, menu_win.getbegyx()[1] + 4,
menu_win.getbegyx()[0] + 3 + visible_height,
menu_win.getbegyx()[1] + menu_win.getmaxyx()[1] - 4)
draw_arrows(menu_win, visible_height, max_index, state)
draw_arrows(menu_win, visible_height, max_index, menu_state)
def draw_arrows(win, visible_height, max_index, state):
def draw_arrows(
win: curses.window,
visible_height: int,
max_index: int,
menu_state: any
) -> None:
mi = max_index - (2 if state.show_save_option else 0)
mi = max_index - (2 if menu_state.show_save_option else 0)
if visible_height < mi:
if state.start_index[-1] > 0:
if menu_state.start_index[-1] > 0:
win.addstr(3, 2, "", get_color("settings_default"))
else:
win.addstr(3, 2, " ", get_color("settings_default"))
if mi - state.start_index[-1] >= visible_height + (0 if state.show_save_option else 1) :
if mi - menu_state.start_index[-1] >= visible_height + (0 if menu_state.show_save_option else 1) :
win.addstr(visible_height + 3, 2, "", get_color("settings_default"))
else:
win.addstr(visible_height + 3, 2, " ", get_color("settings_default"))
def json_editor(stdscr, state):
def json_editor(stdscr: curses.window, menu_state: Any) -> None:
state.selected_index = 0 # Track the selected option
menu_state.selected_index = 0 # Track the selected option
script_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.abspath(os.path.join(script_dir, os.pardir))
file_path = os.path.join(parent_dir, "config.json")
state.show_save_option = True # Always show the Save button
menu_state.show_save_option = True # Always show the Save button
# Ensure the file exists
if not os.path.exists(file_path):
@@ -241,41 +253,41 @@ def json_editor(stdscr, state):
json.dump({}, f)
# Load JSON data
with open(file_path, "r") as f:
with open(file_path, "r", encoding="utf-8") as f:
original_data = json.load(f)
data = original_data # Reference to the original data
state.current_menu = data # Track the current level of the menu
menu_state.current_menu = data # Track the current level of the menu
# Render the menu
menu_win, menu_pad, options = display_menu(state)
menu_win, menu_pad, options = display_menu(menu_state)
need_redraw = True
while True:
if(need_redraw):
menu_win, menu_pad, options = display_menu(state)
menu_win, menu_pad, options = display_menu(menu_state)
menu_win.refresh()
need_redraw = False
max_index = len(options) + (1 if state.show_save_option else 0) - 1
max_index = len(options) + (1 if menu_state.show_save_option else 0) - 1
key = menu_win.getch()
if key == curses.KEY_UP:
old_selected_index = state.selected_index
state.selected_index = max_index if state.selected_index == 0 else state.selected_index - 1
move_highlight(old_selected_index, options, menu_win, menu_pad, state)
old_selected_index = menu_state.selected_index
menu_state.selected_index = max_index if menu_state.selected_index == 0 else menu_state.selected_index - 1
move_highlight(old_selected_index, options, menu_win, menu_pad, menu_state)
elif key == curses.KEY_DOWN:
old_selected_index = state.selected_index
state.selected_index = 0 if state.selected_index == max_index else state.selected_index + 1
move_highlight(old_selected_index, options, menu_win, menu_pad, state)
old_selected_index = menu_state.selected_index
menu_state.selected_index = 0 if menu_state.selected_index == max_index else menu_state.selected_index + 1
move_highlight(old_selected_index, options, menu_win, menu_pad, menu_state)
elif key == ord("\t") and state.show_save_option:
old_selected_index = state.selected_index
state.selected_index = max_index
move_highlight(old_selected_index, options, menu_win, menu_pad, state)
elif key == ord("\t") and menu_state.show_save_option:
old_selected_index = menu_state.selected_index
menu_state.selected_index = max_index
move_highlight(old_selected_index, options, menu_win, menu_pad, menu_state)
elif key in (curses.KEY_RIGHT, 10, 13): # 10 = \n, 13 = carriage return
@@ -283,41 +295,41 @@ def json_editor(stdscr, state):
menu_win.erase()
menu_win.refresh()
if state.selected_index < len(options): # Handle selection of a menu item
selected_key = options[state.selected_index]
state.menu_path.append(str(selected_key))
state.start_index.append(0)
state.menu_index.append(state.selected_index)
if menu_state.selected_index < len(options): # Handle selection of a menu item
selected_key = options[menu_state.selected_index]
menu_state.menu_path.append(str(selected_key))
menu_state.start_index.append(0)
menu_state.menu_index.append(menu_state.selected_index)
# Handle nested data
if isinstance(state.current_menu, dict):
if selected_key in state.current_menu:
selected_data = state.current_menu[selected_key]
if isinstance(menu_state.current_menu, dict):
if selected_key in menu_state.current_menu:
selected_data = menu_state.current_menu[selected_key]
else:
continue # Skip invalid key
elif isinstance(state.current_menu, list):
selected_data = state.current_menu[int(selected_key.strip("[]"))]
elif isinstance(menu_state.current_menu, list):
selected_data = menu_state.current_menu[int(selected_key.strip("[]"))]
if isinstance(selected_data, list) and len(selected_data) == 2:
# Edit color pair
new_value = edit_color_pair(selected_key, selected_data)
state.menu_path.pop()
state.start_index.pop()
state.menu_index.pop()
state.current_menu[selected_key] = new_value
menu_state.menu_path.pop()
menu_state.start_index.pop()
menu_state.menu_index.pop()
menu_state.current_menu[selected_key] = new_value
elif isinstance(selected_data, (dict, list)):
# Navigate into nested data
state.current_menu = selected_data
state.selected_index = 0 # Reset the selected index
menu_state.current_menu = selected_data
menu_state.selected_index = 0 # Reset the selected index
else:
# General value editing
new_value = edit_value(selected_key, selected_data, state)
state.menu_path.pop()
state.menu_index.pop()
state.start_index.pop()
state.current_menu[selected_key] = new_value
new_value = edit_value(selected_key, selected_data)
menu_state.menu_path.pop()
menu_state.menu_index.pop()
menu_state.start_index.pop()
menu_state.current_menu[selected_key] = new_value
need_redraw = True
else:
@@ -332,16 +344,16 @@ def json_editor(stdscr, state):
menu_win.erase()
menu_win.refresh()
# state.selected_index = state.menu_index[-1]
# menu_state.selected_index = menu_state.menu_index[-1]
# Navigate back in the menu
if len(state.menu_path) > 2:
state.menu_path.pop()
state.start_index.pop()
state.current_menu = data
if len(menu_state.menu_path) > 2:
menu_state.menu_path.pop()
menu_state.start_index.pop()
menu_state.current_menu = data
for path in state.menu_path[2:]:
state.current_menu = state.current_menu[path] if isinstance(state.current_menu, dict) else state.current_menu[int(path.strip("[]"))]
for path in menu_state.menu_path[2:]:
menu_state.current_menu = menu_state.current_menu[path] if isinstance(menu_state.current_menu, dict) else menu_state.current_menu[int(path.strip("[]"))]
else:
# Exit the editor
@@ -351,23 +363,23 @@ def json_editor(stdscr, state):
break
def save_json(file_path, data):
def save_json(file_path: str, data: dict[str, Any]) -> None:
formatted_json = format_json_single_line_arrays(data)
with open(file_path, "w", encoding="utf-8") as f:
f.write(formatted_json)
setup_colors(reinit=True)
def main(stdscr):
def main(stdscr: curses.window) -> None:
from contact.ui.ui_state import MenuState
state = MenuState()
if len(state.menu_path) == 0:
state.menu_path = ["App Settings"] # Initialize if not set
menu_state = MenuState()
if len(menu_state.menu_path) == 0:
menu_state.menu_path = ["App Settings"] # Initialize if not set
curses.curs_set(0)
stdscr.keypad(True)
setup_colors()
json_editor(stdscr, state)
json_editor(stdscr, menu_state)
if __name__ == "__main__":
curses.wrapper(main)

View File

@@ -1,7 +1,7 @@
import argparse
from argparse import ArgumentParser
def setup_parser():
parser = argparse.ArgumentParser(
def setup_parser() -> ArgumentParser:
parser = ArgumentParser(
add_help=True,
epilog="If no connection arguments are specified, we attempt a serial connection and then a TCP connection to localhost.")

View File

@@ -1,9 +1,8 @@
import yaml
import logging
from typing import List
from google.protobuf.json_format import MessageToDict
from meshtastic import BROADCAST_ADDR, mt_config
from meshtastic import mt_config
from meshtastic.util import camel_to_snake, snake_to_camel, fromStr
# defs are from meshtastic/python/main
@@ -20,9 +19,9 @@ def traverseConfig(config_root, config, interface_config) -> bool:
return True
def splitCompoundName(comp_name: str) -> List[str]:
def splitCompoundName(comp_name: str) -> list[str]:
"""Split compound (dot separated) preference name into parts"""
name: List[str] = comp_name.split(".")
name: list[str] = comp_name.split(".")
if len(name) < 2:
name[0] = comp_name
name.append(comp_name)

View File

@@ -1,10 +1,11 @@
import re
def parse_ini_file(ini_file_path: str) -> tuple[dict[str, str], dict[str, str]]:
"""Parses an INI file and returns a mapping of keys to human-readable names and help text."""
def parse_ini_file(ini_file_path):
field_mapping = {}
help_text = {}
current_section = None
field_mapping: dict[str, str] = {}
help_text: dict[str, str] = {}
current_section: str | None = None
with open(ini_file_path, 'r', encoding='utf-8') as f:
for line in f:
@@ -46,14 +47,14 @@ def parse_ini_file(ini_file_path):
return field_mapping, help_text
def transform_menu_path(menu_path):
def transform_menu_path(menu_path: list[str]) -> list[str]:
"""Applies path replacements and normalizes entries in the menu path."""
path_replacements = {
"Radio Settings": "config",
"Module Settings": "module"
}
transformed_path = []
transformed_path: list[str] = []
for part in menu_path[1:]: # Skip 'Main Menu'
# Apply fixed replacements
part = path_replacements.get(part, part)

View File

@@ -7,14 +7,15 @@ from contact.utilities.utils import decimal_to_hex
import contact.ui.default_config as config
import contact.globals as globals
def get_table_name(channel):
def get_table_name(channel: str) -> str:
# Construct the table name
table_name = f"{str(globals.myNodeNum)}_{channel}_messages"
quoted_table_name = f'"{table_name}"' # Quote the table name becuase we begin with numerics and contain spaces
return quoted_table_name
def save_message_to_db(channel, user_id, message_text):
def save_message_to_db(channel: str, user_id: str, message_text: str) -> int | None:
"""Save messages to the database, ensuring the table exists."""
try:
quoted_table_name = get_table_name(channel)
@@ -47,7 +48,7 @@ def save_message_to_db(channel, user_id, message_text):
logging.error(f"Unexpected error in save_message_to_db: {e}")
def update_ack_nak(channel, timestamp, message, ack):
def update_ack_nak(channel: str, timestamp: int, message: str, ack: str) -> None:
try:
with sqlite3.connect(config.db_file_path) as db_connection:
db_cursor = db_connection.cursor()
@@ -69,7 +70,7 @@ def update_ack_nak(channel, timestamp, message, ack):
logging.error(f"Unexpected error in update_ack_nak: {e}")
def load_messages_from_db():
def load_messages_from_db() -> None:
"""Load messages from the database for all channels and update globals.all_messages and globals.channel_list."""
try:
with sqlite3.connect(config.db_file_path) as db_connection:
@@ -142,7 +143,7 @@ def load_messages_from_db():
logging.error(f"SQLite error in load_messages_from_db: {e}")
def init_nodedb():
def init_nodedb() -> None:
"""Initialize the node database and update it with nodes from the interface."""
try:
@@ -172,7 +173,7 @@ def init_nodedb():
logging.error(f"Unexpected error in init_nodedb: {e}")
def maybe_store_nodeinfo_in_db(packet):
def maybe_store_nodeinfo_in_db(packet: dict[str, object]) -> None:
"""Save nodeinfo unless that record is already there, updating if necessary."""
try:
user_id = packet['from']
@@ -190,7 +191,17 @@ def maybe_store_nodeinfo_in_db(packet):
except Exception as e:
logging.error(f"Unexpected error in maybe_store_nodeinfo_in_db: {e}")
def update_node_info_in_db(user_id, long_name=None, short_name=None, hw_model=None, is_licensed=None, role=None, public_key=None, chat_archived=None):
def update_node_info_in_db(
user_id: int | str,
long_name: str | None = None,
short_name: str | None = None,
hw_model: str | None = None,
is_licensed: str | int | None = None,
role: str | None = None,
public_key: str | None = None,
chat_archived: int | None = None
) -> None:
"""Update or insert node information into the database, preserving unchanged fields."""
try:
ensure_node_table_exists() # Ensure the table exists before any operation
@@ -249,7 +260,7 @@ def update_node_info_in_db(user_id, long_name=None, short_name=None, hw_model=No
logging.error(f"Unexpected error in update_node_info_in_db: {e}")
def ensure_node_table_exists():
def ensure_node_table_exists() -> None:
"""Ensure the node database table exists."""
table_name = f'"{globals.myNodeNum}_nodedb"' # Quote for safety
schema = '''
@@ -265,7 +276,7 @@ def ensure_node_table_exists():
ensure_table_exists(table_name, schema)
def ensure_table_exists(table_name, schema):
def ensure_table_exists(table_name: str, schema: str) -> None:
"""Ensure the given table exists in the database."""
try:
with sqlite3.connect(config.db_file_path) as db_connection:
@@ -279,7 +290,7 @@ def ensure_table_exists(table_name, schema):
logging.error(f"Unexpected error in ensure_table_exists({table_name}): {e}")
def get_name_from_database(user_id, type="long"):
def get_name_from_database(user_id: int, type: str = "long") -> str:
"""
Retrieve a user's name (long or short) from the node database.
@@ -313,7 +324,7 @@ def get_name_from_database(user_id, type="long"):
logging.error(f"Unexpected error in get_name_from_database: {e}")
return "Unknown"
def is_chat_archived(user_id):
def is_chat_archived(user_id: int) -> int:
try:
with sqlite3.connect(config.db_file_path) as db_connection:
db_cursor = db_connection.cursor()

View File

@@ -3,9 +3,10 @@ import binascii
import curses
import ipaddress
import re
from typing import Any, Optional
from contact.ui.colors import get_color
def wrap_text(text, wrap_width):
def wrap_text(text: str, wrap_width: int) -> list[str]:
"""Wraps text while preserving spaces and breaking long words."""
words = re.findall(r'\S+|\s+', text) # Capture words and spaces separately
wrapped_lines = []
@@ -40,7 +41,7 @@ def wrap_text(text, wrap_width):
return wrapped_lines
def get_text_input(prompt):
def get_text_input(prompt: str) -> Optional[str]:
"""Handles user input with wrapped text for long prompts."""
height = 8
width = 80
@@ -128,7 +129,7 @@ def get_text_input(prompt):
return user_input
def get_admin_key_input(current_value):
def get_admin_key_input(current_value: list[bytes]) -> Optional[list[str]]:
def to_base64(byte_strings):
"""Convert byte values to Base64-encoded strings."""
return [base64.b64encode(b).decode() for b in byte_strings]
@@ -212,7 +213,7 @@ def get_admin_key_input(current_value):
def get_repeated_input(current_value):
def get_repeated_input(current_value: list[str]) -> Optional[str]:
height = 9
width = 80
start_y = (curses.LINES - height) // 2
@@ -279,7 +280,7 @@ def get_repeated_input(current_value):
pass # Ignore invalid character inputs
def get_fixed32_input(current_value):
def get_fixed32_input(current_value: int) -> int:
cvalue = current_value
current_value = str(ipaddress.IPv4Address(current_value))
height = 10
@@ -336,7 +337,7 @@ def get_fixed32_input(current_value):
pass # Ignore invalid inputs
def get_list_input(prompt, current_option, list_options):
def get_list_input(prompt: str, current_option: Optional[str], list_options: list[str]) -> Optional[str]:
"""
Displays a scrollable list of list_options for the user to choose from.
"""
@@ -399,7 +400,13 @@ def get_list_input(prompt, current_option, list_options):
return current_option
def move_highlight(old_idx, new_idx, options, list_win, list_pad):
def move_highlight(
old_idx: int,
new_idx: int,
options: list[str],
list_win: curses.window,
list_pad: curses.window
) -> int:
global scroll_offset
if 'scroll_offset' not in globals():
@@ -439,7 +446,12 @@ def move_highlight(old_idx, new_idx, options, list_win, list_pad):
return scroll_offset # Return updated scroll_offset to be stored externally
def draw_arrows(win, visible_height, max_index, start_index):
def draw_arrows(
win: curses.window,
visible_height: int,
max_index: int,
start_index: int
) -> None:
if visible_height < max_index:
if start_index > 0:

View File

@@ -1,23 +1,41 @@
import logging
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
import contact.globals as globals
def initialize_interface(args):
try:
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)
try:
if ":" in args.host:
tcp_hostname, tcp_port = args.host.split(':')
else:
tcp_hostname = args.host
tcp_port = meshtastic.tcp_interface.DEFAULT_TCP_PORT
return meshtastic.tcp_interface.TCPInterface(tcp_hostname, portNumber=tcp_port)
except Exception as ex:
logging.error(f"Error connecting to {args.host}. {ex}")
else:
try:
return meshtastic.serial_interface.SerialInterface(args.port)
client = meshtastic.serial_interface.SerialInterface(args.port)
except FileNotFoundError as ex:
logging.error(f"The serial device at '{args.port}' was not found. {ex}")
except PermissionError as ex:
logging.error(f"You probably need to add yourself to the `dialout` group to use a serial connection. {ex}")
except Exception as ex:
logging.error(f"Unexpected error initializing interface: {ex}")
if globals.interface.devPath is None:
return meshtastic.tcp_interface.TCPInterface("localhost")
except OSError as ex:
logging.error(f"The serial device couldn't be opened, it might be in use by another process. {ex}")
if client.devPath is None:
try:
client = meshtastic.tcp_interface.TCPInterface("localhost")
except Exception as ex:
logging.error(f"Error connecting to localhost:{ex}")
return client
except Exception as ex:
logging.critical(f"Fatal error initializing interface: {ex}")

View File

@@ -4,7 +4,7 @@ import logging
import base64
import time
def save_changes(interface, modified_settings, state):
def save_changes(interface, modified_settings, menu_state):
"""
Save changes to the device based on modified settings.
:param interface: Meshtastic interface instance
@@ -52,8 +52,8 @@ def save_changes(interface, modified_settings, state):
if not modified_settings:
return
if state.menu_path[1] == "Radio Settings" or state.menu_path[1] == "Module Settings":
config_category = state.menu_path[2].lower() # for radio and module configs
if menu_state.menu_path[1] == "Radio Settings" or menu_state.menu_path[1] == "Module Settings":
config_category = menu_state.menu_path[2].lower() # for radio and module configs
if {'latitude', 'longitude', 'altitude'} & modified_settings.keys():
lat = float(modified_settings.get('latitude', 0.0))
@@ -64,7 +64,7 @@ def save_changes(interface, modified_settings, state):
logging.info(f"Updated {config_category} with Latitude: {lat} and Longitude {lon} and Altitude {alt}")
return
elif state.menu_path[1] == "User Settings": # for user configs
elif menu_state.menu_path[1] == "User Settings": # for user configs
config_category = "User Settings"
long_name = modified_settings.get("longName")
short_name = modified_settings.get("shortName")
@@ -77,11 +77,11 @@ def save_changes(interface, modified_settings, state):
return
elif state.menu_path[1] == "Channels": # for channel configs
elif menu_state.menu_path[1] == "Channels": # for channel configs
config_category = "Channels"
try:
channel = state.menu_path[-1]
channel = menu_state.menu_path[-1]
channel_num = int(channel.split()[-1]) - 1
except (IndexError, ValueError) as e:
channel_num = None

View File

@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.3.4"
version = "1.3.5"
description = "This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores."
authors = [
{name = "Ben Lipsey",email = "ben@pdxlocations.com"}