Bump version to 1.6.3 in pyproject.toml

Add 'isUnmessagable' setting to user configuration

contact/localisations/en.ini

@@ -99,6 +99,7 @@ user, "User"
 longName, "Node long name", "If you are a licensed HAM operator and have enabled HAM mode, this must be set to your HAM operator call sign."
 shortName, "Node short name", "Must be up to 4 bytes. Usually this is 4 characters, if using latin characters and no emojis."
 isLicensed, "Enable licensed amateur (HAM) mode", "IMPORTANT: Read Meshtastic help documentation before enabling."
+isUnmessagable, "Disable incoming messages", "When enabled, this node advertises that it cannot receive messages."
 
 [app_settings]
 title, "App Settings", ""

contact/ui/contact_ui.py

@@ -15,11 +15,13 @@ from contact.utilities.utils import (
     build_reply_prefix,
 )
 from contact.settings import settings_menu
+from contact.ui.control_ui import RemoteAdminCancelled
 from contact.message_handlers.tx_handler import send_message, send_traceroute
 from contact.utilities.utils import parse_protobuf
 from contact.ui.colors import get_color
 from contact.utilities.db_handler import get_name_from_database, update_node_info_in_db, is_chat_archived, load_older_messages
 from contact.utilities.input_handlers import get_list_input
+from contact.utilities.interfaces import interface_is_connected
 from contact.utilities.i18n import t
 from contact.utilities.emoji_utils import normalize_message_text
 import contact.ui.default_config as config
@@ -385,11 +387,7 @@ def main_ui(stdscr: curses.window) -> None:
 
     while True:
         interface = interface_state.interface
-        if (
-            hasattr(interface, "stream")
-            and interface.stream is None
-            and not ui_state.reconnect_attempted
-        ):
+        if not interface_is_connected(interface) and not ui_state.reconnect_attempted:
             ui_state.reconnect_attempted = True
             status_win = show_connection_status(stdscr, "Disconnected", "Trying to reconnect…")
             try:
@@ -1023,6 +1021,15 @@ def handle_backtick(stdscr: curses.window) -> None:
                 wait_win = show_remote_admin_wait(stdscr, get_name_from_database(remote_node_num))
             update_remote_admin_wait(wait_win, message)
 
+        def remote_cancel_requested() -> bool:
+            if wait_win is None:
+                return False
+            try:
+                wait_win.timeout(0)
+                return wait_win.getch() == 27
+            finally:
+                wait_win.timeout(-1)
+
         try:
             settings_menu(
                 stdscr,
@@ -1032,7 +1039,11 @@ def handle_backtick(stdscr: curses.window) -> None:
                 node=interface.getNode(remote_node_num, False),
                 remote=True,
                 status_callback=remote_status,
+                cancel_callback=remote_cancel_requested,
             )
+        except RemoteAdminCancelled:
+            logging.info("Remote admin request cancelled for %s", remote_node_num)
+            remote_status(None)
         except SystemExit as exc:
             logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
             if wait_win is not None:

contact/ui/control_ui.py

@@ -41,6 +41,10 @@ help_win = None
 sensitive_settings = ["Reboot", "Reset Node DB", "Shutdown", "Factory Reset", "factory_reset_config"]
 
 
+class RemoteAdminCancelled(Exception):
+    """Raised when the user dismisses an in-progress remote-admin request."""
+
+
 # Compute the effective menu width for the current terminal
 def get_menu_width() -> int:
     # Leave at least 2 columns for borders; clamp to >= 20 for usability
@@ -280,7 +284,7 @@ def redraw_main_ui_after_reconnect(stdscr: object) -> None:
         logging.debug("Skipping main UI redraw after reconnect", exc_info=True)
 
 
-def _request_remote_with_timeout(request, *args) -> None:
+def _request_remote_with_timeout(request, *args, cancel_callback=None) -> None:
     """Run a Meshtastic remote request without allowing its ACK wait to hang the UI."""
     failure = []
 
@@ -292,24 +296,32 @@ def _request_remote_with_timeout(request, *args) -> None:
 
     worker = threading.Thread(target=run_request, name="remote-admin-request", daemon=True)
     worker.start()
