Enable --info to return data as JSON

meshtastic/__main__.py

@@ -44,6 +44,8 @@
 from meshtastic import BROADCAST_ADDR, mt_config, remote_hardware
 from meshtastic.ble_interface import BLEInterface
 from meshtastic.mesh_interface import MeshInterface
+from meshtastic.formatter import InfoFormatter
+
 try:
     from meshtastic.powermon import (
         PowerMeter,
@@ -978,47 +980,43 @@ def setSimpleConfig(modem_preset):
             ringtone = interface.getNode(args.dest, **getNode_kwargs).get_ringtone()
             print(f"ringtone:{ringtone}")
 
+        if args.get:
+            closeNow = True
+            node = interface.getNode(args.dest, False, **getNode_kwargs)
+            for pref in args.get:
+                found = getPref(node, pref[0])
+
+            if found:
+                print("Completed getting preferences")
+
         if args.info:
-            print("")
             # If we aren't trying to talk to our local node, don't show it
             if args.dest == BROADCAST_ADDR:
-                interface.showInfo()
-                print("")
-                interface.getNode(args.dest, **getNode_kwargs).showInfo()
+                infodata = interface.getInfo()
+                infodata.update(interface.getNode(args.dest, **getNode_kwargs).getInfo())
+                InfoFormatter().format(infodata, args.fmt)
                 closeNow = True
-                print("")
-                pypi_version = meshtastic.util.check_if_newer_version()
-                if pypi_version:
-                    print(
-                        f"*** A newer version v{pypi_version} is available!"
-                        ' Consider running "pip install --upgrade meshtastic" ***\n'
-                    )
             else:
                 print("Showing info of remote node is not supported.")
                 print(
                     "Use the '--get' command for a specific configuration (e.g. 'lora') instead."
                 )
 
-        if args.get:
-            closeNow = True
-            node = interface.getNode(args.dest, False, **getNode_kwargs)
-            for pref in args.get:
-                found = getPref(node, pref[0])
-
-            if found:
-                print("Completed getting preferences")
-
         if args.nodes:
             closeNow = True
             if args.dest != BROADCAST_ADDR:
                 print("Showing node list of a remote node is not supported.")
                 return
-            interface.showNodes(True, args.show_fields)
+            interface.showNodes(True, showFields=args.show_fields, printFmt=args.fmt)
 
         if args.show_fields and not args.nodes:
             print("--show-fields can only be used with --nodes")
             return
 
+        if args.fmt and not (args.nodes or args.info):
+            print("--fmt can only be used with --nodes or --info")
+            return
+
         if args.qr or args.qr_all:
             closeNow = True
             url = interface.getNode(args.dest, True, **getNode_kwargs).getURL(includeAll=args.qr_all)
@@ -1832,6 +1830,13 @@ def addLocalActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentPars
         default=None
     )
 
