From 66256f0a4c6d032fb5b757430b30267e0812bb4d Mon Sep 17 00:00:00 2001 From: pdxlocations Date: Sat, 25 Jul 2026 22:30:33 -0700 Subject: [PATCH] Enhance config import/export to support selected remote nodes --- contact/ui/control_ui.py | 4 +-- contact/utilities/config_io.py | 51 +++++++++++++++++++--------------- tests/test_config_io.py | 47 ++++++++++++++++++++++++++++++- 3 files changed, 76 insertions(+), 26 deletions(-) diff --git a/contact/ui/control_ui.py b/contact/ui/control_ui.py index d387514..1dbc217 100644 --- a/contact/ui/control_ui.py +++ b/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 diff --git a/contact/utilities/config_io.py b/contact/utilities/config_io.py index 35c366c..6a5d9bf 100644 --- a/contact/utilities/config_io.py +++ b/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 = {} diff --git a/tests/test_config_io.py b/tests/test_config_io.py index f354f57..17e3b25 100644 --- a/tests/test_config_io.py +++ b/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()