Compare commits

..

22 Commits
tests ... 1.5.6

Author SHA1 Message Date
pdxlocations
bc69c709ff Bump version to 1.5.6 in pyproject.toml 2026-03-22 21:37:12 -07:00
pdxlocations
6453865cba Merge pull request #262 from SpudGunMan:main
refactor exit
2026-03-22 21:36:27 -07:00
pdxlocations
090e89d92c Merge pull request #263 from pdxlocations:add-french-language
Add French localization file for user interface and settings
2026-03-22 21:16:15 -07:00
pdxlocations
07716c1719 Add French localization file for user interface and settings 2026-03-22 21:15:58 -07:00
Kelly
da5b102047 Update __main__.py 2026-03-22 14:44:22 -07:00
pdxlocations
1668b68c4f Bump version to 1.5.5 in pyproject.toml 2026-03-21 22:08:57 -07:00
pdxlocations
fd4b9e2174 Add tests for handling interface absence and keyboard interrupts in start function 2026-03-21 22:08:41 -07:00
pdxlocations
8f32e2c99c Merge pull request #260 from pdxlocations:reset
Add factory reset config option and tryfix factory reset
2026-03-21 21:28:29 -07:00
pdxlocations
b53dab1840 Add factory reset config option and tryfix factory reset 2026-03-21 21:28:15 -07:00
pdxlocations
f2904eb550 Merge pull request #259 from pdxlocations:fix-shutdown
Fix interface shutdown handling
2026-03-21 21:14:38 -07:00
pdxlocations
480c32ba56 try fix shutdown 2026-03-21 21:13:53 -07:00
pdxlocations
b4b084b627 bump version to 1.5.4 in pyproject.toml 2026-03-19 15:47:51 -07:00
pdxlocations
5940c9b02b fix content margins 2026-03-19 15:19:24 -07:00
pdxlocations
c492c96685 bump version to 1.5.3 in pyproject.toml 2026-03-19 14:37:29 -07:00
pdxlocations
90376d35f3 Single pane mode fix 2026-03-19 14:37:14 -07:00
pdxlocations
4b35a74e2c bump version to 1.5.2 in pyproject.toml 2026-03-19 14:07:16 -07:00
pdxlocations
ecc5308dad Refactor UI redraw handling and improve message drawing logic 2026-03-19 14:06:01 -07:00
pdxlocations
8f376edabe bump version to 1.5.1 in pyproject.toml 2026-03-19 11:41:08 -07:00
pdxlocations
e5ef87ec19 Merge pull request #258 from pdxlocations:reconnect
Reconnect after config changes
2026-03-19 11:40:41 -07:00
pdxlocations
1b6d269d50 Reconnect after config changes 2026-03-19 11:40:24 -07:00
pdxlocations
1d95dae536 Merge pull request #257 from pdxlocations:tests
Add tests
2026-03-19 11:05:03 -07:00
pdxlocations
02b4866a38 Update README.md 2026-03-18 22:58:41 -07:00
28 changed files with 1543 additions and 223 deletions

View File

@@ -16,8 +16,7 @@ pip install contact
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.
<img width="846" alt="Contact - Main UI Screenshot" src="https://github.com/user-attachments/assets/d2996bfb-2c6d-46a8-b820-92a9143375f4">
<img width="991" height="516" alt="Contact - Main UI Screenshot" src="https://github.com/user-attachments/assets/76722145-e8a4-4f01-8898-f4ae794b5d7b" />
<br><br>
The settings dialogue can be accessed within the client or may be run standalone to configure your node by launching `contact --settings` or `contact -c`

View File

