Unify impedance CSV output across streaming and non-streaming paths

setup.py

@@ -29,7 +29,7 @@ def stamp_api_version():
 
 setup(
     name="science-synapse",
-    version="2.7.5",
+    version="2.7.6",
     description="Client library and CLI for the Synapse API",
     author="Science Team",
     author_email="team@science.xyz",

synapse/cli/impedance_csv.py

@@ -0,0 +1,67 @@
+"""Shared helpers for writing impedance-measurement CSV files.
+
+Both the non-streaming (`synapsectl query`) and streaming (`--stream`) paths
+emit the same CSV so downstream tooling can parse either identically:
+
+    Peripheral,<name>
+    Electrode ID,Magnitude (Ohms),Phase (degrees),Status
+    <electrode_id>,<magnitude>,<phase>,<status>
+    ...
+
+`Status` is 1 for a successful measurement and 0 for a failed one.
+"""
+
+import csv
+
+from synapse.api.device_pb2 import Peripheral
+
+CSV_COLUMNS = ["Electrode ID", "Magnitude (Ohms)", "Phase (degrees)", "Status"]
+
+STATUS_OK = 1
+STATUS_FAILED = 0
+
+# Force LF so the streaming (csv.writer) and non-streaming paths produce
+# byte-identical files regardless of platform.
+_LINE_TERMINATOR = "\n"
+
+
+def resolve_peripheral_name(device, impedance_query) -> str:
+    """Best-effort name of the peripheral the measurement ran on, for the CSV header.
+
+    Prefers the ``peripheral_id`` named in the query (if the proto carries one);
+    otherwise falls back to the device's broadband (recording) source, then the
+    first peripheral. Returns "Unknown" if it can't be resolved.
+    """
+    info = device.info() if device is not None else None
+    if not info or not info.peripherals:
+        return "Unknown"
+
+    # Command-range ids (e.g. 2 = "first broadband source") won't match a
+    # concrete peripheral_id and fall through to the broadband lookup below.
+    peripheral_id = getattr(impedance_query, "peripheral_id", 0)
+    if peripheral_id:
+        for p in info.peripherals:
+            if p.peripheral_id == peripheral_id:
+                return p.name
+
+    for p in info.peripherals:
+        if p.type == Peripheral.kBroadbandSource:
+            return p.name
+
+    return info.peripherals[0].name
+
+
+def write_header(filename, peripheral_name):
+    """Create (truncate) the CSV and write the peripheral line + column header."""
+    with open(filename, "w", newline="") as f:
+        writer = csv.writer(f, lineterminator=_LINE_TERMINATOR)
+        writer.writerow(["Peripheral", peripheral_name])
+        writer.writerow(CSV_COLUMNS)
+
+
+def append_measurements(filename, measurements, status=STATUS_OK):
+    """Append measurement rows to an existing CSV created by `write_header`."""
+    with open(filename, "a", newline="") as f:
+        writer = csv.writer(f, lineterminator=_LINE_TERMINATOR)
+        for m in measurements:
+            writer.writerow([m.electrode_id, m.magnitude, m.phase, status])

synapse/cli/query.py

@@ -1,11 +1,11 @@
 #!/usr/bin/env python3
 import asyncio
-import csv
 from threading import Thread
 import time
 import sys
 import synapse as syn
 from synapse.api.query_pb2 import QueryRequest, StreamQueryRequest
+from synapse.cli import impedance_csv
 from google.protobuf.json_format import Parse
 
 from rich.progress import (
@@ -154,10 +154,9 @@ def handle_impedance_stream(self, request):
         failed_measurements = []
 
         # Create a CSV file to read from at the beginning
+        peripheral_name = impedance_csv.resolve_peripheral_name(self.device, query)
         filename = f"impedance_measurements_{time.strftime('%Y%m%d-%H%M%S')}.csv"
-        with open(filename, "w", newline="") as f:
-            writer = csv.writer(f)
-            writer.writerow(["Electrode ID", "Magnitude", "Phase"])
+        impedance_csv.write_header(filename, peripheral_name)
         self.console.print(f"[green] Started saving measurements to {filename}")
 
         progress = Progress(
@@ -217,6 +216,9 @@ def update_progress():
                         progress.console.log(
                             f"electrode id (mag, phase): {sample.electrode_id}\t {sample.magnitude},{sample.phase}"
                         )
+                    self.save_measurement_batch(
+                        filename, failed_batch, status=impedance_csv.STATUS_FAILED
+                    )
                     measurements_received += len(failed_batch)
                     progress.update(
                         task, completed=min(measurements_received, electrode_count)
@@ -257,8 +259,8 @@ def update_progress():
 
     def display_impedance_results(self, measurements):
         table = Table(title="Impedance Measurements")
-        table.add_column("Electorde ID", justify="right")
-        table.add_column("Magnitude (kΩ)", justify="right")
+        table.add_column("Electrode ID", justify="right")
+        table.add_column("Magnitude (Ω)", justify="right")
         table.add_column("Phase (°)", justify="right")
 
         for measurement in measurements:
@@ -269,14 +271,11 @@ def display_impedance_results(self, measurements):
             )
         self.console.print(table)
 
-    def save_measurement_batch(self, filename, measurements):
+    def save_measurement_batch(
+        self, filename, measurements, status=impedance_csv.STATUS_OK
+    ):
         # Save a batch of measurements as they come in
-        with open(filename, "a", newline="") as f:
-            writer = csv.writer(f)
-            for measurement in measurements:
-                writer.writerow(
-                    [measurement.electrode_id, measurement.magnitude, measurement.phase]
-                )
+        impedance_csv.append_measurements(filename, measurements, status=status)
 
 
 def load_config_from_file(path_to_config):

synapse/cli/rpc.py

@@ -13,6 +13,7 @@
 from rich.console import Console
 
 from synapse.cli.query import StreamingQueryClient
+from synapse.cli import impedance_csv
 from synapse.utils.log import log_entry_to_str
 from synapse.cli.device_info_display import DeviceInfoDisplay
 from synapse.utils.proto import load_device_config
@@ -140,9 +141,18 @@ def load_query_request(path_to_config):
             console.print("Running query:")
             console.print(query_proto)
 
-            result: QueryResponse = syn.Device(args.uri, args.verbose).query(
-                query_proto
-            )
+            device = syn.Device(args.uri, args.verbose)
+
+            # Resolve the peripheral name before running the query, matching the
+            # streaming path: if the probe un-enumerates as a result of the query
+            # we can still label the CSV correctly.
+            peripheral_name = None
+            if query_proto.HasField("impedance_query"):
+                peripheral_name = impedance_csv.resolve_peripheral_name(
+                    device, query_proto.impedance_query
+                )
+
+            result: QueryResponse = device.query(query_proto)
             if result:
                 console.print(text_format.MessageToString(result))
 
@@ -152,14 +162,10 @@ def load_query_request(path_to_config):
                     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"
-                                )
+                        impedance_csv.write_header(filename, peripheral_name)
+                        impedance_csv.append_measurements(
+                            filename, measurements.measurements
+                        )
                         console.print(
                             f"[green]Impedance measurements saved to {filename}[/green]"
                         )