Compare commits

..
Author SHA1 Message Date
pdxlocations ac23c2bbbb Merge remote-tracking branch 'origin/main' into remote-admin
# Conflicts:
#	contact/ui/contact_ui.py
#	contact/ui/control_ui.py
2026-07-25 21:55:13 -07:00
pdxlocations e378aba526 Refactor remote admin handling to improve status updates and timeout management 2026-07-25 21:53:36 -07:00
pdxlocations ec35cbd3e7 Merge pull request #294 from pdxlocations:reconnect-01
Reconnect after disconnect
2026-07-25 21:47:05 -07:00
pdxlocations 7f6fa49385 Enhance reconnect logic and error handling in UI and interface modules
- Implement automatic reconnection attempts in the main UI when disconnected.
- Add a non-blocking connection status dialog to inform users during reconnection.
- Improve error handling when sending messages and during interface reconnections.
- Update the reconnect_interface function to increase the number of attempts and refine connection checks.
2026-07-25 21:46:27 -07:00
pdxlocations 4ed3cbf370 bump version to 1.6.2 in pyproject.toml 2026-07-25 14:31:22 -07:00
pdxlocations 7afc025b9d Refactor settings menu to improve channel request handling and status updates 2026-07-25 14:30:58 -07:00
pdxlocations ece37f2a36 bump version to 1.6.1 in pyproject.toml 2026-07-24 23:40:07 -07:00
pdxlocations 0cee9777bc Improve sound playback handling by adding support for multiple audio players and enhancing error logging 2026-07-24 23:39:46 -07:00
pdxlocations 72956321da bump version to 1.6.0 in pyproject.toml 2026-07-24 23:16:02 -07:00
pdxlocations 6a61f7dcd5 Merge pull request #290 from pdxlocations:notifications-2
Add notification sound feature and update configuration
2026-07-24 23:12:45 -07:00
pdxlocations b1b9f22203 Add notification sound feature and update configuration 2026-07-24 23:12:21 -07:00
pdxlocations f63c27e9ad typo 2026-07-24 22:57:30 -07:00
pdxlocations 8ea5646f6c Merge pull request #284 from pdxlocations:update-en.ini-with-2.8-fields
Add new configuration options for device UI and modules
2026-07-24 22:56:27 -07:00
pdxlocations 41d703ad9b Merge pull request #289 from pdxlocations:remote-admin
Remote-admin
2026-07-24 22:52:43 -07:00
pdxlocations f4bd0c8614 add remote admin 2026-07-24 22:50:39 -07:00
pdxlocations 68b5d81c1f Enhance settings menu and remote admin handling
- Refactor settings_menu to support remote node configurations.
- Add show_remote_admin_wait function for displaying progress during remote admin requests.
- Update save_changes to accept a node parameter for better flexibility.
- Improve error handling for remote admin rejections in handle_backtick.
- Add unit tests for new remote admin functionalities and error scenarios.
2026-07-24 15:47:06 -07:00
pdxlocations 61bbc24d29 Merge pull request #287 from pdxlocations:display-bug
Clear stale reply context in draw_text_field function and add corresponding test
2026-07-24 15:25:38 -07:00
pdxlocations 7273e80639 Add new configuration options for device UI and modules 2026-07-24 09:55:53 -07:00
15 changed files with 482 additions and 76 deletions

No files matched your search