@@ -19,6 +19,7 @@ import subprocess
import sys
import threading
import traceback
from typing import Optional
# Third-party
from pubsub import pub
@@ -36,7 +37,7 @@ from contact.utilities.demo_data import build_demo_interface, configure_demo_dat
from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
from contact.ui.dialog import dialog
from contact.utilities.interfaces import initialize_interface
from contact.utilities.interfaces import initialize_interface, reconnect_interface
from contact.utilities.utils import get_channels, get_nodeNum, get_node_list
from contact.utilities.singleton import ui_state, interface_state, app_state
@@ -56,17 +57,79 @@ logging.basicConfig(
app_state.lock = threading.Lock()
DEFAULT_CLOSE_TIMEOUT_SECONDS = 5.0
# ------------------------------------------------------------------------------
# Main Program Logic
# ------------------------------------------------------------------------------
def prompt_region_if_unset(args: object) -> None:
def prompt_region_if_unset(args: object, stdscr: Optional[curses.window] = None) -> None:
"""Prompt user to set region if it is unset."""
confirmation = get_list_input("Your region is UNSET. Set it now?", "Yes", ["Yes", "No"])
if confirmation == "Yes":
set_region(interface_state.interface)
interface_state.interface.close()
interface_state.interface = initialize_interface(args)
close_interface(interface_state.interface)
if stdscr is not None:
draw_splash(stdscr)
interface_state.interface = reconnect_interface(args)
def close_interface(interface: object, timeout_seconds: float = DEFAULT_CLOSE_TIMEOUT_SECONDS) -> bool:
if interface is None:
return True
close_errors = []
def _close_target() -> None:
try:
interface.close()
except BaseException as error: # Keep shutdown resilient even for KeyboardInterrupt/SystemExit from libraries.
close_errors.append(error)
close_thread = threading.Thread(target=_close_target, name="meshtastic-interface-close", daemon=True)
close_thread.start()
close_thread.join(timeout_seconds)
if close_thread.is_alive():
logging.warning("Timed out closing interface after %.1fs; continuing shutdown", timeout_seconds)
return False
if not close_errors:
return True
error = close_errors[0]
if isinstance(error, KeyboardInterrupt):
logging.info("Interrupted while closing interface; continuing shutdown")
return True
logging.warning("Ignoring error while closing interface: %r", error)
return True
def interface_is_ready(interface: object) -> bool:
try:
return getattr(interface, "localNode", None) is not None and interface.localNode.localConfig is not None
except Exception:
return False
def initialize_runtime_interface_with_retry(stdscr: curses.window, args: object):
while True:
interface = initialize_runtime_interface(args)
if getattr(args, "demo_screenshot", False) or interface_is_ready(interface):
return interface
choice = get_list_input(
t("ui.prompt.node_not_found", default="No node found. Retry connection?"),
"Retry",
["Retry", "Close"],
mandatory=True,
)
close_interface(interface)
if choice == "Close":
return None
draw_splash(stdscr)
def initialize_globals(seed_demo: bool = False) -> None:
@@ -117,10 +180,12 @@ def main(stdscr: curses.window) -> None:
logging.info("Initializing interface...")
with app_state.lock:
interface_state.interface = initialize_runtime_interface(args)
interface_state.interface = initialize_runtime_interface_with_retry(stdscr, args)
if interface_state.interface is None:
return
if not getattr(args, "demo_screenshot", False) and interface_state.interface.localNode.localConfig.lora.region == 0:
prompt_region_if_unset(args)
prompt_region_if_unset(args, stdscr)
initialize_globals(seed_demo=getattr(args, "demo_screenshot", False))
logging.info("Starting main UI")
@@ -167,23 +232,32 @@ def start() -> None:
setup_parser().print_help()
sys.exit(0)
interrupted = False
fatal_error = None
try:
curses.wrapper(main)
interface_state.interface.close()
except KeyboardInterrupt:
interrupted = True
logging.info("User exited with Ctrl+C")
interface_state.interface.close()
sys.exit(0)
except Exception as e:
fatal_error = e
logging.critical("Fatal error", exc_info=True)
try:
curses.endwin()
except Exception:
pass
print("Fatal error:", e)
finally:
close_interface(interface_state.interface)
if fatal_error is not None:
print("Fatal error:", fatal_error)
traceback.print_exc()
sys.exit(1)
if interrupted:
sys.exit(0)
if __name__ == "__main__":
start()

View File

@@ -12,6 +12,7 @@ Reboot, "Reboot", ""
Reset Node DB, "Reset Node DB", ""
Shutdown, "Shutdown", ""
Factory Reset, "Factory Reset", ""
factory_reset_config, "Factory Reset Config", ""
Exit, "Exit", ""
Yes, "Yes", ""
No, "No", ""
@@ -55,6 +56,7 @@ confirm.reboot, "Are you sure you want to Reboot?", ""
confirm.reset_node_db, "Are you sure you want to Reset Node DB?", ""
confirm.shutdown, "Are you sure you want to Shutdown?", ""
confirm.factory_reset, "Are you sure you want to Factory Reset?", ""
confirm.factory_reset_config, "Are you sure you want to Factory Reset Config?", ""
confirm.save_before_exit_section, "You have unsaved changes in {section}. Save before exiting?", ""
prompt.select_region, "Select your region:", ""
dialog.slow_down_title, "Slow down", ""

View File

@@ -0,0 +1,197 @@
##field_name, "Nom du champ lisible avec première lettre en majuscule", "Texte d'aide avec [warning]avertissements[/warning], [note]notes[/note], [underline]soulignements[/underline], \033[31mcodes couleur ANSI\033[0m et \nsauts de ligne."
Main Menu, "Menu principal", ""
User Settings, "Paramètres utilisateur", ""
Channels, "Canaux", ""
Radio Settings, "Paramètres radio", ""
Module Settings, "Paramètres des modules", ""
App Settings, "Paramètres de l'application", ""
Export Config File, "Exporter le fichier de configuration", ""
Load Config File, "Charger le fichier de configuration", ""
Config URL, "URL de configuration", ""
Reboot, "Redémarrer", ""
Reset Node DB, "Réinitialiser la base de données des nœuds", ""
Shutdown, "Éteindre", ""
Factory Reset, "Réinitialisation d'usine", ""
factory_reset_config, "Réinitialiser la configuration d'usine", ""
Exit, "Quitter", ""
Yes, "Oui", ""
No, "Non", ""
Cancel, "Annuler", ""
[ui]
save_changes, "Enregistrer les modifications", ""
dialog.invalid_input, "Entrée invalide", ""
prompt.enter_new_value, "Entrer une nouvelle valeur : ", ""
error.value_empty, "La valeur ne peut pas être vide.", ""
error.value_exact_length, "La valeur doit comporter exactement {length} caractères.", ""
error.value_min_length, "La valeur doit comporter au moins {length} caractères.", ""
error.value_max_length, "La valeur ne doit pas dépasser {length} caractères.", ""
error.digits_only, "Seuls les chiffres (0-9) sont autorisés.", ""
error.number_range, "Entrez un nombre entre {min_value} et {max_value}.", ""
error.float_invalid, "Doit être un nombre à virgule flottante valide.", ""
prompt.edit_admin_keys, "Modifier jusqu'à 3 clés administrateur :", ""
label.admin_key, "Clé administrateur", ""
error.admin_key_invalid, "Erreur : chaque clé doit être en Base64 valide et faire 32 octets !", ""
prompt.edit_values, "Modifier jusqu'à 3 valeurs :", ""
label.value, "Valeur", ""
prompt.enter_ip, "Entrez une adresse IP (xxx.xxx.xxx.xxx) :", ""
label.current, "Actuel", ""
label.new_value, "Nouvelle valeur", ""
label.editing, "Modification de {label}", ""
label.current_value, "Valeur actuelle :", ""
error.ip_invalid, "Adresse IP invalide. Réessayez.", ""
prompt.select_foreground_color, "Sélectionner la couleur de premier plan pour {label}", ""
prompt.select_background_color, "Sélectionner la couleur d'arrière-plan pour {label}", ""
prompt.select_value, "Sélectionner {label}", ""
confirm.save_before_exit, "Vous avez des modifications non enregistrées. Enregistrer avant de quitter ?", ""
prompt.config_filename, "Entrez un nom de fichier pour le fichier de configuration", ""
confirm.overwrite_file, "{filename} existe déjà. Écraser ?", ""
dialog.config_saved_title, "Fichier de configuration enregistré :", ""
dialog.no_config_files, " Aucun fichier de configuration trouvé. Exportez-en un d'abord.", ""
prompt.choose_config_file, "Choisissez un fichier de configuration", ""
confirm.load_config_file, "Êtes-vous sûr de vouloir charger {filename} ?", ""
prompt.config_url_current, "L'URL de configuration est actuellement : {value}", ""
confirm.load_config_url, "Êtes-vous sûr de vouloir charger cette configuration ?", ""
confirm.reboot, "Êtes-vous sûr de vouloir redémarrer ?", ""
confirm.reset_node_db, "Êtes-vous sûr de vouloir réinitialiser la base de données des nœuds ?", ""
confirm.shutdown, "Êtes-vous sûr de vouloir éteindre ?", ""
confirm.factory_reset, "Êtes-vous sûr de vouloir effectuer une réinitialisation d'usine ?", ""
confirm.factory_reset_config, "Êtes-vous sûr de vouloir réinitialiser la configuration d'usine ?", ""
confirm.save_before_exit_section, "Vous avez des modifications non enregistrées dans {section}. Enregistrer avant de quitter ?", ""
prompt.select_region, "Sélectionnez votre région :", ""
dialog.slow_down_title, "Ralentissez", ""
dialog.slow_down_body, "Veuillez attendre 2 secondes entre les messages.", ""
dialog.node_details_title, "📡 Détails du nœud : {name}", ""
dialog.traceroute_not_sent_title, "Traceroute non envoyé", ""
dialog.traceroute_not_sent_body, "Veuillez attendre {seconds} secondes avant d'envoyer un autre traceroute.", ""
dialog.traceroute_sent_title, "Traceroute envoyé à : {name}", ""
dialog.traceroute_sent_body, "Les résultats apparaîtront dans la fenêtre des messages.", ""
dialog.help_title, "Aide - Raccourcis clavier", ""
help.scroll, "Haut/Bas = Défilement", ""
help.switch_window, "Gauche/Droite = Changer de fenêtre", ""
help.jump_windows, "F1/F2/F3 = Aller à Canal/Messages/Nœuds", ""
help.enter, "ENTRÉE = Envoyer / Sélectionner", ""
help.settings, "` ou F12 = Paramètres", ""
help.quit, "ESC = Quitter", ""
help.packet_log, "Ctrl+P = Activer/désactiver le journal des paquets", ""
help.traceroute, "Ctrl+T ou F4 = Traceroute", ""
help.node_info, "F5 = Informations complètes du nœud", ""
help.archive_chat, "Ctrl+D = Archiver la discussion / supprimer le nœud", ""
help.favorite, "Ctrl+F = Favori", ""
help.ignore, "Ctrl+G = Ignorer", ""
help.search, "Ctrl+/ ou / = Rechercher", ""
help.help, "Ctrl+K = Aide", ""
help.no_help, "Aucune aide disponible.", ""
[User Settings]
user, "Utilisateur", ""
longName, "Nom long du nœud", "Si vous êtes un opérateur radioamateur agréé et avez activé le mode HAM, cela doit être votre indicatif."
shortName, "Nom court du nœud", "Doit contenir jusqu'à 4 octets."
isLicensed, "Activer le mode radioamateur (HAM)", "IMPORTANT : lire la documentation Meshtastic avant d'activer."
[app_settings]
title, "Paramètres de l'application", ""
channel_list_16ths, "Largeur de la liste des canaux", ""
node_list_16ths, "Largeur de la liste des nœuds", ""
single_pane_mode, "Mode panneau unique", ""
db_file_path, "Chemin du fichier de base de données", ""
log_file_path, "Chemin du fichier journal", ""
node_configs_file_path, "Chemin des configurations des nœuds", ""
language, "Langue", ""
message_prefix, "Préfixe des messages", ""
sent_message_prefix, "Préfixe des messages envoyés", ""
notification_symbol, "Symbole de notification", ""
notification_sound, "Son de notification", ""
ack_implicit_str, "ACK (implicite)", ""
ack_str, "ACK", ""
nak_str, "NAK", ""
ack_unknown_str, "ACK (inconnu)", ""
node_sort, "Tri des nœuds", ""
theme, "Thème", ""
[config.device]
title, "Appareil", ""
role, "Rôle", ""
serial_enabled, "Activer la console série", ""
button_gpio, "GPIO bouton", ""
buzzer_gpio, "GPIO buzzer", ""
rebroadcast_mode, "Mode de rediffusion", ""
node_info_broadcast_secs, "Intervalle diffusion infos nœud", ""
double_tap_as_button_press, "Double tap = bouton", ""
is_managed, "Activer mode géré", ""
disable_triple_click, "Désactiver triple clic", ""
tzdef, "Fuseau horaire", ""
led_heartbeat_disabled, "Désactiver LED heartbeat", ""
buzzer_mode, "Mode buzzer", ""
[config.network]
title, "Réseau", ""
wifi_enabled, "Wi-Fi activé", ""
wifi_ssid, "SSID Wi-Fi", ""
wifi_psk, "Mot de passe Wi-Fi", ""
ntp_server, "Serveur NTP", ""
eth_enabled, "Ethernet activé", ""
address_mode, "Mode IPv4", ""
ip, "Adresse IP", ""
gateway, "Passerelle", ""
subnet, "Sous-réseau", ""
dns, "DNS", ""
[config.display]
title, "Affichage", ""
screen_on_secs, "Durée écran actif", ""
gps_format, "Format GPS", ""
auto_screen_carousel_secs, "Intervalle carrousel", ""
compass_north_top, "Toujours nord en haut", ""
flip_screen, "Retourner écran", ""
units, "Unités préférées", ""
[config.bluetooth]
title, "Bluetooth", ""
enabled, "Activé", ""
mode, "Mode d'appairage", ""
fixed_pin, "Code PIN fixe", ""
[module.mqtt]
title, "MQTT", ""
enabled, "Activé", ""
address, "Adresse serveur", ""
username, "Nom d'utilisateur", ""
password, "Mot de passe", ""
encryption_enabled, "Chiffrement activé", ""
json_enabled, "JSON activé", ""
tls_enabled, "TLS activé", ""
[module.serial]
title, "Série", ""
enabled, "Activé", ""
echo, "Écho", ""
rxd, "GPIO réception", ""
txd, "GPIO transmission", ""
baud, "Débit en bauds", ""
timeout, "Délai", ""
[module.telemetry]
title, "Télémétrie", ""
device_update_interval, "Intervalle métriques appareil", ""
environment_update_interval, "Intervalle métriques environnement", ""
environment_measurement_enabled, "Télémétrie environnement activée", ""
[module.audio]
title, "Audio", ""
codec2_enabled, "Activé", ""
ptt_pin, "GPIO PTT", ""
bitrate, "Débit audio", ""
[module.remote_hardware]
title, "Matériel distant", ""
enabled, "Activé", ""
available_pins, "Broches disponibles", ""
[module.ambient_lighting]
title, "Éclairage ambiant", ""
led_state, "État LED", ""
current, "Courant", ""
red, "Rouge", ""
green, "Vert", ""
blue, "Bleu", ""

View File

@@ -12,6 +12,7 @@ Reboot, "Перезагрузить", ""
Reset Node DB, "Сбросить БД узлов", ""
Shutdown, "Выключить", ""
Factory Reset, "Сброс до заводских", ""
factory_reset_config, "Сбросить только конфигурацию", ""
Exit, "Выход", ""
Yes, "Да", ""
No, "Нет", ""
@@ -55,6 +56,7 @@ confirm.reboot, "Перезагрузить устройство?", ""
confirm.reset_node_db, "Сбросить БД узлов?", ""
confirm.shutdown, "Выключить устройство?", ""
confirm.factory_reset, "Сбросить до заводских настроек?", ""
confirm.factory_reset_config, "Сбросить только конфигурацию?", ""
confirm.save_before_exit_section, "Есть несохраненные изменения в {section}. Сохранить перед выходом?", ""
prompt.select_region, "Выберите ваш регион:", ""
dialog.slow_down_title, "Подождите", ""

View File

@@ -48,11 +48,8 @@ from contact.utilities.utils import (
add_new_message,
)
from contact.ui.contact_ui import (
draw_packetlog_win,
draw_node_list,
draw_messages_window,
draw_channel_list,
add_notification,
request_ui_redraw,
)
from contact.utilities.db_handler import (
save_message_to_db,
@@ -121,7 +118,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
ui_state.packet_buffer = ui_state.packet_buffer[-20:]
if ui_state.display_log:
draw_packetlog_win()
request_ui_redraw(packetlog=True)
if ui_state.current_window == 4:
menu_state.need_redraw = True
@@ -132,7 +129,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
# Assume any incoming packet could update the last seen time for a node
changed = refresh_node_list()
if changed:
draw_node_list()
request_ui_redraw(nodes=True)
if packet["decoded"]["portnum"] == "NODEINFO_APP":
if "user" in packet["decoded"] and "longName" in packet["decoded"]["user"]:
@@ -186,9 +183,9 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
add_new_message(channel_id, f"{config.message_prefix} [{hops}] {message_from_string} ", message_string)
if refresh_channels:
draw_channel_list()
request_ui_redraw(channels=True)
if refresh_messages:
draw_messages_window(True)
request_ui_redraw(messages=True, scroll_messages_to_bottom=True)
save_message_to_db(channel_id, message_from_id, message_string)

View File

@@ -15,7 +15,7 @@ from contact.utilities.db_handler import (
)
import contact.ui.default_config as config
from contact.utilities.singleton import ui_state, interface_state
from contact.utilities.singleton import ui_state, interface_state, app_state
from contact.utilities.utils import add_new_message
@@ -28,145 +28,141 @@ def onAckNak(packet: Dict[str, Any]) -> None:
"""
Handles incoming ACK/NAK response packets.
"""
from contact.ui.contact_ui import draw_messages_window
from contact.ui.contact_ui import request_ui_redraw
request = packet["decoded"]["requestId"]
if request not in ack_naks:
return
with app_state.lock:
request = packet["decoded"]["requestId"]
if request not in ack_naks:
return
acknak = ack_naks.pop(request)
message = ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]][1]
acknak = ack_naks.pop(request)
message = ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]][1]
confirm_string = " "
ack_type = None
if packet["decoded"]["routing"]["errorReason"] == "NONE":
if packet["from"] == interface_state.myNodeNum: # Ack "from" ourself means implicit ACK
confirm_string = config.ack_implicit_str
ack_type = "Implicit"
confirm_string = " "
ack_type = None
if packet["decoded"]["routing"]["errorReason"] == "NONE":
if packet["from"] == interface_state.myNodeNum: # Ack "from" ourself means implicit ACK
confirm_string = config.ack_implicit_str
ack_type = "Implicit"
else:
confirm_string = config.ack_str
ack_type = "Ack"
else:
confirm_string = config.ack_str
ack_type = "Ack"
else:
confirm_string = config.nak_str
ack_type = "Nak"
confirm_string = config.nak_str
ack_type = "Nak"
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
message,
)
ui_state.all_messages[acknak["channel"]][acknak["messageIndex"]] = (
time.strftime("[%H:%M:%S] ") + config.sent_message_prefix + confirm_string + ": ",
message,
)
update_ack_nak(acknak["channel"], acknak["timestamp"], message, ack_type)
update_ack_nak(acknak["channel"], acknak["timestamp"], message, ack_type)
channel_number = ui_state.channel_list.index(acknak["channel"])
if ui_state.channel_list[channel_number] == ui_state.channel_list[ui_state.selected_channel]:
draw_messages_window()
channel_number = ui_state.channel_list.index(acknak["channel"])
if ui_state.channel_list[channel_number] == ui_state.channel_list[ui_state.selected_channel]:
request_ui_redraw(messages=True)
def on_response_traceroute(packet: Dict[str, Any]) -> None:
"""
Handle traceroute response packets and render the route visually in the UI.
"""
from contact.ui.contact_ui import draw_channel_list, draw_messages_window, add_notification
from contact.ui.contact_ui import add_notification, request_ui_redraw
refresh_channels = False
refresh_messages = False
with app_state.lock:
refresh_channels = False
refresh_messages = False
UNK_SNR = -128 # Value representing unknown SNR
UNK_SNR = -128 # Value representing unknown SNR
route_discovery = mesh_pb2.RouteDiscovery()
route_discovery.ParseFromString(packet["decoded"]["payload"])
msg_dict = google.protobuf.json_format.MessageToDict(route_discovery)
route_discovery = mesh_pb2.RouteDiscovery()
route_discovery.ParseFromString(packet["decoded"]["payload"])
msg_dict = google.protobuf.json_format.MessageToDict(route_discovery)
msg_str = "Traceroute to:\n"
msg_str = "Traceroute to:\n"
route_str = (
get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}"
) # Start with destination of response
# SNR list should have one more entry than the route, as the final destination adds its SNR also
lenTowards = 0 if "route" not in msg_dict else len(msg_dict["route"])
snrTowardsValid = "snrTowards" in msg_dict and len(msg_dict["snrTowards"]) == lenTowards + 1
if lenTowards > 0: # Loop through hops in route and add SNR if available
for idx, node_num in enumerate(msg_dict["route"]):
route_str += (
" --> "
+ (get_name_from_database(node_num, "short") or f"{node_num:08x}")
+ " ("
+ (
str(msg_dict["snrTowards"][idx] / 4)
if snrTowardsValid and msg_dict["snrTowards"][idx] != UNK_SNR
else "?"
)
+ "dB)"
)
# End with origin of response
route_str += (
" --> "
+ (get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}")
+ " ("
+ (str(msg_dict["snrTowards"][-1] / 4) if snrTowardsValid and msg_dict["snrTowards"][-1] != UNK_SNR else "?")
+ "dB)"
)
msg_str += route_str + "\n" # Print the route towards destination
# Only if hopStart is set and there is an SNR entry (for the origin) it's valid, even though route might be empty (direct connection)
lenBack = 0 if "routeBack" not in msg_dict else len(msg_dict["routeBack"])
backValid = "hopStart" in packet and "snrBack" in msg_dict and len(msg_dict["snrBack"]) == lenBack + 1
if backValid:
msg_str += "Back:\n"
route_str = (
get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}"
) # Start with origin of response
get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}"
) # Start with destination of response
if lenBack > 0: # Loop through hops in routeBack and add SNR if available
for idx, node_num in enumerate(msg_dict["routeBack"]):
lenTowards = 0 if "route" not in msg_dict else len(msg_dict["route"])
snrTowardsValid = "snrTowards" in msg_dict and len(msg_dict["snrTowards"]) == lenTowards + 1
if lenTowards > 0:
for idx, node_num in enumerate(msg_dict["route"]):
route_str += (
" --> "
+ (get_name_from_database(node_num, "short") or f"{node_num:08x}")
+ " ("
+ (str(msg_dict["snrBack"][idx] / 4) if msg_dict["snrBack"][idx] != UNK_SNR else "?")
+ (
str(msg_dict["snrTowards"][idx] / 4)
if snrTowardsValid and msg_dict["snrTowards"][idx] != UNK_SNR
else "?"
)
+ "dB)"
)
# End with destination of response (us)
route_str += (
" --> "
+ (get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}")
+ (get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}")
+ " ("
+ (str(msg_dict["snrBack"][-1] / 4) if msg_dict["snrBack"][-1] != UNK_SNR else "?")
+ (str(msg_dict["snrTowards"][-1] / 4) if snrTowardsValid and msg_dict["snrTowards"][-1] != UNK_SNR else "?")
+ "dB)"
)
msg_str += route_str + "\n" # Print the route back to us
msg_str += route_str + "\n"
if packet["from"] not in ui_state.channel_list:
ui_state.channel_list.append(packet["from"])
refresh_channels = True
lenBack = 0 if "routeBack" not in msg_dict else len(msg_dict["routeBack"])
backValid = "hopStart" in packet and "snrBack" in msg_dict and len(msg_dict["snrBack"]) == lenBack + 1
if backValid:
msg_str += "Back:\n"
route_str = get_name_from_database(packet["from"], "short") or f"{packet['from']:08x}"
if is_chat_archived(packet["from"]):
update_node_info_in_db(packet["from"], chat_archived=False)
if lenBack > 0:
for idx, node_num in enumerate(msg_dict["routeBack"]):
route_str += (
" --> "
+ (get_name_from_database(node_num, "short") or f"{node_num:08x}")
+ " ("
+ (str(msg_dict["snrBack"][idx] / 4) if msg_dict["snrBack"][idx] != UNK_SNR else "?")
+ "dB)"
)
channel_number = ui_state.channel_list.index(packet["from"])
channel_id = ui_state.channel_list[channel_number]
route_str += (
" --> "
+ (get_name_from_database(packet["to"], "short") or f"{packet['to']:08x}")
+ " ("
+ (str(msg_dict["snrBack"][-1] / 4) if msg_dict["snrBack"][-1] != UNK_SNR else "?")
+ "dB)"
)
if channel_id == ui_state.channel_list[ui_state.selected_channel]:
refresh_messages = True
else:
add_notification(channel_number)
refresh_channels = True
msg_str += route_str + "\n"
message_from_string = get_name_from_database(packet["from"], type="short") + ":\n"
if packet["from"] not in ui_state.channel_list:
ui_state.channel_list.append(packet["from"])
refresh_channels = True
add_new_message(channel_id, f"{config.message_prefix} {message_from_string}", msg_str)
if is_chat_archived(packet["from"]):
update_node_info_in_db(packet["from"], chat_archived=False)
if refresh_channels:
draw_channel_list()
if refresh_messages:
draw_messages_window(True)
channel_number = ui_state.channel_list.index(packet["from"])
channel_id = ui_state.channel_list[channel_number]
save_message_to_db(channel_id, packet["from"], msg_str)
if channel_id == ui_state.channel_list[ui_state.selected_channel]:
refresh_messages = True
else:
add_notification(channel_number)
refresh_channels = True
message_from_string = get_name_from_database(packet["from"], type="short") + ":\n"
add_new_message(channel_id, f"{config.message_prefix} {message_from_string}", msg_str)
if refresh_channels:
request_ui_redraw(channels=True)
if refresh_messages:
request_ui_redraw(messages=True, scroll_messages_to_bottom=True)
save_message_to_db(channel_id, packet["from"], msg_str)
def send_message(message: str, destination: int = BROADCAST_NUM, channel: int = 0) -> None:
"""

View File

@@ -6,19 +6,26 @@ import sys
import traceback
import contact.ui.default_config as config
from contact.utilities.input_handlers import get_list_input
from contact.utilities.i18n import t
from contact.ui.dialog import dialog
from contact.utilities.i18n import t
from contact.ui.colors import setup_colors
from contact.ui.splash import draw_splash
from contact.ui.control_ui import set_region, settings_menu
from contact.ui.dialog import dialog
from contact.ui.splash import draw_splash
from contact.utilities.arg_parser import setup_parser
from contact.utilities.interfaces import initialize_interface
from contact.utilities.i18n import t
from contact.utilities.input_handlers import get_list_input
from contact.utilities.interfaces import initialize_interface, reconnect_interface
def close_interface(interface: object) -> None:
if interface is None:
return
with contextlib.suppress(Exception):
interface.close()
def main(stdscr: curses.window) -> None:
output_capture = io.StringIO()
interface = None
try:
with contextlib.redirect_stdout(output_capture), contextlib.redirect_stderr(output_capture):
setup_colors()
@@ -39,8 +46,9 @@ def main(stdscr: curses.window) -> None:
)
if confirmation == "Yes":
set_region(interface)
interface.close()
interface = initialize_interface(args)
close_interface(interface)
draw_splash(stdscr)
interface = reconnect_interface(args)
stdscr.clear()
stdscr.refresh()
settings_menu(stdscr, interface)
@@ -51,6 +59,8 @@ def main(stdscr: curses.window) -> None:
logging.error("Traceback: %s", traceback.format_exc())
logging.error("Console output before crash:\n%s", console_output)
raise
finally:
close_interface(interface)
def ensure_min_rows(stdscr: curses.window, min_rows: int = 11) -> None:

View File

@@ -15,8 +15,15 @@ from contact.utilities.i18n import t
from contact.utilities.emoji_utils import normalize_message_text
import contact.ui.default_config as config
import contact.ui.dialog
from contact.ui.nav_utils import move_main_highlight, draw_main_arrows, get_msg_window_lines, wrap_text
from contact.utilities.singleton import ui_state, interface_state, menu_state
from contact.ui.nav_utils import (
move_main_highlight,
draw_main_arrows,
get_msg_window_lines,
wrap_text,
truncate_with_ellipsis,
pad_to_width,
)
from contact.utilities.singleton import ui_state, interface_state, menu_state, app_state
MIN_COL = 1 # "effectively zero" without breaking curses
@@ -25,6 +32,53 @@ root_win = None
nodes_pad = None
def request_ui_redraw(
*,
channels: bool = False,
messages: bool = False,
nodes: bool = False,
packetlog: bool = False,
full: bool = False,
scroll_messages_to_bottom: bool = False,
) -> None:
ui_state.redraw_channels = ui_state.redraw_channels or channels
ui_state.redraw_messages = ui_state.redraw_messages or messages
ui_state.redraw_nodes = ui_state.redraw_nodes or nodes
ui_state.redraw_packetlog = ui_state.redraw_packetlog or packetlog
ui_state.redraw_full_ui = ui_state.redraw_full_ui or full
ui_state.scroll_messages_to_bottom = ui_state.scroll_messages_to_bottom or scroll_messages_to_bottom
def process_pending_ui_updates(stdscr: curses.window) -> None:
if ui_state.redraw_full_ui:
ui_state.redraw_full_ui = False
ui_state.redraw_channels = False
ui_state.redraw_messages = False
ui_state.redraw_nodes = False
ui_state.redraw_packetlog = False
ui_state.scroll_messages_to_bottom = False
handle_resize(stdscr, False)
return
if ui_state.redraw_channels:
ui_state.redraw_channels = False
draw_channel_list()
if ui_state.redraw_nodes:
ui_state.redraw_nodes = False
draw_node_list()
if ui_state.redraw_messages:
scroll_to_bottom = ui_state.scroll_messages_to_bottom
ui_state.redraw_messages = False
ui_state.scroll_messages_to_bottom = False
draw_messages_window(scroll_to_bottom)
if ui_state.redraw_packetlog:
ui_state.redraw_packetlog = False
draw_packetlog_win()
# Draw arrows for a specific window id (0=channel,1=messages,2=nodes).
def draw_window_arrows(window_id: int) -> None:
@@ -131,6 +185,43 @@ def get_node_display_name(node_num: int, node: dict) -> str:
return user.get("longName") or get_name_from_database(node_num, "long")
def get_selected_channel_title() -> str:
if not ui_state.channel_list:
return ""
channel = ui_state.channel_list[min(ui_state.selected_channel, len(ui_state.channel_list) - 1)]
if isinstance(channel, int):
return get_name_from_database(channel, "long") or get_name_from_database(channel, "short") or str(channel)
return str(channel)
def get_window_title(window: int) -> str:
if window == 2:
return f"Nodes: {len(ui_state.node_list)}"
if ui_state.single_pane_mode and window == 1:
return get_selected_channel_title()
return ""
def draw_frame_title(box: curses.window, title: str) -> None:
if not title:
return
_, box_w = box.getmaxyx()
max_title_width = max(0, box_w - 6)
if max_title_width <= 0:
return
clipped_title = truncate_with_ellipsis(title, max_title_width).rstrip()
if not clipped_title:
return
try:
box.addstr(0, 2, f" {clipped_title} ", curses.A_BOLD)
except curses.error:
pass
def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
"""Handle terminal resize events and redraw the UI accordingly."""
global messages_pad, messages_win, nodes_pad, nodes_win, channel_pad, channel_win, packetlog_win, entry_win
@@ -139,11 +230,19 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
height, width = stdscr.getmaxyx()
if ui_state.single_pane_mode:
channel_width, messages_width, nodes_width = compute_widths(width, ui_state.current_window)
channel_width = width
messages_width = width
nodes_width = width
channel_x = 0
messages_x = 0
nodes_x = 0
else:
channel_width = int(config.channel_list_16ths) * (width // 16)
nodes_width = int(config.node_list_16ths) * (width // 16)
messages_width = width - channel_width - nodes_width
channel_x = 0
messages_x = channel_width
nodes_x = channel_width + messages_width
channel_width = max(MIN_COL, channel_width)
messages_width = max(MIN_COL, messages_width)
@@ -151,7 +250,7 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
# Ensure the three widths sum exactly to the terminal width by adjusting the focused pane
total = channel_width + messages_width + nodes_width
if total != width:
if not ui_state.single_pane_mode and total != width:
delta = total - width
if ui_state.current_window == 0:
channel_width = max(MIN_COL, channel_width - delta)
@@ -168,11 +267,11 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
if firstrun:
entry_win = curses.newwin(entry_height, width, height - entry_height, 0)
channel_win = curses.newwin(content_h, channel_width, 0, 0)
messages_win = curses.newwin(content_h, messages_width, 0, channel_width)
nodes_win = curses.newwin(content_h, nodes_width, 0, channel_width + messages_width)
channel_win = curses.newwin(content_h, channel_width, 0, channel_x)
messages_win = curses.newwin(content_h, messages_width, 0, messages_x)
nodes_win = curses.newwin(content_h, nodes_width, 0, nodes_x)
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height, channel_width)
packetlog_win = curses.newwin(pkt_h, messages_width, height - pkt_h - entry_height, messages_x)
# Will be resized to what we need when drawn
messages_pad = curses.newpad(1, 1)
@@ -199,23 +298,30 @@ def handle_resize(stdscr: curses.window, firstrun: bool) -> None:
entry_win.mvwin(height - entry_height, 0)
channel_win.resize(content_h, channel_width)
channel_win.mvwin(0, 0)
channel_win.mvwin(0, channel_x)
messages_win.resize(content_h, messages_width)
messages_win.mvwin(0, channel_width)
messages_win.mvwin(0, messages_x)
nodes_win.resize(content_h, nodes_width)
nodes_win.mvwin(0, channel_width + messages_width)
nodes_win.mvwin(0, nodes_x)
packetlog_win.resize(pkt_h, messages_width)
packetlog_win.mvwin(height - pkt_h - entry_height, channel_width)
packetlog_win.mvwin(height - pkt_h - entry_height, messages_x)
# Draw window borders
for win in [channel_win, entry_win, nodes_win, messages_win]:
windows_to_draw = [entry_win]
if ui_state.single_pane_mode:
windows_to_draw.append([channel_win, messages_win, nodes_win][ui_state.current_window])
else:
windows_to_draw.extend([channel_win, nodes_win, messages_win])
for win in windows_to_draw:
win.box()
win.refresh()
entry_win.keypad(True)
entry_win.timeout(200)
curses.curs_set(1)
try:
@@ -261,14 +367,19 @@ def main_ui(stdscr: curses.window) -> None:
handle_resize(stdscr, True)
while True:
with app_state.lock:
process_pending_ui_updates(stdscr)
draw_text_field(entry_win, f"Message: {(input_text or '')[-(stdscr.getmaxyx()[1] - 10):]}", get_color("input"))
# Get user input from entry window
if queued_char is None:
char = entry_win.get_wch()
else:
char = queued_char
queued_char = None
try:
if queued_char is None:
char = entry_win.get_wch()
else:
char = queued_char
queued_char = None
except curses.error:
continue
# draw_debug(f"Keypress: {char}")
@@ -774,6 +885,7 @@ def handle_backtick(stdscr: curses.window) -> None:
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)
@@ -960,7 +1072,7 @@ def draw_channel_list() -> None:
channel_pad.erase()
win_width = channel_win.getmaxyx()[1]
channel_pad.resize(len(ui_state.all_messages), channel_win.getmaxyx()[1])
channel_pad.resize(max(1, len(ui_state.channel_list)), channel_win.getmaxyx()[1])
idx = 0
for channel in ui_state.channel_list:
@@ -977,9 +1089,7 @@ def draw_channel_list() -> None:
notification = " " + config.notification_symbol if idx in ui_state.notifications else ""
# Truncate the channel name if it's too long to fit in the window
truncated_channel = (
(channel[: win_width - 5] + "-" if len(channel) > win_width - 5 else channel) + notification
).ljust(win_width - 3)
truncated_channel = truncate_with_ellipsis(f"{channel}{notification}", win_width - 4)
color = get_color("channel_list")
if idx == ui_state.selected_channel:
@@ -1074,8 +1184,7 @@ def draw_node_list() -> None:
node_name = get_node_display_name(node_num, node)
# Future node name custom formatting possible
node_str = f"{status_icon} {node_name}"
node_str = node_str.ljust(box_width - 4)[: box_width - 2]
node_str = truncate_with_ellipsis(f"{status_icon} {node_name}", box_width - 4)
nodes_pad.addstr(i, 1, node_str, get_node_row_color(i))
paint_frame(nodes_win, selected=(ui_state.current_window == 2))
@@ -1297,8 +1406,7 @@ def refresh_pad(window: int) -> None:
pad = messages_pad
box = messages_win
lines = get_msg_window_lines(messages_win, packetlog_win)
selected_item = ui_state.selected_message
start_index = ui_state.selected_message
start_index = ui_state.start_index[1]
if ui_state.display_log:
packetlog_win.box()
@@ -1308,7 +1416,6 @@ def refresh_pad(window: int) -> None:
pad = nodes_pad
box = nodes_win
lines = box.getmaxyx()[0] - 2
box.addstr(0, 2, (f"Nodes: {len(ui_state.node_list)}"), curses.A_BOLD)
selected_item = ui_state.selected_node
start_index = max(0, selected_item - (win_height - 3)) # Leave room for borders
@@ -1345,6 +1452,9 @@ def refresh_pad(window: int) -> None:
if bottom < top or right < left:
return
draw_frame_title(box, get_window_title(window))
box.refresh()
pad.refresh(
start_index,
0,

View File

@@ -5,10 +5,12 @@ import logging
import os
import sys
from typing import List
from meshtastic.protobuf import admin_pb2
from contact.utilities.save_to_radio import save_changes
import contact.ui.default_config as config
from contact.utilities.config_io import config_export, config_import
from contact.utilities.interfaces import reconnect_interface
from contact.utilities.control_utils import transform_menu_path
from contact.utilities.i18n import t
from contact.utilities.ini_utils import parse_ini_file
@@ -23,15 +25,17 @@ 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.singleton import menu_state
from contact.utilities.arg_parser import setup_parser
from contact.utilities.singleton import interface_state, menu_state
# Setup Variables
MAX_MENU_WIDTH = 80 # desired max; will shrink on small terminals
save_option = "Save Changes"
max_help_lines = 0
help_win = None
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
# Compute the effective menu width for the current terminal
@@ -40,7 +44,7 @@ def get_menu_width() -> int:
return max(20, min(MAX_MENU_WIDTH, curses.COLS - 2))
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset"]
sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
# Get the parent directory of the script
script_dir = os.path.dirname(os.path.abspath(__file__))
@@ -222,6 +226,59 @@ def get_input_type_for_field(field) -> type:
return str
def reconnect_interface_with_splash(stdscr: object, interface: object) -> object:
try:
interface.close()
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)
return new_interface
def reconnect_after_admin_action(stdscr: object, interface: object, action, log_message: str) -> object:
action()
logging.info(log_message)
return reconnect_interface_with_splash(stdscr, interface)
def request_factory_reset(node: object, full: bool = False):
try:
return node.factoryReset(full=full)
except TypeError as ex:
field_name = "factory_reset_device" if full else "factory_reset_config"
field = admin_pb2.AdminMessage.DESCRIPTOR.fields_by_name[field_name]
if field.cpp_type != field.CPPTYPE_INT32:
raise
node.ensureSessionKey()
message = admin_pb2.AdminMessage()
setattr(message, field_name, 1)
if node == node.iface.localNode:
on_response = None
else:
on_response = node.onAckNak
return node._sendAdmin(message, onResponse=on_response)
def redraw_main_ui_after_reconnect(stdscr: object) -> None:
try:
from contact.ui import contact_ui
from contact.utilities.utils import get_channels, refresh_node_list
get_channels()
refresh_node_list()
contact_ui.handle_resize(stdscr, False)
except Exception:
logging.debug("Skipping main UI redraw after reconnect", exc_info=True)
def settings_menu(stdscr: object, interface: object) -> None:
curses.update_lines_cols()
@@ -330,9 +387,12 @@ def settings_menu(stdscr: object, interface: object) -> None:
help_win.refresh()
if menu_state.show_save_option and menu_state.selected_index == len(options):
save_changes(interface, modified_settings, menu_state)
reconnect_required = save_changes(interface, modified_settings, menu_state)
modified_settings.clear()
logging.info("Changes Saved")
if reconnect_required:
interface = reconnect_interface_with_splash(stdscr, interface)
menu = generate_menu_from_protobuf(interface)
if len(menu_state.menu_path) > 1:
menu_state.menu_path.pop()
@@ -460,8 +520,10 @@ def settings_menu(stdscr: object, interface: object) -> None:
t("ui.confirm.reboot", default="Are you sure you want to Reboot?"), None, ["Yes", "No"]
)
if confirmation == "Yes":
interface.localNode.reboot()
logging.info(f"Node Reboot Requested by menu")
interface = reconnect_after_admin_action(
stdscr, interface, interface.localNode.reboot, "Node Reboot Requested by menu"
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
continue
@@ -472,8 +534,10 @@ def settings_menu(stdscr: object, interface: object) -> None:
["Yes", "No"],
)
if confirmation == "Yes":
interface.localNode.resetNodeDb()
logging.info(f"Node DB Reset Requested by menu")
interface = reconnect_after_admin_action(
stdscr, interface, interface.localNode.resetNodeDb, "Node DB Reset Requested by menu"
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
continue
@@ -494,8 +558,30 @@ def settings_menu(stdscr: object, interface: object) -> None:
["Yes", "No"],
)
if confirmation == "Yes":
interface.localNode.factoryReset()
logging.info(f"Factory Reset Requested by menu")
interface = reconnect_after_admin_action(
stdscr,
interface,
lambda: request_factory_reset(interface.localNode, full=True),
"Factory Reset Requested by menu",
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
continue
elif selected_option == "factory_reset_config":
confirmation = get_list_input(
t("ui.confirm.factory_reset_config", default="Are you sure you want to Factory Reset Config?"),
None,
["Yes", "No"],
)
if confirmation == "Yes":
interface = reconnect_after_admin_action(
stdscr,
interface,
lambda: request_factory_reset(interface.localNode, full=False),
"Factory Reset Config Requested by menu",
)
menu = rebuild_menu_at_current_path(interface, menu_state)
menu_state.start_index.pop()
continue
@@ -655,8 +741,10 @@ def settings_menu(stdscr: object, interface: object) -> None:
if save_prompt == "Cancel":
continue # Stay in the menu without doing anything
elif save_prompt == "Yes":
save_changes(interface, modified_settings, menu_state)
reconnect_required = save_changes(interface, modified_settings, menu_state)
logging.info("Changes Saved")
if reconnect_required:
interface = reconnect_interface_with_splash(stdscr, interface)
modified_settings.clear()
menu = rebuild_menu_at_current_path(interface, menu_state)

View File

@@ -133,6 +133,7 @@ def generate_menu_from_protobuf(interface: object) -> Dict[str, Any]:
"Reset Node DB": None,
"Shutdown": None,
"Factory Reset": None,
"factory_reset_config": None,
"Exit": None,
}
)

View File

@@ -331,6 +331,51 @@ def text_width(text: str) -> int:
return sum(2 if east_asian_width(c) in "FW" else 1 for c in text)
def slice_to_width(text: str, max_width: int) -> str:
if max_width <= 0:
return ""
width = 0
chars = []
for char in text:
char_width = text_width(char)
if width + char_width > max_width:
break
chars.append(char)
width += char_width
return "".join(chars)
def pad_to_width(text: str, width: int) -> str:
clipped = slice_to_width(text, width)
return clipped + (" " * max(0, width - text_width(clipped)))
def truncate_with_ellipsis(text: str, width: int) -> str:
if width <= 0:
return ""
if text_width(text) <= width:
return pad_to_width(text, width)
if width == 1:
return ""
return pad_to_width(slice_to_width(text, width - 1) + "", width)
def split_text_to_width_chunks(text: str, width: int) -> List[str]:
if width <= 0:
return [""]
chunks = []
remaining = text
while remaining:
chunk = slice_to_width(remaining, width)
if not chunk:
break
chunks.append(chunk)
remaining = remaining[len(chunk) :]
return chunks or [""]
def wrap_text(text: str, wrap_width: int) -> List[str]:
"""Wraps text while preserving spaces and breaking long words."""
@@ -353,8 +398,7 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
wrapped_lines.append(line_buffer.strip())
line_buffer = ""
line_length = 0
for i in range(0, word_length, wrap_width):
wrapped_lines.append(word[i : i + wrap_width])
wrapped_lines.extend(split_text_to_width_chunks(word, wrap_width))
continue
if line_length + word_length > wrap_width and word.strip():

View File

@@ -33,6 +33,12 @@ class ChatUIState:
show_save_option: bool = False
menu_path: List[str] = field(default_factory=list)
single_pane_mode: bool = False
redraw_channels: bool = False
redraw_messages: bool = False
redraw_nodes: bool = False
redraw_packetlog: bool = False
redraw_full_ui: bool = False
scroll_messages_to_bottom: bool = False
@dataclass

View File

@@ -1,4 +1,5 @@
import logging
import time
import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
@@ -41,3 +42,21 @@ def initialize_interface(args):
except Exception as ex:
logging.critical(f"Fatal error initializing interface: {ex}")
def reconnect_interface(args, attempts: int = 15, delay_seconds: float = 1.0):
last_error = None
for attempt in range(attempts):
try:
interface = initialize_interface(args)
if interface is not None:
return interface
last_error = RuntimeError("initialize_interface returned None")
except Exception as ex:
last_error = ex
if attempt < attempts - 1:
time.sleep(delay_seconds)
raise RuntimeError("Failed to reconnect to the Meshtastic node") from last_error

View File

@@ -4,6 +4,79 @@ import logging
import base64
import time
DEVICE_REBOOT_KEYS = {"button_gpio", "buzzer_gpio", "role", "rebroadcast_mode"}
POWER_REBOOT_KEYS = {
"device_battery_ina_address",
"is_power_saving",
"ls_secs",
"min_wake_secs",
"on_battery_shutdown_after_secs",
"sds_secs",
"wait_bluetooth_secs",
}
DISPLAY_REBOOT_KEYS = {"screen_on_secs", "flip_screen", "oled", "displaymode"}
LORA_REBOOT_KEYS = {
"use_preset",
"region",
"modem_preset",
"bandwidth",
"spread_factor",
"coding_rate",
"tx_power",
"frequency_offset",
"override_frequency",
"channel_num",
"sx126x_rx_boosted_gain",
}
SECURITY_NON_REBOOT_KEYS = {"debug_log_api_enabled", "serial_enabled"}
USER_RECONNECT_KEYS = {"longName", "shortName", "isLicensed", "is_licensed"}
def _collect_changed_keys(modified_settings):
changed = set()
for key, value in modified_settings.items():
if isinstance(value, dict):
changed.update(_collect_changed_keys(value))
else:
changed.add(key)
return changed
def _requires_reconnect(menu_state, modified_settings) -> bool:
if not modified_settings or len(menu_state.menu_path) < 2:
return False
section = menu_state.menu_path[1]
changed_keys = _collect_changed_keys(modified_settings)
if section == "Module Settings":
return True
if section == "User Settings":
return bool(changed_keys & USER_RECONNECT_KEYS)
if section == "Channels":
return False
if section != "Radio Settings" or len(menu_state.menu_path) < 3:
return False
config_category = menu_state.menu_path[2].lower()
if config_category in {"network", "bluetooth"}:
return True
if config_category == "security":
return not changed_keys.issubset(SECURITY_NON_REBOOT_KEYS)
if config_category == "device":
return bool(changed_keys & DEVICE_REBOOT_KEYS)
if config_category == "power":
return bool(changed_keys & POWER_REBOOT_KEYS)
if config_category == "display":
return bool(changed_keys & DISPLAY_REBOOT_KEYS)
if config_category == "lora":
return bool(changed_keys & LORA_REBOOT_KEYS)
# Firmware defaults most config writes to reboot-required unless a handler
# explicitly clears that flag.
return True
def save_changes(interface, modified_settings, menu_state):
"""
@@ -15,7 +88,7 @@ def save_changes(interface, modified_settings, menu_state):
try:
if not modified_settings:
logging.info("No changes to save. modified_settings is empty.")
return
return False
node = interface.getNode("^local")
admin_key_backup = None
@@ -51,7 +124,7 @@ def save_changes(interface, modified_settings, menu_state):
# Return early if there are no other settings left to process
if not modified_settings:
return
return _requires_reconnect(menu_state, {"admin_key": admin_key_backup})
if menu_state.menu_path[1] == "Radio Settings" or menu_state.menu_path[1] == "Module Settings":
config_category = menu_state.menu_path[2].lower() # for radio and module configs
@@ -63,7 +136,7 @@ def save_changes(interface, modified_settings, menu_state):
interface.localNode.setFixedPosition(lat, lon, alt)
logging.info(f"Updated {config_category} with Latitude: {lat} and Longitude {lon} and Altitude {alt}")
return
return False
elif menu_state.menu_path[1] == "User Settings": # for user configs
config_category = "User Settings"
@@ -78,7 +151,7 @@ def save_changes(interface, modified_settings, menu_state):
f"Updated {config_category} with Long Name: {long_name}, Short Name: {short_name}, Licensed Mode: {is_licensed}"
)
return
return _requires_reconnect(menu_state, modified_settings)
elif menu_state.menu_path[1] == "Channels": # for channel configs
config_category = "Channels"
@@ -107,7 +180,7 @@ def save_changes(interface, modified_settings, menu_state):
logging.info(f"Updated Channel {channel_num} in {config_category}")
logging.info(node.channels)
return
return False
else:
config_category = None
@@ -120,7 +193,7 @@ def save_changes(interface, modified_settings, menu_state):
config_container = getattr(node.moduleConfig, config_category)
else:
logging.warning(f"Config category '{config_category}' not found in config.")
return
return False
if len(menu_state.menu_path) >= 4:
nested_key = menu_state.menu_path[3]
@@ -164,8 +237,11 @@ def save_changes(interface, modified_settings, menu_state):
if admin_key_backup is not None:
modified_settings["admin_key"] = admin_key_backup
return _requires_reconnect(menu_state, modified_settings)
except Exception as e:
logging.error(f"Failed to write configuration for category '{config_category}': {e}")
return False
except Exception as e:
logging.error(f"Error saving changes: {e}")
return False

