Compare commits

...
31 Commits
Author SHA1 Message Date
pdxlocations afbde3c2d6 0 = Default 2026-09-09 22:55:35 -07:00
pdxlocations 51351d316f Merge pull request #310 from pdxlocations:ham
Add ham params
2026-09-09 22:50:22 -07:00
pdxlocations 888108f14e Add ham params 2026-09-09 22:50:02 -07:00
pdxlocations 72f97bfde7 Enhance telemetry beautification by adding local stats parsing and new metrics; update tests for coverage. 2026-09-06 00:02:25 -07:00
pdxlocations 31bff496f1 Bump version to 1.6.12 in pyproject.toml 2026-09-05 13:51:05 -07:00
pdxlocations b7c3ed20f2 retore missing commit 2026-09-05 13:49:56 -07:00
pdxlocations 7f7e0598d2 restore missing commits 2026-09-05 13:48:09 -07:00
pdxlocations 82e7456248 Bump version to 1.6.11 in pyproject.toml 2026-09-05 13:41:36 -07:00
pdxlocations 7b21c9c886 Enhance channel list rendering to display configured names without altering slot keys; add unit test for verification. 2026-09-05 13:41:19 -07:00
pdxlocations 392d236a40 Bump version to 1.6.10 in pyproject.toml 2026-09-05 13:36:37 -07:00
pdxlocations 546fab099c Merge pull request #308 from pdxlocations:flicker-fix
Refactor paint_frame and refresh_node_selection functions to include optional refresh parameter; add stage_node_details_background function for improved window management in F5 key handling tests.
2026-09-05 13:36:06 -07:00
pdxlocations a89d4f8c0d Refactor paint_frame and refresh_node_selection functions to include optional refresh parameter; add stage_node_details_background function for improved window management in F5 key handling tests. 2026-09-05 13:35:36 -07:00
pdxlocations 808fee4fe0 Bump version to 1.6.9 in pyproject.toml 2026-09-05 00:00:11 -07:00
pdxlocations 94d0a9053e Enhance interface initialization and splash screen status display
- Updated `initialize_runtime_interface` to accept `stdscr` for better integration with the UI.
- Modified `initialize_interface` to include a status callback for improved user feedback during connection attempts.
- Refactored `draw_splash` to dynamically display status messages, improving user experience during startup.
2026-09-04 23:59:53 -07:00
pdxlocations f38167db1a Bump version to 1.6.8 in pyproject.toml 2026-09-04 23:51:29 -07:00
pdxlocations 57caf463bc Merge pull request #305 from pdxlocations:edit-message-text
Edit message text with arrows
2026-09-04 23:30:23 -07:00
pdxlocations 36408a1f37 Edit message text with arrows 2026-09-04 23:30:08 -07:00
pdxlocations 07b9ed22d8 Bump version to 1.6.7 in pyproject.toml 2026-07-30 22:26:46 -07:00
pdxlocations bdba69677b Refactor error handling in start function to use print_exception for better traceback output and enhance tests for channel name resolution with new modem presets 2026-07-30 22:18:25 -07:00
pdxlocations 1f88a37ac1 Refactor save_changes to streamline admin key updates and enhance test coverage for key replacement 2026-07-26 20:34:55 -07:00
pdxlocations 8f33cbc22d Refactor handle_backtick to improve local settings selection 2026-07-26 20:32:21 -07:00
pdxlocations 40665ff076 Bump version to 1.6.6 in pyproject.toml 2026-07-26 14:26:01 -07:00
pdxlocations d1b6da6f26 Refactor channel settings to rename 'is_client_muted' to 'is_muted' and update related save functionality and tests 2026-07-26 14:23:11 -07:00
pdxlocations ac90b4af3e Enhance remote admin verification to establish authorized sessions and update related tests 2026-07-26 14:07:09 -07:00
pdxlocations 7af693ddea Enhance bot response to include reply context in message sending 2026-07-26 13:47:51 -07:00
pdxlocations dae6a43652 Bump version to 1.6.5 in pyproject.toml 2026-07-26 11:09:26 -07:00
pdxlocations 22e5e89828 Bump version to 1.6.4 in pyproject.toml 2026-07-26 11:08:57 -07:00
pdxlocations c2db76d5ec Merge pull request #298 from pdxlocations:log-viewer
Add log viewer functionality and enhance reconnect handling
2026-07-26 11:08:27 -07:00
pdxlocations d8dc4876b6 Add log viewer functionality and enhance reconnect handling
- Introduced a log viewer in the UI to display live logs.
- Added functionality to load the last few lines of the log file.
- Enhanced reconnect logic to handle UI state during connection attempts.
- Updated UI state management to track log viewer status.
- Added tests for log loading functionality.
2026-07-26 11:07:33 -07:00
pdxlocations 0ae644e3fb Keep Meshtastic logs out of the curses terminal 2026-07-26 10:44:02 -07:00
pdxlocations bfd20bfa49 Verify remote admin access before opening settings 2026-07-26 10:31:00 -07:00
30 changed files with 1325 additions and 195 deletions

No files matched your search