+    group.add_argument(
+        "--fmt",
+        help="Specify format to show when using --nodes/--info",
+        type=str,
+        default=None
+    )
+
     return parser
 
 def addRemoteActionArgs(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:

meshtastic/formatter.py

@@ -0,0 +1,110 @@
+import json
+
+from meshtastic.util import pskToString, check_if_newer_version
+from meshtastic.protobuf import channel_pb2
+
+"""Defines the formatting of outputs using factories"""
+
+class FormatterFactory:
+    """Factory of formatters"""
+    def __init__(self):
+        self.infoFormatters = {
+            "json": FormatAsJson,
+            "default": FormatAsText
+        }
+
+    def getInfoFormatter(self, formatSpec: str = 'default'):
+        """returns the formatter for info data. If no valid formatter is found, default to text"""
+        return self.infoFormatters.get(formatSpec.lower(), self.infoFormatters["default"])
+
+
+class InfoFormatter:
+    """responsible to format info data"""
+    def format(self, data: dict, formatSpec: str | None = None) -> str:
+        """returns formatted string according to formatSpec for info data"""
+        if not formatSpec:
+            formatSpec = 'default'
+        formatter = FormatterFactory().getInfoFormatter(formatSpec)
+        return formatter().formatInfo(data)
+
+
+class AbstractFormatter:
+    """Abstract base class for all derived formatters"""
+    @property
+    def getType(self) -> str:
+        return type(self).__name__
+
+    def formatInfo(self, data: dict):
+        """interface definition for formatting info data
+        Need to be implemented in the derived class"""
+        raise NotImplementedError
+
+
+class FormatAsJson(AbstractFormatter):
+    """responsible to return the data as JSON string"""
+    def formatInfo(self, data: dict) -> str:
+        """Info as JSON"""
+
+        # Remove the bytes entry of PSK before serialization of JSON
+        if 'Channels' in data:
+            for c in data['Channels']:
+                if '__psk__' in c:
+                    del c['__psk__']
+        jsonData = json.dumps(data, indent=2)
+        print(jsonData)
+        return jsonData
+
+
+class FormatAsText(AbstractFormatter):
+    """responsible to print the data. No string return"""
+    def formatInfo(self, data: dict) -> str:
+        """Info printed as plain text"""
+        print("")
+        self.showMeshInfo(data)
+        self.showNodeInfo(data)
+        print("")
+        pypi_version = check_if_newer_version()
+        if pypi_version:
+            print(
+                f"*** A newer version v{pypi_version} is available!"
+                ' Consider running "pip install --upgrade meshtastic" ***\n'
+            )
+        return ""
+
+    def showMeshInfo(self, data: dict):
+        """Show human-readable summary about mesh interface data"""
+        owner = f"Owner: {data['Owner'][0]}({data['Owner'][1]})"
+
+        myinfo = ""
+        if (dx := data.get('MyInfo', None)) is not None:
+            myinfo = f"My info: {json.dumps(dx)}"
+
+        metadata = ""
+        if (dx := data.get('Metadata', None)) is not None:
+            metadata = f"Metadata: {json.dumps(dx)}"
+
+        mesh = f"\nNodes in mesh:{json.dumps(data.get('Nodes', {}), indent=2)}"
+
+        infos = f"{owner}\n{myinfo}\n{metadata}\n{mesh}"
+        print(infos)
+
+    def showNodeInfo(self, data: dict):
+        """Show human-readable description of our node"""
+        if (dx := data.get('Preferences', None)) is not None:
+            print(f"Preferences: {json.dumps(dx, indent=2)}")
+
+        if (dx := data.get('ModulePreferences', None)) is not None:
+            print(f"Module preferences: {json.dumps(dx, indent=2)}")
+
+        if (dx := data.get('Channels', None)) is not None:
+            print("Channels:")
+            for idx, c in enumerate(dx):
+                if channel_pb2.Channel.Role.Name(c['role']) != "DISABLED":
+                    print(f"  Index {idx}: {channel_pb2.Channel.Role.Name(c['role'])} psk={pskToString(c['__psk__'])} {json.dumps(c['settings'])}")
+            print("")
+            publicURL = data.get('publicURL', None)
+            if publicURL:
+                print(f"\nPrimary channel URL: {publicURL}")
+            adminURL = data.get('adminURL', None)
+            if adminURL and adminURL != publicURL:
+                print(f"Complete URL (includes all channels): {adminURL}")

meshtastic/mesh_interface.py

@@ -17,6 +17,7 @@
 from typing import Any, Callable, Dict, List, Optional, Union
 
 import google.protobuf.json_format
+from google.protobuf.json_format import MessageToDict
 
 try:
     import print_color  # type: ignore[import-untyped]
@@ -192,40 +193,49 @@ def _handleLogRecord(self, record: mesh_pb2.LogRecord) -> None:
         # For now we just try to format the line as if it had come in over the serial port
         self._handleLogLine(record.message)
 
-    def showInfo(self, file=sys.stdout) -> str:  # pylint: disable=W0613
-        """Show human readable summary about this object"""
-        owner = f"Owner: {self.getLongName()} ({self.getShortName()})"
-        myinfo = ""
+    def getInfo(self) -> dict:
+        """retrieve object data"""
+        objData: dict[str, Any] = {
+            "Owner": [self.getLongName(), self.getShortName()],
+            "MyInfo": {},
+            "Metadata": {},
+            "Nodes": {}
+        }
+
         if self.myInfo:
-            myinfo = f"\nMy info: {message_to_json(self.myInfo)}"
-        metadata = ""
+            objData["MyInfo"] = MessageToDict(self.myInfo)
         if self.metadata:
-            metadata = f"\nMetadata: {message_to_json(self.metadata)}"
-        mesh = "\n\nNodes in mesh: "
-        nodes = {}
+            objData["Metadata"] = MessageToDict(self.metadata)
+
+        keys_to_remove = ("raw", "decoded", "payload")
         if self.nodes:
             for n in self.nodes.values():
                 # when the TBeam is first booted, it sometimes shows the raw data
                 # so, we will just remove any raw keys
-                keys_to_remove = ("raw", "decoded", "payload")
                 n2 = remove_keys_from_dict(keys_to_remove, n)
 
                 # if we have 'macaddr', re-format it
-                if "macaddr" in n2["user"]:
-                    val = n2["user"]["macaddr"]
-                    # decode the base64 value
-                    addr = convert_mac_addr(val)
-                    n2["user"]["macaddr"] = addr
+                self.reformatMAC(n2)
 
                 # use id as dictionary key for correct json format in list of nodes
                 nodeid = n2["user"]["id"]
-                nodes[nodeid] = n2
-        infos = owner + myinfo + metadata + mesh + json.dumps(nodes, indent=2)
-        print(infos)
-        return infos
+                # nodes[nodeid] = n2
+                objData["Nodes"][nodeid] = n2
+        return objData
 
-    def showNodes(
-        self, includeSelf: bool = True, showFields: Optional[List[str]] = None
+    @staticmethod
+    def reformatMAC(n2: dict):
+        """reformat MAC address to hex format"""
+        if "macaddr" in n2["user"]:
+            val = n2["user"]["macaddr"]
+            # decode the base64 value
+            addr = convert_mac_addr(val)
+            n2["user"]["macaddr"] = addr
+
+    def showNodes(self,
+                  includeSelf: bool = True,
+                  showFields: Optional[List[str]] = None,
+                  printFmt: Optional[str] = None
     ) -> str:  # pylint: disable=W0613
         """Show table summary of nodes in mesh
 
@@ -372,7 +382,16 @@ def getNestedValue(node_dict: Dict[str, Any], key_path: str) -> Any:
         for i, row in enumerate(rows):
             row["N"] = i + 1
 
-        table = tabulate(rows, headers="keys", missingval="N/A", tablefmt="fancy_grid")
+        if not printFmt or len(printFmt) == 0:
+            printFmt = "fancy_grid"
+        if printFmt.lower() == 'json':
+            headers = []
+            if len(rows) > 0:
+                headers = list(rows[0].keys())
+            outDict = {'headers': headers, 'nodes': rows}
+            table = json.dumps(outDict)
+        else:
+            table = tabulate(rows, headers="keys", missingval="N/A", tablefmt=printFmt)
         print(table)
         return table
 

meshtastic/node.py

@@ -2,11 +2,14 @@
 """
 
 import base64
+import json
 import logging
 import time
 
 from typing import Optional, Union, List
 
+from google.protobuf.json_format import MessageToDict
+
 from meshtastic.protobuf import admin_pb2, apponly_pb2, channel_pb2, config_pb2, localonly_pb2, mesh_pb2, portnums_pb2
 from meshtastic.util import (
     Timeout,
@@ -75,35 +78,29 @@ def module_available(self, excluded_bit: int) -> bool:
         except Exception:
             return True
 
-    def showChannels(self):
-        """Show human readable description of our channels."""
-        print("Channels:")
+    def getChannelInfo(self) -> dict:
+        """Return description of our channels as dict."""
+        # print("Channels:")
+        chanConfig = {}
         if self.channels:
             logger.debug(f"self.channels:{self.channels}")
-            for c in self.channels:
-                cStr = message_to_json(c.settings)
-                # don't show disabled channels
-                if channel_pb2.Channel.Role.Name(c.role) != "DISABLED":
-                    print(
-                        f"  Index {c.index}: {channel_pb2.Channel.Role.Name(c.role)} psk={pskToString(c.settings.psk)} {cStr}"
-                    )
+            chanConfig = [{"role": c.role, "__psk__": c.settings.psk, "settings": MessageToDict(c.settings, always_print_fields_with_no_presence=True)} for c in self.channels]
         publicURL = self.getURL(includeAll=False)
         adminURL = self.getURL(includeAll=True)
-        print(f"\nPrimary channel URL: {publicURL}")
-        if adminURL != publicURL:
-            print(f"Complete URL (includes all channels): {adminURL}")
+        return {"Channels": chanConfig, "publicURL": publicURL, "adminURL": adminURL}
 
-    def showInfo(self):
-        """Show human readable description of our node"""
-        prefs = ""
+    def getInfo(self) ->dict:
+        """Return preferences of our node as dictionary"""
+        locConfig = {}
         if self.localConfig:
-            prefs = message_to_json(self.localConfig, multiline=True)
-        print(f"Preferences: {prefs}\n")
-        prefs = ""
+            locConfig = MessageToDict(self.localConfig)
+        modConfig = {}
         if self.moduleConfig:
-            prefs = message_to_json(self.moduleConfig, multiline=True)
-        print(f"Module preferences: {prefs}\n")
-        self.showChannels()
+            modConfig = MessageToDict(self.moduleConfig)
+        chanConfig = self.getChannelInfo()
+        info = {"Preferences": locConfig, "ModulePreferences": modConfig}
+        info.update(chanConfig)
+        return info
 
     def setChannels(self, channels):
         """Set the channels for this node"""