View File

@@ -10,35 +10,50 @@ from contact.utilities.singleton import ui_state, interface_state
import contact.utilities.telemetry_beautifier as tb
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
return convert_to_camel_case(modem_preset_string)
def get_channels():
"""Retrieve channels from the node and update ui_state.channel_list and ui_state.all_messages."""
"""Retrieve channels from the node and rebuild named channel state."""
node = interface_state.interface.getNode("^local")
device_channels = node.channels
previous_channel_list = list(ui_state.channel_list)
previous_messages = dict(ui_state.all_messages)
# Clear and rebuild channel list
# ui_state.channel_list = []
named_channels = []
for device_channel in device_channels:
if device_channel.role:
# Use the channel name if available, otherwise use the modem preset
if device_channel.settings.name:
channel_name = device_channel.settings.name
else:
# If channel name is blank, use the modem preset
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
channel_name = convert_to_camel_case(modem_preset_string)
named_channels.append(_get_channel_name(device_channel, node))
# Add channel to ui_state.channel_list if not already present
if channel_name not in ui_state.channel_list:
ui_state.channel_list.append(channel_name)
previous_named_channels = [channel for channel in previous_channel_list if isinstance(channel, str)]
preserved_direct_channels = [channel for channel in previous_channel_list if isinstance(channel, int)]
rebuilt_messages = {}
# Initialize ui_state.all_messages[channel_name] if it doesn't exist
if channel_name not in ui_state.all_messages:
ui_state.all_messages[channel_name] = []
for index, channel_name in enumerate(named_channels):
previous_name = previous_named_channels[index] if index < len(previous_named_channels) else channel_name
if previous_name in previous_messages:
rebuilt_messages[channel_name] = previous_messages[previous_name]
elif channel_name in previous_messages:
rebuilt_messages[channel_name] = previous_messages[channel_name]
else:
rebuilt_messages[channel_name] = []
for channel in preserved_direct_channels:
if channel in previous_messages:
rebuilt_messages[channel] = previous_messages[channel]
ui_state.channel_list = named_channels + preserved_direct_channels
ui_state.all_messages = rebuilt_messages
if ui_state.channel_list:
ui_state.selected_channel = max(0, min(ui_state.selected_channel, len(ui_state.channel_list) - 1))
return ui_state.channel_list

