Merge pull request #298 from pdxlocations:log-viewer

Add log viewer functionality and enhance reconnect handling

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", ""

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 ''}"

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)

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)

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)

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

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()

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()