make the info message print more readably on the cli

setup.py

@@ -40,20 +40,21 @@
     python_requires=">=3.9",
     install_requires=[
         "coolname",
+        "crcmod",
+        "dearpygui",
         "grpcio-tools",
-        "protoletariat",
+        "numexpr>=2.8.7",
         "numpy >=2.0.0",
-        "pyserial",
-        "scipy",
-        "crcmod",
-        "rich",
-        "pyqtgraph",
-        "pyqt5",
         "pandas >=2.2.0",
-        "dearpygui",
-        "pyzmq",
+        "paramiko >=3.5.1",
+        "protoletariat",
+        "pyqt5",
+        "pyqtgraph",
+        "pyserial",
         "pyyaml",
-        "paramiko >=3.5.1"
+        "pyzmq",
+        "rich",
+        "scipy",
     ],
     entry_points={
         "console_scripts": [

synapse/cli/device_info_display.py

@@ -0,0 +1,116 @@
+import time
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+from rich.tree import Tree
+from google.protobuf.json_format import MessageToDict
+from synapse.client.device import Device
+
+
+def visualize_configuration(info_dict):
+    config = info_dict.get("configuration", {})
+    if config:
+        tree = Tree("Configuration")
+        for node in config.get("nodes", []):
+            node_type = node.get("type", "").replace("k", "")
+            node_name = node.get("name", "Unknown")
+            node_tree = tree.add(f"{node_name}")
+            node_tree.add(f"ID: {node.get('id', 'Unknown')}")
+            node_tree.add(f"Type: {node_type}")
+
+            if node_type == "Application":
+                app = node.get("application", {})
+                name = app.get("name", "Unknown")
+                running = app.get("running", False)
+                status = "[green]Running[/green]" if running else "[red]Stopped[/red]"
+                node_tree.add(f"Name: {name}")
+                node_tree.add(f"Status: {status}")
+            elif node_type == "BroadbandSource":
+                source = node.get("broadband_source", {})
+                name = source.get("name", "Unknown")
+                running = source.get("running", False)
+                status = "[green]Running[/green]" if running else "[red]Stopped[/red]"
+                node_tree.add(f"Name: {name}")
+                node_tree.add(f"Status: {status}")
+                if "signal" in source and "electrode" in source["signal"]:
+                    channels = source["signal"]["electrode"].get("channels", [])
+                    electrode_ids = [
+                        str(ch.get("electrode_id", "?")) for ch in channels
+                    ]
+                    node_tree.add(
+                        f"Electrodes ({len(channels)}): {', '.join(electrode_ids)}"
+                    )
+
+        return tree
+
+
+def visualize_peripherals(info_dict):
+    tree = Tree("Peripherals")
+    peripherals = info_dict.get("peripherals", [])
+    if peripherals:
+        for peripheral in peripherals:
+            peripheral_tree = tree.add(
+                f"{peripheral.get('name', 'Unknown')} ({peripheral.get('vendor', 'Unknown')})"
+            )
+            peripheral_tree.add(f"ID: {peripheral.get('peripheral_id', 'Unknown')}")
+            peripheral_tree.add(f"Type: {peripheral.get('type', 'Unknown')}")
+    else:
+        tree.add("No peripherals found")
+    return tree
+
+
+class DeviceInfoDisplay:
+    """A class for displaying device information."""
+
+    def __init__(self):
+        self.console = Console()
+
+    def summary(self, device: Device):
+        info = device.info()
+        info_dict = MessageToDict(info, preserving_proto_field_name=True)
+
+        status = info_dict.get("status", {})
+
+        self.console.print(
+            f"Name: [bold cyan]{info_dict.get('name', 'Unknown')}[/bold cyan]",
+        )
+
+        if status:
+            state = status.get("state", "Unknown").replace("k", "")
+            state = {
+                "Running": "[green]Running[/green]",
+                "Stopped": "[red]Stopped[/red]",
+                "Unknown": "[yellow]Unknown[/yellow]",
+            }.get(state, "[yellow]Unknown[/yellow]")
+            self.console.print(f"Status: {state}")
+
+        self.console.print(
+            f"Serial: {info_dict.get('serial', 'Unknown')}",
+            highlight=False,
+        )
+        self.console.print(
+            f"Synapse Version: {info_dict.get('synapse_version', 'Unknown')}",
+            highlight=False,
+        )
+        self.console.print(
+            f"Firmware Version: {info_dict.get('firmware_version', 'Unknown')}",
+            highlight=False,
+        )
+
+        if status:
+            if "power" in status:
+                battery = status["power"].get("battery_level_percent", "N/A")
+                self.console.print(f"Battery: {battery}%", highlight=False)
+
+            if "storage" in status:
+                storage = status["storage"]
+                total = float(storage.get("total_gb", 0))
+                used = float(storage.get("used_gb", 0))
+                used_percent = (used / total * 100) if total > 0 else 0
+                self.console.print(
+                    f"Storage: {used_percent:.1f}% used ({used:.1f}GB / {total:.1f}GB)",
+                    highlight=False,
+                )
+
+            self.console.print(visualize_peripherals(info_dict))
+            self.console.print(visualize_configuration(info_dict))

synapse/cli/rpc.py

@@ -16,6 +16,7 @@
 
 from synapse.cli.query import StreamingQueryClient
 from synapse.utils.log import log_entry_to_str
+from synapse.cli.device_info_display import DeviceInfoDisplay
 
 
 def add_commands(subparsers):
@@ -99,14 +100,9 @@ def add_commands(subparsers):
 
 
 def info(args):
-    console = Console()
-    with console.status("Getting device information...", spinner="bouncingBall"):
-        info = syn.Device(args.uri, args.verbose).info()
-
-    if not info:
-        console.print(f"[bold red]Failed to get device information from {args.uri}")
-        return
-    pprint(info)
+    device = syn.Device(args.uri, args.verbose)
+    display = DeviceInfoDisplay()
+    display.summary(device)
 
 
 def query(args):
@@ -116,41 +112,65 @@ def load_query_request(path_to_config):
                 data = f.read()
                 proto = Parse(data, QueryRequest())
                 return proto
-        except Exception:
-            print(f"Failed to open {path_to_config}")
+        except FileNotFoundError:
+            console.print(f"[red]Failed to open {path_to_config}: File not found[/red]")
+            return None
+        except Exception as e:
+            console.print(f"[red]Failed to parse query file: {str(e)}[/red]")
             return None
 
+    console = Console()
     if args.stream:
         client = StreamingQueryClient(args.uri, args.verbose)
         query_proto = load_query_request(args.query_file)
         if not query_proto:
             return False
-        return client.stream_query(StreamQueryRequest(request=query_proto))
+        try:
+            return client.stream_query(StreamQueryRequest(request=query_proto))
+        except Exception as e:
+            console.print(f"[red]Error streaming query: {str(e)}[/red]")
+            return False
 
     if Path(args.query_file).suffix != ".json":
-        print("Query file must be a JSON file")
+        console.print("[red]Query file must be a JSON file[/red]")
         return False
 
-    with open(args.query_file) as query_json:
-        query_proto = Parse(query_json.read(), QueryRequest())
-        print("Running query:")
-        print(query_proto)
-
-        result: QueryResponse = syn.Device(args.uri, args.verbose).query(query_proto)
-        if result:
-            print(text_format.MessageToString(result))
-
-            if result.HasField("impedance_response"):
-                measurements = result.impedance_response
-                # Write impedance measurements to a CSV file
-                with open(
-                    f"impedance_measurements_{time.strftime('%Y%m%d-%H%M%S')}.csv", "w"
-                ) as f:
-                    f.write("Electrode ID,Magnitude (Ohms),Phase (degrees),Status\n")
-                    for measurement in measurements.measurements:
-                        f.write(
-                            f"{measurement.electrode_id},{measurement.magnitude},{measurement.phase},1\n"
+    try:
+        with open(args.query_file) as query_json:
+            query_proto = Parse(query_json.read(), QueryRequest())
+            console.print("Running query:")
+            console.print(query_proto)
+
+            result: QueryResponse = syn.Device(args.uri, args.verbose).query(
+                query_proto
+            )
+            if result:
+                console.print(text_format.MessageToString(result))
+
+                if result.HasField("impedance_response"):
+                    measurements = result.impedance_response
+                    # Write impedance measurements to a CSV file
+                    timestamp = time.strftime("%Y%m%d-%H%M%S")
+                    filename = f"impedance_measurements_{timestamp}.csv"
+                    try:
+                        with open(filename, "w") as f:
+                            f.write(
+                                "Electrode ID,Magnitude (Ohms),Phase (degrees),Status\n"
+                            )
+                            for measurement in measurements.measurements:
+                                f.write(
+                                    f"{measurement.electrode_id},{measurement.magnitude},{measurement.phase},1\n"
+                                )
+                        console.print(
+                            f"[green]Impedance measurements saved to {filename}[/green]"
                         )
+                    except IOError as e:
+                        console.print(
+                            f"[red]Error writing impedance measurements: {str(e)}[/red]"
+                        )
+    except Exception as e:
+        console.print(f"[red]Error executing query: {str(e)}[/red]")
+        return False
 
 
 def start(args):
@@ -181,7 +201,9 @@ def start(args):
             cfg_proto = Parse(json_text, DeviceConfiguration())
             config_obj = syn.Config.from_proto(cfg_proto)
         except Exception as e:
-            console.print(f"[bold red]Failed to parse configuration file[/bold red]: {e}")
+            console.print(
+                f"[bold red]Failed to parse configuration file[/bold red]: {e}"
+            )
             return
 
     device = syn.Device(args.uri, args.verbose)
@@ -194,7 +216,9 @@ def start(args):
                 console.print("[bold red]Internal error configuring device")
                 return
             if cfg_ret.code != StatusCode.kOk:
-                console.print(f"[bold red]Error configuring device[/bold red]\n{cfg_ret.message}")
+                console.print(
+                    f"[bold red]Error configuring device[/bold red]\n{cfg_ret.message}"
+                )
                 return
         console.print("[green]Device Configured")
 
@@ -204,7 +228,9 @@ def start(args):
             console.print("[bold red]Internal error starting device")
             return
         if start_ret.code != StatusCode.kOk:
-            console.print(f"[bold red]Error starting device[/bold red]\n{start_ret.message}")
+            console.print(
+                f"[bold red]Error starting device[/bold red]\n{start_ret.message}"
+            )
             return
 
     console.print("[green]Device Started")