View File

@@ -1,6 +1,6 @@
[project]
name = "contact"
version = "1.5.0"
version = "1.5.6"
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"}

202
tests/test_contact_ui.py Normal file
View File

@@ -0,0 +1,202 @@
import unittest
from unittest import mock
import contact.ui.default_config as config
from contact.ui import contact_ui
from contact.ui.nav_utils import text_width
from contact.utilities.singleton import ui_state
from tests.test_support import reset_singletons, restore_config, snapshot_config
class ContactUiTests(unittest.TestCase):
def setUp(self) -> None:
reset_singletons()
self.saved_config = snapshot_config("single_pane_mode")
def tearDown(self) -> None:
restore_config(self.saved_config)
reset_singletons()
def test_handle_backtick_refreshes_channels_after_settings_menu(self) -> None:
stdscr = mock.Mock()
ui_state.current_window = 1
config.single_pane_mode = "False"
with mock.patch.object(contact_ui.curses, "curs_set") as curs_set:
with mock.patch.object(contact_ui, "settings_menu") as settings_menu:
with mock.patch.object(contact_ui, "get_channels") as get_channels:
with mock.patch.object(contact_ui, "refresh_node_list") as refresh_node_list:
with mock.patch.object(contact_ui, "handle_resize") as handle_resize:
contact_ui.handle_backtick(stdscr)
settings_menu.assert_called_once()
get_channels.assert_called_once_with()
refresh_node_list.assert_called_once_with()
handle_resize.assert_called_once_with(stdscr, False)
self.assertEqual(curs_set.call_args_list[0].args, (0,))
self.assertEqual(curs_set.call_args_list[-1].args, (1,))
self.assertEqual(ui_state.current_window, 1)
def test_process_pending_ui_updates_draws_requested_windows(self) -> None:
stdscr = mock.Mock()
ui_state.redraw_channels = True
ui_state.redraw_messages = True
ui_state.redraw_nodes = True
ui_state.redraw_packetlog = True
ui_state.scroll_messages_to_bottom = True
with mock.patch.object(contact_ui, "draw_channel_list") as draw_channel_list:
with mock.patch.object(contact_ui, "draw_messages_window") as draw_messages_window:
with mock.patch.object(contact_ui, "draw_node_list") as draw_node_list:
with mock.patch.object(contact_ui, "draw_packetlog_win") as draw_packetlog_win:
contact_ui.process_pending_ui_updates(stdscr)
draw_channel_list.assert_called_once_with()
draw_messages_window.assert_called_once_with(True)
draw_node_list.assert_called_once_with()
draw_packetlog_win.assert_called_once_with()
def test_process_pending_ui_updates_full_redraw_uses_handle_resize(self) -> None:
stdscr = mock.Mock()
ui_state.redraw_full_ui = True
ui_state.redraw_channels = True
ui_state.redraw_messages = True
with mock.patch.object(contact_ui, "handle_resize") as handle_resize:
contact_ui.process_pending_ui_updates(stdscr)
handle_resize.assert_called_once_with(stdscr, False)
self.assertFalse(ui_state.redraw_channels)
self.assertFalse(ui_state.redraw_messages)
def test_refresh_node_selection_reserves_scroll_arrow_column(self) -> None:
ui_state.node_list = [101, 202]
ui_state.selected_node = 1
ui_state.start_index = [0, 0, 0]
contact_ui.nodes_pad = mock.Mock()
contact_ui.nodes_pad.getmaxyx.return_value = (4, 20)
contact_ui.nodes_win = mock.Mock()
contact_ui.nodes_win.getmaxyx.return_value = (10, 20)
interface = mock.Mock()
interface.nodesByNum = {101: {}, 202: {}}
with mock.patch.object(contact_ui, "refresh_pad") as refresh_pad:
with mock.patch.object(contact_ui, "draw_window_arrows") as draw_window_arrows:
with mock.patch.object(contact_ui, "get_node_row_color", side_effect=[11, 22]):
with mock.patch("contact.ui.contact_ui.interface_state.interface", interface):
contact_ui.refresh_node_selection(old_index=0, highlight=True)
self.assertEqual(
contact_ui.nodes_pad.chgat.call_args_list,
[mock.call(0, 1, 16, 11), mock.call(1, 1, 16, 22)],
)
refresh_pad.assert_called_once_with(2)
draw_window_arrows.assert_called_once_with(2)
def test_draw_channel_list_reserves_scroll_arrow_column(self) -> None:
ui_state.channel_list = ["VeryLongChannelName"]
ui_state.notifications = []
ui_state.selected_channel = 0
ui_state.current_window = 0
contact_ui.channel_pad = mock.Mock()
contact_ui.channel_win = mock.Mock()
contact_ui.channel_win.getmaxyx.return_value = (10, 20)
with mock.patch.object(contact_ui, "get_color", return_value=1):
with mock.patch.object(contact_ui, "paint_frame"):
with mock.patch.object(contact_ui, "refresh_pad"):
with mock.patch.object(contact_ui, "draw_window_arrows"):
with mock.patch.object(contact_ui, "remove_notification"):
contact_ui.draw_channel_list()
text = contact_ui.channel_pad.addstr.call_args.args[2]
self.assertEqual(len(text), 16)
def test_draw_node_list_reserves_scroll_arrow_column(self) -> None:
ui_state.node_list = [101]
ui_state.current_window = 2
contact_ui.nodes_pad = mock.Mock()
contact_ui.nodes_win = mock.Mock()
contact_ui.nodes_win.getmaxyx.return_value = (10, 20)
contact_ui.entry_win = mock.Mock()
interface = mock.Mock()
interface.nodesByNum = {101: {"user": {"longName": "VeryLongNodeName", "publicKey": ""}}}
with mock.patch("contact.ui.contact_ui.interface_state.interface", interface):
with mock.patch.object(contact_ui, "get_node_row_color", return_value=1):
with mock.patch.object(contact_ui.curses, "curs_set"):
with mock.patch.object(contact_ui, "paint_frame"):
with mock.patch.object(contact_ui, "refresh_pad"):
with mock.patch.object(contact_ui, "draw_window_arrows"):
contact_ui.draw_node_list()
text = contact_ui.nodes_pad.addstr.call_args.args[2]
self.assertEqual(text_width(text), 16)
self.assertIn("", text)
def test_handle_resize_single_pane_keeps_full_width_windows(self) -> None:
stdscr = mock.Mock()
stdscr.getmaxyx.return_value = (24, 80)
ui_state.single_pane_mode = True
ui_state.current_window = 1
contact_ui.entry_win = mock.Mock()
contact_ui.channel_win = mock.Mock()
contact_ui.messages_win = mock.Mock()
contact_ui.nodes_win = mock.Mock()
contact_ui.packetlog_win = mock.Mock()
contact_ui.messages_pad = mock.Mock()
contact_ui.nodes_pad = mock.Mock()
contact_ui.channel_pad = mock.Mock()
with mock.patch.object(contact_ui.curses, "curs_set"):
with mock.patch.object(contact_ui, "draw_channel_list") as draw_channel_list:
with mock.patch.object(contact_ui, "draw_messages_window") as draw_messages_window:
with mock.patch.object(contact_ui, "draw_node_list") as draw_node_list:
with mock.patch.object(contact_ui, "draw_window_arrows") as draw_window_arrows:
contact_ui.handle_resize(stdscr, False)
contact_ui.channel_win.resize.assert_called_once_with(21, 80)
contact_ui.messages_win.resize.assert_called_once_with(21, 80)
contact_ui.nodes_win.resize.assert_called_once_with(21, 80)
contact_ui.channel_win.mvwin.assert_called_once_with(0, 0)
contact_ui.messages_win.mvwin.assert_called_once_with(0, 0)
contact_ui.nodes_win.mvwin.assert_called_once_with(0, 0)
contact_ui.channel_win.box.assert_not_called()
contact_ui.nodes_win.box.assert_not_called()
contact_ui.messages_win.box.assert_called_once_with()
draw_channel_list.assert_called_once_with()
draw_messages_window.assert_called_once_with(True)
draw_node_list.assert_called_once_with()
draw_window_arrows.assert_called_once_with(1)
def test_get_window_title_uses_selected_channel_only_for_messages_in_single_pane_mode(self) -> None:
ui_state.single_pane_mode = True
ui_state.channel_list = ["Primary"]
ui_state.selected_channel = 0
self.assertEqual(contact_ui.get_window_title(0), "")
self.assertEqual(contact_ui.get_window_title(1), "Primary")
def test_refresh_pad_draws_selected_channel_title_on_message_frame(self) -> None:
ui_state.single_pane_mode = True
ui_state.current_window = 1
ui_state.channel_list = ["Primary"]
ui_state.selected_channel = 0
ui_state.start_index = [0, 0, 0]
ui_state.display_log = False
contact_ui.channel_win = mock.Mock()
contact_ui.channel_win.getmaxyx.return_value = (10, 20)
contact_ui.messages_pad = mock.Mock()
contact_ui.messages_pad.getmaxyx.return_value = (5, 20)
contact_ui.messages_win = mock.Mock()
contact_ui.messages_win.getbegyx.return_value = (0, 0)
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)
contact_ui.messages_win.addstr.assert_called_once_with(0, 2, " Primary ", contact_ui.curses.A_BOLD)

