Add cli option to export nodes

meshtastic/__main__.py

@@ -1018,6 +1018,16 @@ def setSimpleConfig(modem_preset):
             print("--show-fields can only be used with --nodes")
             return
 
+        if args.export_nodes:
+            closeNow = True
+            if args.dest != BROADCAST_ADDR:
+                print("Exporting nodes of a remote node is not supported.")
+                return
+            # Determine format from file extension
+            filename = args.export_nodes
+            format_type = "csv" if filename.lower().endswith('.csv') else "json"
+            interface.exportNodeDb(filename, format=format_type)
+
         if args.qr or args.qr_all:
             closeNow = True
             url = interface.getNode(args.dest, True, **getNode_kwargs).getURL(includeAll=args.qr_all)
@@ -1822,6 +1832,12 @@ def addLocalActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPars
         default=None
     )
 
+    group.add_argument(
+        "--export-nodes",
+        help="Export node database to a file. Specify filename with .json or .csv extension to determine format (default: nodes.json)",
+        metavar="FILENAME",
+    )
+
     return parser
 
 def addRemoteActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:

meshtastic/mesh_interface.py

@@ -3,6 +3,7 @@
 # pylint: disable=R0917,C0302
 
 import collections
+import csv
 import json
 import logging
 import math
@@ -375,6 +376,98 @@ def getNestedValue(node_dict: Dict[str, Any], key_path: str) -> Any:
         print(table)
         return table
 
+    def exportNodeDb(self, filename: str, format: str = "json") -> None:
+        """Export the node database to a file
+
+        Args:
+            filename: Path to the output file
+            format: Export format - either "json" or "csv" (default: "json")
+        """
+        if not self.nodesByNum:
+            logger.warning("No nodes in database to export")
+            print("Warning: No nodes in database to export")
+            return
+
+        # Prepare node data for export
+        nodes_data = []
+        for node in self.nodesByNum.values():
+            # Remove raw protobuf data and other non-serializable fields
+            keys_to_remove = ("raw", "decoded", "payload")
+            node_clean = remove_keys_from_dict(keys_to_remove, node.copy())
+
+            # Flatten the structure for easier access
+            export_node = {
+                "num": node_clean.get("num"),
+                "user_id": node_clean.get("user", {}).get("id"),
+                "long_name": node_clean.get("user", {}).get("longName"),
+                "short_name": node_clean.get("user", {}).get("shortName"),
+                "hw_model": node_clean.get("user", {}).get("hwModel"),
+                "role": node_clean.get("user", {}).get("role"),
+                "latitude": node_clean.get("position", {}).get("latitude"),
+                "longitude": node_clean.get("position", {}).get("longitude"),
+                "altitude": node_clean.get("position", {}).get("altitude"),
+                "last_heard": node_clean.get("lastHeard"),
+                "snr": node_clean.get("snr"),
+                "hops_away": node_clean.get("hopsAway"),
+                "channel": node_clean.get("channel", 0),
+                "via_mqtt": node_clean.get("viaMqtt", False),
+                "is_favorite": node_clean.get("isFavorite", False),
+            }
+
+            # Add device metrics if available
+            if "deviceMetrics" in node_clean:
+                metrics = node_clean["deviceMetrics"]
+                export_node["battery_level"] = metrics.get("batteryLevel")
+                export_node["voltage"] = metrics.get("voltage")
+                export_node["channel_utilization"] = metrics.get("channelUtilization")
+                export_node["air_util_tx"] = metrics.get("airUtilTx")
+
+            nodes_data.append(export_node)
+
+        # Export based on format
+        if format.lower() == "csv":
+            self._exportNodeDbCsv(filename, nodes_data)
+        else:
+            self._exportNodeDbJson(filename, nodes_data)
+
+    def _exportNodeDbJson(self, filename: str, nodes_data: List[Dict]) -> None:
+        """Export node database as JSON"""
+        try:
+            with open(filename, 'w', encoding='utf-8') as f:
+                json.dump({
+                    "export_date": datetime.now().isoformat(),
+                    "nodes": nodes_data
+                }, f, indent=2)
+            print(f"Node database exported to {filename} ({len(nodes_data)} nodes)")
+            logger.info(f"Node database exported to {filename}")
+        except Exception as e:
+            logger.error(f"Failed to export node database: {e}")
+            print(f"Error: Failed to export node database: {e}")
+
+    def _exportNodeDbCsv(self, filename: str, nodes_data: List[Dict]) -> None:
+        """Export node database as CSV"""
+        try:
+            if not nodes_data:
+                print("Warning: No nodes to export")
+                return
+
+            # Get all possible field names
+            fieldnames = set()
+            for node in nodes_data:
+                fieldnames.update(node.keys())
+            fieldnames = sorted(fieldnames)
+
+            with open(filename, 'w', newline='', encoding='utf-8') as f:
+                writer = csv.DictWriter(f, fieldnames=fieldnames)
+                writer.writeheader()
+                writer.writerows(nodes_data)
+
+            print(f"Node database exported to {filename} ({len(nodes_data)} nodes)")
+            logger.info(f"Node database exported to {filename}")
+        except Exception as e:
+            logger.error(f"Failed to export node database: {e}")
+            print(f"Error: Failed to export node database: {e}")
+
     def getNode(
         self, nodeId: str, requestChannels: bool = True, requestChannelAttempts: int = 3, timeout: int = 300
     ) -> meshtastic.node.Node: