Record real electrode ids in broadband HDF5 output

synapse/cli/offline_hdf5_plotter.py

@@ -105,14 +105,20 @@ def load_h5_data(data_file, console, time_range=None):
         # List immediate groups (top-level only)
         print_tree(f, console)
 
-        # Get channel information
+        # Get channel information. `id` is one id per entry per frame, in frame
+        # order — physical electrode ids for recordings that could resolve them
+        # (see id_source), logical channel ids or row indices otherwise.
         channels = f["/general/extracellular_ephys/electrodes/"]
         channel_ids = channels["id"][:].tolist()
         number_of_channels = len(channel_ids)
 
         sample_rate = float(attributes["sample_rate_hz"])
         console.print(f"Sample rate: {sample_rate} Hz")
-        console.print(f"Found {number_of_channels} channels")
+        id_source = channels.attrs.get("id_source")
+        if id_source is not None:
+            console.print(f"Found {number_of_channels} channels (id source: {id_source})")
+        else:
+            console.print(f"Found {number_of_channels} channels")
 
         # Get frame data info
         frame_data = f["/acquisition/ElectricalSeries"]
@@ -204,8 +210,11 @@ def load_h5_data(data_file, console, time_range=None):
                 actual_samples_per_channel, number_of_channels
             )
 
-            # Create DataFrame
-            df = pd.DataFrame(reshaped_data, columns=range(number_of_channels))
+            # Create DataFrame, labeling each column with the channel id of the
+            # row it came from. filter_channels selects by label (--channels
+            # takes channel ids), so positional column labels quietly selected
+            # the wrong traces for any recording whose ids aren't 0..N-1.
+            df = pd.DataFrame(reshaped_data, columns=channel_ids)
 
         return PlotData(data=df, sample_rate=sample_rate, channel_ids=channel_ids)
 

synapse/cli/streaming.py

@@ -16,6 +16,12 @@
 from synapse.api.node_pb2 import NodeType
 from synapse.client.taps import Tap
 from synapse.utils.proto import load_device_config
+from synapse.utils.electrode_ids import (
+    GPIO_CHANNEL_ID_OFFSET,
+    ElectrodeRowIds,
+    build_channel_to_electrode_map,
+    derive_electrode_row_ids,
+)
 from synapse.api.datatype_pb2 import BroadbandFrame
 
 
@@ -252,7 +258,7 @@ def get_stats(self):
     def set_attributes(
         self,
         sample_rate_hz: float,
-        channels: list,
+        electrode_rows: ElectrodeRowIds,
         broadband_lsb_uv: float,
         session_description: str = "",
     ):
@@ -269,7 +275,22 @@ def set_attributes(
         electrodes_group = self.file.create_group(
             "general/extracellular_ephys/electrodes"
         )
-        electrodes_group.create_dataset("id", data=channels, dtype="uint32")
+        # `id` holds physical electrode ids where we could resolve them (see
+        # derive_electrode_row_ids), and degrades to logical channel ids or row
+        # indices otherwise — `id_source` says which. The logical channel id and
+        # channel type are written alongside it, as equal-length datasets, so
+        # downstream analysis can reconstruct the full mapping.
+        electrodes_group.create_dataset(
+            "id", data=electrode_rows.ids, dtype="uint32"
+        )
+        electrodes_group.create_dataset(
+            "channel_id", data=electrode_rows.channel_ids, dtype="uint32"
+        )
+        electrodes_group.create_dataset(
+            "channel_type", data=electrode_rows.types, dtype="uint32"
+        )
+        electrodes_group.attrs["id_source"] = electrode_rows.source
+        electrodes_group.attrs["gpio_channel_id_offset"] = GPIO_CHANNEL_ID_OFFSET
 
     def start(self):
         """Start the writer thread"""
@@ -685,8 +706,15 @@ def list_available_taps(args, device, console):
     )
 
 
-def detect_stream_parameters(broadband_tap, console):
-    """Detect sample rate and available channels from the first message"""
+def detect_stream_parameters(broadband_tap, console, channel_to_electrode=None):
+    """Detect sample rate and channel identity from the first message
+
+    Returns (sample_rate, electrode_rows, first_frame). `electrode_rows` carries
+    one id per frame_data entry: the physical electrode where the device
+    configuration lets us resolve it, and a documented fallback otherwise. It
+    replaces the old range(num_channels), which labeled every row by its
+    position and so mislabeled any sparse or non-ascending configuration.
+    """
     console.log("[cyan]Detecting stream parameters from first message...[/cyan]")
 
     try:
@@ -704,15 +732,23 @@ def detect_stream_parameters(broadband_tap, console):
 
         # Extract parameters
         sample_rate = first_frame.sample_rate_hz
-        num_channels = len(first_frame.frame_data)
-        available_channels = list(range(num_channels))
+        electrode_rows = derive_electrode_row_ids(first_frame, channel_to_electrode)
+        if electrode_rows is None:
+            console.print(
+                "[bold red]First message carried no channel data[/bold red]"
+            )
+            return None, None, None
 
+        num_channels = len(electrode_rows.ids)
         console.log(f"[green]Detected sample rate: {sample_rate} Hz[/green]")
         console.log(
-            f"[green]Detected {num_channels} channels (0-{num_channels - 1})[/green]"
+            f"[green]Detected {num_channels} channels "
+            f"(id source: {electrode_rows.source})[/green]"
         )
