Unify impedance CSV output across streaming and non-streaming paths

Both synapsectl query and synapsectl query --stream now write identical
impedance CSVs via a shared synapse.cli.impedance_csv module:

  Electrode ID,Magnitude (Ohms),Phase (degrees),Status
  <electrode_id>,<magnitude>,<phase>,<status>

The CSV body is unchanged from the legacy non-streaming format (column
header on the first line), so existing parsers keep working. The
peripheral the measurement ran on is encoded in the *filename* instead:

  impedance_measurements_<peripheral>_<timestamp>.csv

Changes:
- Fix streaming CSV discrepancies: it previously wrote a different header
  ('Electrode ID,Magnitude,Phase'), omitted units, and had no Status
  column. It now matches the non-streaming format exactly.
- Streaming now also writes failed measurements with Status=0 (success=1);
  previously failed measurements were dropped from the CSV entirely.
- Peripheral name is resolved from the device's peripheral list (preferring
  the query's peripheral_id, else the broadband recording source) and
  embedded in the filename.

Author: namthor9

synapse/cli/impedance_csv.py

@@ -0,0 +1,79 @@
+"""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:
+
+    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. The
+peripheral the measurement ran on is encoded in the filename (see
+`make_filename`) rather than in the file body, so the CSV stays compatible
+with parsers that expect the column header on the first line.
+"""
+
+import csv
+import re
+import time
+
+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 filename.
+
+    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 make_filename(peripheral_name) -> str:
+    """Timestamped CSV filename with the peripheral name embedded, e.g.
+    ``impedance_measurements_NYX1-512_20260624-153012.csv``. The name is
+    sanitized so it's always a safe filename component."""
+    safe = re.sub(r"[^A-Za-z0-9._-]+", "_", peripheral_name or "Unknown").strip("_")
+    safe = safe or "Unknown"
+    timestamp = time.strftime("%Y%m%d-%H%M%S")
+    return f"impedance_measurements_{safe}_{timestamp}.csv"
+
+
+def write_header(filename):
+    """Create (truncate) the CSV and write the column header."""
+    with open(filename, "w", newline="") as f:
+        csv.writer(f, lineterminator=_LINE_TERMINATOR).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,10 @@
 #!/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 (
@@ -153,11 +152,11 @@ def handle_impedance_stream(self, request):
         all_measurements = []
         failed_measurements = []
 
-        # Create a CSV file to read from at the beginning
-        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"])
+        # Create a CSV file to read from at the beginning (peripheral name is
+        # encoded in the filename).
+        peripheral_name = impedance_csv.resolve_peripheral_name(self.device, query)
+        filename = impedance_csv.make_filename(peripheral_name)
+        impedance_csv.write_header(filename)
         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)
@@ -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

@@ -1,5 +1,4 @@
 from pathlib import Path
-import time
 from datetime import datetime
 from typing import Optional
 
@@ -13,6 +12,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,26 +140,24 @@ 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)
+            result: QueryResponse = device.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"
+                    peripheral_name = impedance_csv.resolve_peripheral_name(
+                        device, query_proto.impedance_query
+                    )
+                    # Write impedance measurements to a CSV file (peripheral name
+                    # is encoded in the filename).
+                    filename = impedance_csv.make_filename(peripheral_name)
                     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)
+                        impedance_csv.append_measurements(
+                            filename, measurements.measurements
+                        )
                         console.print(
                             f"[green]Impedance measurements saved to {filename}[/green]"
                         )