-    worker.join(REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS)
+    deadline = time.monotonic() + REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS
+    while worker.is_alive() and time.monotonic() < deadline:
+        worker.join(0.1)
+        if cancel_callback and cancel_callback():
+            raise RemoteAdminCancelled()
     if worker.is_alive():
         raise TimeoutError("Timed out waiting for the remote node to answer the admin request.")
     if failure:
         raise failure[0]
 
 
-def _request_remote_channels_with_timeout(node: object) -> None:
+def _request_remote_channels_with_timeout(node: object, cancel_callback=None) -> None:
     """Retrieve a remote node's channel list without leaving the UI waiting forever."""
     node.requestChannels()
     deadline = time.monotonic() + REMOTE_ADMIN_REQUEST_TIMEOUT_SECONDS
     while getattr(node, "channels", None) is None:
+        if cancel_callback and cancel_callback():
+            raise RemoteAdminCancelled()
         if time.monotonic() >= deadline:
             raise TimeoutError("Timed out waiting for the remote node's channel list.")
         time.sleep(0.1)
 
 
-def _request_remote_section(node: object, menu_path: List[str], selected_option: str, status_callback=None) -> bool:
+def _request_remote_section(
+    node: object, menu_path: List[str], selected_option: str, status_callback=None, cancel_callback=None
+) -> bool:
     """Fetch exactly the remote-admin section the user is opening.
 
     Meshtastic returns radio and module settings one protobuf section at a
@@ -319,7 +331,7 @@ def _request_remote_section(node: object, menu_path: List[str], selected_option:
     if menu_path == ["Main Menu"] and selected_option == "Channels":
         if status_callback:
             status_callback("Requesting channel list…")
-        _request_remote_channels_with_timeout(node)
+        _request_remote_channels_with_timeout(node, cancel_callback=cancel_callback)
         return True
 
     if len(menu_path) != 2:
@@ -338,12 +350,17 @@ def _request_remote_section(node: object, menu_path: List[str], selected_option:
         return False
     if status_callback:
         status_callback(f"Requesting {selected_option} config…")
-    _request_remote_with_timeout(node.requestConfig, field)
+    _request_remote_with_timeout(node.requestConfig, field, cancel_callback=cancel_callback)
     return True
 
 
 def settings_menu(
-    stdscr: object, interface: object, node: object = None, remote: bool = False, status_callback=None
+    stdscr: object,
+    interface: object,
+    node: object = None,
+    remote: bool = False,
+    status_callback=None,
+    cancel_callback=None,
 ) -> None:
     curses.update_lines_cols()
     global admin_target_label
@@ -495,7 +512,7 @@ def settings_menu(
                     filename += ".yaml"
 
                 try:
-                    config_text = config_export(interface)
+                    config_text = config_export(interface, node=node)
                     yaml_file_path = os.path.join(config_folder, filename)
 
                     if os.path.exists(yaml_file_path):
@@ -560,7 +577,7 @@ def settings_menu(
                         ["Yes", "No"],
                     )
                     if overwrite == "Yes":
-                        config_import(interface, file_path)
+                        config_import(interface, file_path, node=node)
                 menu_state.start_index.pop()
                 continue
 
@@ -700,7 +717,7 @@ def settings_menu(
                     menu_state.need_redraw = True
                     continue
 
-                if selected_option in ["longName", "shortName", "isLicensed"]:
+                if selected_option in ["longName", "shortName", "isLicensed", "isUnmessagable"]:
                     if selected_option in ["longName", "shortName"]:
                         new_value = get_text_input(
                             f"{human_readable_name} is currently: {current_value}", selected_option, None
@@ -708,7 +725,7 @@ def settings_menu(
                         new_value = current_value if new_value is None else new_value
                         menu_state.current_menu[selected_option] = (field, new_value)
 
-                    elif selected_option == "isLicensed":
+                    elif selected_option in ["isLicensed", "isUnmessagable"]:
                         new_value = get_list_input(
                             f"{human_readable_name} is currently: {current_value}",
                             str(current_value),
@@ -812,7 +829,7 @@ def settings_menu(
                 if remote:
                     try:
                         requested_remote_section = _request_remote_section(
-                            node, menu_state.menu_path, selected_option, status_callback
+                            node, menu_state.menu_path, selected_option, status_callback, cancel_callback
                         )
                     finally:
                         if status_callback:

contact/ui/menus.py

@@ -76,6 +76,7 @@ def generate_menu_from_protobuf(interface: object, node: Any = None, include_app
                 "longName": (None, current_user_config.get("longName", "Not Set")),
                 "shortName": (None, current_user_config.get("shortName", "Not Set")),
                 "isLicensed": (None, current_user_config.get("isLicensed", "False")),
+                "isUnmessagable": (None, current_user_config.get("isUnmessagable", "False")),
             }
         else:
             logging.info("User settings not found in Node Info")

contact/utilities/config_io.py

@@ -134,47 +134,47 @@ def setPref(config, comp_name, raw_val) -> bool:
     return True
 
 
-def config_import(interface, filename):
+def config_import(interface, filename, node=None):
+    """Apply a configuration file to the selected local or remote node."""
+    node = node or interface.localNode
     with open(filename, encoding="utf8") as file:
         configuration = yaml.safe_load(file)
         closeNow = True
 
-        interface.getNode("^local", False).beginSettingsTransaction()
+        node.beginSettingsTransaction()
 
         if "owner" in configuration:
             logging.info(f"Setting device owner to {configuration['owner']}")
             waitForAckNak = True
-            interface.getNode("^local", False).setOwner(configuration["owner"])
+            node.setOwner(configuration["owner"])
             time.sleep(0.5)
 
         if "owner_short" in configuration:
             logging.info(f"Setting device owner short to {configuration['owner_short']}")
             waitForAckNak = True
-            interface.getNode("^local", False).setOwner(long_name=None, short_name=configuration["owner_short"])
+            node.setOwner(long_name=None, short_name=configuration["owner_short"])
             time.sleep(0.5)
 
         if "ownerShort" in configuration:
             logging.info(f"Setting device owner short to {configuration['ownerShort']}")
             waitForAckNak = True
-            interface.getNode("^local", False).setOwner(long_name=None, short_name=configuration["ownerShort"])
+            node.setOwner(long_name=None, short_name=configuration["ownerShort"])
             time.sleep(0.5)
 
         if "channel_url" in configuration:
             logging.info(f"Setting channel url to {configuration['channel_url']}")
-            interface.getNode("^local").setURL(configuration["channel_url"])
+            node.setURL(configuration["channel_url"])
             time.sleep(0.5)
 
         if "channelUrl" in configuration:
             logging.info(f"Setting channel url to {configuration['channelUrl']}")
-            interface.getNode("^local").setURL(configuration["channelUrl"])
+            node.setURL(configuration["channelUrl"])
             time.sleep(0.5)
 
         if "location" in configuration:
             alt = 0
             lat = 0.0
             lon = 0.0
-            localConfig = interface.localNode.localConfig
-
             if "alt" in configuration["location"]:
                 alt = int(configuration["location"]["alt"] or 0)
                 logging.info(f"Fixing altitude at {alt} meters")
@@ -185,39 +185,44 @@ def config_import(interface, filename):
                 lon = float(configuration["location"]["lon"] or 0)
                 logging.info(f"Fixing longitude at {lon} degrees")
             logging.info("Setting device position")
-            interface.localNode.setFixedPosition(lat, lon, alt)
+            node.setFixedPosition(lat, lon, alt)
             time.sleep(0.5)
 
         if "config" in configuration:
-            localConfig = interface.getNode("^local").localConfig
+            localConfig = node.localConfig
             for section in configuration["config"]:
                 traverseConfig(section, configuration["config"][section], localConfig)
-                interface.getNode("^local").writeConfig(camel_to_snake(section))
+                node.writeConfig(camel_to_snake(section))
                 time.sleep(0.5)
 
         if "module_config" in configuration:
-            moduleConfig = interface.getNode("^local").moduleConfig
+            moduleConfig = node.moduleConfig
             for section in configuration["module_config"]:
                 traverseConfig(
                     section,
                     configuration["module_config"][section],
                     moduleConfig,
                 )
-                interface.getNode("^local").writeConfig(camel_to_snake(section))
+                node.writeConfig(camel_to_snake(section))
                 time.sleep(0.5)
 
-        interface.getNode("^local", False).commitSettingsTransaction()
+        node.commitSettingsTransaction()
         logging.info("Writing modified configuration to device")
 
 
-def config_export(interface) -> str:
-    """used in --export-config"""
+def config_export(interface, node=None) -> str:
+    """Export the selected node's configuration as Meshtastic configure YAML."""
     configObj = {}