+        for warning in electrode_rows.warnings:
+            console.print(f"[yellow]{warning}[/yellow]")
 
-        return sample_rate, available_channels, first_frame
+        return sample_rate, electrode_rows, first_frame
 
     except Exception as e:
         console.print(f"[bold red]Error detecting stream parameters: {e}[/bold red]")
@@ -944,19 +980,24 @@ def read(args):
         console.print("[bold red]Failed to get broadband tap[/bold red]")
         return
 
+    # Get the latest info on the device. The stream identifies its rows only by
+    # logical channel id, so the applied configuration is the only place the
+    # physical electrode behind each row is recorded.
+    device_info = device.info()
+    channel_to_electrode = build_channel_to_electrode_map(device_info)
+
     # Detect stream parameters from the first message
-    sample_rate, available_channels, first_frame = detect_stream_parameters(
-        broadband_tap, console
+    sample_rate, electrode_rows, first_frame = detect_stream_parameters(
+        broadband_tap, console, channel_to_electrode
     )
     if sample_rate is None:
         console.print("[bold red]Failed to detect stream parameters[/bold red]")
         return
+    available_channels = electrode_rows.ids
 
     # Setup our HDF5 writer if output is requested
     writer = None
     if args.output:
-        # Get the latest info on the device
-        device_info = device.info()
         broadband_node = get_broadband_node_status(device_info)
         if broadband_node is None:
             console.print(
@@ -967,7 +1008,7 @@ def read(args):
         writer = BroadbandFrameWriter(args.output)
         writer.set_attributes(
             sample_rate_hz=sample_rate,
-            channels=available_channels,
+            electrode_rows=electrode_rows,
             broadband_lsb_uv=broadband_lsb_uv,
         )
         writer.start()

synapse/tests/cli/test_broadband_frame_writer.py

@@ -0,0 +1,76 @@
+import h5py
+
+from synapse.api.channel_pb2 import ChannelRange, ChannelType
+from synapse.api.datatype_pb2 import BroadbandFrame
+from synapse.cli.streaming import BroadbandFrameWriter
+from synapse.utils.electrode_ids import (
+    GPIO_CHANNEL_ID_OFFSET,
+    derive_electrode_row_ids,
+)
+
+
+def write_attributes(tmp_path, frame, channel_to_electrode=None):
+    """Run set_attributes for a frame and hand back the resulting file."""
+    rows = derive_electrode_row_ids(frame, channel_to_electrode)
+    writer = BroadbandFrameWriter(str(tmp_path))
+    try:
+        writer.set_attributes(
+            sample_rate_hz=30000.0,
+            electrode_rows=rows,
+            broadband_lsb_uv=0.195,
+        )
+    finally:
+        writer.file.close()
+    return writer.filename
+
+
+def test_writes_physical_electrode_ids(tmp_path):
+    frame = BroadbandFrame(
+        frame_data=[0, 0, 0, 0, 0],
+        channel_ranges=[
+            ChannelRange(
+                type=ChannelType.ELECTRODE, count=5, channel_ids=[0, 1, 2, 3, 4]
+            )
+        ],
+    )
+    filename = write_attributes(
+        tmp_path, frame, {0: 6, 1: 16, 2: 10, 3: 14, 4: 20}
+    )
+
+    with h5py.File(filename, "r") as f:
+        electrodes = f["general/extracellular_ephys/electrodes"]
+        assert electrodes["id"][:].tolist() == [6, 16, 10, 14, 20]
+        assert electrodes["channel_id"][:].tolist() == [0, 1, 2, 3, 4]
+        assert electrodes["channel_type"][:].tolist() == [0, 0, 0, 0, 0]
+        assert electrodes.attrs["id_source"] == "electrode_map"
+        assert electrodes.attrs["gpio_channel_id_offset"] == GPIO_CHANNEL_ID_OFFSET
+        assert f.attrs["sample_rate_hz"] == 30000.0
+        assert f.attrs["lsb_uv"] == 0.195
+
+
+def test_electrode_table_length_matches_the_per_frame_entry_count(tmp_path):
+    """GPIO rows are part of ElectricalSeries, so they need table rows too."""
+    frame = BroadbandFrame(
+        frame_data=[0, 0, 0],
+        channel_ranges=[
+            ChannelRange(type=ChannelType.ELECTRODE, count=2, channel_ids=[0, 1]),
+            ChannelRange(type=ChannelType.GPIO, count=1, channel_ids=[1]),
+        ],
+    )
+    filename = write_attributes(tmp_path, frame, {0: 6, 1: 16})
+
+    with h5py.File(filename, "r") as f:
+        electrodes = f["general/extracellular_ephys/electrodes"]
+        assert len(electrodes["id"]) == len(frame.frame_data)
+        assert electrodes["id"][:].tolist() == [6, 16, GPIO_CHANNEL_ID_OFFSET + 1]
+        assert electrodes["channel_type"][:].tolist() == [0, 0, 1]
+
+
+def test_records_the_legacy_fallback_in_id_source(tmp_path):
+    frame = BroadbandFrame(frame_data=[0, 0, 0])
+    filename = write_attributes(tmp_path, frame)
+
+    with h5py.File(filename, "r") as f:
+        electrodes = f["general/extracellular_ephys/electrodes"]
+        assert electrodes["id"][:].tolist() == [0, 1, 2]
+        assert electrodes.attrs["id_source"] == "positional"