Enhance config import/export to support selected remote nodes

contact/ui/control_ui.py

@@ -512,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):
@@ -577,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
 

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 = {}

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