+8
View File
@@ -17,6 +17,14 @@
"cwd": "${workspaceFolder}",
"module": "contact.__main__",
"args": ["--host","192.168.86.69"]
},
{
"name": "Python Debugger: tcp-remote",
"type": "debugpy",
"request": "launch",
"cwd": "${workspaceFolder}",
"module": "contact.__main__",
"args": ["--host","100.89.187.29"]
}
]
}
+8
View File
@@ -45,6 +45,14 @@ All messages will saved in a SQLite DB and restored upon relaunch of the app. Y
By navigating to Settings -> App Settings, you may customize your UI's icons, colors, and more!
### Notification sounds
Add audio files to Contact's `contact/sounds/` folder, then choose one in Settings → App Settings → Notification sound. `alert.mp3` is the default selection; choose `None` to disable notification audio. Supported file types are MP3, WAV, OGG, AIFF, and FLAC.
### Remote administration
With the Nodes pane selected, press `` ` `` or `F12` and choose **Remote admin** for the highlighted node. Contact requests the node's settings through Meshtastic Remote Admin and displays an error if the node does not authorize your public key. Remote App Settings are intentionally unavailable because they apply only to the local Contact client.
For smaller displays you may wish to enable `single_pane_mode`:
<img width="486" height="194" alt="Screenshot 2025-08-22 at 11 15 54 PM" src="https://github.com/user-attachments/assets/447c5d30-0850-4a4f-b0d4-976e4c5e329d" />
+66 -2
View File
@@ -112,7 +112,7 @@ language, "Language", "UI language for labels and help text."
message_prefix, "Message prefix", ""
sent_message_prefix, "Sent message prefix", ""
notification_symbol, "Notification symbol", ""
notification_sound, "Notification sound", ""
notification_sound, "Notification sound", "Select a sound file from Contact's sounds folder, or None to disable notification audio."
ack_implicit_str, "ACK (implicit)", ""
ack_str, "ACK", ""
nak_str, "NAK", ""
@@ -252,6 +252,8 @@ heading_bold, "Heading bold", "The heading can be hard to read when 'INVERTED' o
wake_on_tap_or_motion, "Wake on tap or motion", "This option enables the ability to wake the device screen when motion, such as a tap on the device, is detected via an attached accelerometer, or a capacitive touch button."
compass_orientation, "Compass orientation", "Whether to rotate the compass."
use_12h_clock, "Use 12 hour clock"
use_long_node_name, "Use long node name", "Show each node's long name on the device display when space permits."
enable_message_bubbles, "Enable message bubbles", "Display messages in chat-style bubbles on supported device displays."
[config.device_ui]
title, "Device UI"
@@ -270,6 +272,10 @@ node_filter, "Node Filter", ""
node_highlight, "Node Highlight", ""
calibration_data, "Calibration Data", ""
map_data, "Map Data", ""
compass_mode, "Compass mode", "Controls whether the compass is dynamic, fixed, or freezes its heading."
screen_rgb_color, "Screen RGB color", "Sets the display color on supported RGB screens."
is_clockface_analog, "Analog clock face", "Use an analog clock face on supported device displays."
gps_format, "GPS coordinate format", "Select the coordinate format used by the device UI."
[config.device_ui.node_filter]
title, "Node Filter"
@@ -295,6 +301,12 @@ home, "Home", ""
style, "Style", ""
follow_gps, "Follow GPS", ""
[config.device_ui.map_data.home]
title, "Home location", ""
zoom, "Zoom", "Map zoom level for the home location."
latitude, "Latitude", "Latitude of the home location."
longitude, "Longitude", "Longitude of the home location."
[config.lora]
title, "LoRa"
use_preset, "Use modem preset", "Presets are pre-defined modem settings (Bandwidth, Spread Factor, and Coding Rate) which influence both message speed and range. The vast majority of users use a preset."
@@ -314,6 +326,8 @@ override_frequency, "Override frequency in MHz", "Overrides frequency slot. May
pa_fan_disabled, "PA Fan Disabled", "If true, disable the build-in PA FAN using pin define in RF95_FAN_EN"
ignore_mqtt, "Ignore MQTT", "Ignores any messages it receives via LoRa that came via MQTT somewhere along the path towards the device."
config_ok_to_mqtt, "OK to MQTT", "Indicates that the user approves their packets to be uplinked to MQTT brokers."
fem_lna_mode, "FEM LNA mode", "Configures the front-end module low-noise amplifier mode on supported hardware."
serial_hal_only, "Serial HAL only", "Use only the serial hardware abstraction layer on supported hardware."
[config.bluetooth]
title, "Bluetooth"
@@ -330,6 +344,7 @@ serial_enabled, "Enable serial console", ""
debug_log_api_enabled, "Enable debug log", "Set this to true to continue outputting live debug logs over serial or Bluetooth when the API is active."
admin_channel_enabled, "Enable legacy admin channel", "If the node you need to administer or be administered by is running 2.4.x or earlier, you should set this to enabled. Requires a secondary channel named 'admin' be present on both nodes."
admin_key, "Admin keys", "The public key(s) authorized to send administrative messages to this node. Only messages signed by these keys will be accepted for administrative control. Up to 3."
packet_signature_policy, "Packet signature policy", "Controls how the node validates packet signatures."
[module.mqtt]
title, "MQTT"
@@ -392,6 +407,7 @@ title, "Range Test"
enabled, "Enabled", "Enables the module."
sender, "Sender interval", "How long to wait between sending sequential test packets in seconds. 0 is default which disables sending messages."
save, "Save CSV file", "If enabled, all received messages are saved to the device's flash memory in a file named rangetest.csv. Leave disabled when using the Android or Apple apps. Saves directly to the device's flash memory (without the need for a smartphone). [warning]Only available on ESP32-based devices.[/warning]"
clear_on_reboot, "Clear CSV on reboot", "Delete the RangeTest.csv file when the device reboots. [warning]Only available on ESP32-based devices.[/warning]"
[module.telemetry]
title, "Telemetry"
@@ -409,6 +425,7 @@ health_measurement_enabled, "Health telemetry interval", "This option is used to
health_update_interval, "Health telemetry enabled", "This option is used to enable/disable the sending of health data from an attached supported sensor over the mesh network."
health_screen_enabled, "Health screen enabled", "Show the health telemetry data on the device display."
device_telemetry_enabled, "Device telemetry enabled", "Enable the Device Telemetry"
air_quality_screen_enabled, "Air quality screen enabled", "Show air-quality telemetry on the device display."
[module.canned_message]
title, "Canned Message"
@@ -470,5 +487,52 @@ use_pullup, "Use pull-up", "Whether or not use INPUT_PULLUP mode for GPIO pin. O
title, "Paxcounter"
enabled, "Enabled", "Enables the module."
paxcounter_update_interval, "Update interval", "The interval in seconds of how often we can send a message to the mesh when a state change is detected."
Wi-Fi_threshold, "Wi-Fi Threshold", "WiFi RSSI threshold. Defaults to -80"
wifi_threshold, "Wi-Fi threshold", "Wi-Fi RSSI threshold. Defaults to -80."
ble_threshold, "BLE Threshold", "BLE RSSI threshold. Defaults to -80"
[module.statusmessage]
title, "Status Message"
node_status, "Node status", ""
[module.traffic_management]
title, "Traffic Management"
position_min_interval_secs, "Position minimum interval", "Minimum seconds between position updates from the same node. Set to 0 to disable position-update suppression."
nodeinfo_direct_response_max_hops, "NodeInfo direct-response maximum hops", "Maximum hop distance at which cached NodeInfo requests receive a direct response. Set to 0 to disable direct responses."
rate_limit_window_secs, "Rate-limit window", "Per-node rate-limit window in seconds. Set to 0 to disable rate limiting."
rate_limit_max_packets, "Rate-limit maximum packets", "Maximum packets allowed per node during the rate-limit window. Set to 0 to disable rate limiting."
unknown_packet_threshold, "Unknown-packet threshold", "Maximum unknown or undecryptable packets per rate-limit window before packets from that source are dropped. Set to 0 to disable filtering."
[module.tak]
title, "Tak"
team, "Team", "Your team color."
role, "Role", "Your role in the team."
[module.mesh_beacon]
title, "Mesh Beacon", ""
flags, "Beacon options", "Bitwise combination of the listen, broadcast, and legacy-split options."
broadcast_send_as_node, "Broadcast as node", "Optional node ID to use as the sender of beacon packets. Leave unset to send as this node."
broadcast_message, "Broadcast message", "Message included in each beacon broadcast. Maximum 100 bytes."
broadcast_offer_channel, "Offered channel", "Optional channel name and key to advertise in the beacon offer."
broadcast_offer_region, "Offered region", "Optional LoRa region to advertise in the beacon offer."
broadcast_offer_preset, "Offered modem preset", "Optional modem preset to advertise in the beacon offer."
broadcast_on_channel, "Broadcast channel", "Optional channel settings to use when transmitting beacons. Uses the primary channel when unset."
broadcast_on_region, "Broadcast region", "Optional LoRa region to use when transmitting beacons."
broadcast_on_preset, "Broadcast modem preset", "Optional modem preset to use when transmitting beacons."
broadcast_interval_secs, "Broadcast interval", "How often to broadcast beacons in seconds. The minimum and default is 3600 seconds."
broadcast_targets, "Broadcast targets", "Optional list of channel, region, and modem-preset targets for beacon broadcasts."
[module.mesh_beacon.broadcast_offer_channel]
title, "Offered channel", ""
psk, "PSK", "Encryption key for the offered channel."
name, "Name", "Name of the offered channel."
[module.mesh_beacon.broadcast_on_channel]
title, "Broadcast channel", ""
psk, "PSK", "Encryption key for the broadcast channel."
name, "Name", "Name of the broadcast channel."
[module.mesh_beacon.broadcast_targets]
title, "Broadcast targets", ""
preset, "Modem preset", "Optional modem preset for this beacon target."
region, "Region", "LoRa region for this beacon target."
channel_index, "Channel index", "Index of the configured channel to use for this beacon target."
+21 -24
View File
@@ -66,37 +66,34 @@ from contact.message_handlers.bot_handler import bot_respond
def play_sound():
try:
system = platform.system()
sound_path = None
executable = None
selected_sound = str(getattr(config, "notification_sound", "None"))
if selected_sound.casefold() in {"none", "false", ""}:
return
sound_path = os.path.join(config.parent_dir, "sounds", selected_sound)
if not os.path.isfile(sound_path):
logging.warning("Configured notification sound is unavailable: %s", sound_path)
return
if system == "Darwin": # macOS
sound_path = "/System/Library/Sounds/Ping.aiff"
executable = "afplay"
elif system == "Linux":
ogg_path = "/usr/share/sounds/freedesktop/stereo/complete.oga"
wav_path = "/usr/share/sounds/alsa/Front_Center.wav" # common fallback
if shutil.which("paplay") and os.path.exists(ogg_path):
executable = "paplay"
sound_path = ogg_path
elif shutil.which("ffplay") and os.path.exists(ogg_path):
executable = "ffplay"
sound_path = ogg_path
elif shutil.which("aplay") and os.path.exists(wav_path):
executable = "aplay"
sound_path = wav_path
else:
logging.warning("No suitable sound player or sound file found on Linux")
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; install ffmpeg (ffplay), mpg123, mpv, or VLC."
)
except subprocess.CalledProcessError as e:
logging.error(f"Sound playback failed: {e}")
@@ -144,7 +141,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
hops = hop_start - hop_limit
if config.notification_sound == "True":
if str(config.notification_sound).casefold() not in {"none", "false", ""}:
schedule_notification_sound()
message_bytes = packet["decoded"]["payload"]
Binary file not shown.
+165 -7
View File
@@ -1,5 +1,6 @@
import curses
import logging
from typing import Optional
import time
import traceback
from numbers import Real
@@ -383,6 +384,41 @@ def main_ui(stdscr: curses.window) -> None:
handle_resize(stdscr, True)
while True:
interface = interface_state.interface
if (
hasattr(interface, "stream")
and interface.stream is None
and not ui_state.reconnect_attempted
):
ui_state.reconnect_attempted = True
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
handle_resize(stdscr, False)
except Exception:
if status_win is not None:
status_win.erase()
status_win.refresh()
handle_resize(stdscr, False)
logging.exception("Automatic reconnect after transport disconnect failed")
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)
with app_state.lock:
process_pending_ui_updates(stdscr)
entry_display = f"{ui_state.reply_context}{input_text or ''}"
@@ -652,12 +688,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
@@ -938,7 +989,65 @@ def handle_backtick(stdscr: curses.window) -> None:
curses.curs_set(0)
previous_window = ui_state.current_window
ui_state.current_window = 4
settings_menu(stdscr, interface_state.interface)
interface = interface_state.interface
if previous_window != 2:
settings_menu(stdscr, interface)
ui_state.current_window = previous_window
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
curses.curs_set(1)
get_channels()
refresh_node_list()
handle_resize(stdscr, False)
return
options = ["Local settings"]
remote_node_num = None
if ui_state.node_list:
remote_node_num = ui_state.node_list[ui_state.selected_node]
options.append(f"Remote admin: {get_name_from_database(remote_node_num)}")
choice = get_list_input("Open settings for", None, options)
if choice == "Local settings":
settings_menu(stdscr, interface)
elif remote_node_num is not None:
wait_win = None
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)
try:
settings_menu(
stdscr,
interface,
# Settings requests configs explicitly; avoid getNode's
# implicit channel request racing that setup.
node=interface.getNode(remote_node_num, False),
remote=True,
status_callback=remote_status,
)
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))
ui_state.current_window = previous_window
ui_state.single_pane_mode = config.single_pane_mode.lower() == "true"
curses.curs_set(1)
@@ -947,6 +1056,55 @@ 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:
"""Show progress while Meshtastic retrieves remote-admin settings."""
message = f"Attempting remote admin of {node_name}, waiting for response..."
try:
height, width = stdscr.getmaxyx()
box_width = min(width - 4, max(len(message) + 4, 36))
box_height = 5
wait_win = curses.newwin(box_height, box_width, max(0, (height - box_height) // 2), max(0, (width - box_width) // 2))
wait_win.bkgd(get_color("background"))
wait_win.attrset(get_color("window_frame"))
wait_win.border()
wait_win.addstr(0, 2, " Remote Admin ", get_color("settings_default"))
wait_win.addstr(2, 2, message[: box_width - 4], get_color("settings_default"))
wait_win.refresh()
return wait_win
except curses.error:
return None
def show_connection_status(stdscr: curses.window, title: str, message: str) -> curses.window | None:
"""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: curses.window | None, message: str) -> None:
"""Append the current remote-admin request to the waiting overlay."""
if wait_win is None:
return
try:
_, width = wait_win.getmaxyx()
wait_win.addstr(3, 2, " " * (width - 4), get_color("settings_default"))
wait_win.addstr(3, 2, message[: width - 4], get_color("settings_default"))
wait_win.refresh()
except curses.error:
pass
def handle_ctrl_p() -> None:
"""Handle Ctrl + P key events to toggle the packet log display."""
# Display packet log
+114 -20
View File
@@ -4,6 +4,8 @@ import ipaddress
import logging
import os
import sys
import threading
import time
from typing import List
from meshtastic.protobuf import admin_pb2
@@ -25,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
@@ -34,6 +35,8 @@ from contact.utilities.singleton import interface_state, menu_state
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
save_option = "Save Changes"
max_help_lines = 0
REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS = 15
admin_target_label = ""
help_win = None
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
@@ -86,7 +89,8 @@ def get_translated_header(menu_path: List[str]) -> str:
continue
full_key = ".".join(transformed_path[:idx])
translated_parts.append(field_mapping.get(full_key, part))
return " > ".join(translated_parts)
breadcrumb = " > ".join(translated_parts)
return f"{admin_target_label} | {breadcrumb}" if admin_target_label else breadcrumb
def display_menu() -> tuple[object, object]:
@@ -232,9 +236,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)
@@ -279,10 +280,80 @@ def redraw_main_ui_after_reconnect(stdscr: object) -> None:
logging.debug("Skipping main UI redraw after reconnect", exc_info=True)
def settings_menu(stdscr: object, interface: object) -> None:
curses.update_lines_cols()
def _request_remote_with_timeout(request, *args) -> None:
"""Run a Meshtastic remote request without allowing its ACK wait to hang the UI."""
failure = []
menu = generate_menu_from_protobuf(interface)
def run_request() -> None:
try:
request(*args)
except BaseException as exc:
failure.append(exc)
worker = threading.Thread(target=run_request, name="remote-admin-request", daemon=True)
worker.start()
worker.join(REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS)
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) -> 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 time.monotonic() >= deadline:
raise TimeoutError("Timed out waiting for the remote node's channel list.")
time.sleep(0.1)
def _request_remote_section(node: object, menu_path: List[str], selected_option: str, status_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)
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)
return True
def settings_menu(
stdscr: object, interface: object, node: object = None, remote: bool = False, status_callback=None
) -> None:
curses.update_lines_cols()
global admin_target_label
node = node or interface.localNode
node_num = getattr(node, "nodeNum", 0)
node_info = interface.getMyNodeInfo() if not remote else getattr(interface, "nodesByNum", {}).get(node_num, {})
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}"
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"]
@@ -387,12 +458,14 @@ def settings_menu(stdscr: object, interface: object) -> None:
help_win.refresh()
if menu_state.show_save_option and menu_state.selected_index == len(options):
reconnect_required = save_changes(interface, modified_settings, menu_state)
reconnect_required = save_changes(interface, modified_settings, menu_state, node=node)
modified_settings.clear()
logging.info("Changes Saved")
if remote:
break
if reconnect_required:
interface = reconnect_interface_with_splash(stdscr, interface)
menu = generate_menu_from_protobuf(interface)
menu = generate_menu_from_protobuf(interface, node=node, include_app_settings=not remote)
if len(menu_state.menu_path) > 1:
menu_state.menu_path.pop()
@@ -492,7 +565,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
continue
elif selected_option == "Config URL":
current_value = interface.localNode.getURL()
current_value = node.getURL()
new_value = get_text_input(
t(
"ui.prompt.config_url_current",
@@ -510,7 +583,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
["Yes", "No"],
)
if overwrite == "Yes":
interface.localNode.setURL(new_value)
node.setURL(new_value)
logging.info(f"New Config URL sent to node")
menu_state.start_index.pop()
continue
@@ -521,7 +594,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
if confirmation == "Yes":
interface = reconnect_after_admin_action(
stdscr, interface, interface.localNode.reboot, "Node Reboot Requested by menu"
stdscr, interface, node.reboot, "Node Reboot Requested by menu"
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
@@ -535,7 +608,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
if confirmation == "Yes":
interface = reconnect_after_admin_action(
stdscr, interface, interface.localNode.resetNodeDb, "Node DB Reset Requested by menu"
stdscr, interface, node.resetNodeDb, "Node DB Reset Requested by menu"
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
@@ -546,7 +619,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
t("ui.confirm.shutdown", default="Are you sure you want to Shutdown?"), None, ["Yes", "No"]
)
if confirmation == "Yes":
interface.localNode.shutdown()
node.shutdown()
logging.info(f"Node Shutdown Requested by menu")
menu_state.start_index.pop()
continue
@@ -561,7 +634,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
interface = reconnect_after_admin_action(
stdscr,
interface,
lambda: request_factory_reset(interface.localNode, full=True),
lambda: request_factory_reset(node, full=True),
"Factory Reset Requested by menu",
)
menu = rebuild_menu_at_current_path(interface, menu_state)
@@ -578,7 +651,7 @@ def settings_menu(stdscr: object, interface: object) -> None:
interface = reconnect_after_admin_action(
stdscr,
interface,
lambda: request_factory_reset(interface.localNode, full=False),
lambda: request_factory_reset(node, full=False),
"Factory Reset Config Requested by menu",
)
menu = rebuild_menu_at_current_path(interface, menu_state)
@@ -613,8 +686,8 @@ def settings_menu(stdscr: object, interface: object) -> None:
is_ringtone = selected_option == "ringtone"
getter_name = "get_ringtone" if is_ringtone else "get_canned_message"
setter_name = "set_ringtone" if is_ringtone else "set_canned_message"
getter = getattr(interface.localNode, getter_name, None)
setter = getattr(interface.localNode, setter_name, None)
getter = getattr(node, getter_name, None)
setter = getattr(node, setter_name, None)
fetched_value = getter() if callable(getter) else None
current_value = fetched_value if fetched_value is not None else current_value
new_value = get_text_input(
@@ -735,7 +808,28 @@ def settings_menu(stdscr: object, interface: object) -> None:
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
)
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
+10 -1
View File
@@ -235,7 +235,7 @@ def initialize_config() -> Dict[str, object]:
"message_prefix": ">>",
"sent_message_prefix": ">> Sent",
"notification_symbol": "*",
"notification_sound": "True",
"notification_sound": "alert.mp3",
"ack_implicit_str": "[◌]",
"ack_str": "[✓]",
"nak_str": "[x]",
@@ -264,6 +264,15 @@ def initialize_config() -> Dict[str, object]:
# Check and add missing variables
updated = update_dict(default_config_variables, loaded_config)
# Migrate the legacy boolean notification preference to the file-based
# sound selection introduced in 1.5.12.
if loaded_config.get("notification_sound") == "True":
loaded_config["notification_sound"] = "alert.mp3"
updated = True
elif loaded_config.get("notification_sound") == "False":
loaded_config["notification_sound"] = "None"
updated = True
# Update the JSON file if any variables were missing
if updated:
formatted_json = format_json_single_line_arrays(loaded_config)
+18 -14
View File
@@ -57,14 +57,17 @@ def extract_fields(
return menu
def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
def generate_menu_from_protobuf(interface: object, node: Any = None, include_app_settings: bool = True) -> Dict[str, Any]:
"""
Builds the full settings menu structure from the protobuf definitions.
"""
menu_structure = {"Main Menu": {}}
# Add User Settings
current_node_info = interface.getMyNodeInfo() if interface else None
node = node or (interface.localNode if interface else None)
current_node_info = interface.getMyNodeInfo() if interface and node is interface.localNode else None
if current_node_info is None and interface and node is not None:
current_node_info = getattr(interface, "nodesByNum", {}).get(getattr(node, "nodeNum", None))
if current_node_info:
current_user_config = current_node_info.get("user", None)
@@ -79,28 +82,28 @@ def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
menu_structure["Main Menu"]["User Settings"] = "No user settings available"
else:
logging.info("Node Info not available")
menu_structure["Main Menu"]["User Settings"] = "Node Info not available"
menu_structure["Main Menu"]["User Settings"] = {}
# Add Channels
channel = channel_pb2.ChannelSettings()
menu_structure["Main Menu"]["Channels"] = {}
if interface:
if node:
for i in range(8):
current_channel = interface.localNode.getChannelByChannelIndex(i)
current_channel = node.getChannelByChannelIndex(i)
if current_channel:
channel_config = extract_fields(channel, current_channel.settings)
menu_structure["Main Menu"]["Channels"][f"Channel {i + 1}"] = channel_config
# Add Radio Settings
radio = config_pb2.Config()
current_radio_config = interface.localNode.localConfig if interface else None
current_radio_config = node.localConfig if node else None
menu_structure["Main Menu"]["Radio Settings"] = extract_fields(radio, current_radio_config)
# Add Lat/Lon/Alt
position_data = {
"latitude": (None, current_node_info["position"].get("latitude", 0.0)),
"longitude": (None, current_node_info["position"].get("longitude", 0.0)),
"altitude": (None, current_node_info["position"].get("altitude", 0)),
"latitude": (None, (current_node_info or {}).get("position", {}).get("latitude", 0.0)),
"longitude": (None, (current_node_info or {}).get("position", {}).get("longitude", 0.0)),
"altitude": (None, (current_node_info or {}).get("position", {}).get("altitude", 0)),
}
existing_position_menu = menu_structure["Main Menu"]["Radio Settings"].get("position", {})
@@ -117,23 +120,24 @@ def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
# Add Module Settings
module = module_config_pb2.ModuleConfig()
current_module_config = interface.localNode.moduleConfig if interface else None
current_module_config = node.moduleConfig if node else None
module_settings = extract_fields(module, current_module_config)
if interface and interface.localNode:
if node:
# These values use dedicated admin requests rather than ModuleConfig
# fields, so expose them alongside their related module settings.
module_settings.setdefault("external_notification", {})["ringtone"] = (
None,
getattr(interface.localNode, "ringtone", "") or "",
getattr(node, "ringtone", "") or "",
)
module_settings.setdefault("canned_message", {})["messages"] = (
None,
getattr(interface.localNode, "cannedPluginMessage", "") or "",
getattr(node, "cannedPluginMessage", "") or "",
)
menu_structure["Main Menu"]["Module Settings"] = module_settings
# Add App Settings
menu_structure["Main Menu"]["App Settings"] = {"Open": "app_settings"}
if include_app_settings:
menu_structure["Main Menu"]["App Settings"] = {"Open": "app_settings"}
# Additional settings options
menu_structure["Main Menu"].update(
+1
View File
@@ -39,6 +39,7 @@ class ChatUIState:
redraw_full_ui: bool = False
scroll_messages_to_bottom: bool = False
preserve_message_selection: bool = False
reconnect_attempted: 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)
+15 -1
View File
@@ -115,9 +115,23 @@ def edit_color_pair(key: str, display_label: str, current_value: List[str]) -> L
def edit_value(key: str, display_label: str, current_value: str) -> str:
if key in ("notification_sound", "single_pane_mode", "enabled"):
if key in ("single_pane_mode", "enabled"):
return get_list_input(display_label, current_value, ["True", "False"])
if key == "notification_sound":
sounds_dir = os.path.join(config.parent_dir, "sounds")
try:
sound_options = sorted(
filename for filename in os.listdir(sounds_dir)
if filename.lower().endswith((".mp3", ".wav", ".ogg", ".aiff", ".flac"))
)
except OSError:
sound_options = []
sound_options = ["None"] + sound_options
if current_value not in sound_options:
sound_options.append(current_value)
return get_list_input(display_label, current_value, sound_options)
w = get_effective_width()
height = 10
input_width = w - 16 # Allow space for "New Value: "
+9 -3
View File
@@ -44,15 +44,21 @@ 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
for attempt in range(attempts):
try:
interface = initialize_interface(args)
if interface is not None:
if interface is not None and getattr(interface, "localNode", None) is not None and getattr(
interface.localNode, "localConfig", None
) is not None and (not hasattr(interface, "stream") or interface.stream is not None):
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
+3 -3
View File
@@ -78,7 +78,7 @@ def _requires_reconnect(menu_state, modified_settings) -> bool:
return True
def save_changes(interface, modified_settings, menu_state):
def save_changes(interface, modified_settings, menu_state, node=None):
"""
Save changes to the device based on modified settings.
:param interface: Meshtastic interface instance
@@ -90,7 +90,7 @@ def save_changes(interface, modified_settings, menu_state):
logging.info("No changes to save. modified_settings is empty.")
return False
node = interface.getNode("^local")
node = node or interface.getNode("^local")
admin_key_backup = None
if "admin_key" in modified_settings:
# Get reference to security config
@@ -134,7 +134,7 @@ def save_changes(interface, modified_settings, menu_state):
lon = float(modified_settings.get("longitude", 0.0))
alt = int(modified_settings.get("altitude", 0))
interface.localNode.setFixedPosition(lat, lon, alt)
node.setFixedPosition(lat, lon, alt)
logging.info(f"Updated {config_category} with Latitude: {lat} and Longitude {lon} and Altitude {alt}")
return False
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.5.13"
version = "1.6.2"
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"}
+43
View File
@@ -39,6 +39,49 @@ class ContactUiTests(unittest.TestCase):
self.assertEqual(curs_set.call_args_list[-1].args, (1,))
self.assertEqual(ui_state.current_window, 1)
def test_handle_backtick_shows_error_when_remote_admin_is_rejected(self) -> None:
stdscr = mock.Mock()
ui_state.current_window = 2
ui_state.node_list = [123]
ui_state.selected_node = 0
contact_ui.interface_state.interface = mock.Mock()
contact_ui.interface_state.interface.getNode.return_value = mock.Mock()
with mock.patch.object(contact_ui.curses, "curs_set"):
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"):
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)
dialog.assert_called_once_with(
"Remote admin rejected",
"The selected node did not authorize this admin request.",
)
def test_show_remote_admin_wait_draws_status(self) -> None:
stdscr = mock.Mock()
stdscr.getmaxyx.return_value = (24, 100)
wait_win = mock.Mock()
with mock.patch.object(contact_ui.curses, "newwin", return_value=wait_win):
result = contact_ui.show_remote_admin_wait(stdscr, "Remote")
self.assertIs(result, wait_win)
self.assertIn("Attempting remote admin of Remote", wait_win.addstr.call_args_list[-1].args[2])
wait_win.refresh.assert_called_once()
def test_update_remote_admin_wait_replaces_status_line(self) -> None:
wait_win = mock.Mock()
wait_win.getmaxyx.return_value = (5, 40)
contact_ui.update_remote_admin_wait(wait_win, "Requesting lora config…")
self.assertIn("Requesting lora config", wait_win.addstr.call_args_list[-1].args[2])
def test_process_pending_ui_updates_draws_requested_windows(self) -> None:
stdscr = mock.Mock()
ui_state.redraw_channels = True