+    node = node or interface.localNode
 
-    owner = interface.getLongName()
-    owner_short = interface.getShortName()
-    channel_url = interface.localNode.getURL()
-    myinfo = interface.getMyNodeInfo()
+    if node is interface.localNode:
+        myinfo = interface.getMyNodeInfo()
+    else:
+        myinfo = getattr(interface, "nodesByNum", {}).get(getattr(node, "nodeNum", None), {})
+    user = myinfo.get("user", {}) if isinstance(myinfo, dict) else {}
+    owner = user.get("longName")
+    owner_short = user.get("shortName")
+    channel_url = node.getURL()
     pos = myinfo.get("position")
     lat = None
     lon = None
@@ -242,7 +247,7 @@ def config_export(interface) -> str:
         if alt:
             configObj["location"]["alt"] = alt
 
-    config = MessageToDict(interface.localNode.localConfig)  # checkme - Used as a dictionary here and a string below
+    config = MessageToDict(node.localConfig)  # checkme - Used as a dictionary here and a string below
     if config:
         # Convert inner keys to correct snake/camelCase
         prefs = {}
@@ -265,7 +270,7 @@ def config_export(interface) -> str:
         else:
             configObj["config"] = config
 
-    module_config = MessageToDict(interface.localNode.moduleConfig)
+    module_config = MessageToDict(node.moduleConfig)
     if module_config:
         # Convert inner keys to correct snake/camelCase
         prefs = {}

