Compare commits

...

19 Commits

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
21 changed files with 983 additions and 178 deletions

View File

@@ -57,6 +57,8 @@ logging.basicConfig(
app_state.lock = threading.Lock()
DEFAULT_CLOSE_TIMEOUT_SECONDS = 5.0
# ------------------------------------------------------------------------------
# Main Program Logic
@@ -72,11 +74,36 @@ def prompt_region_if_unset(args: object, stdscr: Optional[curses.window] = None)
interface_state.interface = reconnect_interface(args)
def close_interface(interface: object) -> None:
def close_interface(interface: object, timeout_seconds: float = DEFAULT_CLOSE_TIMEOUT_SECONDS) -> bool:
if interface is None:
return
with contextlib.suppress(Exception):
interface.close()
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:
@@ -205,23 +232,32 @@ def start() -> None:
setup_parser().print_help()
sys.exit(0)
interrupted = False
fatal_error = None
try:
curses.wrapper(main)
close_interface(interface_state.interface)
except KeyboardInterrupt:
interrupted = True
logging.info("User exited with Ctrl+C")
close_interface(interface_state.interface)
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.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,7 +46,7 @@ def main(stdscr: curses.window) -> None:
)
if confirmation == "Yes":
set_region(interface)
interface.close()
close_interface(interface)
draw_splash(stdscr)
interface = reconnect_interface(args)
stdscr.clear()
@@ -52,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}")
@@ -961,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:
@@ -978,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:
@@ -1075,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))
@@ -1298,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()
@@ -1309,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
@@ -1346,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,6 +5,7 @@ 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
@@ -34,7 +35,7 @@ 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
@@ -43,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__))
@@ -246,6 +247,26 @@ def reconnect_after_admin_action(stdscr: object, interface: object, action, log_
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
@@ -538,7 +559,27 @@ def settings_menu(stdscr: object, interface: object) -> None:
)
if confirmation == "Yes":
interface = reconnect_after_admin_action(
stdscr, interface, interface.localNode.factoryReset, "Factory Reset Requested by menu"
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()

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

View File

@@ -3,6 +3,7 @@ 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
@@ -36,3 +37,166 @@ class ContactUiTests(unittest.TestCase):
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)

View File

@@ -1,4 +1,5 @@
from argparse import Namespace
from types import SimpleNamespace
import unittest
from unittest import mock
@@ -63,3 +64,49 @@ class ControlUiTests(unittest.TestCase):
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"])

View File

@@ -186,6 +186,15 @@ 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)
@@ -215,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)

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