112
tests/test_control_ui.py Normal file
View File

@@ -0,0 +1,112 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
from contact.ui import control_ui
from contact.utilities.singleton import interface_state
from tests.test_support import reset_singletons
class ControlUiTests(unittest.TestCase):
def setUp(self) -> None:
reset_singletons()
def tearDown(self) -> None:
reset_singletons()
def test_reconnect_interface_with_splash_replaces_interface(self) -> None:
old_interface = mock.Mock()
new_interface = mock.Mock()
stdscr = mock.Mock()
parser = mock.Mock()
parser.parse_args.return_value = Namespace()
with mock.patch.object(control_ui, "setup_parser", return_value=parser):
with mock.patch.object(control_ui, "draw_splash") as draw_splash:
with mock.patch.object(control_ui, "reconnect_interface", return_value=new_interface) as reconnect:
with mock.patch.object(control_ui, "redraw_main_ui_after_reconnect") as redraw:
result = control_ui.reconnect_interface_with_splash(stdscr, old_interface)
old_interface.close.assert_called_once_with()
stdscr.clear.assert_called_once_with()
stdscr.refresh.assert_called_once_with()
draw_splash.assert_called_once_with(stdscr)
reconnect.assert_called_once_with(parser.parse_args.return_value)
redraw.assert_called_once_with(stdscr)
self.assertIs(result, new_interface)
self.assertIs(interface_state.interface, new_interface)
def test_reconnect_after_admin_action_runs_action_then_reconnects(self) -> None:
stdscr = mock.Mock()
interface = mock.Mock()
new_interface = mock.Mock()
action = mock.Mock()
with mock.patch.object(control_ui, "reconnect_interface_with_splash", return_value=new_interface) as reconnect:
result = control_ui.reconnect_after_admin_action(
stdscr, interface, action, "Factory Reset Requested by menu"
)
action.assert_called_once_with()
reconnect.assert_called_once_with(stdscr, interface)
self.assertIs(result, new_interface)
def test_redraw_main_ui_after_reconnect_refreshes_channels_nodes_and_layout(self) -> None:
stdscr = mock.Mock()
with mock.patch("contact.utilities.utils.get_channels") as get_channels:
with mock.patch("contact.utilities.utils.refresh_node_list") as refresh_node_list:
with mock.patch("contact.ui.contact_ui.handle_resize") as handle_resize:
control_ui.redraw_main_ui_after_reconnect(stdscr)
get_channels.assert_called_once_with()
refresh_node_list.assert_called_once_with()
handle_resize.assert_called_once_with(stdscr, False)
def test_request_factory_reset_uses_library_helper_when_supported(self) -> None:
node = mock.Mock()
control_ui.request_factory_reset(node)
node.factoryReset.assert_called_once_with(full=False)
node.ensureSessionKey.assert_not_called()
node._sendAdmin.assert_not_called()
def test_request_factory_reset_uses_library_helper_for_full_reset_when_supported(self) -> None:
node = mock.Mock()
control_ui.request_factory_reset(node, full=True)
node.factoryReset.assert_called_once_with(full=True)
node.ensureSessionKey.assert_not_called()
node._sendAdmin.assert_not_called()
def test_request_factory_reset_falls_back_to_int_valued_admin_message(self) -> None:
node = mock.Mock()
node.factoryReset.side_effect = TypeError(
"Field meshtastic.protobuf.AdminMessage.factory_reset_config: Expected an int, got a boolean."
)
node.iface = SimpleNamespace(localNode=node)
control_ui.request_factory_reset(node)
node.ensureSessionKey.assert_called_once_with()
sent_message = node._sendAdmin.call_args.args[0]
self.assertEqual(sent_message.factory_reset_config, 1)
self.assertIsNone(node._sendAdmin.call_args.kwargs["onResponse"])
def test_request_factory_reset_full_falls_back_to_int_valued_admin_message(self) -> None:
node = mock.Mock()
node.factoryReset.side_effect = TypeError(
"Field meshtastic.protobuf.AdminMessage.factory_reset_device: Expected an int, got a boolean."
)
node.iface = SimpleNamespace(localNode=node)
control_ui.request_factory_reset(node, full=True)
node.ensureSessionKey.assert_called_once_with()
sent_message = node._sendAdmin.call_args.args[0]
self.assertEqual(sent_message.factory_reset_device, 1)
self.assertIsNone(node._sendAdmin.call_args.kwargs["onResponse"])