contact/utilities/interfaces.py

@@ -3,6 +3,28 @@ import time
 import meshtastic.serial_interface, meshtastic.tcp_interface, meshtastic.ble_interface
 
 
+def interface_is_connected(interface) -> bool:
+    """Return the transport-independent connection state for an interface.
+
+    TCPInterface deliberately leaves ``stream`` set to ``None`` and uses its
+    ``socket`` instead.  Checking only ``stream`` therefore treats every TCP
+    connection (including meshtasticd) as disconnected.
+    """
+    if interface is None:
+        return False
+
+    connected = getattr(interface, "isConnected", None)
+    if hasattr(connected, "is_set"):
+        return connected.is_set()
+    if isinstance(connected, bool):
+        return connected
+    if hasattr(interface, "socket"):
+        return getattr(interface, "socket", None) is not None
+    if hasattr(interface, "stream"):
+        return getattr(interface, "stream", None) is not None
+    return getattr(interface, "localNode", None) is not None
+
+
 def initialize_interface(args):
     try:
 
@@ -50,9 +72,9 @@ def reconnect_interface(args, attempts: int = 20, delay_seconds: float = 1.0):
     for attempt in range(attempts):
         try:
             interface = initialize_interface(args)
-            if interface is not None and getattr(interface, "localNode", None) is not None and getattr(
+            if interface_is_connected(interface) and getattr(interface, "localNode", None) is not None and getattr(
                 interface.localNode, "localConfig", None
-            ) is not None and (not hasattr(interface, "stream") or interface.stream is not None):
+            ) is not None:
                 return interface
             last_error = RuntimeError("interface did not complete connection setup")
             try:

contact/utilities/save_to_radio.py

@@ -29,7 +29,7 @@ LORA_REBOOT_KEYS = {
     "sx126x_rx_boosted_gain",
 }
 SECURITY_NON_REBOOT_KEYS = {"debug_log_api_enabled", "serial_enabled"}
-USER_RECONNECT_KEYS = {"longName", "shortName", "isLicensed", "is_licensed"}
+USER_RECONNECT_KEYS = {"longName", "shortName", "isLicensed", "is_licensed", "isUnmessagable", "is_unmessagable"}
 
 
 def _collect_changed_keys(modified_settings):
@@ -143,12 +143,16 @@ def save_changes(interface, modified_settings, menu_state, node=None):
             long_name = modified_settings.get("longName")
             short_name = modified_settings.get("shortName")
             is_licensed = modified_settings.get("isLicensed")
+            is_unmessagable = modified_settings.get("isUnmessagable")
             is_licensed = is_licensed == "True" or is_licensed is True  # Normalize boolean
+            if is_unmessagable is not None:
+                is_unmessagable = is_unmessagable == "True" or is_unmessagable is True
 
