mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
Compare commits
19 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bc69c709ff | ||
|
|
6453865cba | ||
|
|
090e89d92c | ||
|
|
07716c1719 | ||
|
|
da5b102047 | ||
|
|
1668b68c4f | ||
|
|
fd4b9e2174 | ||
|
|
8f32e2c99c | ||
|
|
b53dab1840 | ||
|
|
f2904eb550 | ||
|
|
480c32ba56 | ||
|
|
b4b084b627 | ||
|
|
5940c9b02b | ||
|
|
c492c96685 | ||
|
|
90376d35f3 | ||
|
|
4b35a74e2c | ||
|
|
ecc5308dad | ||
|
|
8f376edabe | ||
|
|
e5ef87ec19 |
@@ -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()
|
||||
|
||||
@@ -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", ""
|
||||
|
||||
197
contact/localisations/fr.ini
Normal file
197
contact/localisations/fr.ini
Normal 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", ""
|
||||
@@ -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, "Подождите", ""
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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:
|
||||
"""
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"])
|
||||
|
||||
@@ -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
28
tests/test_menus.py
Normal 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
36
tests/test_nav_utils.py
Normal 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)],
|
||||
)
|
||||
@@ -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
75
tests/test_settings.py
Normal 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()
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user