26
tests/test_interfaces.py Normal file
View File

@@ -0,0 +1,26 @@
from argparse import Namespace
import unittest
from unittest import mock
from contact.utilities.interfaces import reconnect_interface
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:
with mock.patch("contact.utilities.interfaces.time.sleep") as sleep:
result = reconnect_interface(args, attempts=3, delay_seconds=0.25)
self.assertEqual(result, "iface")
self.assertEqual(initialize.call_count, 3)
self.assertEqual(sleep.call_count, 2)
def test_reconnect_interface_raises_after_exhausting_attempts(self) -> None:
args = Namespace()
with mock.patch("contact.utilities.interfaces.initialize_interface", return_value=None):
with mock.patch("contact.utilities.interfaces.time.sleep"):
with self.assertRaises(RuntimeError):
reconnect_interface(args, attempts=2, delay_seconds=0)

View File

@@ -1,4 +1,5 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
@@ -40,20 +41,58 @@ class MainRuntimeTests(unittest.TestCase):
self.assertEqual(result, "live-interface")
initialize_interface.assert_called_once_with(args)
def test_interface_is_ready_detects_missing_local_node(self) -> None:
self.assertFalse(entrypoint.interface_is_ready(object()))
self.assertTrue(entrypoint.interface_is_ready(SimpleNamespace(localNode=SimpleNamespace(localConfig=mock.Mock()))))
def test_initialize_runtime_interface_with_retry_retries_until_node_is_ready(self) -> None:
args = Namespace(demo_screenshot=False)
stdscr = mock.Mock()
bad_interface = mock.Mock(spec=["close"])
good_interface = SimpleNamespace(localNode=SimpleNamespace(localConfig=mock.Mock()))
with mock.patch.object(entrypoint, "initialize_runtime_interface", side_effect=[bad_interface, good_interface]):
with mock.patch.object(entrypoint, "get_list_input", return_value="Retry") as get_list_input:
with mock.patch.object(entrypoint, "draw_splash") as draw_splash:
result = entrypoint.initialize_runtime_interface_with_retry(stdscr, args)
self.assertIs(result, good_interface)
get_list_input.assert_called_once()
bad_interface.close.assert_called_once_with()
draw_splash.assert_called_once_with(stdscr)
def test_initialize_runtime_interface_with_retry_returns_none_when_user_closes(self) -> None:
args = Namespace(demo_screenshot=False)
stdscr = mock.Mock()
bad_interface = mock.Mock(spec=["close"])
with mock.patch.object(entrypoint, "initialize_runtime_interface", return_value=bad_interface):
with mock.patch.object(entrypoint, "get_list_input", return_value="Close") as get_list_input:
with mock.patch.object(entrypoint, "draw_splash") as draw_splash:
result = entrypoint.initialize_runtime_interface_with_retry(stdscr, args)
self.assertIsNone(result)
get_list_input.assert_called_once()
bad_interface.close.assert_called_once_with()
draw_splash.assert_not_called()
def test_prompt_region_if_unset_reinitializes_interface_after_confirmation(self) -> None:
args = Namespace()
old_interface = mock.Mock()
new_interface = mock.Mock()
stdscr = mock.Mock()
interface_state.interface = old_interface
with mock.patch.object(entrypoint, "get_list_input", return_value="Yes"):
with mock.patch.object(entrypoint, "set_region") as set_region:
with mock.patch.object(entrypoint, "initialize_interface", return_value=new_interface) as initialize:
entrypoint.prompt_region_if_unset(args)
with mock.patch.object(entrypoint, "draw_splash") as draw_splash:
with mock.patch.object(entrypoint, "reconnect_interface", return_value=new_interface) as reconnect:
entrypoint.prompt_region_if_unset(args, stdscr)
set_region.assert_called_once_with(old_interface)
old_interface.close.assert_called_once_with()
initialize.assert_called_once_with(args)
draw_splash.assert_called_once_with(stdscr)
reconnect.assert_called_once_with(args)
self.assertIs(interface_state.interface, new_interface)
def test_prompt_region_if_unset_leaves_interface_unchanged_when_declined(self) -> None:
@@ -63,11 +102,11 @@ class MainRuntimeTests(unittest.TestCase):
with mock.patch.object(entrypoint, "get_list_input", return_value="No"):
with mock.patch.object(entrypoint, "set_region") as set_region:
with mock.patch.object(entrypoint, "initialize_interface") as initialize:
with mock.patch.object(entrypoint, "reconnect_interface") as reconnect:
entrypoint.prompt_region_if_unset(args)
set_region.assert_not_called()
initialize.assert_not_called()
reconnect.assert_not_called()
interface.close.assert_not_called()
self.assertIs(interface_state.interface, interface)
@@ -147,6 +186,30 @@ class MainRuntimeTests(unittest.TestCase):
wrapper.assert_called_once_with(entrypoint.main)
interface.close.assert_called_once_with()
def test_start_does_not_crash_when_wrapper_returns_without_interface(self) -> None:
interface_state.interface = None
with mock.patch.object(entrypoint.sys, "argv", ["contact"]):
with mock.patch.object(entrypoint.curses, "wrapper") as wrapper:
entrypoint.start()
wrapper.assert_called_once_with(entrypoint.main)
def test_main_returns_cleanly_when_user_closes_missing_node_dialog(self) -> None:
stdscr = mock.Mock()
args = Namespace(settings=False, demo_screenshot=False)
with mock.patch.object(entrypoint, "setup_colors"):
with mock.patch.object(entrypoint, "ensure_min_rows"):
with mock.patch.object(entrypoint, "draw_splash"):
with mock.patch.object(entrypoint, "setup_parser") as setup_parser:
with mock.patch.object(entrypoint, "initialize_runtime_interface_with_retry", return_value=None):
with mock.patch.object(entrypoint, "initialize_globals") as initialize_globals:
setup_parser.return_value.parse_args.return_value = args
entrypoint.main(stdscr)
initialize_globals.assert_not_called()
def test_start_handles_keyboard_interrupt(self) -> None:
interface = mock.Mock()
interface_state.interface = interface
@@ -161,6 +224,18 @@ class MainRuntimeTests(unittest.TestCase):
interface.close.assert_called_once_with()
exit_mock.assert_called_once_with(0)
def test_start_handles_keyboard_interrupt_with_no_interface(self) -> None:
interface_state.interface = None
with mock.patch.object(entrypoint.sys, "argv", ["contact"]):
with mock.patch.object(entrypoint.curses, "wrapper", side_effect=KeyboardInterrupt):
with mock.patch.object(entrypoint.sys, "exit", side_effect=SystemExit(0)) as exit_mock:
with self.assertRaises(SystemExit) as raised:
entrypoint.start()
self.assertEqual(raised.exception.code, 0)
exit_mock.assert_called_once_with(0)
def test_start_handles_fatal_exception_and_exits_one(self) -> None:
with mock.patch.object(entrypoint.sys, "argv", ["contact"]):
with mock.patch.object(entrypoint.curses, "wrapper", side_effect=RuntimeError("boom")):