-            node.setOwner(long_name, short_name, is_licensed)
+            node.setOwner(long_name, short_name, is_licensed, is_unmessagable)
 
             logging.info(
-                f"Updated {config_category} with Long Name: {long_name}, Short Name: {short_name}, Licensed Mode: {is_licensed}"
+                f"Updated {config_category} with Long Name: {long_name}, Short Name: {short_name}, "
+                f"Licensed Mode: {is_licensed}, Unmessageable: {is_unmessagable}"
             )
 
             return _requires_reconnect(menu_state, modified_settings)

pyproject.toml

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

tests/test_config_io.py

@@ -1,6 +1,12 @@
 import unittest
+from types import SimpleNamespace
+from tempfile import NamedTemporaryFile
+from unittest.mock import Mock
 
-from contact.utilities.config_io import _is_repeated_field, splitCompoundName
+import yaml
+from meshtastic.protobuf import config_pb2, module_config_pb2
+
+from contact.utilities.config_io import _is_repeated_field, config_export, config_import, splitCompoundName
 
 
 class ConfigIoTests(unittest.TestCase):
@@ -19,3 +25,42 @@ class ConfigIoTests(unittest.TestCase):
         field_type = type("Field", (), {"label": 3, "LABEL_REPEATED": 3})
 
         self.assertTrue(_is_repeated_field(field_type()))
+
+    def test_config_export_uses_selected_remote_node(self) -> None:
+        local_node = SimpleNamespace(nodeNum=1)
+        remote_node = SimpleNamespace(
+            nodeNum=2,
+            localConfig=config_pb2.Config(),
+            moduleConfig=module_config_pb2.ModuleConfig(),
+            getURL=lambda: "remote-channel-url",
+        )
+        interface = SimpleNamespace(
+            localNode=local_node,
+            nodesByNum={
+                2: {
+                    "user": {"longName": "Remote Node", "shortName": "REM"},
+                    "position": {"latitude": 1.25, "longitude": 2.5, "altitude": 12},
+                }
+            },
+        )
+
+        exported = yaml.safe_load(config_export(interface, node=remote_node))
+
+        self.assertEqual(exported["owner"], "Remote Node")
+        self.assertEqual(exported["owner_short"], "REM")
+        self.assertEqual(exported["channel_url"], "remote-channel-url")
+        self.assertEqual(exported["location"], {"lat": 1.25, "lon": 2.5, "alt": 12})
+
+    def test_config_import_uses_selected_remote_node(self) -> None:
+        remote_node = Mock()
+        interface = SimpleNamespace(localNode=Mock())
+        with NamedTemporaryFile("w", suffix=".yaml") as config_file:
+            yaml.safe_dump({"owner": "Remote Node"}, config_file)
+            config_file.flush()
+
+            config_import(interface, config_file.name, node=remote_node)
+
+        remote_node.beginSettingsTransaction.assert_called_once_with()
+        remote_node.setOwner.assert_called_once_with("Remote Node")
+        remote_node.commitSettingsTransaction.assert_called_once_with()
+        interface.localNode.setOwner.assert_not_called()

tests/test_menus.py

@@ -44,3 +44,20 @@ class MenusTests(unittest.TestCase):
 
         self.assertEqual(module_settings["external_notification"]["ringtone"], (None, "tone"))
         self.assertEqual(module_settings["canned_message"]["messages"], (None, "Hi|Bye"))
+
+    def test_user_settings_include_unmessageable_toggle(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", "isUnmessagable": True},
+                "position": {},
+            },
+        )
+
+        user_settings = generate_menu_from_protobuf(interface)["Main Menu"]["User Settings"]
+        self.assertEqual(user_settings["isUnmessagable"], (None, True))

tests/test_save_to_radio.py

@@ -112,3 +112,12 @@ class SaveToRadioTests(unittest.TestCase):
 
         self.assertTrue(reconnect_required)
         node.setOwner.assert_called_once()
+
+    def test_save_changes_sends_user_unmessageable_setting(self) -> None:
+        interface, node = self.build_interface()
+        menu_state = SimpleNamespace(menu_path=["Main Menu", "User Settings"])
+
+        reconnect_required = save_changes(interface, {"isUnmessagable": True}, menu_state)
+
+        self.assertTrue(reconnect_required)
+        self.assertEqual(node.setOwner.call_args.args[3], True)