+7 -6
View File
@@ -38,6 +38,7 @@ from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
from contact.ui.dialog import dialog
from contact.utilities.interfaces import initialize_interface, reconnect_interface
from contact.utilities.logging_utils import configure_logging
from contact.utilities.utils import get_channels, get_nodeNum, get_node_list
from contact.utilities.singleton import ui_state, interface_state, app_state
@@ -51,9 +52,7 @@ os.environ.setdefault("TERM", "xterm-256color")
if os.environ.get("COLORTERM") == "gnome-terminal":
os.environ["TERM"] = "xterm-256color"
logging.basicConfig(
filename=config.log_file_path, level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
configure_logging(logging.INFO)
app_state.lock = threading.Lock()
@@ -115,7 +114,7 @@ def interface_is_ready(interface: object) -> bool:
def initialize_runtime_interface_with_retry(stdscr: curses.window, args: object):
while True:
interface = initialize_runtime_interface(args)
interface = initialize_runtime_interface(args, stdscr)
if getattr(args, "demo_screenshot", False) or interface_is_ready(interface):
return interface
@@ -156,10 +155,12 @@ def initialize_globals(seed_demo: bool = False) -> None:
load_messages_from_db()
def initialize_runtime_interface(args: object):
def initialize_runtime_interface(args: object, stdscr=None):
if getattr(args, "demo_screenshot", False):
configure_demo_database()
return build_demo_interface()
if stdscr is not None:
return initialize_interface(args, status_callback=lambda status: draw_splash(stdscr, status))
return initialize_interface(args)
@@ -252,7 +253,7 @@ def start() -> None:
if fatal_error is not None:
print("Fatal error:", fatal_error)
traceback.print_exc()
traceback.print_exception(type(fatal_error), fatal_error, fatal_error.__traceback__)
sys.exit(1)
if interrupted:
+9 -2
View File
@@ -98,7 +98,12 @@ dialog.resize_body, "Please resize the terminal to at least {rows} rows.", ""
user, "User"
longName, "Node long name", "If you are a licensed HAM operator and have enabled HAM mode, this must be set to your HAM operator call sign."
shortName, "Node short name", "Must be up to 4 bytes. Usually this is 4 characters, if using latin characters and no emojis."
isLicensed, "Enable licensed amateur (HAM) mode", "IMPORTANT: Read Meshtastic help documentation before enabling."
isLicensed, "Enable licensed amateur (HAM) mode", "Shows HAM parameters for editing. Saving applies licensed operation, including radio and channel changes. Disabling does not restore previous settings."
call_sign, "Callsign", "Licensed operator callsign."
tx_power, "TX power in dBm", "Transmit power at the LoRa transceiver, before amplification. Use a value supported by your radio."
frequency, "Frequency in MHz", "Operating frequency override. Zero uses automatic frequency selection."
short_name, "HAM short name", "Optional short display name."
long_name, "Name appended to callsign", "Optional name appended to the callsign. Requires supporting firmware."
isUnmessagable, "Disable incoming messages", "When enabled, this node advertises that it cannot receive messages."
[app_settings]
@@ -114,6 +119,7 @@ message_prefix, "Message prefix", ""
sent_message_prefix, "Sent message prefix", ""
notification_symbol, "Notification symbol", ""
notification_sound, "Notification sound", "Select a sound file from Contact's sounds folder, or None to disable notification audio."
view_log, "View Log", "Open a scrollable live view of Contact's log file."
ack_implicit_str, "ACK (implicit)", ""
ack_str, "ACK", ""
nak_str, "NAK", ""
@@ -159,6 +165,7 @@ node_ignored, "Node ignored", ""
[Channels.channel]
title, "Channels"
enabled, "Enabled", "Enable or disable this channel without clearing its name or encryption key."
channel_num, "Channel number", "The index number of this channel."
psk, "PSK", "The channel's encryption key."
name, "Name", "The channel's name."
@@ -167,7 +174,7 @@ uplink_enabled, "Uplink enabled", "Let this channel's data be sent to the MQTT s
downlink_enabled, "Downlink enabled", "Let data from the MQTT server configured on this node be sent to this channel."
module_settings, "Module settings", "Position precision and Client Mute."
module_settings.position_precision, "Position precision", "The precision level of location data sent on this channel."
module_settings.is_client_muted, "Is Client Muted", "Controls whether or not the phone / clients should mute the current channel. Useful for noisy public channels you don't necessarily want to disable."
module_settings.is_muted, "Is muted", "Controls whether or not the client or device should mute this channel. Useful for noisy public channels you do not necessarily want to disable."
[config.device]
title, "Device"
+8 -1
View File
@@ -6,6 +6,7 @@ from typing import Any, Dict
import contact.ui.default_config as config
from contact.utilities.singleton import app_state, interface_state
from contact.utilities.utils import get_reply_context
from contact.message_handlers.tx_handler import send_message
BOT_RESPONSE_DELAY_SECONDS = 2.3
@@ -81,7 +82,13 @@ def bot_respond(packet: Dict[str, Any], message: str, send_channel: int) -> bool
# Use the triggering packet's ID so Meshtastic clients render this
# automatic response as a native reply when that ID is available.
send_message(response_data_string, channel=send_channel, reply_id=packet_id)
reply_context = get_reply_context(packet_id) if packet_id is not None else ""
send_message(
response_data_string,
channel=send_channel,
reply_id=packet_id,
reply_context=reply_context,
)
# Import locally to avoid circular import at module import time.
from contact.ui.contact_ui import request_ui_redraw
+2 -2
View File
@@ -1,6 +1,6 @@
import time
from typing import Any, Dict
from typing import Any, Dict, Optional
import google.protobuf.json_format
from meshtastic import BROADCAST_NUM
@@ -168,7 +168,7 @@ def send_message(
message: str,
destination: int = BROADCAST_NUM,
channel: int = 0,
reply_id: int | None = None,
reply_id: Optional[int] = None,
reply_context: str = "",
) -> None:
"""
+3 -6
View File
@@ -14,6 +14,7 @@ from contact.utilities.arg_parser import setup_parser
from contact.utilities.i18n import t
from contact.utilities.input_handlers import get_list_input
from contact.utilities.interfaces import initialize_interface, reconnect_interface
from contact.utilities.logging_utils import configure_logging
def close_interface(interface: object) -> None:
@@ -36,7 +37,7 @@ def main(stdscr: curses.window) -> None:
parser = setup_parser()
args = parser.parse_args()
interface = initialize_interface(args)
interface = initialize_interface(args, status_callback=lambda status: draw_splash(stdscr, status))
if interface.localNode.localConfig.lora.region == 0:
confirmation = get_list_input(
@@ -81,11 +82,7 @@ def ensure_min_rows(stdscr: curses.window, min_rows: int = 11) -> None:
stdscr.refresh()
logging.basicConfig( # Run `tail -f client.log` in another terminal to view live
filename=config.log_file_path,
level=logging.WARNING, # DEBUG, INFO, WARNING, ERROR, CRITICAL)
format="%(asctime)s - %(levelname)s - %(message)s",
)
configure_logging(logging.WARNING)
if __name__ == "__main__":
log_file = config.log_file_path
+361 -98
View File
@@ -1,5 +1,6 @@
import curses
import logging
import os
from typing import Optional
import time
import traceback
@@ -15,7 +16,8 @@ from contact.utilities.utils import (
build_reply_prefix,
)
from contact.settings import settings_menu
from contact.ui.control_ui import RemoteAdminCancelled
from contact.ui.control_ui import RemoteAdminCancelled, verify_remote_admin
from contact.ui.user_config import load_log_tail
from contact.message_handlers.tx_handler import send_message, send_traceroute
from contact.utilities.utils import parse_protobuf
from contact.ui.colors import get_color
@@ -33,6 +35,8 @@ from contact.ui.nav_utils import (
wrap_text,
truncate_with_ellipsis,
pad_to_width,
text_width,
slice_to_width,
)
from contact.utilities.singleton import ui_state, interface_state, menu_state, app_state
@@ -41,6 +45,117 @@ MIN_COL = 1 # "effectively zero" without breaking curses
RESIZE_DEBOUNCE_MS = 250
root_win = None
nodes_pad = None
entry_win = None
input_text = ""
def refresh_log_viewer() -> None:
"""Refresh the in-app log overlay only when the log file changes."""
try:
stat = os.stat(config.log_file_path)
signature = (stat.st_mtime_ns, stat.st_size)
except OSError:
signature = None
if not ui_state.log_viewer_loaded or signature != ui_state.log_viewer_signature:
ui_state.log_viewer_lines = load_log_tail(config.log_file_path)
ui_state.log_viewer_signature = signature
ui_state.log_viewer_loaded = True
def flush_log_handlers() -> None:
"""Ensure a just-written status event is visible to the log tailer."""
for handler in logging.getLogger().handlers:
try:
handler.flush()
except Exception:
pass
def draw_log_viewer(stdscr: curses.window) -> int:
"""Draw the log overlay and return the number of visible log rows."""
refresh_log_viewer()
height, width = stdscr.getmaxyx()
content_height = max(1, height - 4)
max_start = max(0, len(ui_state.log_viewer_lines) - content_height)
if ui_state.log_viewer_follow:
ui_state.log_viewer_start_line = max_start
else:
ui_state.log_viewer_start_line = min(ui_state.log_viewer_start_line, max_start)
try:
stdscr.erase()
stdscr.bkgd(get_color("background"))
stdscr.attrset(get_color("window_frame"))
stdscr.border()
state = "Following" if ui_state.log_viewer_follow else "Paused"
title = " View Log — {} ".format(state)
stdscr.addstr(0, 2, title[: max(0, width - 4)], get_color("settings_breadcrumbs", bold=True))
for row, line in enumerate(
ui_state.log_viewer_lines[ui_state.log_viewer_start_line : ui_state.log_viewer_start_line + content_height],
start=1,
):
stdscr.addstr(row, 1, line[: max(0, width - 2)], get_color("settings_default"))
hint = " Esc Close Up/Down Scroll PgUp/PgDn Page F Follow "
stdscr.addstr(height - 2, 1, hint[: max(0, width - 2)], get_color("commands"))
stdscr.refresh()
except curses.error:
pass
return content_height
def handle_log_viewer_key(key: int, content_height: int) -> bool:
"""Handle a log-overlay key. Returns True when the overlay closes."""
if ui_state.reconnect_prompt_open:
if key in (ord("r"), ord("R")):
ui_state.reconnect_prompt_open = False
ui_state.reconnect_attempted = False
elif key in (ord("c"), ord("C"), 27):
ui_state.reconnect_prompt_open = False
return False
if key == 27:
ui_state.log_viewer_open = False
return True
if key in (ord("f"), ord("F")):
ui_state.log_viewer_follow = not ui_state.log_viewer_follow
return False
max_start = max(0, len(ui_state.log_viewer_lines) - content_height)
if key == curses.KEY_UP:
ui_state.log_viewer_follow = False
ui_state.log_viewer_start_line = max(0, ui_state.log_viewer_start_line - 1)
elif key == curses.KEY_DOWN:
ui_state.log_viewer_start_line = min(max_start, ui_state.log_viewer_start_line + 1)
ui_state.log_viewer_follow = ui_state.log_viewer_start_line >= max_start
elif key == curses.KEY_PPAGE:
ui_state.log_viewer_follow = False
ui_state.log_viewer_start_line = max(0, ui_state.log_viewer_start_line - content_height)
elif key == curses.KEY_NPAGE:
ui_state.log_viewer_start_line = min(max_start, ui_state.log_viewer_start_line + content_height)
ui_state.log_viewer_follow = ui_state.log_viewer_start_line >= max_start
return False
def draw_log_reconnect_prompt(stdscr: curses.window) -> None:
"""Draw a non-blocking reconnect choice over the live log overlay."""
if not ui_state.reconnect_prompt_open:
return
try:
height, width = stdscr.getmaxyx()
message = "Could not reconnect after 20 seconds."
hint = "R Retry C Cancel"
box_width = min(width - 4, max(len(message) + 4, len(hint) + 4, 42))
win = curses.newwin(6, box_width, max(0, (height - 6) // 2), max(0, (width - box_width) // 2))
win.bkgd(get_color("background"))
win.attrset(get_color("window_frame"))
win.border()
win.addstr(0, 2, " Disconnected ", get_color("settings_default"))
win.addstr(2, 2, message[: box_width - 4], get_color("settings_default"))
win.addstr(4, 2, hint[: box_width - 4], get_color("commands"))
win.refresh()
except curses.error:
pass
def request_ui_redraw(
@@ -131,11 +246,14 @@ def compute_widths(total_w: int, focus: int):
return MIN_COL, MIN_COL, total_w - 2 * MIN_COL
def paint_frame(win, selected: bool) -> None:
def paint_frame(win, selected: bool, refresh: bool = True) -> None:
if win is not entry_win and ui_state.input_focused:
selected = False
win.attrset(get_color("window_frame_selected") if selected else get_color("window_frame"))
win.box()
win.attrset(get_color("window_frame"))
win.refresh()
if refresh:
win.refresh()
def get_channel_row_color(index: int) -> int:
@@ -158,7 +276,7 @@ def get_node_row_color(index: int, highlight: bool = False) -> int:
return get_color(color, reverse=reverse)
def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None:
def refresh_node_selection(old_index: int = -1, highlight: bool = False, refresh: bool = True) -> None:
if nodes_pad is None or not ui_state.node_list:
return
@@ -177,8 +295,9 @@ def refresh_node_selection(old_index: int = -1, highlight: bool = False) -> None
pass
ui_state.start_index[2] = max(0, ui_state.selected_node - (nodes_win.getmaxyx()[0] - 3))
refresh_pad(2)
draw_window_arrows(2)
if refresh:
refresh_pad(2)
draw_window_arrows(2)
def refresh_main_window(window_id: int, selected: bool) -> None:
@@ -189,7 +308,7 @@ def refresh_main_window(window_id: int, selected: bool) -> None:
channel_pad.chgat(ui_state.selected_channel, 1, width, get_channel_row_color(ui_state.selected_channel))
refresh_pad(0)
elif window_id == 1:
paint_frame(messages_win, selected=selected)
paint_frame(messages_win, selected=selected, refresh=False)
refresh_pad(1)
elif window_id == 2:
paint_frame(nodes_win, selected=selected)
@@ -380,6 +499,8 @@ def main_ui(stdscr: curses.window) -> None:
root_win = stdscr
input_text = ""
ui_state.input_cursor = 0
ui_state.input_focused = False
queued_char = None
stdscr.keypad(True)
get_channels()
@@ -389,7 +510,15 @@ def main_ui(stdscr: curses.window) -> None:
interface = interface_state.interface
if not interface_is_connected(interface) and not ui_state.reconnect_attempted:
ui_state.reconnect_attempted = True
status_win = show_connection_status(stdscr, "Disconnected", "Trying to reconnect…")
logging.warning("Meshtastic connection disconnected; attempting automatic reconnect")
flush_log_handlers()
if ui_state.log_viewer_open:
# Reconnect is synchronous. Repaint once before entering it so
# the disconnect event remains visible during the attempt.
draw_log_viewer(stdscr)
status_win = None
if not ui_state.log_viewer_open:
status_win = show_connection_status(stdscr, "Disconnected", "Trying to reconnect…")
try:
from contact.ui.control_ui import reconnect_interface_with_splash
@@ -398,13 +527,18 @@ def main_ui(stdscr: curses.window) -> None:
status_win.erase()
status_win.refresh()
ui_state.reconnect_attempted = False
handle_resize(stdscr, False)
if not ui_state.log_viewer_open:
handle_resize(stdscr, False)
except Exception:
if status_win is not None:
status_win.erase()
status_win.refresh()
handle_resize(stdscr, False)
if not ui_state.log_viewer_open:
handle_resize(stdscr, False)
logging.exception("Automatic reconnect after transport disconnect failed")
if ui_state.log_viewer_open:
ui_state.reconnect_prompt_open = True
continue
retry = get_list_input(
"Contact could not reconnect after 20 seconds.",
"Retry",
@@ -417,10 +551,21 @@ def main_ui(stdscr: curses.window) -> None:
continue
handle_resize(stdscr, False)
if ui_state.log_viewer_open:
content_height = draw_log_viewer(stdscr)
draw_log_reconnect_prompt(stdscr)
stdscr.timeout(200)
try:
key = stdscr.getch()
except curses.error:
continue
if key != -1 and handle_log_viewer_key(key, content_height):
handle_resize(stdscr, False)
continue
with app_state.lock:
process_pending_ui_updates(stdscr)
entry_display = f"{ui_state.reply_context}{input_text or ''}"
draw_text_field(entry_win, f"Message: {entry_display[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
draw_message_input(input_text)
# Get user input from entry window
try:
@@ -460,6 +605,7 @@ def main_ui(stdscr: curses.window) -> None:
elif char in (chr(curses.KEY_ENTER), chr(10), chr(13)):
input_text = handle_enter(input_text)
ui_state.input_cursor = min(ui_state.input_cursor, len(input_text))
elif char in (curses.KEY_F4, chr(20)): # Ctrl + t and F4 for Traceroute
handle_ctrl_t(stdscr)
@@ -467,9 +613,14 @@ def main_ui(stdscr: curses.window) -> None:
elif char == curses.KEY_F5:
handle_f5_key(stdscr)
elif char in (curses.KEY_BACKSPACE, chr(127)):
elif char in (curses.KEY_BACKSPACE, chr(127), chr(8)):
input_text = handle_backspace(entry_win, input_text)
elif char == curses.KEY_DC:
focus_message_input()
position = ui_state.input_cursor
input_text = input_text[:position] + input_text[position + 1:]
elif char in (curses.KEY_F12, "`"): # ` Launch the settings interface
handle_backtick(stdscr)
@@ -483,7 +634,6 @@ def main_ui(stdscr: curses.window) -> None:
entry_win.erase()
elif char == curses.KEY_RESIZE:
input_text = ""
queued_char = drain_resize_events(entry_win)
handle_resize(stdscr, False)
continue
@@ -509,15 +659,66 @@ def main_ui(stdscr: curses.window) -> None:
break
else:
# Append typed character to input text
if isinstance(char, str):
input_text += char
else:
input_text += chr(char)
if isinstance(char, str) and char.isprintable():
input_text = insert_input_text(input_text, char)
def focus_message_input() -> None:
if ui_state.input_focused:
return
# Keep the conversation visible in single-pane mode while composing.
handle_function_keys(curses.KEY_F2)
ui_state.input_focused = True
refresh_message_highlight()
refresh_main_window(1, selected=False)
def leave_message_input() -> None:
ui_state.input_focused = False
ui_state.current_window = 1
refresh_message_highlight()
refresh_main_window(1, selected=True)
def insert_input_text(draft: str, text: str) -> str:
focus_message_input()
position = min(ui_state.input_cursor, len(draft))
ui_state.input_cursor = position + len(text)
return draft[:position] + text + draft[position:]
def draw_message_input(draft: str) -> None:
"""Redraw the whole input row and scroll it to keep the draft cursor visible."""
width = entry_win.getmaxyx()[1]
prompt = slice_to_width("Message: ", max(0, width - 3))
available = max(1, width - 2 - text_width(prompt))
ui_state.input_cursor = max(0, min(ui_state.input_cursor, len(draft)))
content = ui_state.reply_context + draft
cursor = len(ui_state.reply_context) + ui_state.input_cursor
start = cursor
used = 0
while start > 0 and used + text_width(content[start - 1]) < available:
start -= 1
used += text_width(content[start])
visible = slice_to_width(content[start:], available)
draw_text_field(entry_win, prompt + visible, get_color("input"))
paint_frame(entry_win, selected=ui_state.input_focused, refresh=False)
try:
hint = " Ctrl+K Help "
if width >= len(hint) + 4:
entry_win.addstr(0, width - len(hint) - 2, hint, get_color("commands"))
curses.curs_set(1 if ui_state.input_focused else 0)
entry_win.move(1, min(width - 2, 1 + text_width(prompt) + used))
except curses.error:
pass
entry_win.refresh()
def handle_up() -> None:
"""Handle key up events to scroll the current window."""
if ui_state.input_focused:
leave_message_input()
return
if ui_state.current_window == 0:
scroll_channels(-1)
elif ui_state.current_window == 1:
@@ -528,6 +729,9 @@ def handle_up() -> None:
def handle_down() -> None:
"""Handle key down events to scroll the current window."""
if ui_state.input_focused:
leave_message_input()
return
if ui_state.current_window == 0:
scroll_channels(1)
elif ui_state.current_window == 1:
@@ -538,6 +742,9 @@ def handle_down() -> None:
def handle_home() -> None:
"""Handle home key events to select the first item in the current window."""
if ui_state.input_focused:
ui_state.input_cursor = 0
return
if ui_state.current_window == 0:
select_channel(0)
elif ui_state.current_window == 1:
@@ -552,6 +759,9 @@ def handle_home() -> None:
def handle_end() -> None:
"""Handle end key events to select the last item in the current window."""
if ui_state.input_focused:
ui_state.input_cursor = len(input_text)
return
if ui_state.current_window == 0:
select_channel(len(ui_state.channel_list) - 1)
elif ui_state.current_window == 1:
@@ -592,6 +802,9 @@ def handle_pagedown() -> None:
def handle_leftright(char: int) -> None:
"""Handle left/right key events to switch between windows."""
delta = -1 if char == curses.KEY_LEFT else 1
if ui_state.input_focused:
ui_state.input_cursor = max(0, min(len(input_text), ui_state.input_cursor + delta))
return
old_window = ui_state.current_window
ui_state.current_window = (ui_state.current_window + delta) % 3
if ui_state.single_pane_mode:
@@ -612,6 +825,8 @@ def handle_leftright(char: int) -> None:
def handle_function_keys(char: int) -> None:
"""Switch windows using F1/F2/F3."""
if ui_state.input_focused:
leave_message_input()
if char == curses.KEY_F1:
target = 0
elif char == curses.KEY_F2:
@@ -722,6 +937,28 @@ def handle_enter(input_text: str) -> str:
return input_text
def stage_node_details_background(stdscr, previous_window):
"""Restore the panes behind a moved dialog without displaying an intermediate frame."""
current_window = ui_state.current_window
try:
ui_state.current_window = previous_window
stdscr.touchwin()
stdscr.noutrefresh()
windows = (channel_win, messages_win, nodes_win)
for index, win in enumerate(windows):
if ui_state.single_pane_mode and index != previous_window:
continue
win.touchwin()
refresh_pad(index, commit=False)
if ui_state.display_log and (not ui_state.single_pane_mode or previous_window == 1):
packetlog_win.touchwin()
packetlog_win.noutrefresh()
entry_win.touchwin()
entry_win.noutrefresh()
finally:
ui_state.current_window = current_window
def handle_f5_key(stdscr: curses.window) -> None:
if not ui_state.node_list:
return
@@ -734,7 +971,10 @@ def handle_f5_key(stdscr: curses.window) -> None:
message_parts.append(f"• Device: {node.get('user', {}).get('longName', 'Unknown')}")
message_parts.append(f"• Short name: {node.get('user', {}).get('shortName', 'Unknown')}")
message_parts.append(f"• Hardware: {node.get('user', {}).get('hwModel', 'Unknown')}")
message_parts.append(f"• Role: {node.get('user', {}).get('role', 'Unknown')}")
user = node.get('user', {})
# Protobuf dictionaries omit CLIENT because it is the default enum value.
role = user.get('role', 'CLIENT') if user else 'Unknown'
message_parts.append(f"• Role: {role}")
message_parts.append(f"Public key: {node.get('user', {}).get('publicKey')}")
message_parts.append(f"• Node ID: {node.get('num', 'Unknown')}")
@@ -819,6 +1059,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
ui_state.current_window = 4
scroll_offset = 0
dialog_win = None
previous_layout = None
curses.curs_set(0)
refresh_node_selection(highlight=True)
@@ -838,11 +1079,15 @@ def handle_f5_key(stdscr: curses.window) -> None:
max_scroll = max(0, len(message_lines) - viewport_h)
scroll_offset = max(0, min(scroll_offset, max_scroll))
layout = (dialog_height, dialog_width, y, x, ui_state.selected_node, height, width)
if previous_layout is not None and layout != previous_layout:
stage_node_details_background(stdscr, previous_window)
previous_layout = layout
if dialog_win is None:
dialog_win = curses.newwin(dialog_height, dialog_width, y, x)
else:
dialog_win.erase()
dialog_win.refresh()
dialog_win.resize(dialog_height, dialog_width)
dialog_win.mvwin(y, x)
@@ -887,7 +1132,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
except curses.error:
pass
dialog_win.refresh()
dialog_win.noutrefresh()
msg_win.noutrefresh()
curses.doupdate()
@@ -907,12 +1152,12 @@ def handle_f5_key(stdscr: curses.window) -> None:
old_selected_node = ui_state.selected_node
ui_state.selected_node = (ui_state.selected_node - 1) % len(ui_state.node_list)
scroll_offset = 0
refresh_node_selection(old_selected_node, highlight=True)
refresh_node_selection(old_selected_node, highlight=True, refresh=False)
elif char == curses.KEY_DOWN:
old_selected_node = ui_state.selected_node
ui_state.selected_node = (ui_state.selected_node + 1) % len(ui_state.node_list)
scroll_offset = 0
refresh_node_selection(old_selected_node, highlight=True)
refresh_node_selection(old_selected_node, highlight=True, refresh=False)
elif char == curses.KEY_PPAGE:
scroll_offset = max(0, scroll_offset - viewport_h)
elif char == curses.KEY_NPAGE:
@@ -971,14 +1216,12 @@ def handle_ctrl_t(stdscr: curses.window) -> None:
def handle_backspace(entry_win: curses.window, input_text: str) -> str:
"""Handle backspace key events to remove the last character from input text."""
if input_text:
input_text = input_text[:-1]
y, x = entry_win.getyx()
entry_win.move(y, x - 1)
entry_win.addch(" ") #
entry_win.move(y, x - 1)
entry_win.refresh()
"""Delete before the draft cursor; the next frame redraws the entire line."""
focus_message_input()
position = ui_state.input_cursor
if position > 0:
input_text = input_text[:position - 1] + input_text[position:]
ui_state.input_cursor -= 1
return input_text
@@ -998,67 +1241,83 @@ def handle_backtick(stdscr: curses.window) -> None:
handle_resize(stdscr, False)
return
options = ["Local settings"]
remote_node_num = None
selected_node_num = None
if ui_state.node_list:
remote_node_num = ui_state.node_list[ui_state.selected_node]
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
choice = get_list_input("Open settings for", None, options)
if choice == "Local settings":
settings_menu(stdscr, interface)
elif remote_node_num is not None:
wait_win = None
selected_node_num = ui_state.node_list[ui_state.selected_node]
def remote_status(message: Optional[str]) -> None:
nonlocal wait_win
if message is None:
# The local node is already the target for normal settings; do not offer a
# misleading remote-admin choice when it is selected in the node pane.
if selected_node_num == interface_state.myNodeNum:
settings_menu(stdscr, interface)
else:
options = ["Local settings"]
remote_node_num = None
if selected_node_num is not None:
remote_node_num = selected_node_num
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
choice = get_list_input("Open settings for", None, options)
if choice == "Local settings":
settings_menu(stdscr, interface)
elif remote_node_num is not None:
wait_win = None
def remote_status(message: Optional[str]) -> None:
nonlocal wait_win
if message is None:
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
wait_win = None
return
if wait_win is None:
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
update_remote_admin_wait(wait_win, message)
def remote_cancel_requested() -> bool:
if wait_win is None:
return False
try:
wait_win.timeout(0)
return wait_win.getch() == 27
finally:
wait_win.timeout(-1)
try:
remote_node = interface.getNode(remote_node_num, False)
verify_remote_admin(
remote_node,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
remote_status(None)
settings_menu(
stdscr,
interface,
# Settings requests configs explicitly; avoid getNode's
# implicit channel request racing that setup.
node=remote_node,
remote=True,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
except RemoteAdminCancelled:
logging.info("Remote admin request cancelled for %s", remote_node_num)
remote_status(None)
except (SystemExit, PermissionError) as exc:
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
wait_win = None
return
if wait_win is None:
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
update_remote_admin_wait(wait_win, message)
def remote_cancel_requested() -> bool:
if wait_win is None:
return False
try:
wait_win.timeout(0)
return wait_win.getch() == 27
finally:
wait_win.timeout(-1)
try:
settings_menu(
stdscr,
interface,
# Settings requests configs explicitly; avoid getNode's
# implicit channel request racing that setup.
node=interface.getNode(remote_node_num, False),
remote=True,
status_callback=remote_status,
cancel_callback=remote_cancel_requested,
)
except RemoteAdminCancelled:
logging.info("Remote admin request cancelled for %s", remote_node_num)
remote_status(None)
except SystemExit as exc:
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
contact.ui.dialog.dialog(
"Remote admin rejected",
"The selected node did not authorize this admin request.",
)
except Exception as exc:
logging.exception("Remote admin failed for %s", remote_node_num)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
contact.ui.dialog.dialog("Remote admin failed", str(exc))
contact.ui.dialog.dialog(
"Remote admin rejected",
"The selected node did not authorize this admin request.",
)
except Exception as exc:
logging.exception("Remote admin failed for %s", remote_node_num)
if wait_win is not None:
wait_win.erase()
wait_win.refresh()
contact.ui.dialog.dialog("Remote admin failed", str(exc))
ui_state.current_window = previous_window
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
curses.curs_set(1)
@@ -1067,7 +1326,7 @@ def handle_backtick(stdscr: curses.window) -> None:
handle_resize(stdscr, False)
def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> curses.window | None:
def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> Optional[curses.window]:
"""Show progress while Meshtastic retrieves remote-admin settings."""
message = f"Attempting remote admin of {node_name}, waiting for response..."
try:
@@ -1086,7 +1345,7 @@ def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> curses.wind
return None
def show_connection_status(stdscr: curses.window, title: str, message: str) -> curses.window | None:
def show_connection_status(stdscr: curses.window, title: str, message: str) -> Optional[curses.window]:
"""Display a non-blocking connection-status dialog over the current UI."""
try:
height, width = stdscr.getmaxyx()
@@ -1103,7 +1362,7 @@ def show_connection_status(stdscr: curses.window, title: str, message: str) -> c
return None
def update_remote_admin_wait(wait_win: curses.window | None, message: str) -> None:
def update_remote_admin_wait(wait_win: Optional[curses.window], message: str) -> None:
"""Append the current remote-admin request to the waiting overlay."""
if wait_win is None:
return
@@ -1415,7 +1674,7 @@ def draw_messages_window(scroll_to_bottom: bool = False, preserve_selection: boo
messages_pad.addstr(row, 1, line, color)
ui_state.message_line_ranges[channel] = message_ranges
paint_frame(messages_win, selected=(ui_state.current_window == 1))
paint_frame(messages_win, selected=(ui_state.current_window == 1), refresh=False)
visible_lines = get_msg_window_lines(messages_win, packetlog_win)
@@ -1430,7 +1689,6 @@ def draw_messages_window(scroll_to_bottom: bool = False, preserve_selection: boo
refresh_message_highlight()
messages_win.refresh()
refresh_pad(1)
draw_packetlog_win()
draw_window_arrows(1)
@@ -1444,7 +1702,7 @@ def refresh_message_highlight() -> None:
previous_range = ui_state.highlighted_message_range
width = max(0, messages_win.getmaxyx()[1] - 2)
if not ui_state.channel_list or ui_state.current_window != 1:
if not ui_state.channel_list or ui_state.current_window != 1 or ui_state.input_focused:
if previous_range:
start, end, color = previous_range
for row in range(start, end):
@@ -1760,7 +2018,7 @@ def search(win: int) -> None:
entry_win.erase()
def refresh_pad(window: int) -> None:
def refresh_pad(window: int, commit: bool = True) -> None:
# If in single-pane mode and this isn't the focused window, skip refreshing its (collapsed) pad
if ui_state.single_pane_mode and window != ui_state.current_window:
@@ -1777,7 +2035,7 @@ def refresh_pad(window: int) -> None:
if ui_state.display_log:
packetlog_win.box()
packetlog_win.refresh()
packetlog_win.noutrefresh()
elif window == 2:
pad = nodes_pad
@@ -1820,9 +2078,12 @@ def refresh_pad(window: int) -> None:
return
draw_frame_title(box, get_window_title(window))
box.refresh()
box.noutrefresh()
pad.refresh(
# The frame overlaps the pad's viewport. Recompose even unchanged pad
# rows before committing, so its text cannot flash blank between draws.
pad.touchwin()
pad.noutrefresh(
start_index,
0,
top,
@@ -1830,6 +2091,8 @@ def refresh_pad(window: int) -> None:
bottom,
right,
)
if commit:
curses.doupdate()
def add_notification(channel_number: int) -> None:
+61 -7
View File
@@ -25,7 +25,7 @@ from contact.utilities.input_handlers import (
)
from contact.ui.colors import get_color
from contact.ui.dialog import dialog
from contact.ui.menus import generate_menu_from_protobuf
from contact.ui.menus import generate_menu_from_protobuf, update_ham_fields
from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window
from contact.ui.user_config import json_editor
from contact.utilities.arg_parser import setup_parser
@@ -141,6 +141,12 @@ def display_menu() -> tuple[object, object]:
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)
if menu_state.menu_path == ["Main Menu", "Channels"] and isinstance(field_info, dict):
# Keep slot keys for navigation/saving, but label rows with the
# configured name. Empty names retain their slot label.
channel_name = field_info.get("name", (None, ""))[1]
if isinstance(channel_name, str) and channel_name.strip():
display_name = channel_name
if full_key.startswith("config.network.ipv4_config.") and option in {"ip", "gateway", "subnet", "dns"}:
if isinstance(current_value, int):
@@ -279,7 +285,8 @@ def redraw_main_ui_after_reconnect(stdscr: object) -> None:
get_channels()
refresh_node_list()
contact_ui.handle_resize(stdscr, False)
if not contact_ui.ui_state.log_viewer_open:
contact_ui.handle_resize(stdscr, False)
except Exception:
logging.debug("Skipping main UI redraw after reconnect", exc_info=True)
@@ -319,6 +326,25 @@ def _request_remote_channels_with_timeout(node: object, cancel_callback=None) ->
time.sleep(0.1)
def verify_remote_admin(node: object, status_callback=None, cancel_callback=None) -> None:
"""Establish and verify an authorized remote-admin session.
Config reads alone are not a sufficient authorization check. The
Meshtastic admin-session exchange only provides a session passkey when
the remote node accepts this client's administrative credentials.
"""
if status_callback:
status_callback("Checking remote-admin permission…")
_request_remote_with_timeout(node.ensureSessionKey, cancel_callback=cancel_callback)
try:
remote_session = node.iface._getOrCreateByNum(node.nodeNum).get("adminSessionPassKey")
except (AttributeError, TypeError):
remote_session = None
if not remote_session:
raise PermissionError("The selected node did not authorize this admin request.")
def _request_remote_section(
node: object, menu_path: List[str], selected_option: str, status_callback=None, cancel_callback=None
) -> bool:
@@ -334,6 +360,14 @@ def _request_remote_section(
_request_remote_channels_with_timeout(node, cancel_callback=cancel_callback)
return True
if menu_path == ["Main Menu"] and selected_option == "User Settings":
# HAM inputs derive their initial power/frequency from this node's LoRa config.
field = node.localConfig.DESCRIPTOR.fields_by_name["lora"]
if status_callback:
status_callback("Requesting lora config…")
_request_remote_with_timeout(node.requestConfig, field, cancel_callback=cancel_callback)
return True
if len(menu_path) != 2:
return False
@@ -482,6 +516,7 @@ def settings_menu(
break
if reconnect_required:
interface = reconnect_interface_with_splash(stdscr, interface)
node = interface.localNode
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=not remote)
if len(menu_state.menu_path) > 1:
@@ -680,12 +715,14 @@ def settings_menu(
menu_win.refresh()
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
open_log_viewer = json_editor(stdscr, menu_state) # Open the App Settings menu
reload_translations()
menu_state.current_menu = menu["Main Menu"]
menu_state.menu_path = ["Main Menu"]
menu_state.start_index.pop()
menu_state.selected_index = 4
if open_log_viewer:
break
continue
field_info = menu_state.current_menu.get(selected_option)
@@ -734,9 +771,6 @@ def settings_menu(
new_value = new_value == "True"
menu_state.current_menu[selected_option] = (field, new_value)
for option, (field, value) in menu_state.current_menu.items():
modified_settings[option] = value
menu_state.start_index.pop()
elif selected_option in ["latitude", "longitude", "altitude"]:
@@ -757,6 +791,11 @@ def settings_menu(
new_value = current_value if new_values is None else [base64.b64decode(key) for key in new_values]
menu_state.start_index.pop()
elif field is None and selected_option == "enabled" and menu_state.menu_path[1] == "Channels":
new_value = get_list_input(human_readable_name, str(current_value), ["True", "False"])
new_value = new_value == "True" or new_value is True
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"])
if new_value == "Not Set":
@@ -798,8 +837,17 @@ def settings_menu(
else: # Handle other field types
input_type = get_input_type_for_field(field)
allow_empty = (
menu_state.menu_path == ["Main Menu", "User Settings"]
and selected_option in ("short_name", "long_name")
) or (
selected_option == "name"
and len(menu_state.menu_path) == 3
and menu_state.menu_path[1] == "Channels"
)
new_value = get_text_input(
f"{human_readable_name} is currently: {current_value}", selected_option, input_type
f"{human_readable_name} is currently: {current_value}", selected_option, input_type,
allow_empty=allow_empty,
)
new_value = current_value if new_value is None else new_value
menu_state.start_index.pop()
@@ -824,6 +872,12 @@ def settings_menu(
new_value = enum_value_descriptor.name if enum_value_descriptor else new_value
menu_state.current_menu[selected_option] = (field, new_value)
if menu_state.menu_path == ["Main Menu", "User Settings"]:
update_ham_fields(menu_state.current_menu, node)
# Save the full HAM command, including fields that were not edited.
modified_settings.clear()
modified_settings.update({name: setting[1] for name, setting in menu_state.current_menu.items()})
menu_state.selected_index = list(menu_state.current_menu).index(selected_option)
else:
requested_remote_section = False
if remote:
+38 -1
View File
@@ -5,7 +5,7 @@ from collections import OrderedDict
from typing import Any, Union, Dict
from google.protobuf.message import Message
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
from meshtastic.protobuf import admin_pb2, channel_pb2, config_pb2, module_config_pb2
def encode_if_bytes(value: Any) -> str:
"""Encode byte values to base64 string."""
@@ -57,6 +57,38 @@ def extract_fields(
return menu
def update_ham_fields(user_menu, node):
"""Reflect the selected license flag using the installed HAM descriptors."""
params = admin_pb2.HamParameters()
fields = params.DESCRIPTOR.fields_by_name
licensed = user_menu.get("isLicensed", (None, False))[1] in (True, "True")
if not licensed:
call_sign = user_menu.get("call_sign", (None, ""))[1]
suffix = user_menu.get("long_name", (None, ""))[1]
user_menu.setdefault("longName", (None, " ".join(filter(None, (call_sign, suffix)))))
user_menu.setdefault("shortName", (None, user_menu.get("short_name", (None, ""))[1]))
for name in fields:
user_menu.pop(name, None)
return
owner_name = user_menu.get("longName", (None, ""))[1]
call_sign, _, suffix = owner_name.partition(" ")
defaults = {
"call_sign": call_sign,
"short_name": user_menu.get("shortName", (None, ""))[1],
"long_name": suffix,
"tx_power": node.localConfig.lora.tx_power,
"frequency": node.localConfig.lora.override_frequency,
}
for field in params.DESCRIPTOR.fields:
if field.name in defaults:
setattr(params, field.name, defaults[field.name])
for name, setting in extract_fields(params, params).items():
user_menu.setdefault(name, setting)
user_menu.pop("longName", None)
user_menu.pop("shortName", None)
def generate_menu_from_protobuf(interface: object, node: Any = None, include_app_settings: bool = True) -> Dict[str, Any]:
"""
Builds the full settings menu structure from the protobuf definitions.
@@ -78,6 +110,7 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
"isLicensed": (None, current_user_config.get("isLicensed", "False")),
"isUnmessagable": (None, current_user_config.get("isUnmessagable", "False")),
}
update_ham_fields(menu_structure["Main Menu"]["User Settings"], node)
else:
logging.info("User settings not found in Node Info")
menu_structure["Main Menu"]["User Settings"] = "No user settings available"
@@ -93,6 +126,10 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
current_channel = node.getChannelByChannelIndex(i)
if current_channel:
channel_config = extract_fields(channel, current_channel.settings)
channel_config = {
"enabled": (None, current_channel.role != channel_pb2.Channel.Role.DISABLED),
**channel_config,
}
menu_structure["Main Menu"]["Channels"][f"Channel {i + 1}"] = channel_config
# Add Radio Settings
+10 -5
View File
@@ -1,8 +1,9 @@
import curses
import textwrap
from contact.ui.colors import get_color
def draw_splash(stdscr: object) -> None:
def draw_splash(stdscr: object, status: str = "connecting...") -> None:
"""Draw the splash screen with a logo and connecting message."""
curses.curs_set(0)
@@ -13,16 +14,20 @@ def draw_splash(stdscr: object) -> None:
message_1 = "/ Λ"
message_2 = "/ / \\"
message_3 = "P W R D"
message_4 = "connecting..."
start_x = width // 2 - len(message_1) // 2
start_x2 = width // 2 - len(message_4) // 2
start_y = height // 2 - 1
stdscr.addstr(start_y, start_x, message_1, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 1, start_x - 1, message_2, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 2, start_x - 2, message_3, get_color("splash_logo", bold=True))
stdscr.addstr(start_y + 4, start_x2, message_4, get_color("splash_text"))
stdscr.move(start_y + 5, start_x2)
status_lines = []
for line in status.splitlines():
status_lines.extend(textwrap.wrap(line, max(1, width - 4)) or [""])
for offset, line in enumerate(status_lines):
row = start_y + 4 + offset
if row >= height - 1:
break
stdscr.addstr(row, max(1, (width - len(line)) // 2), line, get_color("splash_text"))
stdscr.attrset(get_color("window_frame"))
stdscr.box()
+9
View File
@@ -25,6 +25,8 @@ class ChatUIState:
selected_message: int = 0
selected_node: int = 0
current_window: int = 0
input_focused: bool = False
input_cursor: int = 0
last_sent_time: float = 0.0
last_traceroute_time: float = 0.0
selected_index: int = 0
@@ -40,6 +42,13 @@ class ChatUIState:
scroll_messages_to_bottom: bool = False
preserve_message_selection: bool = False
reconnect_attempted: bool = False
reconnect_prompt_open: bool = False
log_viewer_open: bool = False
log_viewer_follow: bool = True
log_viewer_start_line: int = 0
log_viewer_lines: List[str] = field(default_factory=list)
log_viewer_signature: Any = None
log_viewer_loaded: bool = False
oldest_message_rowid: Dict[Union[str, int], int] = field(default_factory=dict)
has_older_messages: Dict[Union[str, int], bool] = field(default_factory=dict)
message_line_ranges: Dict[Union[str, int], List[tuple]] = field(default_factory=dict)
+48 -5
View File
@@ -9,12 +9,14 @@ from contact.ui.nav_utils import move_highlight, draw_arrows, update_help_window
from contact.utilities.ini_utils import parse_ini_file
from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
from contact.utilities.singleton import menu_state
from contact.utilities.singleton import menu_state, ui_state
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
max_help_lines = 6
save_option = "Save Changes"
VIEW_LOG_OPTION = "__view_log__"
LOG_VIEWER_LINE_LIMIT = 500
translation_file = config.get_localisation_file(config.language)
field_mapping, help_text = parse_ini_file(translation_file)
translation_language = config.language
@@ -86,6 +88,32 @@ def get_effective_width() -> int:
return max(20, min(MAX_MENU_WIDTH, curses.COLS - 2))
def load_log_tail(path: str, max_lines: int = LOG_VIEWER_LINE_LIMIT) -> List[str]:
"""Read the final log lines without loading an arbitrarily large log file."""
if max_lines <= 0:
return []
try:
with open(path, "rb") as log_file:
log_file.seek(0, os.SEEK_END)
position = log_file.tell()
chunks = []
newline_count = 0
while position > 0 and newline_count <= max_lines:
read_size = min(8192, position)
position -= read_size
log_file.seek(position)
chunk = log_file.read(read_size)
chunks.append(chunk)
newline_count += chunk.count(b"\n")
data = b"".join(reversed(chunks)).decode("utf-8", errors="replace")
lines = data.splitlines()
if position > 0 and lines:
lines = lines[1:]
return lines[-max_lines:]
except OSError as exc:
return [f"Unable to read log: {exc}"]
def edit_color_pair(key: str, display_label: str, current_value: List[str]) -> List[str]:
"""
Allows the user to select a foreground and background color for a key.
@@ -273,6 +301,8 @@ def display_menu() -> tuple[Any, Any, List[str]]:
# Determine menu items based on the type of current_menu
if isinstance(menu_state.current_menu, dict):
options = list(menu_state.current_menu.keys())
if len(menu_state.menu_path) <= 2:
options.insert(0, VIEW_LOG_OPTION)
elif isinstance(menu_state.current_menu, list):
options = [f"[{i}]" for i in range(len(menu_state.current_menu))]
else:
@@ -311,10 +341,13 @@ def display_menu() -> tuple[Any, Any, List[str]]:
menu_state.current_menu[key]
if isinstance(menu_state.current_menu, dict)
else menu_state.current_menu[int(key.strip("[]"))]
)
) if key != VIEW_LOG_OPTION else ""
if isinstance(menu_state.current_menu, dict):
full_key = get_app_settings_key(menu_state.menu_path, key)
display_key = lookup_app_settings_label(full_key, key)
if key == VIEW_LOG_OPTION:
display_key = field_mapping.get("app_settings.view_log", "View Log")
else:
full_key = get_app_settings_key(menu_state.menu_path, key)
display_key = lookup_app_settings_label(full_key, key)
else:
display_key = key
display_key = f"{display_key}"[: w // 2 - 2]
@@ -391,7 +424,7 @@ def update_app_settings_help(menu_win: curses.window, options: List[str]) -> Non
)
def json_editor(stdscr: curses.window, menu_state: Any) -> None:
def json_editor(stdscr: curses.window, menu_state: Any) -> bool:
menu_state.selected_index = 0 # Track the selected option
made_changes = False # Track if any changes were made
@@ -470,6 +503,14 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
if menu_state.selected_index < len(options): # Handle selection of a menu item
selected_key = options[menu_state.selected_index]
if selected_key == VIEW_LOG_OPTION:
ui_state.log_viewer_open = True
ui_state.log_viewer_follow = True
ui_state.log_viewer_start_line = 0
ui_state.log_viewer_lines = []
ui_state.log_viewer_signature = None
ui_state.log_viewer_loaded = False
return True
menu_state.menu_path.append(str(selected_key))
menu_state.start_index.append(0)
menu_state.menu_index.append(menu_state.selected_index)
@@ -573,6 +614,8 @@ def json_editor(stdscr: curses.window, menu_state: Any) -> None:
break
return False
def save_json(file_path: str, data: Dict[str, Any]) -> None:
formatted_json = config.format_json_single_line_arrays(data)
+4 -1
View File
@@ -38,7 +38,7 @@ def invalid_input(window: curses.window, message: str, redraw_func: Optional[cal
curses.curs_set(1)
def get_text_input(prompt: str, selected_config: str, input_type: str) -> Optional[str]:
def get_text_input(prompt: str, selected_config: str, input_type: str, *, allow_empty: bool = False) -> Optional[str]:
"""Handles user input with wrapped text for long prompts."""
def redraw_input_win():
@@ -127,6 +127,9 @@ def get_text_input(prompt: str, selected_config: str, input_type: str) -> Option
menu_state.need_redraw = True
if not user_input.strip():
if allow_empty:
user_input = ""
break
invalid_input(
input_win,
t("ui.error.value_empty", default="Value cannot be empty."),
+18 -2
View File
@@ -1,4 +1,6 @@
import logging
import contextlib
import io
import time
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
@@ -25,7 +27,7 @@ def interface_is_connected(interface) -> bool:
return getattr(interface, "localNode", None) is not None
def initialize_interface(args):
def initialize_interface(args, status_callback=None):
try:
if args.ble:
@@ -43,7 +45,10 @@ def initialize_interface(args):
logging.error(f"Error connecting to {args.host}. {ex}")
else:
try:
client = meshtastic.serial_interface.SerialInterface(args.port)
# The library prints its fallback notice directly to stdout,
# which would overwrite the curses splash screen.
with contextlib.redirect_stdout(io.StringIO()):
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:
@@ -56,6 +61,8 @@ def initialize_interface(args):
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:
if status_callback:
status_callback("No serial device found.\nConnecting via TCP to localhost…")
client = meshtastic.tcp_interface.TCPInterface("localhost")
except Exception as ex:
logging.error(f"Error connecting to localhost:{ex}")
@@ -68,13 +75,19 @@ def initialize_interface(args):
def reconnect_interface(args, attempts: int = 20, delay_seconds: float = 1.0):
last_error = None
target = getattr(args, "host", None) or getattr(args, "port", None) or getattr(args, "ble", None) or "auto"
for attempt in range(attempts):
attempt_number = attempt + 1
logging.info("Reconnect attempt %d/%d to %s", attempt_number, attempts, target)
try:
interface = initialize_interface(args)
if interface_is_connected(interface) and getattr(interface, "localNode", None) is not None and getattr(
interface.localNode, "localConfig", None
) is not None:
node_num = getattr(interface.localNode, "nodeNum", None)
node_label = f" !{node_num:08x}" if isinstance(node_num, int) else ""
logging.info("Reconnected to Meshtastic node%s on attempt %d/%d", node_label, attempt_number, attempts)
return interface
last_error = RuntimeError("interface did not complete connection setup")
try:
@@ -84,7 +97,10 @@ def reconnect_interface(args, attempts: int = 20, delay_seconds: float = 1.0):
except Exception as ex:
last_error = ex
logging.warning("Reconnect attempt %d/%d failed: %s", attempt_number, attempts, last_error)
if attempt < attempts - 1:
time.sleep(delay_seconds)
logging.error("Unable to reconnect to Meshtastic node after %d attempts", attempts)
raise RuntimeError("Failed to reconnect to the Meshtastic node") from last_error
+26
View File
@@ -0,0 +1,26 @@
import logging
import os
import sys
import contact.ui.default_config as config
def configure_logging(level: int) -> None:
"""Set up file logging or exit with instructions for fixing its path."""
try:
logging.basicConfig(
filename=config.log_file_path,
level=level,
format="%(asctime)s - %(levelname)s - %(message)s",
force=True,
)
except OSError as error:
print(
f"Unable to open log file: {os.path.abspath(config.log_file_path)}\n"
f"{error}\n"
f"Please edit config.json at: {os.path.abspath(config.json_file_path)}\n"
'Set "log_file_path" to a file in an existing, writable directory, '
"then restart Contact.",
file=sys.stderr,
)
raise SystemExit(1) from None
+39 -27
View File
@@ -1,8 +1,7 @@
from meshtastic.protobuf import channel_pb2
from meshtastic.protobuf import admin_pb2, channel_pb2
from google.protobuf.message import Message
import logging
import base64
import time
DEVICE_REBOOT_KEYS = {"button_gpio", "buzzer_gpio", "role", "rebroadcast_mode"}
POWER_REBOOT_KEYS = {
@@ -100,22 +99,14 @@ def save_changes(interface, modified_settings, menu_state, node=None):
# Filter out empty keys
valid_keys = [key for key in admin_keys if key and key.strip() and key != b""]
if not valid_keys:
logging.warning("No valid admin keys provided. Skipping admin key update.")
else:
# Clear existing keys if needed
if security_config.admin_key:
logging.info("Clearing existing admin keys...")
del security_config.admin_key[:]
node.writeConfig("security")
time.sleep(2) # Give time for device to process
# Append new keys
for key in valid_keys:
logging.info(f"Adding admin key: {key}")
security_config.admin_key.append(key)
node.writeConfig("security")
logging.info("Admin keys updated successfully!")
# Update the repeated field in memory, then send one complete
# security config. Sending an empty list first can revoke the key
# authorizing this remote-admin session before a second write can
# restore the remaining keys.
del security_config.admin_key[:]
security_config.admin_key.extend(valid_keys)
node.writeConfig("security")
logging.info("Admin keys updated successfully (%d key(s)).", len(valid_keys))
# Backup 'admin_key' before removing it
admin_key_backup = modified_settings.get("admin_key", None)
@@ -148,6 +139,21 @@ def save_changes(interface, modified_settings, menu_state, node=None):
if is_unmessagable is not None:
is_unmessagable = is_unmessagable == "True" or is_unmessagable is True
if is_licensed and "call_sign" in modified_settings:
message = admin_pb2.AdminMessage()
for field in message.set_ham_mode.DESCRIPTOR.fields:
if field.name in modified_settings:
setattr(message.set_ham_mode, field.name, modified_settings[field.name])
node.ensureSessionKey()
# Keep the separate user flag and HAM command in one settings transaction.
node.beginSettingsTransaction()
if is_unmessagable is not None:
node.setOwner(None, None, True, is_unmessagable)
callback = None if node is interface.localNode else node.onAckNak
node._sendAdmin(message, onResponse=callback)
node.commitSettingsTransaction()
return True
node.setOwner(long_name, short_name, is_licensed, is_unmessagable)
logging.info(
@@ -161,25 +167,31 @@ def save_changes(interface, modified_settings, menu_state, node=None):
config_category = "Channels"
try:
channel = menu_state.menu_path[-1]
channels_index = menu_state.menu_path.index("Channels")
channel = menu_state.menu_path[channels_index + 1]
channel_num = int(channel.split()[-1]) - 1
except (IndexError, ValueError) as e:
channel_num = None
channel = node.channels[channel_num]
for key, value in modified_settings.items():
if key == "psk": # Special case: decode Base64 for psk
if key == "enabled":
if not value:
channel.role = channel_pb2.Channel.Role.DISABLED
elif channel.role == channel_pb2.Channel.Role.DISABLED:
channel.role = (
channel_pb2.Channel.Role.PRIMARY if channel_num == 0
else channel_pb2.Channel.Role.SECONDARY
)
elif key == "psk": # Special case: decode Base64 for psk
channel.settings.psk = base64.b64decode(value)
elif key == "position_precision": # Special case: module_settings
channel.settings.module_settings.position_precision = value
elif key in {"position_precision", "is_muted"}:
# These fields belong to ChannelSettings.module_settings,
# not ChannelSettings itself.
setattr(channel.settings.module_settings, key, value)
else:
setattr(channel.settings, key, value) # Use setattr for other fields
if channel_num == 0:
channel.role = channel_pb2.Channel.Role.PRIMARY
else:
channel.role = channel_pb2.Channel.Role.SECONDARY
node.writeChannel(channel_num)
logging.info(f"Updated Channel {channel_num} in {config_category}")
+34 -8
View File
@@ -13,9 +13,29 @@ sensors = {
'channel_utilization': {'icon':'ChUtil:', 'unit':'%'},
'air_util_tx': {'icon':'AirUtil:', 'unit':'%'},
'uptime_seconds': {'icon':'🆙 ', 'unit':'h'},
'num_packets_tx': {'icon':'📤 Tx:', 'unit':''},
'num_packets_rx': {'icon':'📥 Rx:', 'unit':''},
'num_packets_rx_bad': {'icon':'⚠️ RxBad:', 'unit':''},
'num_online_nodes': {'icon':'📡 Online:', 'unit':''},
'num_total_nodes': {'icon':'📡 Total:', 'unit':''},
'num_rx_dupe': {'icon':'📥 Dupe:', 'unit':''},
'num_tx_relay': {'icon':'📤 Relay:', 'unit':''},
'num_tx_relay_canceled': {'icon':'📤 Canceled:', 'unit':''},
'num_tx_dropped': {'icon':'⚠️ Dropped:', 'unit':''},
'heap_total_bytes': {'icon':'💾 Total:', 'unit':'B'},
'heap_free_bytes': {'icon':'💾 Free:', 'unit':'B'},
'noise_floor': {'icon':'📶 Noise:', 'unit':'dBm'},
'latitude_i': {'icon':'🌍 ', 'unit':''},
'longitude_i': {'icon':'', 'unit':''},
'altitude': {'icon':'⬆️ ', 'unit':'m'},
'location_source': {'icon':'📍 ', 'unit':''},
'altitude_source': {'icon':'⬆️ Source:', 'unit':''},
'ground_speed': {'icon':'💨 ', 'unit':'m/s'},
'ground_track': {'icon':'🧭 ', 'unit':'°'},
'PDOP': {'icon':'🎯 PDOP:', 'unit':''},
'HDOP': {'icon':'🎯 HDOP:', 'unit':''},
'VDOP': {'icon':'🎯 VDOP:', 'unit':''},
'sats_in_view': {'icon':'🛰️ ', 'unit':''},
'time': {'icon':'🕔 ', 'unit':''}
}
@@ -55,18 +75,20 @@ def get_chunks(data):
parsed=""
for item in reading:
key, value = item.split(":")
key, value = item.split(":", 1)
key = key.strip()
value = value.strip()
# If value is float, round it to the 1 digit after point
# else make it int
if "." in value:
value = round(float(value.strip()),1)
else:
try:
value = int(value)
except ValueError:
try:
value = int(value.strip())
except Exception:
# Leave it string as last resort
value = value
value = round(float(value), 1)
except ValueError:
# Leave nonnumeric values intact, including strings with periods.
pass
# Python 3.9-compatible alternative to match/case.
if key == "uptime_seconds":
@@ -79,6 +101,10 @@ def get_chunks(data):
elif key == "wind_direction":
# Convert wind direction from degrees to abbreviation
value = humanize_wind_direction(value)
elif key in ("PDOP", "HDOP", "VDOP"):
value = round(value / 100, 2)
elif key == "ground_track":
value = round(value / 100, 2)
elif key == "time":
value = datetime.datetime.fromtimestamp(int(value)).strftime("%d.%m.%Y %H:%m")
+35 -3
View File
@@ -1,6 +1,8 @@
import datetime
import logging
import re
import time
from functools import lru_cache
from typing import Optional, Union
from google.protobuf.message import DecodeError
@@ -10,14 +12,41 @@ import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state
import contact.utilities.telemetry_beautifier as tb
NEW_MODEM_PRESET_NAMES = {
14: "TINY_FAST",
15: "TINY_SLOW",
16: "MEDIUM_TURBO",
}
@lru_cache(maxsize=None)
def _get_modem_preset_name(modem_preset_enum: int) -> str:
try:
return config_pb2.Config.LoRaConfig.ModemPreset.Name(modem_preset_enum)
except ValueError:
compatibility_name = NEW_MODEM_PRESET_NAMES.get(modem_preset_enum)
if compatibility_name is not None:
logging.warning(
"Modem preset %s is newer than the installed Meshtastic protobuf schema; "
"using compatibility name %s",
modem_preset_enum,
compatibility_name,
)
return compatibility_name
logging.warning(
"Modem preset %s is not recognized by the installed Meshtastic protobuf schema",
modem_preset_enum,
)
return f"UNKNOWN_PRESET_{modem_preset_enum}"
def _get_channel_name(device_channel, node):
if device_channel.settings.name:
return device_channel.settings.name
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
modem_preset_enum = node.localConfig.lora.modem_preset
modem_preset_string = _get_modem_preset_name(modem_preset_enum)
return convert_to_camel_case(modem_preset_string)
@@ -267,6 +296,9 @@ def parse_protobuf(packet: dict) -> Union[str, dict]:
if hasattr(pb, "environment_metrics") and pb.HasField("environment_metrics"):
return tb.get_chunks(str(pb.environment_metrics))
if hasattr(pb, "local_stats") and pb.HasField("local_stats"):
return tb.get_chunks(str(pb.local_stats))
# For other data, without implemented beautification, fallback to just printing the object
return str(pb).replace("\n", " ").replace("\r", " ").strip()
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.6.3"
version = "1.6.12"
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"}
+11 -10
View File
@@ -40,18 +40,19 @@ class BotHandlerTests(unittest.TestCase):
packet = {"id": 900, "from": 222, "decoded": {}}
with mock.patch.object(config, "ping_bot_enabled", "True"):
with mock.patch.object(self.bot_handler.threading, "Thread") as thread:
self.assertTrue(self.bot_handler.bot_respond(packet, "ping", 0))
send_response = thread.call_args.kwargs["target"]
with mock.patch.object(self.bot_handler.time, "sleep"):
with mock.patch.dict(
sys.modules,
{"contact.ui.contact_ui": types.SimpleNamespace(request_ui_redraw=mock.Mock())},
):
send_response()
with mock.patch.object(self.bot_handler, "get_reply_context", return_value="<Re: NODE: ping> "):
with mock.patch.object(self.bot_handler.threading, "Thread") as thread:
self.assertTrue(self.bot_handler.bot_respond(packet, "ping", 0))
send_response = thread.call_args.kwargs["target"]
with mock.patch.object(self.bot_handler.time, "sleep"):
with mock.patch.dict(
sys.modules,
{"contact.ui.contact_ui": types.SimpleNamespace(request_ui_redraw=mock.Mock())},
):
send_response()
self.bot_handler.send_message.assert_called_once_with(
"Pong!", channel=0, reply_id=900
"Pong!", channel=0, reply_id=900, reply_context="<Re: NODE: ping> "
)
def test_bot_response_requires_enabled_app_setting(self) -> None:
+84 -4
View File
@@ -1,4 +1,5 @@
import unittest
from contextlib import ExitStack
from types import SimpleNamespace
from unittest import mock
@@ -11,6 +12,52 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
class ContactUiTests(unittest.TestCase):
def test_f5_restores_background_when_switching_between_different_size_nodes(self):
ui_state.node_list = [101, 102]
ui_state.current_window = 2
interface = SimpleNamespace(nodesByNum={
101: {"user": {"shortName": "BIG", "longName": "A" * 60}, "position": {"altitude": 100}},
102: {"user": {"shortName": "SMALL", "longName": "B"}},
})
win = mock.Mock()
win.getch.side_effect = [contact_ui.curses.KEY_DOWN, contact_ui.curses.KEY_UP, 27]
events = mock.Mock()
events.attach_mock(win.noutrefresh, "dialog")
stdscr = mock.Mock()
with ExitStack() as stack:
stack.enter_context(mock.patch.object(contact_ui.interface_state, "interface", interface))
for name, value in (("LINES", 40), ("COLS", 100)):
stack.enter_context(mock.patch.object(contact_ui.curses, name, value, create=True))
for name in ("curs_set", "update_lines_cols", "doupdate"):
stack.enter_context(mock.patch.object(contact_ui.curses, name))
stack.enter_context(mock.patch.object(contact_ui.curses, "newwin", return_value=win))
for name in ("get_color", "refresh_node_selection", "handle_resize"):
stack.enter_context(mock.patch.object(contact_ui, name, return_value=0))
restore = stack.enter_context(mock.patch.object(contact_ui, "stage_node_details_background"))
events.attach_mock(restore, "restore")
contact_ui.handle_f5_key(stdscr)
self.assertEqual(events.mock_calls, [
mock.call.dialog(), mock.call.restore(stdscr, 2), mock.call.dialog(),
mock.call.restore(stdscr, 2), mock.call.dialog(),
])
self.assertNotEqual(win.resize.call_args_list[0], win.resize.call_args_list[1])
self.assertEqual(win.refresh.call_count, 1) # Only on close.
def test_background_restore_stages_panes_without_flushing(self):
ui_state.current_window = 4
ui_state.single_pane_mode = True
stdscr = mock.Mock()
with ExitStack() as stack:
for name in ("channel_win", "messages_win", "nodes_win", "entry_win"):
stack.enter_context(mock.patch.object(contact_ui, name, mock.Mock(), create=True))
refresh = stack.enter_context(mock.patch.object(contact_ui, "refresh_pad"))
update = stack.enter_context(mock.patch.object(contact_ui.curses, "doupdate"))
contact_ui.stage_node_details_background(stdscr, 2)
refresh.assert_called_once_with(2, commit=False)
update.assert_not_called()
stdscr.noutrefresh.assert_called_once()
self.assertEqual(ui_state.current_window, 4)
def setUp(self) -> None:
reset_singletons()
self.saved_config = snapshot_config("single_pane_mode")
@@ -53,14 +100,35 @@ class ContactUiTests(unittest.TestCase):
with mock.patch.object(contact_ui, "handle_resize"):
with mock.patch.object(contact_ui, "get_list_input", return_value="Remote admin: node"):
with mock.patch.object(contact_ui, "get_name_from_database", return_value="node"):
with mock.patch.object(contact_ui, "settings_menu", side_effect=SystemExit(1)):
with mock.patch("contact.ui.dialog.dialog") as dialog:
contact_ui.handle_backtick(stdscr)
with mock.patch.object(contact_ui, "verify_remote_admin", side_effect=SystemExit(1)):
with mock.patch.object(contact_ui, "settings_menu") as settings_menu:
with mock.patch("contact.ui.dialog.dialog") as dialog:
contact_ui.handle_backtick(stdscr)
dialog.assert_called_once_with(
"Remote admin rejected",
"The selected node did not authorize this admin request.",
)
settings_menu.assert_not_called()
def test_handle_backtick_opens_local_settings_for_selected_local_node(self) -> None:
stdscr = mock.Mock()
ui_state.current_window = 2
ui_state.node_list = [123]
ui_state.selected_node = 0
contact_ui.interface_state.myNodeNum = 123
contact_ui.interface_state.interface = mock.Mock()
with mock.patch.object(contact_ui.curses, "curs_set"):
with mock.patch.object(contact_ui, "get_list_input") as chooser:
with mock.patch.object(contact_ui, "settings_menu") as settings_menu:
with mock.patch.object(contact_ui, "get_channels"):
with mock.patch.object(contact_ui, "refresh_node_list"):
with mock.patch.object(contact_ui, "handle_resize"):
contact_ui.handle_backtick(stdscr)
chooser.assert_not_called()
settings_menu.assert_called_once_with(stdscr, contact_ui.interface_state.interface)
def test_show_remote_admin_wait_draws_status(self) -> None:
stdscr = mock.Mock()
@@ -416,6 +484,8 @@ class ContactUiTests(unittest.TestCase):
self.assertEqual(contact_ui.get_window_title(1), "Primary")
def test_refresh_pad_draws_selected_channel_title_on_message_frame(self) -> None:
self.addCleanup(mock.patch.stopall)
mock.patch.object(contact_ui, "packetlog_win", mock.Mock(), create=True).start()
ui_state.single_pane_mode = True
ui_state.current_window = 1
ui_state.channel_list = ["Primary"]
@@ -432,9 +502,16 @@ class ContactUiTests(unittest.TestCase):
contact_ui.messages_win.getmaxyx.return_value = (10, 20)
with mock.patch.object(contact_ui, "get_msg_window_lines", return_value=4):
contact_ui.refresh_pad(1)
with mock.patch.object(contact_ui.curses, "doupdate") as update:
contact_ui.refresh_pad(1)
contact_ui.messages_win.addstr.assert_called_once_with(0, 2, " Primary ", contact_ui.curses.A_BOLD)
contact_ui.messages_win.refresh.assert_not_called()
contact_ui.messages_pad.refresh.assert_not_called()
contact_ui.messages_win.noutrefresh.assert_called_once()
contact_ui.messages_pad.touchwin.assert_called_once()
contact_ui.messages_pad.noutrefresh.assert_called_once()
update.assert_called_once()
def test_search_ignores_no_input_from_curses(self) -> None:
ui_state.node_list = [101]
@@ -487,6 +564,9 @@ class ContactUiTests(unittest.TestCase):
contact_ui.handle_f5_key(stdscr)
self.assertEqual(dialog_win.getch.call_count, 2)
# Idle polling must never present an erased dialog before its text.
self.assertEqual(dialog_win.noutrefresh.call_count, 2)
self.assertEqual(dialog_win.refresh.call_count, 1) # Closing the dialog only.
handle_resize.assert_called_once_with(stdscr, False)
def test_f5_node_details_tolerates_none_metrics(self) -> None:
+57
View File
@@ -1,4 +1,5 @@
from argparse import Namespace
from contextlib import ExitStack
from types import SimpleNamespace
import unittest
from unittest import mock
@@ -10,6 +11,36 @@ from tests.test_support import reset_singletons
class ControlUiTests(unittest.TestCase):
def test_channel_list_renders_names_without_changing_slot_keys(self):
from meshtastic.protobuf import channel_pb2, config_pb2, module_config_pb2
from contact.ui.menus import generate_menu_from_protobuf
from contact.utilities.singleton import menu_state
channels = [channel_pb2.Channel(settings=channel_pb2.ChannelSettings(name=name))
for name in ("Local", "Local", "", " ")]
node = SimpleNamespace(
localConfig=config_pb2.Config(), moduleConfig=module_config_pb2.ModuleConfig(),
getChannelByChannelIndex=lambda i: channels[i] if i < len(channels) else None,
)
menu_state.menu_path = ["Main Menu", "Channels"]
menu_state.current_menu = generate_menu_from_protobuf(None, node=node)["Main Menu"]["Channels"]
win, pad = mock.Mock(), mock.Mock()
win.getbegyx.return_value = (2, 2)
win.getmaxyx.return_value = (10, 80)
with ExitStack() as stack:
for name, value in (("LINES", 24), ("COLS", 100)):
stack.enter_context(mock.patch.object(control_ui.curses, name, value, create=True))
stack.enter_context(mock.patch.object(control_ui.curses, "newwin", return_value=win))
stack.enter_context(mock.patch.object(control_ui.curses, "newpad", return_value=pad))
stack.enter_context(mock.patch.object(control_ui.curses, "curs_set"))
stack.enter_context(mock.patch.object(control_ui, "get_menu_width", return_value=80))
for name in ("get_color", "draw_help_window", "draw_arrows"):
stack.enter_context(mock.patch.object(control_ui, name, return_value=0))
control_ui.display_menu()
labels = [call.args[2].strip() for call in pad.addstr.call_args_list]
self.assertEqual(labels, ["Local", "Local", "Channel 3", "Channel 4"])
self.assertEqual(list(menu_state.current_menu), [f"Channel {i}" for i in range(1, 5)])
def setUp(self) -> None:
reset_singletons()
@@ -53,6 +84,32 @@ class ControlUiTests(unittest.TestCase):
reconnect.assert_called_once_with(stdscr, interface)
self.assertIs(result, new_interface)
def test_verify_remote_admin_establishes_an_authorized_session(self) -> None:
node = SimpleNamespace(
nodeNum=123,
ensureSessionKey=mock.Mock(),
iface=SimpleNamespace(_getOrCreateByNum=mock.Mock(return_value={"adminSessionPassKey": b"key"})),
)
status_callback = mock.Mock()
with mock.patch.object(control_ui, "_request_remote_with_timeout") as request:
control_ui.verify_remote_admin(node, status_callback=status_callback)
status_callback.assert_called_once_with("Checking remote-admin permission…")
request.assert_called_once_with(node.ensureSessionKey, cancel_callback=None)
node.iface._getOrCreateByNum.assert_called_once_with(123)
def test_verify_remote_admin_rejects_missing_session_key(self) -> None:
node = SimpleNamespace(
nodeNum=123,
ensureSessionKey=mock.Mock(),
iface=SimpleNamespace(_getOrCreateByNum=mock.Mock(return_value={})),
)
with mock.patch.object(control_ui, "_request_remote_with_timeout"):
with self.assertRaises(PermissionError):
control_ui.verify_remote_admin(node)
def test_redraw_main_ui_after_reconnect_refreshes_channels_nodes_and_layout(self) -> None:
stdscr = mock.Mock()
+146
View File
@@ -0,0 +1,146 @@
from contextlib import ExitStack
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
from meshtastic.protobuf import admin_pb2, config_pb2, module_config_pb2
from contact.ui import control_ui
from contact.ui.menus import generate_menu_from_protobuf, update_ham_fields
from contact.utilities.save_to_radio import save_changes
from tests.test_support import reset_singletons
class HamSettingsTests(unittest.TestCase):
def build_interface(self, licensed=False):
node = Mock(nodeNum=2, localConfig=config_pb2.Config(), moduleConfig=module_config_pb2.ModuleConfig())
node.localConfig.lora.tx_power = 20
node.localConfig.lora.override_frequency = 915.5
node.getChannelByChannelIndex.return_value = None
info = {'user': {'longName': 'KD2ABC', 'shortName': 'HAM', 'isLicensed': licensed}}
interface = SimpleNamespace(localNode=node, getMyNodeInfo=lambda: info, nodesByNum={2: info})
node.iface = interface
return interface, node
def test_fields_follow_license_selection_and_installed_descriptors(self):
interface, node = self.build_interface()
user = generate_menu_from_protobuf(interface)['Main Menu']['User Settings']
descriptors = admin_pb2.HamParameters.DESCRIPTOR.fields
self.assertTrue(all(field.name not in user for field in descriptors))
user['isLicensed'] = (None, True)
update_ham_fields(user, node)
for field in descriptors:
self.assertIs(user[field.name][0], field)
self.assertEqual(user['call_sign'][1], 'KD2ABC')
self.assertEqual(user['tx_power'][1], 20)
self.assertEqual(user['frequency'][1], 915.5)
self.assertNotIn('longName', user)
user['frequency'] = (user['frequency'][0], 916.0)
update_ham_fields(user, node)
self.assertEqual(user['frequency'][1], 916.0)
user['isLicensed'] = (None, False)
update_ham_fields(user, node)
self.assertTrue(all(field.name not in user for field in descriptors))
self.assertEqual(user['longName'][1], 'KD2ABC')
self.assertEqual(user['shortName'][1], 'HAM')
def test_remote_menu_uses_remote_license_and_radio_values(self):
interface, node = self.build_interface(True)
interface.localNode = Mock()
user = generate_menu_from_protobuf(interface, node=node)['Main Menu']['User Settings']
self.assertEqual(user['frequency'][1], 915.5)
self.assertEqual(user['call_sign'][1], 'KD2ABC')
def test_save_full_ham_message_on_selected_node(self):
for remote in (False, True):
interface, node = self.build_interface(True)
if remote:
interface.localNode = Mock()
user = generate_menu_from_protobuf(interface, node=node)['Main Menu']['User Settings']
changes = {name: value for name, (_, value) in user.items()}
changes['frequency'] = 916.0
state = SimpleNamespace(menu_path=['Main Menu', 'User Settings'])
self.assertTrue(save_changes(interface, changes, state, node=node))
message = node._sendAdmin.call_args.args[0]
self.assertEqual(message.WhichOneof('payload_variant'), 'set_ham_mode')
self.assertEqual(message.set_ham_mode.call_sign, 'KD2ABC')
self.assertEqual(message.set_ham_mode.short_name, 'HAM')
self.assertEqual(message.set_ham_mode.tx_power, 20)
self.assertEqual(message.set_ham_mode.frequency, 916.0)
self.assertEqual(node._sendAdmin.call_args.kwargs['onResponse'], node.onAckNak if remote else None)
self.assertEqual([c[0] for c in node.method_calls], [
*['getChannelByChannelIndex'] * 8, 'ensureSessionKey', 'beginSettingsTransaction',
'setOwner', '_sendAdmin', 'commitSettingsTransaction'])
def test_disabling_license_does_not_send_ham_command(self):
interface, node = self.build_interface(True)
user = generate_menu_from_protobuf(interface)['Main Menu']['User Settings']
user['isLicensed'] = (None, False)
update_ham_fields(user, node)
state = SimpleNamespace(menu_path=['Main Menu', 'User Settings'])
self.assertTrue(save_changes(interface, {k: v[1] for k, v in user.items()}, state, node=node))
node._sendAdmin.assert_not_called()
node.setOwner.assert_called_once_with('KD2ABC', 'HAM', False, False)
def test_edit_through_normal_menu_after_selecting_licensed(self):
reset_singletons()
interface, node = self.build_interface()
win = Mock()
# Open User Settings, select licensed, edit the now-visible callsign, save.
win.getch.side_effect = [10, control_ui.curses.KEY_DOWN, control_ui.curses.KEY_DOWN,
10, control_ui.curses.KEY_DOWN, control_ui.curses.KEY_DOWN, 10, 9, 10, 27]
saved = {}
def capture(_interface, values, _state, **kwargs):
saved.update(values)
return False
try:
with ExitStack() as stack:
stack.enter_context(patch.object(control_ui.curses, 'update_lines_cols'))
stack.enter_context(patch.object(control_ui, 'display_menu', return_value=(win, Mock())))
stack.enter_context(patch.object(control_ui, 'move_highlight'))
stack.enter_context(patch.object(control_ui, 'get_list_input', return_value='True'))
text = stack.enter_context(patch.object(control_ui, 'get_text_input', return_value='KD2XYZ'))
stack.enter_context(patch.object(control_ui, 'save_changes', side_effect=capture))
control_ui.settings_menu(Mock(), interface)
self.assertEqual(text.call_args.args[1], 'call_sign')
self.assertEqual(saved['call_sign'], 'KD2XYZ')
self.assertTrue(saved['isLicensed'])
self.assertEqual(saved['tx_power'], 20)
finally:
reset_singletons()
def test_new_library_fields_are_discovered_edited_and_saved(self):
from google.protobuf import descriptor_pb2, descriptor_pool, message_factory
schema = descriptor_pb2.FileDescriptorProto(name='future_ham.proto', syntax='proto3')
params = schema.message_type.add(name='HamParameters')
for field in admin_pb2.HamParameters.DESCRIPTOR.fields:
params.field.add(name=field.name, number=field.number, type=field.type)
if 'long_name' not in admin_pb2.HamParameters.DESCRIPTOR.fields_by_name:
params.field.add(name='long_name', number=5, type=9)
params.field.add(name='future_option', number=100, type=8)
message = schema.message_type.add(name='AdminMessage')
message.field.add(name='set_ham_mode', number=18, type=11, type_name='.HamParameters')
pool = descriptor_pool.DescriptorPool()
pool.Add(schema)
future_params = message_factory.GetMessageClass(pool.FindMessageTypeByName('HamParameters'))
future_message = message_factory.GetMessageClass(pool.FindMessageTypeByName('AdminMessage'))
interface, node = self.build_interface(True)
with patch.object(admin_pb2, 'HamParameters', future_params), patch.object(admin_pb2, 'AdminMessage', future_message):
user = generate_menu_from_protobuf(interface)['Main Menu']['User Settings']
self.assertIn('long_name', user)
self.assertEqual(user['future_option'][1], False)
changes = {name: setting[1] for name, setting in user.items()}
changes.update(long_name='Alice', future_option=True)
state = SimpleNamespace(menu_path=['Main Menu', 'User Settings'])
self.assertTrue(save_changes(interface, changes, state, node=node))
sent = node._sendAdmin.call_args.args[0]
parsed = future_message.FromString(sent.SerializeToString())
self.assertEqual(parsed.set_ham_mode.long_name, 'Alice')
self.assertTrue(parsed.set_ham_mode.future_option)
def test_remote_user_settings_fetch_lora_before_showing_ham_defaults(self):
_, node = self.build_interface(True)
with patch.object(control_ui, '_request_remote_with_timeout') as request:
self.assertTrue(control_ui._request_remote_section(node, ['Main Menu'], 'User Settings'))
request.assert_called_once_with(node.requestConfig,
node.localConfig.DESCRIPTOR.fields_by_name['lora'],
cancel_callback=None)
+9 -3
View File
@@ -1,4 +1,5 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
@@ -9,13 +10,18 @@ class InterfacesTests(unittest.TestCase):
def test_reconnect_interface_retries_until_connection_succeeds(self) -> None:
args = Namespace()
with mock.patch("contact.utilities.interfaces.initialize_interface", side_effect=[None, None, "iface"]) as initialize:
ready_interface = SimpleNamespace(localNode=SimpleNamespace(localConfig=object(), nodeNum=123))
with mock.patch(
"contact.utilities.interfaces.initialize_interface", side_effect=[None, None, ready_interface]
) as initialize:
with mock.patch("contact.utilities.interfaces.time.sleep") as sleep:
result = reconnect_interface(args, attempts=3, delay_seconds=0.25)
with mock.patch("contact.utilities.interfaces.logging.info") as info:
result = reconnect_interface(args, attempts=3, delay_seconds=0.25)
self.assertEqual(result, "iface")
self.assertIs(result, ready_interface)
self.assertEqual(initialize.call_count, 3)
self.assertEqual(sleep.call_count, 2)
self.assertTrue(any("Reconnected to Meshtastic node" in call.args[0] for call in info.call_args_list))
def test_reconnect_interface_raises_after_exhausting_attempts(self) -> None:
args = Namespace()
+5 -2
View File
@@ -240,7 +240,7 @@ class MainRuntimeTests(unittest.TestCase):
with mock.patch.object(entrypoint.sys, "argv", ["contact"]):
with mock.patch.object(entrypoint.curses, "wrapper", side_effect=RuntimeError("boom")):
with mock.patch.object(entrypoint.curses, "endwin") as endwin:
with mock.patch.object(entrypoint.traceback, "print_exc") as print_exc:
with mock.patch.object(entrypoint.traceback, "print_exception") as print_exception:
with mock.patch("builtins.print") as print_mock:
with mock.patch.object(entrypoint.sys, "exit", side_effect=SystemExit(1)) as exit_mock:
with self.assertRaises(SystemExit) as raised:
@@ -248,6 +248,9 @@ class MainRuntimeTests(unittest.TestCase):
self.assertEqual(raised.exception.code, 1)
endwin.assert_called_once_with()
print_exc.assert_called_once_with()
exception_type, exception, exception_traceback = print_exception.call_args_list[-1].args
self.assertIs(exception_type, RuntimeError)
self.assertEqual(str(exception), "boom")
self.assertIsNotNone(exception_traceback)
print_mock.assert_any_call("Fatal error:", mock.ANY)
exit_mock.assert_called_once_with(1)
+95
View File
@@ -0,0 +1,95 @@
import unittest
from contextlib import ExitStack
from unittest import mock
from contact.ui import contact_ui
from contact.utilities.singleton import ui_state
from tests.test_support import reset_singletons
class MessageInputTests(unittest.TestCase):
def setUp(self):
reset_singletons()
ui_state.current_window = 1
for name in ("refresh_message_highlight", "refresh_main_window", "get_color"):
patcher = mock.patch.object(contact_ui, name, return_value=1)
patcher.start()
self.addCleanup(patcher.stop)
self.addCleanup(reset_singletons)
def test_keyboard_loop_sends_edited_draft_after_resize(self):
win = mock.Mock()
win.get_wch.side_effect = [
"a", "b", "c", contact_ui.curses.KEY_LEFT,
contact_ui.curses.KEY_DC, "d", contact_ui.curses.KEY_BACKSPACE,
"e", contact_ui.curses.KEY_RESIZE, "\n", "\x1b",
]
with ExitStack() as stack:
stack.enter_context(mock.patch.object(contact_ui, "entry_win", win))
for name in ("get_channels", "handle_resize", "process_pending_ui_updates", "draw_message_input"):
stack.enter_context(mock.patch.object(contact_ui, name))
stack.enter_context(mock.patch.object(contact_ui, "interface_is_connected", return_value=True))
stack.enter_context(mock.patch.object(contact_ui, "drain_resize_events", return_value=None))
send = stack.enter_context(mock.patch.object(contact_ui, "handle_enter", return_value=""))
contact_ui.main_ui(mock.Mock())
send.assert_called_once_with("abe")
self.assertEqual(contact_ui.input_text, "")
self.assertEqual(ui_state.input_cursor, 0)
def test_typing_focuses_input_and_edits_at_cursor(self):
draft = contact_ui.insert_input_text("", "helo")
self.assertTrue(ui_state.input_focused)
with mock.patch.object(contact_ui, "input_text", draft):
contact_ui.handle_leftright(contact_ui.curses.KEY_LEFT)
draft = contact_ui.insert_input_text(draft, "l")
self.assertEqual(draft, "hello")
self.assertEqual(ui_state.input_cursor, 4)
draft = contact_ui.handle_backspace(mock.Mock(), draft)
self.assertEqual(draft, "helo")
self.assertEqual(ui_state.input_cursor, 3)
def test_up_and_down_return_to_messages_without_scrolling(self):
for handler in (contact_ui.handle_up, contact_ui.handle_down):
ui_state.input_focused = True
ui_state.input_cursor = 2
with mock.patch.object(contact_ui, "scroll_messages") as scroll:
handler()
self.assertFalse(ui_state.input_focused)
self.assertEqual(ui_state.current_window, 1)
self.assertEqual(ui_state.input_cursor, 2)
scroll.assert_not_called()
def test_cursor_stops_at_draft_boundaries(self):
ui_state.input_focused = True
with mock.patch.object(contact_ui, "input_text", "abc"):
contact_ui.handle_leftright(contact_ui.curses.KEY_LEFT)
self.assertEqual(ui_state.input_cursor, 0)
self.assertEqual(contact_ui.handle_backspace(mock.Mock(), "abc"), "abc")
contact_ui.handle_end()
contact_ui.handle_leftright(contact_ui.curses.KEY_RIGHT)
self.assertEqual(ui_state.input_cursor, 3)
contact_ui.handle_home()
self.assertEqual(ui_state.input_cursor, 0)
def test_redraw_clears_line_and_shows_text_after_cursor(self):
win = mock.Mock()
win.getmaxyx.return_value = (3, 30)
ui_state.input_focused = True
ui_state.input_cursor = 2
with mock.patch.object(contact_ui, "entry_win", win), mock.patch.object(contact_ui.curses, "curs_set"):
contact_ui.draw_message_input("hello")
win.addstr.assert_any_call(1, 1, " " * 28, 1)
win.addstr.assert_any_call(1, 1, "Message: hello", 1)
win.move.assert_called_once_with(1, 12)
contact_ui.get_color.assert_any_call("window_frame_selected")
def test_long_draft_scrolls_to_cursor_with_reply_context(self):
win = mock.Mock()
win.getmaxyx.return_value = (3, 24)
ui_state.reply_context = "Reply to someone: "
ui_state.input_cursor = 20
with mock.patch.object(contact_ui, "entry_win", win), mock.patch.object(contact_ui.curses, "curs_set"):
contact_ui.draw_message_input("x" * 20)
self.assertLess(win.move.call_args.args[1], 23)
rendered = [call.args[2] for call in win.addstr.call_args_list if call.args[:2] == (1, 1)][-1]
self.assertEqual(rendered, "Message: " + "x" * 12)
+55
View File
@@ -0,0 +1,55 @@
from contextlib import ExitStack
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
import contact.ui.default_config
from contact.utilities import input_handlers
from tests.test_support import reset_singletons
class RestoredSettingsTests(unittest.TestCase):
def test_blank_input_is_unset_only_when_allowed(self):
for keys, allowed, expected in [
(["\n"], True, ""), ([" ", " ", "\n"], True, ""),
(["\x1b"], True, None), (["\n", "\x1b"], False, None),
]:
with self.subTest(keys=keys, allowed=allowed), ExitStack() as stack:
reset_singletons()
stack.callback(reset_singletons)
win = mock.Mock()
win.get_wch.side_effect = keys
stack.enter_context(mock.patch.object(input_handlers.curses, "newwin", return_value=win))
for name, value in (("LINES", 24), ("COLS", 80)):
stack.enter_context(mock.patch.object(input_handlers.curses, name, value, create=True))
stack.enter_context(mock.patch.object(input_handlers.curses, "curs_set"))
stack.enter_context(mock.patch.object(input_handlers, "get_color", return_value=0))
invalid = stack.enter_context(mock.patch.object(input_handlers, "invalid_input"))
result = input_handlers.get_text_input("Channel name", "name", str, allow_empty=allowed)
self.assertEqual(result, expected)
self.assertEqual(invalid.call_count, 0 if allowed else 1)
def test_missing_log_path_shows_configuration_guidance(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory).resolve()
config_path = root / "config.json"
log_path = root / "missing" / "client.log"
for module in ("contact", "contact.settings"):
with self.subTest(module=module):
config_path.write_text(json.dumps({"log_file_path": str(log_path)}))
result = subprocess.run(
[sys.executable, "-m", module],
env={**os.environ, "CONTACT_CONFIG_ROOT": str(root)},
cwd=Path(__file__).resolve().parents[1],
capture_output=True, text=True, timeout=15,
)
self.assertEqual(result.returncode, 1)
self.assertIn(str(log_path), result.stderr)
self.assertIn(f"Please edit config.json at: {config_path}", result.stderr)
self.assertIn('"log_file_path"', result.stderr)
self.assertNotIn("Traceback", result.stdout + result.stderr)
+43
View File
@@ -1,11 +1,28 @@
from types import SimpleNamespace
import unittest
from unittest import mock
from meshtastic.protobuf import channel_pb2
from contact.utilities.save_to_radio import save_changes
class SaveToRadioTests(unittest.TestCase):
def test_channel_enable_disable_and_unrelated_edits_preserve_role(self):
for index in (0, 1):
interface, node = self.build_interface()
node.channels = [channel_pb2.Channel(index=i) for i in range(2)]
channel = node.channels[index]
path = SimpleNamespace(menu_path=["Main Menu", "Channels", f"Channel {index + 1}"])
save_changes(interface, {"name": "Test"}, path)
self.assertEqual(channel.role, channel_pb2.Channel.DISABLED)
save_changes(interface, {"enabled": True}, path)
expected = channel_pb2.Channel.PRIMARY if index == 0 else channel_pb2.Channel.SECONDARY
self.assertEqual(channel.role, expected)
save_changes(interface, {"enabled": False}, path)
self.assertEqual(channel.role, channel_pb2.Channel.DISABLED)
self.assertEqual(channel.settings.name, "Test")
self.assertEqual(node.writeChannel.call_args_list, [mock.call(index)] * 3)
def build_interface(self):
node = mock.Mock()
node.localConfig = SimpleNamespace(
@@ -55,6 +72,20 @@ class SaveToRadioTests(unittest.TestCase):
self.assertTrue(reconnect_required)
def test_save_changes_replaces_remote_admin_keys_in_one_write(self) -> None:
interface, node = self.build_interface()
key_one = b"1" * 32
key_two = b"2" * 32
key_three = b"3" * 32
node.localConfig.security.admin_key = [key_one, key_two, key_three]
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"])
reconnect_required = save_changes(interface, {"admin_key": [key_one, key_two]}, menu_state, node=node)
self.assertTrue(reconnect_required)
self.assertEqual(node.localConfig.security.admin_key, [key_one, key_two])
node.writeConfig.assert_called_once_with("security")
def test_save_changes_returns_true_only_for_rebooting_device_fields(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Device"])
@@ -95,6 +126,18 @@ class SaveToRadioTests(unittest.TestCase):
self.assertTrue(reconnect_required)
self.assertTrue(node.moduleConfig.mqtt.enabled)
def test_save_changes_writes_channel_is_muted_to_module_settings(self) -> None:
interface, node = self.build_interface()
channel = SimpleNamespace(settings=SimpleNamespace(module_settings=SimpleNamespace(is_muted=False)), role=None)
node.channels = [channel]
menu_state = SimpleNamespace(menu_path=["Main Menu", "Channels", "Channel 1", "module_settings"])
reconnect_required = save_changes(interface, {"is_muted": True}, menu_state)
self.assertFalse(reconnect_required)
self.assertTrue(channel.settings.module_settings.is_muted)
node.writeChannel.assert_called_once_with(0)
def test_save_changes_returns_true_for_user_name_changes(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "User Settings"])
+6
View File
@@ -25,3 +25,9 @@ class TelemetryBeautifierTests(unittest.TestCase):
rendered = get_chunks("time:0\n")
self.assertIn("🕔 01.01.1970 00:00", rendered)
def test_get_chunks_normalizes_keys_and_preserves_unknown_strings(self) -> None:
rendered = get_chunks(' uptime_seconds: 7200\nunknown: "v1.2:ok"\n')
self.assertIn("🆙 2.0h", rendered)
self.assertIn('unknown:"v1.2:ok"', rendered)
+12
View File
@@ -1,4 +1,5 @@
import unittest
from tempfile import NamedTemporaryFile
from unittest import mock
import contact.ui.default_config # Initialize config before the color helpers.
@@ -13,6 +14,17 @@ class UserConfigTests(unittest.TestCase):
self.assertEqual(result, "True")
picker.assert_called_once_with("Enabled", "False", ["True", "False"])
def test_load_log_tail_returns_only_requested_recent_lines(self) -> None:
with NamedTemporaryFile("w") as log_file:
log_file.write("".join(f"line {index}\n" for index in range(600)))
log_file.flush()
lines = user_config.load_log_tail(log_file.name, max_lines=500)
self.assertEqual(len(lines), 500)
self.assertEqual(lines[0], "line 100")
self.assertEqual(lines[-1], "line 599")
if __name__ == "__main__":
unittest.main()
+89 -1
View File
@@ -1,17 +1,66 @@
import unittest
from unittest import mock
from meshtastic.protobuf import config_pb2, mesh_pb2, telemetry_pb2
import contact.ui.default_config as config
from contact.utilities.demo_data import DEMO_LOCAL_NODE_NUM, build_demo_interface
from contact.utilities.singleton import interface_state, ui_state
from contact.utilities.utils import add_new_message, get_channels, get_node_list, parse_protobuf
from contact.utilities.utils import (
_get_channel_name,
_get_modem_preset_name,
add_new_message,
get_channels,
get_node_list,
parse_protobuf,
)
from tests.test_support import reset_singletons, restore_config, snapshot_config
class UtilsTests(unittest.TestCase):
def test_parse_protobuf_beautifies_local_stats(self) -> None:
telemetry = telemetry_pb2.Telemetry(time=1788677531)
telemetry.local_stats.uptime_seconds = 7200
telemetry.local_stats.channel_utilization = 2.90333319
telemetry.local_stats.num_packets_tx = 12
telemetry.local_stats.num_packets_rx_bad = 3
telemetry.local_stats.num_online_nodes = 4
telemetry.local_stats.heap_free_bytes = 1024
rendered = parse_protobuf({"decoded": {
"portnum": "TELEMETRY_APP", "payload": telemetry.SerializeToString(),
}})
self.assertIn("🆙 2.0h", rendered)
self.assertIn("ChUtil:2.9%", rendered)
self.assertIn("📤 Tx:12", rendered)
self.assertIn("⚠️ RxBad:3", rendered)
self.assertIn("📡 Online:4", rendered)
self.assertIn("💾 Free:1024B", rendered)
self.assertNotIn("local_stats", rendered)
def test_parse_protobuf_beautifies_position_details(self) -> None:
position = mesh_pb2.Position(
latitude_i=454230020, longitude_i=-1228472320,
location_source=mesh_pb2.Position.LOC_INTERNAL,
ground_speed=2, ground_track=9000, PDOP=211, sats_in_view=8,
)
rendered = parse_protobuf({"decoded": {
"portnum": "POSITION_APP", "payload": position.SerializeToString(),
}})
self.assertIn("🌍 45.423002", rendered)
self.assertIn("-122.847232", rendered)
self.assertIn("📍 LOC_INTERNAL", rendered)
self.assertIn("💨 2m/s", rendered)
self.assertIn("🧭 90.0°", rendered)
self.assertIn("🎯 PDOP:2.11", rendered)
self.assertIn("🛰️ 8", rendered)
def setUp(self) -> None:
reset_singletons()
_get_modem_preset_name.cache_clear()
self.saved_config = snapshot_config("node_sort")
def tearDown(self) -> None:
@@ -60,6 +109,45 @@ class UtilsTests(unittest.TestCase):
self.assertIn("MediumFast", ui_state.all_messages)
self.assertIn("Another Channel", ui_state.all_messages)
def test_get_channel_name_handles_newer_known_modem_preset(self) -> None:
device_channel = mock.Mock()
device_channel.settings.name = ""
node = mock.Mock()
node.localConfig.lora.modem_preset = 14
with mock.patch("contact.utilities.utils.logging.warning") as warning:
self.assertEqual(_get_channel_name(device_channel, node), "TinyFast")
warning.assert_called_once()
def test_get_channel_name_handles_unknown_modem_preset(self) -> None:
device_channel = mock.Mock()
device_channel.settings.name = ""
node = mock.Mock()
node.localConfig.lora.modem_preset = 99
with mock.patch("contact.utilities.utils.logging.warning") as warning:
self.assertEqual(_get_channel_name(device_channel, node), "UnknownPreset99")
warning.assert_called_once()
def test_get_channel_name_prefers_public_protobuf_enum_over_compatibility_map(self) -> None:
device_channel = mock.Mock()
device_channel.settings.name = ""
node = mock.Mock()
node.localConfig.lora.modem_preset = 14
with mock.patch.object(
config_pb2.Config.LoRaConfig.ModemPreset,
"Name",
return_value="TINY_FAST",
) as enum_name:
with mock.patch("contact.utilities.utils.logging.warning") as warning:
self.assertEqual(_get_channel_name(device_channel, node), "TinyFast")
enum_name.assert_called_once_with(14)
warning.assert_not_called()
def test_get_channels_rebuilds_renamed_channels_and_preserves_messages(self) -> None:
interface = build_demo_interface()
interface.localNode.channels[0].settings.name = "Renamed Channel"