28
tests/test_menus.py Normal file
View File

@@ -0,0 +1,28 @@
from types import SimpleNamespace
import unittest
from meshtastic.protobuf import config_pb2, module_config_pb2
from contact.ui.menus import generate_menu_from_protobuf
class MenusTests(unittest.TestCase):
def test_main_menu_includes_factory_reset_config_after_factory_reset(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", "isLicensed": False},
"position": {"latitude": 0.0, "longitude": 0.0, "altitude": 0},
},
)
menu = generate_menu_from_protobuf(interface)
keys = list(menu["Main Menu"].keys())
self.assertLess(keys.index("Factory Reset"), keys.index("factory_reset_config"))
self.assertEqual(keys[keys.index("Factory Reset") + 1], "factory_reset_config")

36
tests/test_nav_utils.py Normal file
View File

@@ -0,0 +1,36 @@
import unittest
from unittest import mock
from contact.ui import nav_utils
from contact.ui.nav_utils import truncate_with_ellipsis, wrap_text
from contact.utilities.singleton import ui_state
class NavUtilsTests(unittest.TestCase):
def setUp(self) -> None:
ui_state.current_window = 0
ui_state.node_list = []
ui_state.start_index = [0, 0, 0]
def test_wrap_text_splits_wide_characters_by_display_width(self) -> None:
self.assertEqual(wrap_text("🔐🔐🔐", 4), ["🔐", "🔐", "🔐"])
def test_truncate_with_ellipsis_respects_display_width(self) -> None:
self.assertEqual(truncate_with_ellipsis("🔐Alpha", 5), "🔐Al…")
def test_highlight_line_reserves_scroll_arrow_column_for_nodes(self) -> None:
ui_state.current_window = 2
ui_state.start_index = [0, 0, 0]
menu_win = mock.Mock()
menu_win.getbegyx.return_value = (0, 0)
menu_win.getmaxyx.return_value = (8, 20)
menu_pad = mock.Mock()
menu_pad.getmaxyx.return_value = (4, 20)
with mock.patch.object(nav_utils, "get_node_color", side_effect=[11, 22]):
nav_utils.highlight_line(menu_win, menu_pad, 0, 1, 5)
self.assertEqual(
menu_pad.chgat.call_args_list,
[mock.call(0, 1, 16, 11), mock.call(1, 1, 16, 22)],
)

View File

