mirror of
https://github.com/pdxlocations/contact.git
synced 2026-10-31 16:21:57 +01:00
Compare commits
35
Commits
1.6.0
...
flicker-fix
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a89d4f8c0d | ||
|
|
808fee4fe0 | ||
|
|
94d0a9053e | ||
|
|
f38167db1a | ||
|
|
57caf463bc | ||
|
|
36408a1f37 | ||
|
|
07b9ed22d8 | ||
|
|
bdba69677b | ||
|
|
1f88a37ac1 | ||
|
|
8f33cbc22d | ||
|
|
40665ff076 | ||
|
|
d1b6da6f26 | ||
|
|
ac90b4af3e | ||
|
|
7af693ddea | ||
|
|
dae6a43652 | ||
|
|
22e5e89828 | ||
|
|
c2db76d5ec | ||
|
|
d8dc4876b6 | ||
|
|
0ae644e3fb | ||
|
|
bfd20bfa49 | ||
|
|
fa2dbccf44 | ||
|
|
66256f0a4c | ||
|
|
e570ce4e48 | ||
|
|
d0d9c1dd55 | ||
|
|
043418bfde | ||
|
|
a8a1527a84 | ||
|
|
7d2092e25f | ||
|
|
ac23c2bbbb | ||
|
|
e378aba526 | ||
|
|
ec35cbd3e7 | ||
|
|
7f6fa49385 | ||
|
|
4ed3cbf370 | ||
|
|
7afc025b9d | ||
|
|
ece37f2a36 | ||
|
|
0cee9777bc |
No files matched your search
+12
-4
@@ -52,7 +52,13 @@ 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"
|
||||
filename=config.log_file_path,
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - %(levelname)s - %(message)s",
|
||||
# Meshtastic or another dependency may have installed a stderr handler
|
||||
# before Contact starts curses. Replace it so background warnings cannot
|
||||
# overwrite the terminal UI.
|
||||
force=True,
|
||||
)
|
||||
|
||||
app_state.lock = threading.Lock()
|
||||
@@ -115,7 +121,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 +162,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 +260,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:
|
||||
|
||||
@@ -99,6 +99,7 @@ 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."
|
||||
isUnmessagable, "Disable incoming messages", "When enabled, this node advertises that it cannot receive messages."
|
||||
|
||||
[app_settings]
|
||||
title, "App Settings", ""
|
||||
@@ -113,6 +114,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", ""
|
||||
@@ -166,7 +168,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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -74,16 +74,26 @@ def play_sound():
|
||||
logging.warning("Configured notification sound is unavailable: %s", sound_path)
|
||||
return
|
||||
|
||||
executable = "afplay" if platform.system() == "Darwin" else shutil.which("ffplay") or shutil.which("mpg123")
|
||||
executable = "afplay" if platform.system() == "Darwin" else next(
|
||||
(shutil.which(player) for player in ("ffplay", "mpg123", "mpv", "cvlc", "paplay") if shutil.which(player)),
|
||||
None,
|
||||
)
|
||||
|
||||
if executable and sound_path:
|
||||
player = os.path.basename(executable)
|
||||
cmd = [executable, sound_path]
|
||||
if executable == "ffplay":
|
||||
if player == "ffplay":
|
||||
cmd = [executable, "-nodisp", "-autoexit", sound_path]
|
||||
elif player == "mpv":
|
||||
cmd = [executable, "--no-video", sound_path]
|
||||
elif player == "cvlc":
|
||||
cmd = [executable, "--play-and-exit", sound_path]
|
||||
|
||||
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
return
|
||||
logging.warning("No suitable sound player found for notification sound")
|
||||
logging.warning(
|
||||
"No suitable sound player found for notification sound; install ffmpeg (ffplay), mpg123, mpv, or VLC."
|
||||
)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
logging.error(f"Sound playback failed: {e}")
|
||||
|
||||
@@ -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:
|
||||
"""
|
||||
|
||||
+2
-1
@@ -36,7 +36,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(
|
||||
@@ -85,6 +85,7 @@ logging.basicConfig( # Run `tail -f client.log` in another terminal to view liv
|
||||
filename=config.log_file_path,
|
||||
level=logging.WARNING, # DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
||||
format="%(asctime)s - %(levelname)s - %(message)s",
|
||||
force=True,
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+427
-74
@@ -1,5 +1,7 @@
|
||||
import curses
|
||||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
import time
|
||||
import traceback
|
||||
from numbers import Real
|
||||
@@ -14,11 +16,14 @@ from contact.utilities.utils import (
|
||||
build_reply_prefix,
|
||||
)
|
||||
from contact.settings import settings_menu
|
||||
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
|
||||
from contact.utilities.db_handler import get_name_from_database, update_node_info_in_db, is_chat_archived, load_older_messages
|
||||
from contact.utilities.input_handlers import get_list_input
|
||||
from contact.utilities.interfaces import interface_is_connected
|
||||
from contact.utilities.i18n import t
|
||||
from contact.utilities.emoji_utils import normalize_message_text
|
||||
import contact.ui.default_config as config
|
||||
@@ -30,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
|
||||
|
||||
@@ -38,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(
|
||||
@@ -128,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:
|
||||
@@ -155,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
|
||||
|
||||
@@ -174,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:
|
||||
@@ -186,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)
|
||||
@@ -377,16 +499,73 @@ 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()
|
||||
handle_resize(stdscr, True)
|
||||
|
||||
while True:
|
||||
interface = interface_state.interface
|
||||
if not interface_is_connected(interface) and not ui_state.reconnect_attempted:
|
||||
ui_state.reconnect_attempted = True
|
||||
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
|
||||
|
||||
reconnect_interface_with_splash(stdscr, interface)
|
||||
if status_win is not None:
|
||||
status_win.erase()
|
||||
status_win.refresh()
|
||||
ui_state.reconnect_attempted = 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()
|
||||
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",
|
||||
["Retry", "Cancel"],
|
||||
mandatory=True,
|
||||
)
|
||||
if retry == "Retry":
|
||||
ui_state.reconnect_attempted = False
|
||||
handle_resize(stdscr, False)
|
||||
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:
|
||||
@@ -426,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)
|
||||
@@ -433,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)
|
||||
|
||||
@@ -449,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
|
||||
@@ -475,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:
|
||||
@@ -494,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:
|
||||
@@ -504,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:
|
||||
@@ -518,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:
|
||||
@@ -558,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:
|
||||
@@ -578,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:
|
||||
@@ -652,12 +901,27 @@ def handle_enter(input_text: str) -> str:
|
||||
)
|
||||
return input_text
|
||||
# Enter key pressed, send user input as message
|
||||
send_message(
|
||||
input_text,
|
||||
channel=ui_state.selected_channel,
|
||||
reply_id=ui_state.reply_id,
|
||||
reply_context=ui_state.reply_context,
|
||||
)
|
||||
try:
|
||||
send_message(
|
||||
input_text,
|
||||
channel=ui_state.selected_channel,
|
||||
reply_id=ui_state.reply_id,
|
||||
reply_context=ui_state.reply_context,
|
||||
)
|
||||
except Exception as exc:
|
||||
logging.warning("Message send failed; reconnecting interface: %s", exc)
|
||||
try:
|
||||
from contact.ui.control_ui import reconnect_interface_with_splash
|
||||
|
||||
reconnect_interface_with_splash(root_win, interface_state.interface)
|
||||
contact.ui.dialog.dialog("Reconnected", "Connection restored. Your message was not sent; press Enter to retry.")
|
||||
except Exception as reconnect_exc:
|
||||
logging.exception("Automatic reconnect failed")
|
||||
contact.ui.dialog.dialog(
|
||||
"Reconnect failed",
|
||||
"Contact could not reconnect after 20 seconds. Check the radio connection and retry your message.",
|
||||
)
|
||||
return input_text
|
||||
ui_state.reply_id = None
|
||||
ui_state.reply_context = ""
|
||||
ui_state.reply_id_unavailable = False
|
||||
@@ -673,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
|
||||
@@ -770,6 +1056,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)
|
||||
@@ -789,11 +1076,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)
|
||||
|
||||
@@ -838,7 +1129,7 @@ def handle_f5_key(stdscr: curses.window) -> None:
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
dialog_win.refresh()
|
||||
dialog_win.noutrefresh()
|
||||
msg_win.noutrefresh()
|
||||
curses.doupdate()
|
||||
|
||||
@@ -858,12 +1149,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:
|
||||
@@ -922,14 +1213,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
|
||||
|
||||
|
||||
@@ -949,40 +1238,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":
|
||||
selected_node_num = ui_state.node_list[ui_state.selected_node]
|
||||
|
||||
# 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)
|
||||
elif remote_node_num is not None:
|
||||
wait_win = None
|
||||
try:
|
||||
wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
|
||||
settings_menu(
|
||||
stdscr,
|
||||
interface,
|
||||
node=interface.getNode(remote_node_num),
|
||||
remote=True,
|
||||
status_callback=lambda message: update_remote_admin_wait(wait_win, message),
|
||||
)
|
||||
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))
|
||||
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()
|
||||
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)
|
||||
@@ -991,7 +1323,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:
|
||||
@@ -1010,7 +1342,24 @@ def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> curses.wind
|
||||
return None
|
||||
|
||||
|
||||
def update_remote_admin_wait(wait_win: curses.window | None, message: str) -> 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()
|
||||
box_width = min(width - 4, max(len(message) + 4, len(title) + 6, 32))
|
||||
win = curses.newwin(5, box_width, max(0, (height - 5) // 2), max(0, (width - box_width) // 2))
|
||||
win.bkgd(get_color("background"))
|
||||
win.attrset(get_color("window_frame"))
|
||||
win.border()
|
||||
win.addstr(0, 2, f" {title} ", get_color("settings_default"))
|
||||
win.addstr(2, 2, message[: box_width - 4], get_color("settings_default"))
|
||||
win.refresh()
|
||||
return win
|
||||
except curses.error:
|
||||
return 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
|
||||
@@ -1322,7 +1671,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)
|
||||
|
||||
@@ -1337,7 +1686,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)
|
||||
@@ -1351,7 +1699,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):
|
||||
@@ -1667,7 +2015,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:
|
||||
@@ -1684,7 +2032,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
|
||||
@@ -1727,9 +2075,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,
|
||||
@@ -1737,6 +2088,8 @@ def refresh_pad(window: int) -> None:
|
||||
bottom,
|
||||
right,
|
||||
)
|
||||
if commit:
|
||||
curses.doupdate()
|
||||
|
||||
|
||||
def add_notification(channel_number: int) -> None:
|
||||
|
||||
+114
-29
@@ -5,6 +5,7 @@ import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from typing import List
|
||||
from meshtastic.protobuf import admin_pb2
|
||||
|
||||
@@ -26,7 +27,6 @@ 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.nav_utils import move_highlight, draw_arrows, update_help_window
|
||||
from contact.ui.splash import draw_splash
|
||||
from contact.ui.user_config import json_editor
|
||||
from contact.utilities.arg_parser import setup_parser
|
||||
from contact.utilities.singleton import interface_state, menu_state
|
||||
@@ -41,6 +41,10 @@ help_win = None
|
||||
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
|
||||
|
||||
|
||||
class RemoteAdminCancelled(Exception):
|
||||
"""Raised when the user dismisses an in-progress remote-admin request."""
|
||||
|
||||
|
||||
# Compute the effective menu width for the current terminal
|
||||
def get_menu_width() -> int:
|
||||
# Leave at least 2 columns for borders; clamp to >= 20 for usability
|
||||
@@ -236,9 +240,6 @@ def reconnect_interface_with_splash(stdscr: object, interface: object) -> object
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
stdscr.clear()
|
||||
stdscr.refresh()
|
||||
draw_splash(stdscr)
|
||||
new_interface = reconnect_interface(setup_parser().parse_args())
|
||||
interface_state.interface = new_interface
|
||||
redraw_main_ui_after_reconnect(stdscr)
|
||||
@@ -278,12 +279,13 @@ 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)
|
||||
|
||||
|
||||
def _request_remote_with_timeout(request, *args) -> None:
|
||||
def _request_remote_with_timeout(request, *args, cancel_callback=None) -> None:
|
||||
"""Run a Meshtastic remote request without allowing its ACK wait to hang the UI."""
|
||||
failure = []
|
||||
|
||||
@@ -295,15 +297,90 @@ def _request_remote_with_timeout(request, *args) -> None:
|
||||
|
||||
worker = threading.Thread(target=run_request, name="remote-admin-request", daemon=True)
|
||||
worker.start()
|
||||
worker.join(REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS)
|
||||
deadline = time.monotonic() + REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS
|
||||
while worker.is_alive() and time.monotonic() < deadline:
|
||||
worker.join(0.1)
|
||||
if cancel_callback and cancel_callback():
|
||||
raise RemoteAdminCancelled()
|
||||
if worker.is_alive():
|
||||
raise TimeoutError("Timed out waiting for the remote node to answer the admin request.")
|
||||
if failure:
|
||||
raise failure[0]
|
||||
|
||||
|
||||
def _request_remote_channels_with_timeout(node: object, cancel_callback=None) -> None:
|
||||
"""Retrieve a remote node's channel list without leaving the UI waiting forever."""
|
||||
node.requestChannels()
|
||||
deadline = time.monotonic() + REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS
|
||||
while getattr(node, "channels", None) is None:
|
||||
if cancel_callback and cancel_callback():
|
||||
raise RemoteAdminCancelled()
|
||||
if time.monotonic() >= deadline:
|
||||
raise TimeoutError("Timed out waiting for the remote node's channel list.")
|
||||
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:
|
||||
"""Fetch exactly the remote-admin section the user is opening.
|
||||
|
||||
Meshtastic returns radio and module settings one protobuf section at a
|
||||
time. Asking for all of them on menu entry is both slow and unreliable on
|
||||
long paths, so leave the top-level menu immediately available.
|
||||
"""
|
||||
if menu_path == ["Main Menu"] and selected_option == "Channels":
|
||||
if status_callback:
|
||||
status_callback("Requesting channel list…")
|
||||
_request_remote_channels_with_timeout(node, cancel_callback=cancel_callback)
|
||||
return True
|
||||
|
||||
if len(menu_path) != 2:
|
||||
return False
|
||||
|
||||
category = menu_path[-1]
|
||||
if category == "Radio Settings":
|
||||
config = node.localConfig
|
||||
elif category == "Module Settings":
|
||||
config = node.moduleConfig
|
||||
else:
|
||||
return False
|
||||
|
||||
field = config.DESCRIPTOR.fields_by_name.get(selected_option)
|
||||
if field is None or field.name in {"version", "sessionkey"}:
|
||||
return False
|
||||
if status_callback:
|
||||
status_callback(f"Requesting {selected_option} config…")
|
||||
_request_remote_with_timeout(node.requestConfig, field, cancel_callback=cancel_callback)
|
||||
return True
|
||||
|
||||
|
||||
def settings_menu(
|
||||
stdscr: object, interface: object, node: object = None, remote: bool = False, status_callback=None
|
||||
stdscr: object,
|
||||
interface: object,
|
||||
node: object = None,
|
||||
remote: bool = False,
|
||||
status_callback=None,
|
||||
cancel_callback=None,
|
||||
) -> None:
|
||||
curses.update_lines_cols()
|
||||
global admin_target_label
|
||||
@@ -313,21 +390,6 @@ def settings_menu(
|
||||
node_name = node_info.get("user", {}).get("longName") if isinstance(node_info, dict) else None
|
||||
admin_target_label = node_name or f"!{node_num:08x}"
|
||||
|
||||
if remote:
|
||||
# Remote nodes do not populate radio config automatically. Request
|
||||
# the standard sections before building the menu. Module requests are
|
||||
# intentionally deferred: firmware may omit modules and never answer
|
||||
# those requests, which would otherwise block the entire UI.
|
||||
for config_type in list(node.localConfig.DESCRIPTOR.fields):
|
||||
if config_type.name in {"version", "sessionkey"}:
|
||||
continue
|
||||
if status_callback:
|
||||
status_callback(f"Requesting {config_type.name} config…")
|
||||
_request_remote_with_timeout(node.requestConfig, config_type)
|
||||
if status_callback:
|
||||
status_callback("Requesting channels…")
|
||||
_request_remote_with_timeout(node.requestChannels)
|
||||
|
||||
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=not remote)
|
||||
menu_state.current_menu = menu["Main Menu"]
|
||||
menu_state.menu_path = ["Main Menu"]
|
||||
@@ -470,7 +532,7 @@ def settings_menu(
|
||||
filename += ".yaml"
|
||||
|
||||
try:
|
||||
config_text = config_export(interface)
|
||||
config_text = config_export(interface, node=node)
|
||||
yaml_file_path = os.path.join(config_folder, filename)
|
||||
|
||||
if os.path.exists(yaml_file_path):
|
||||
@@ -535,7 +597,7 @@ def settings_menu(
|
||||
["Yes", "No"],
|
||||
)
|
||||
if overwrite == "Yes":
|
||||
config_import(interface, file_path)
|
||||
config_import(interface, file_path, node=node)
|
||||
menu_state.start_index.pop()
|
||||
continue
|
||||
|
||||
@@ -638,12 +700,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)
|
||||
@@ -675,7 +739,7 @@ def settings_menu(
|
||||
menu_state.need_redraw = True
|
||||
continue
|
||||
|
||||
if selected_option in ["longName", "shortName", "isLicensed"]:
|
||||
if selected_option in ["longName", "shortName", "isLicensed", "isUnmessagable"]:
|
||||
if selected_option in ["longName", "shortName"]:
|
||||
new_value = get_text_input(
|
||||
f"{human_readable_name} is currently: {current_value}", selected_option, None
|
||||
@@ -683,7 +747,7 @@ def settings_menu(
|
||||
new_value = current_value if new_value is None else new_value
|
||||
menu_state.current_menu[selected_option] = (field, new_value)
|
||||
|
||||
elif selected_option == "isLicensed":
|
||||
elif selected_option in ["isLicensed", "isUnmessagable"]:
|
||||
new_value = get_list_input(
|
||||
f"{human_readable_name} is currently: {current_value}",
|
||||
str(current_value),
|
||||
@@ -783,7 +847,28 @@ def settings_menu(
|
||||
|
||||
menu_state.current_menu[selected_option] = (field, new_value)
|
||||
else:
|
||||
menu_state.current_menu = menu_state.current_menu[selected_option]
|
||||
requested_remote_section = False
|
||||
if remote:
|
||||
try:
|
||||
requested_remote_section = _request_remote_section(
|
||||
node, menu_state.menu_path, selected_option, status_callback, cancel_callback
|
||||
)
|
||||
finally:
|
||||
if status_callback:
|
||||
status_callback(None)
|
||||
if requested_remote_section:
|
||||
# The response updates the node's protobufs. Rebuild
|
||||
# the menu before entering the requested section so
|
||||
# the newly received values are what the user sees.
|
||||
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=False)
|
||||
refreshed_menu = menu["Main Menu"]
|
||||
for path_step in menu_state.menu_path[1:]:
|
||||
refreshed_menu = refreshed_menu[path_step]
|
||||
menu_state.current_menu = refreshed_menu[selected_option]
|
||||
else:
|
||||
menu_state.current_menu = menu_state.current_menu[selected_option]
|
||||
else:
|
||||
menu_state.current_menu = menu_state.current_menu[selected_option]
|
||||
menu_state.menu_path.append(selected_option)
|
||||
menu_state.menu_index.append(menu_state.selected_index)
|
||||
menu_state.selected_index = 0
|
||||
|
||||
@@ -76,6 +76,7 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
|
||||
"longName": (None, current_user_config.get("longName", "Not Set")),
|
||||
"shortName": (None, current_user_config.get("shortName", "Not Set")),
|
||||
"isLicensed": (None, current_user_config.get("isLicensed", "False")),
|
||||
"isUnmessagable": (None, current_user_config.get("isUnmessagable", "False")),
|
||||
}
|
||||
else:
|
||||
logging.info("User settings not found in Node Info")
|
||||
|
||||
+10
-5
@@ -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()
|
||||
|
||||
@@ -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
|
||||
@@ -39,6 +41,14 @@ class ChatUIState:
|
||||
redraw_full_ui: bool = False
|
||||
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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -134,47 +134,47 @@ def setPref(config, comp_name, raw_val) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def config_import(interface, filename):
|
||||
def config_import(interface, filename, node=None):
|
||||
"""Apply a configuration file to the selected local or remote node."""
|
||||
node = node or interface.localNode
|
||||
with open(filename, encoding="utf8") as file:
|
||||
configuration = yaml.safe_load(file)
|
||||
closeNow = True
|
||||
|
||||
interface.getNode("^local", False).beginSettingsTransaction()
|
||||
node.beginSettingsTransaction()
|
||||
|
||||
if "owner" in configuration:
|
||||
logging.info(f"Setting device owner to {configuration['owner']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode("^local", False).setOwner(configuration["owner"])
|
||||
node.setOwner(configuration["owner"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "owner_short" in configuration:
|
||||
logging.info(f"Setting device owner short to {configuration['owner_short']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode("^local", False).setOwner(long_name=None, short_name=configuration["owner_short"])
|
||||
node.setOwner(long_name=None, short_name=configuration["owner_short"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "ownerShort" in configuration:
|
||||
logging.info(f"Setting device owner short to {configuration['ownerShort']}")
|
||||
waitForAckNak = True
|
||||
interface.getNode("^local", False).setOwner(long_name=None, short_name=configuration["ownerShort"])
|
||||
node.setOwner(long_name=None, short_name=configuration["ownerShort"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "channel_url" in configuration:
|
||||
logging.info(f"Setting channel url to {configuration['channel_url']}")
|
||||
interface.getNode("^local").setURL(configuration["channel_url"])
|
||||
node.setURL(configuration["channel_url"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "channelUrl" in configuration:
|
||||
logging.info(f"Setting channel url to {configuration['channelUrl']}")
|
||||
interface.getNode("^local").setURL(configuration["channelUrl"])
|
||||
node.setURL(configuration["channelUrl"])
|
||||
time.sleep(0.5)
|
||||
|
||||
if "location" in configuration:
|
||||
alt = 0
|
||||
lat = 0.0
|
||||
lon = 0.0
|
||||
localConfig = interface.localNode.localConfig
|
||||
|
||||
if "alt" in configuration["location"]:
|
||||
alt = int(configuration["location"]["alt"] or 0)
|
||||
logging.info(f"Fixing altitude at {alt} meters")
|
||||
@@ -185,39 +185,44 @@ def config_import(interface, filename):
|
||||
lon = float(configuration["location"]["lon"] or 0)
|
||||
logging.info(f"Fixing longitude at {lon} degrees")
|
||||
logging.info("Setting device position")
|
||||
interface.localNode.setFixedPosition(lat, lon, alt)
|
||||
node.setFixedPosition(lat, lon, alt)
|
||||
time.sleep(0.5)
|
||||
|
||||
if "config" in configuration:
|
||||
localConfig = interface.getNode("^local").localConfig
|
||||
localConfig = node.localConfig
|
||||
for section in configuration["config"]:
|
||||
traverseConfig(section, configuration["config"][section], localConfig)
|
||||
interface.getNode("^local").writeConfig(camel_to_snake(section))
|
||||
node.writeConfig(camel_to_snake(section))
|
||||
time.sleep(0.5)
|
||||
|
||||
if "module_config" in configuration:
|
||||
moduleConfig = interface.getNode("^local").moduleConfig
|
||||
moduleConfig = node.moduleConfig
|
||||
for section in configuration["module_config"]:
|
||||
traverseConfig(
|
||||
section,
|
||||
configuration["module_config"][section],
|
||||
moduleConfig,
|
||||
)
|
||||
interface.getNode("^local").writeConfig(camel_to_snake(section))
|
||||
node.writeConfig(camel_to_snake(section))
|
||||
time.sleep(0.5)
|
||||
|
||||
interface.getNode("^local", False).commitSettingsTransaction()
|
||||
node.commitSettingsTransaction()
|
||||
logging.info("Writing modified configuration to device")
|
||||
|
||||
|
||||
def config_export(interface) -> str:
|
||||
"""used in --export-config"""
|
||||
def config_export(interface, node=None) -> str:
|
||||
"""Export the selected node's configuration as Meshtastic configure YAML."""
|
||||
configObj = {}
|
||||
node = node or interface.localNode
|
||||
|
||||
owner = interface.getLongName()
|
||||
owner_short = interface.getShortName()
|
||||
channel_url = interface.localNode.getURL()
|
||||
myinfo = interface.getMyNodeInfo()
|
||||
if node is interface.localNode:
|
||||
myinfo = interface.getMyNodeInfo()
|
||||
else:
|
||||
myinfo = getattr(interface, "nodesByNum", {}).get(getattr(node, "nodeNum", None), {})
|
||||
user = myinfo.get("user", {}) if isinstance(myinfo, dict) else {}
|
||||
owner = user.get("longName")
|
||||
owner_short = user.get("shortName")
|
||||
channel_url = node.getURL()
|
||||
pos = myinfo.get("position")
|
||||
lat = None
|
||||
lon = None
|
||||
@@ -242,7 +247,7 @@ def config_export(interface) -> str:
|
||||
if alt:
|
||||
configObj["location"]["alt"] = alt
|
||||
|
||||
config = MessageToDict(interface.localNode.localConfig) # checkme - Used as a dictionary here and a string below
|
||||
config = MessageToDict(node.localConfig) # checkme - Used as a dictionary here and a string below
|
||||
if config:
|
||||
# Convert inner keys to correct snake/camelCase
|
||||
prefs = {}
|
||||
@@ -265,7 +270,7 @@ def config_export(interface) -> str:
|
||||
else:
|
||||
configObj["config"] = config
|
||||
|
||||
module_config = MessageToDict(interface.localNode.moduleConfig)
|
||||
module_config = MessageToDict(node.moduleConfig)
|
||||
if module_config:
|
||||
# Convert inner keys to correct snake/camelCase
|
||||
prefs = {}
|
||||
|
||||
@@ -1,9 +1,33 @@
|
||||
import logging
|
||||
import contextlib
|
||||
import io
|
||||
import time
|
||||
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
|
||||
|
||||
|
||||
def initialize_interface(args):
|
||||
def interface_is_connected(interface) -> bool:
|
||||
"""Return the transport-independent connection state for an interface.
|
||||
|
||||
TCPInterface deliberately leaves ``stream`` set to ``None`` and uses its
|
||||
``socket`` instead. Checking only ``stream`` therefore treats every TCP
|
||||
connection (including meshtasticd) as disconnected.
|
||||
"""
|
||||
if interface is None:
|
||||
return False
|
||||
|
||||
connected = getattr(interface, "isConnected", None)
|
||||
if hasattr(connected, "is_set"):
|
||||
return connected.is_set()
|
||||
if isinstance(connected, bool):
|
||||
return connected
|
||||
if hasattr(interface, "socket"):
|
||||
return getattr(interface, "socket", None) is not None
|
||||
if hasattr(interface, "stream"):
|
||||
return getattr(interface, "stream", None) is not None
|
||||
return getattr(interface, "localNode", None) is not None
|
||||
|
||||
|
||||
def initialize_interface(args, status_callback=None):
|
||||
try:
|
||||
|
||||
if args.ble:
|
||||
@@ -21,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:
|
||||
@@ -34,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}")
|
||||
@@ -44,19 +73,34 @@ def initialize_interface(args):
|
||||
logging.critical(f"Fatal error initializing interface: {ex}")
|
||||
|
||||
|
||||
def reconnect_interface(args, attempts: int = 15, delay_seconds: float = 1.0):
|
||||
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 not None:
|
||||
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("initialize_interface returned None")
|
||||
last_error = RuntimeError("interface did not complete connection setup")
|
||||
try:
|
||||
interface.close()
|
||||
except Exception:
|
||||
pass
|
||||
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
|
||||
@@ -2,7 +2,6 @@ from meshtastic.protobuf import 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 = {
|
||||
@@ -29,7 +28,7 @@ LORA_REBOOT_KEYS = {
|
||||
"sx126x_rx_boosted_gain",
|
||||
}
|
||||
SECURITY_NON_REBOOT_KEYS = {"debug_log_api_enabled", "serial_enabled"}
|
||||
USER_RECONNECT_KEYS = {"longName", "shortName", "isLicensed", "is_licensed"}
|
||||
USER_RECONNECT_KEYS = {"longName", "shortName", "isLicensed", "is_licensed", "isUnmessagable", "is_unmessagable"}
|
||||
|
||||
|
||||
def _collect_changed_keys(modified_settings):
|
||||
@@ -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)
|
||||
@@ -143,12 +134,16 @@ def save_changes(interface, modified_settings, menu_state, node=None):
|
||||
long_name = modified_settings.get("longName")
|
||||
short_name = modified_settings.get("shortName")
|
||||
is_licensed = modified_settings.get("isLicensed")
|
||||
is_unmessagable = modified_settings.get("isUnmessagable")
|
||||
is_licensed = is_licensed == "True" or is_licensed is True # Normalize boolean
|
||||
if is_unmessagable is not None:
|
||||
is_unmessagable = is_unmessagable == "True" or is_unmessagable is True
|
||||
|
||||
node.setOwner(long_name, short_name, is_licensed)
|
||||
node.setOwner(long_name, short_name, is_licensed, is_unmessagable)
|
||||
|
||||
logging.info(
|
||||
f"Updated {config_category} with Long Name: {long_name}, Short Name: {short_name}, Licensed Mode: {is_licensed}"
|
||||
f"Updated {config_category} with Long Name: {long_name}, Short Name: {short_name}, "
|
||||
f"Licensed Mode: {is_licensed}, Unmessageable: {is_unmessagable}"
|
||||
)
|
||||
|
||||
return _requires_reconnect(menu_state, modified_settings)
|
||||
@@ -157,7 +152,8 @@ 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
|
||||
@@ -166,8 +162,10 @@ def save_changes(interface, modified_settings, menu_state, node=None):
|
||||
for key, value in modified_settings.items():
|
||||
if 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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "contact"
|
||||
version = "1.6.0"
|
||||
version = "1.6.9"
|
||||
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
@@ -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:
|
||||
|
||||
+46
-1
@@ -1,6 +1,12 @@
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from tempfile import NamedTemporaryFile
|
||||
from unittest.mock import Mock
|
||||
|
||||
from contact.utilities.config_io import _is_repeated_field, splitCompoundName
|
||||
import yaml
|
||||
from meshtastic.protobuf import config_pb2, module_config_pb2
|
||||
|
||||
from contact.utilities.config_io import _is_repeated_field, config_export, config_import, splitCompoundName
|
||||
|
||||
|
||||
class ConfigIoTests(unittest.TestCase):
|
||||
@@ -19,3 +25,42 @@ class ConfigIoTests(unittest.TestCase):
|
||||
field_type = type("Field", (), {"label": 3, "LABEL_REPEATED": 3})
|
||||
|
||||
self.assertTrue(_is_repeated_field(field_type()))
|
||||
|
||||
def test_config_export_uses_selected_remote_node(self) -> None:
|
||||
local_node = SimpleNamespace(nodeNum=1)
|
||||
remote_node = SimpleNamespace(
|
||||
nodeNum=2,
|
||||
localConfig=config_pb2.Config(),
|
||||
moduleConfig=module_config_pb2.ModuleConfig(),
|
||||
getURL=lambda: "remote-channel-url",
|
||||
)
|
||||
interface = SimpleNamespace(
|
||||
localNode=local_node,
|
||||
nodesByNum={
|
||||
2: {
|
||||
"user": {"longName": "Remote Node", "shortName": "REM"},
|
||||
"position": {"latitude": 1.25, "longitude": 2.5, "altitude": 12},
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
exported = yaml.safe_load(config_export(interface, node=remote_node))
|
||||
|
||||
self.assertEqual(exported["owner"], "Remote Node")
|
||||
self.assertEqual(exported["owner_short"], "REM")
|
||||
self.assertEqual(exported["channel_url"], "remote-channel-url")
|
||||
self.assertEqual(exported["location"], {"lat": 1.25, "lon": 2.5, "alt": 12})
|
||||
|
||||
def test_config_import_uses_selected_remote_node(self) -> None:
|
||||
remote_node = Mock()
|
||||
interface = SimpleNamespace(localNode=Mock())
|
||||
with NamedTemporaryFile("w", suffix=".yaml") as config_file:
|
||||
yaml.safe_dump({"owner": "Remote Node"}, config_file)
|
||||
config_file.flush()
|
||||
|
||||
config_import(interface, config_file.name, node=remote_node)
|
||||
|
||||
remote_node.beginSettingsTransaction.assert_called_once_with()
|
||||
remote_node.setOwner.assert_called_once_with("Remote Node")
|
||||
remote_node.commitSettingsTransaction.assert_called_once_with()
|
||||
interface.localNode.setOwner.assert_not_called()
|
||||
@@ -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:
|
||||
|
||||
@@ -53,6 +53,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()
|
||||
|
||||
|
||||
@@ -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
@@ -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)
|
||||
@@ -44,3 +44,20 @@ class MenusTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(module_settings["external_notification"]["ringtone"], (None, "tone"))
|
||||
self.assertEqual(module_settings["canned_message"]["messages"], (None, "Hi|Bye"))
|
||||
|
||||
def test_user_settings_include_unmessageable_toggle(self) -> None:
|
||||
local_node = SimpleNamespace(
|
||||
localConfig=config_pb2.Config(),
|
||||
moduleConfig=module_config_pb2.ModuleConfig(),
|
||||
getChannelByChannelIndex=lambda _: None,
|
||||
)
|
||||
interface = SimpleNamespace(
|
||||
localNode=local_node,
|
||||
getMyNodeInfo=lambda: {
|
||||
"user": {"longName": "Test User", "shortName": "TU", "isUnmessagable": True},
|
||||
"position": {},
|
||||
},
|
||||
)
|
||||
|
||||
user_settings = generate_menu_from_protobuf(interface)["Main Menu"]["User Settings"]
|
||||
self.assertEqual(user_settings["isUnmessagable"], (None, True))
|
||||
@@ -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,6 +55,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 +109,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"])
|
||||
@@ -112,3 +138,12 @@ class SaveToRadioTests(unittest.TestCase):
|
||||
|
||||
self.assertTrue(reconnect_required)
|
||||
node.setOwner.assert_called_once()
|
||||
|
||||
def test_save_changes_sends_user_unmessageable_setting(self) -> None:
|
||||
interface, node = self.build_interface()
|
||||
menu_state = SimpleNamespace(menu_path=["Main Menu", "User Settings"])
|
||||
|
||||
reconnect_required = save_changes(interface, {"isUnmessagable": True}, menu_state)
|
||||
|
||||
self.assertTrue(reconnect_required)
|
||||
self.assertEqual(node.setOwner.call_args.args[3], True)
|
||||
@@ -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()
|
||||
+50
-1
@@ -1,10 +1,19 @@
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from meshtastic.protobuf import config_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
|
||||
|
||||
@@ -12,6 +21,7 @@ from tests.test_support import reset_singletons, restore_config, snapshot_config
|
||||
class UtilsTests(unittest.TestCase):
|
||||
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 +70,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"
|
||||
|
||||
Reference in new issue
Block a user