diff --git a/contact/localisations/en.ini b/contact/localisations/en.ini index 8642888..fd7ad13 100644 --- a/contact/localisations/en.ini +++ b/contact/localisations/en.ini @@ -114,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", "" diff --git a/contact/ui/contact_ui.py b/contact/ui/contact_ui.py index 026fa04..969ebe2 100644 --- a/contact/ui/contact_ui.py +++ b/contact/ui/contact_ui.py @@ -1,5 +1,6 @@ import curses import logging +import os from typing import Optional import time import traceback @@ -16,6 +17,7 @@ from contact.utilities.utils import ( ) 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 @@ -43,6 +45,115 @@ root_win = None nodes_pad = None +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( *, channels: bool = False, @@ -389,7 +500,15 @@ def main_ui(stdscr: curses.window) -> None: interface = interface_state.interface if not interface_is_connected(interface) and not ui_state.reconnect_attempted: ui_state.reconnect_attempted = True - status_win = show_connection_status(stdscr, "Disconnected", "Trying to reconnect…") + logging.warning("Meshtastic connection disconnected; attempting automatic reconnect") + flush_log_handlers() + if ui_state.log_viewer_open: + # Reconnect is synchronous. Repaint once before entering it so + # the disconnect event remains visible during the attempt. + draw_log_viewer(stdscr) + status_win = None + if not ui_state.log_viewer_open: + status_win = show_connection_status(stdscr, "Disconnected", "Trying to reconnect…") try: from contact.ui.control_ui import reconnect_interface_with_splash @@ -398,13 +517,18 @@ def main_ui(stdscr: curses.window) -> None: status_win.erase() status_win.refresh() ui_state.reconnect_attempted = False - handle_resize(stdscr, False) + if not ui_state.log_viewer_open: + handle_resize(stdscr, False) except Exception: if status_win is not None: status_win.erase() status_win.refresh() - handle_resize(stdscr, False) + if not ui_state.log_viewer_open: + handle_resize(stdscr, False) logging.exception("Automatic reconnect after transport disconnect failed") + if ui_state.log_viewer_open: + ui_state.reconnect_prompt_open = True + continue retry = get_list_input( "Contact could not reconnect after 20 seconds.", "Retry", @@ -417,6 +541,18 @@ def main_ui(stdscr: curses.window) -> None: continue handle_resize(stdscr, False) + if ui_state.log_viewer_open: + content_height = draw_log_viewer(stdscr) + draw_log_reconnect_prompt(stdscr) + stdscr.timeout(200) + try: + key = stdscr.getch() + except curses.error: + continue + if key != -1 and handle_log_viewer_key(key, content_height): + handle_resize(stdscr, False) + continue + with app_state.lock: process_pending_ui_updates(stdscr) entry_display = f"{ui_state.reply_context}{input_text or ''}" diff --git a/contact/ui/control_ui.py b/contact/ui/control_ui.py index 7a6e97f..3605395 100644 --- a/contact/ui/control_ui.py +++ b/contact/ui/control_ui.py @@ -279,7 +279,8 @@ def redraw_main_ui_after_reconnect(stdscr: object) -> None: get_channels() refresh_node_list() - contact_ui.handle_resize(stdscr, False) + if not contact_ui.ui_state.log_viewer_open: + contact_ui.handle_resize(stdscr, False) except Exception: logging.debug("Skipping main UI redraw after reconnect", exc_info=True) @@ -695,12 +696,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) diff --git a/contact/ui/ui_state.py b/contact/ui/ui_state.py index 4c29963..0ca2858 100644 --- a/contact/ui/ui_state.py +++ b/contact/ui/ui_state.py @@ -40,6 +40,13 @@ class ChatUIState: scroll_messages_to_bottom: bool = False preserve_message_selection: bool = False reconnect_attempted: bool = False + reconnect_prompt_open: bool = False + log_viewer_open: bool = False + log_viewer_follow: bool = True + log_viewer_start_line: int = 0 + log_viewer_lines: List[str] = field(default_factory=list) + log_viewer_signature: Any = None + log_viewer_loaded: bool = False oldest_message_rowid: Dict[Union[str, int], int] = field(default_factory=dict) has_older_messages: Dict[Union[str, int], bool] = field(default_factory=dict) message_line_ranges: Dict[Union[str, int], List[tuple]] = field(default_factory=dict) diff --git a/contact/ui/user_config.py b/contact/ui/user_config.py index b0336c7..933db50 100644 --- a/contact/ui/user_config.py +++ b/contact/ui/user_config.py @@ -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) diff --git a/contact/utilities/interfaces.py b/contact/utilities/interfaces.py index a6952df..a4d5262 100644 --- a/contact/utilities/interfaces.py +++ b/contact/utilities/interfaces.py @@ -68,13 +68,19 @@ def initialize_interface(args): def reconnect_interface(args, attempts: int = 20, delay_seconds: float = 1.0): last_error = None + target = getattr(args, "host", None) or getattr(args, "port", None) or getattr(args, "ble", None) or "auto" for attempt in range(attempts): + attempt_number = attempt + 1 + logging.info("Reconnect attempt %d/%d to %s", attempt_number, attempts, target) try: interface = initialize_interface(args) if interface_is_connected(interface) and getattr(interface, "localNode", None) is not None and getattr( interface.localNode, "localConfig", None ) is not None: + node_num = getattr(interface.localNode, "nodeNum", None) + node_label = f" !{node_num:08x}" if isinstance(node_num, int) else "" + logging.info("Reconnected to Meshtastic node%s on attempt %d/%d", node_label, attempt_number, attempts) return interface last_error = RuntimeError("interface did not complete connection setup") try: @@ -84,7 +90,10 @@ def reconnect_interface(args, attempts: int = 20, delay_seconds: float = 1.0): except Exception as ex: last_error = ex + logging.warning("Reconnect attempt %d/%d failed: %s", attempt_number, attempts, last_error) + if attempt < attempts - 1: time.sleep(delay_seconds) + logging.error("Unable to reconnect to Meshtastic node after %d attempts", attempts) raise RuntimeError("Failed to reconnect to the Meshtastic node") from last_error diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py index e594329..0b4491c 100644 --- a/tests/test_interfaces.py +++ b/tests/test_interfaces.py @@ -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() diff --git a/tests/test_user_config.py b/tests/test_user_config.py index 4f232a4..c76d1df 100644 --- a/tests/test_user_config.py +++ b/tests/test_user_config.py @@ -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()