@@ -34,17 +34,13 @@ class RxHandlerTests(unittest.TestCase):
}
with mock.patch.object(rx_handler, "refresh_node_list", return_value=True):
with mock.patch.object(rx_handler, "draw_node_list") as draw_node_list:
with mock.patch.object(rx_handler, "draw_messages_window") as draw_messages_window:
with mock.patch.object(rx_handler, "draw_channel_list") as draw_channel_list:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
with mock.patch.object(rx_handler, "request_ui_redraw") as request_ui_redraw:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
draw_node_list.assert_called_once_with()
draw_messages_window.assert_called_once_with(True)
draw_channel_list.assert_not_called()
self.assertEqual(request_ui_redraw.call_args_list, [mock.call(nodes=True), mock.call(messages=True, scroll_messages_to_bottom=True)])
add_notification.assert_not_called()
save_message_to_db.assert_called_once_with("Primary", 222, "hello")
self.assertEqual(ui_state.all_messages["Primary"][-1][1], "hello")
@@ -66,18 +62,16 @@ class RxHandlerTests(unittest.TestCase):
}
with mock.patch.object(rx_handler, "refresh_node_list", return_value=False):
with mock.patch.object(rx_handler, "draw_messages_window") as draw_messages_window:
with mock.patch.object(rx_handler, "draw_channel_list") as draw_channel_list:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "update_node_info_in_db") as update_node_info_in_db:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
with mock.patch.object(rx_handler, "request_ui_redraw") as request_ui_redraw:
with mock.patch.object(rx_handler, "add_notification") as add_notification:
with mock.patch.object(rx_handler, "update_node_info_in_db") as update_node_info_in_db:
with mock.patch.object(rx_handler, "save_message_to_db") as save_message_to_db:
with mock.patch.object(rx_handler, "get_name_from_database", return_value="SAT2"):
rx_handler.on_receive(packet, interface=None)
self.assertIn(222, ui_state.channel_list)
self.assertIn(222, ui_state.all_messages)
draw_messages_window.assert_not_called()
draw_channel_list.assert_called_once_with()
request_ui_redraw.assert_called_once_with(channels=True)
add_notification.assert_called_once_with(1)
update_node_info_in_db.assert_called_once_with(222, chat_archived=False)
save_message_to_db.assert_called_once_with(222, 222, "dm")
@@ -87,10 +81,10 @@ class RxHandlerTests(unittest.TestCase):
ui_state.display_log = True
ui_state.current_window = 4
with mock.patch.object(rx_handler, "draw_packetlog_win") as draw_packetlog_win:
with mock.patch.object(rx_handler, "request_ui_redraw") as request_ui_redraw:
rx_handler.on_receive({"id": "new"}, interface=None)
draw_packetlog_win.assert_called_once_with()
request_ui_redraw.assert_called_once_with(packetlog=True)
self.assertEqual(len(ui_state.packet_buffer), 20)
self.assertEqual(ui_state.packet_buffer[-1], {"id": "new"})
self.assertTrue(menu_state.need_redraw)

114
tests/test_save_to_radio.py Normal file
View File

@@ -0,0 +1,114 @@
from types import SimpleNamespace
import unittest
from unittest import mock
from contact.utilities.save_to_radio import save_changes
class SaveToRadioTests(unittest.TestCase):
def build_interface(self):
node = mock.Mock()
node.localConfig = SimpleNamespace(
lora=SimpleNamespace(region=0, serial_enabled=False),
device=SimpleNamespace(role="CLIENT", name="node"),
security=SimpleNamespace(debug_log_api_enabled=False, serial_enabled=False, admin_key=[]),
display=SimpleNamespace(flip_screen=False, units=0),
power=SimpleNamespace(is_power_saving=False, adc_enabled=False),
network=SimpleNamespace(wifi_enabled=False),
bluetooth=SimpleNamespace(enabled=False),
)
node.moduleConfig = SimpleNamespace(mqtt=SimpleNamespace(enabled=False))
interface = mock.Mock()
interface.getNode.return_value = node
return interface, node
def test_save_changes_returns_true_for_lora_writes_that_require_reconnect(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Lora"])
reconnect_required = save_changes(interface, {"region": 7}, menu_state)
self.assertTrue(reconnect_required)
self.assertEqual(node.localConfig.lora.region, 7)
node.writeConfig.assert_called_once_with("lora")
def test_save_changes_returns_false_when_nothing_changed(self) -> None:
interface = mock.Mock()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Lora"])
self.assertFalse(save_changes(interface, {}, menu_state))
def test_save_changes_returns_false_for_non_rebooting_security_fields(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"])
reconnect_required = save_changes(interface, {"serial_enabled": True}, menu_state)
self.assertFalse(reconnect_required)
self.assertTrue(node.localConfig.security.serial_enabled)
def test_save_changes_returns_true_for_rebooting_security_fields(self) -> None:
interface, _node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Security"])
reconnect_required = save_changes(interface, {"admin_key": [b"12345678"]}, menu_state)
self.assertTrue(reconnect_required)
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"])
self.assertFalse(save_changes(interface, {"name": "renamed"}, menu_state))
self.assertEqual(node.localConfig.device.name, "renamed")
node.writeConfig.reset_mock()
self.assertTrue(save_changes(interface, {"role": "ROUTER"}, menu_state))
self.assertEqual(node.localConfig.device.role, "ROUTER")
def test_save_changes_returns_true_for_network_settings(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Network"])
reconnect_required = save_changes(interface, {"wifi_enabled": True}, menu_state)
self.assertTrue(reconnect_required)
self.assertTrue(node.localConfig.network.wifi_enabled)
def test_save_changes_returns_true_only_for_rebooting_power_fields(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Radio Settings", "Power"])
self.assertFalse(save_changes(interface, {"adc_enabled": True}, menu_state))
self.assertTrue(node.localConfig.power.adc_enabled)
node.writeConfig.reset_mock()
self.assertTrue(save_changes(interface, {"is_power_saving": True}, menu_state))
self.assertTrue(node.localConfig.power.is_power_saving)
def test_save_changes_returns_true_for_module_settings(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "Module Settings", "Mqtt"])
reconnect_required = save_changes(interface, {"enabled": True}, menu_state)
self.assertTrue(reconnect_required)
self.assertTrue(node.moduleConfig.mqtt.enabled)
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"])
reconnect_required = save_changes(interface, {"longName": "Node"}, menu_state)
self.assertTrue(reconnect_required)
node.setOwner.assert_called_once()
def test_save_changes_returns_true_for_user_license_changes(self) -> None:
interface, node = self.build_interface()
menu_state = SimpleNamespace(menu_path=["Main Menu", "User Settings"])
reconnect_required = save_changes(interface, {"isLicensed": True}, menu_state)
self.assertTrue(reconnect_required)
node.setOwner.assert_called_once()

75
tests/test_settings.py Normal file
View File

@@ -0,0 +1,75 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
import contact.settings as settings
class SettingsRuntimeTests(unittest.TestCase):
def test_main_closes_interface_after_normal_settings_exit(self) -> None:
stdscr = mock.Mock()
args = Namespace()
interface = mock.Mock()
interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=1)))
with mock.patch.object(settings, "setup_colors"):
with mock.patch.object(settings, "ensure_min_rows"):
with mock.patch.object(settings, "draw_splash"):
with mock.patch.object(settings.curses, "curs_set"):
with mock.patch.object(settings, "setup_parser") as setup_parser:
with mock.patch.object(settings, "initialize_interface", return_value=interface):
with mock.patch.object(settings, "settings_menu") as settings_menu:
setup_parser.return_value.parse_args.return_value = args
settings.main(stdscr)
settings_menu.assert_called_once_with(stdscr, interface)
interface.close.assert_called_once_with()
def test_main_closes_reconnected_interface_after_region_reset(self) -> None:
stdscr = mock.Mock()
args = Namespace()
old_interface = mock.Mock()
old_interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=0)))
new_interface = mock.Mock()
new_interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=1)))
with mock.patch.object(settings, "setup_colors"):
with mock.patch.object(settings, "ensure_min_rows"):
with mock.patch.object(settings, "draw_splash"):
with mock.patch.object(settings.curses, "curs_set"):
with mock.patch.object(settings, "setup_parser") as setup_parser:
with mock.patch.object(settings, "initialize_interface", return_value=old_interface):
with mock.patch.object(settings, "get_list_input", return_value="Yes"):
with mock.patch.object(settings, "set_region") as set_region:
with mock.patch.object(
settings, "reconnect_interface", return_value=new_interface
) as reconnect_interface:
with mock.patch.object(settings, "settings_menu") as settings_menu:
setup_parser.return_value.parse_args.return_value = args
settings.main(stdscr)
set_region.assert_called_once_with(old_interface)
reconnect_interface.assert_called_once_with(args)
settings_menu.assert_called_once_with(stdscr, new_interface)
old_interface.close.assert_called_once_with()
new_interface.close.assert_called_once_with()
def test_main_closes_interface_when_settings_menu_raises(self) -> None:
stdscr = mock.Mock()
args = Namespace()
interface = mock.Mock()
interface.localNode = SimpleNamespace(localConfig=SimpleNamespace(lora=SimpleNamespace(region=1)))
with mock.patch.object(settings, "setup_colors"):
with mock.patch.object(settings, "ensure_min_rows"):
with mock.patch.object(settings, "draw_splash"):
with mock.patch.object(settings.curses, "curs_set"):
with mock.patch.object(settings, "setup_parser") as setup_parser:
with mock.patch.object(settings, "initialize_interface", return_value=interface):
with mock.patch.object(settings, "settings_menu", side_effect=RuntimeError("boom")):
setup_parser.return_value.parse_args.return_value = args
with self.assertRaises(RuntimeError):
settings.main(stdscr)
interface.close.assert_called_once_with()

View File

@@ -81,11 +81,11 @@ class TxHandlerTests(unittest.TestCase):
with mock.patch.object(tx_handler, "update_ack_nak") as update_ack_nak:
with mock.patch("contact.message_handlers.tx_handler.time.strftime", return_value="[01:02:03] "):
with mock.patch("contact.ui.contact_ui.draw_messages_window") as draw_messages_window:
with mock.patch("contact.ui.contact_ui.request_ui_redraw") as request_ui_redraw:
tx_handler.onAckNak(packet)
update_ack_nak.assert_called_once_with("Primary", 55, "hello", "Ack")
draw_messages_window.assert_called_once_with()
request_ui_redraw.assert_called_once_with(messages=True)
self.assertIn(config.sent_message_prefix, ui_state.all_messages["Primary"][0][0])
self.assertIn(config.ack_str, ui_state.all_messages["Primary"][0][0])
@@ -100,7 +100,7 @@ class TxHandlerTests(unittest.TestCase):
with mock.patch.object(tx_handler, "update_ack_nak") as update_ack_nak:
with mock.patch("contact.message_handlers.tx_handler.time.strftime", return_value="[01:02:03] "):
with mock.patch("contact.ui.contact_ui.draw_messages_window"):
with mock.patch("contact.ui.contact_ui.request_ui_redraw"):
tx_handler.onAckNak(packet)
update_ack_nak.assert_called_once_with("Primary", 55, "hello", "Implicit")

View File

@@ -60,6 +60,28 @@ class UtilsTests(unittest.TestCase):
self.assertIn("MediumFast", ui_state.all_messages)
self.assertIn("Another Channel", ui_state.all_messages)
def test_get_channels_rebuilds_renamed_channels_and_preserves_messages(self) -> None:
interface = build_demo_interface()
interface.localNode.channels[0].settings.name = "Renamed Channel"
interface_state.interface = interface
ui_state.channel_list = ["MediumFast", "Another Channel", 2701131788]
ui_state.all_messages = {
"MediumFast": [("prefix", "first")],
"Another Channel": [("prefix", "second")],
2701131788: [("prefix", "dm")],
}
ui_state.selected_channel = 2
channels = get_channels()
self.assertEqual(channels[0], "Renamed Channel")
self.assertEqual(channels[1], "Another Channel")
self.assertEqual(channels[2], 2701131788)
self.assertEqual(ui_state.all_messages["Renamed Channel"], [("prefix", "first")])
self.assertEqual(ui_state.all_messages["Another Channel"], [("prefix", "second")])
self.assertEqual(ui_state.all_messages[2701131788], [("prefix", "dm")])
self.assertNotIn("MediumFast", ui_state.all_messages)
def test_parse_protobuf_returns_string_payload_unchanged(self) -> None:
packet = {"decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": "hello"}}