Add more friendly error when writing recording with multiple offsets (#…

…1111)

CHANGELOG.md

@@ -7,6 +7,7 @@
 ## Features
 * Using in-house `GenericDataChunkIterator` [PR #1068](https://github.com/catalystneuro/neuroconv/pull/1068)
 * Data interfaces now perform source (argument inputs) validation with the json schema  [PR #1020](https://github.com/catalystneuro/neuroconv/pull/1020)
+* Improve the error message when writing a recording extractor with multiple offsets [PR #1111](https://github.com/catalystneuro/neuroconv/pull/1111)
 * Added `channels_to_skip` to `EDFRecordingInterface` so the user can skip non-neural channels [PR #1110](https://github.com/catalystneuro/neuroconv/pull/1110)
 
 ## Improvements

src/neuroconv/tools/spikeinterface/spikeinterface.py

@@ -99,7 +99,8 @@ def add_devices_to_nwbfile(nwbfile: pynwb.NWBFile, metadata: Optional[DeepDict]
         metadata["Ecephys"]["Device"] = [defaults]
     for device_metadata in metadata["Ecephys"]["Device"]:
         if device_metadata.get("name", defaults["name"]) not in nwbfile.devices:
-            nwbfile.create_device(**dict(defaults, **device_metadata))
+            device_kwargs = dict(defaults, **device_metadata)
+            nwbfile.create_device(**device_kwargs)
 
 
 def add_electrode_groups(recording: BaseRecording, nwbfile: pynwb.NWBFile, metadata: dict = None):
@@ -778,6 +779,28 @@ def add_electrical_series(
     )
 
 
+def _report_variable_offset(channel_offsets, channel_ids):
+    """
+    Helper function to report variable offsets per channel IDs.
+    Groups the different available offsets per channel IDs and raises a ValueError.
+    """
+    # Group the different offsets per channel IDs
+    offset_to_channel_ids = {}
+    for offset, channel_id in zip(channel_offsets, channel_ids):
+        if offset not in offset_to_channel_ids:
+            offset_to_channel_ids[offset] = []
+        offset_to_channel_ids[offset].append(channel_id)
+
+    # Create a user-friendly message
+    message_lines = ["Recording extractors with heterogeneous offsets are not supported."]
+    message_lines.append("Multiple offsets were found per channel IDs:")
+    for offset, ids in offset_to_channel_ids.items():
+        message_lines.append(f"  Offset {offset}: Channel IDs {ids}")
+    message = "\n".join(message_lines)
+
+    raise ValueError(message)
+
+
 def add_electrical_series_to_nwbfile(
     recording: BaseRecording,
     nwbfile: pynwb.NWBFile,
@@ -905,14 +928,16 @@ def add_electrical_series_to_nwbfile(
     # Spikeinterface guarantees data in micro volts when return_scaled=True. This multiplies by gain and adds offsets
     # In nwb to get traces in Volts we take data*channel_conversion*conversion + offset
     channel_conversion = recording.get_channel_gains()
-    channel_offset = recording.get_channel_offsets()
+    channel_offsets = recording.get_channel_offsets()
 
     unique_channel_conversion = np.unique(channel_conversion)
     unique_channel_conversion = unique_channel_conversion[0] if len(unique_channel_conversion) == 1 else None
 
-    unique_offset = np.unique(channel_offset)
+    unique_offset = np.unique(channel_offsets)
     if unique_offset.size > 1:
-        raise ValueError("Recording extractors with heterogeneous offsets are not supported")
+        channel_ids = recording.get_channel_ids()
+        # This prints a user friendly error where the user is provided with a map from offset to channels
+        _report_variable_offset(channel_offsets, channel_ids)
     unique_offset = unique_offset[0] if unique_offset[0] is not None else 0
 
     micro_to_volts_conversion_factor = 1e-6

tests/test_ecephys/test_tools_spikeinterface.py

@@ -1,3 +1,4 @@
+import re
 import unittest
 from datetime import datetime
 from pathlib import Path
@@ -8,9 +9,11 @@
 import numpy as np
 import psutil
 import pynwb.ecephys
+import pytest
 from hdmf.data_utils import DataChunkIterator
 from hdmf.testing import TestCase
 from pynwb import NWBFile
+from pynwb.testing.mock.file import mock_NWBFile
 from spikeinterface.core.generate import (
     generate_ground_truth_recording,
     generate_recording,
@@ -394,6 +397,36 @@ def test_variable_offsets_assertion(self):
             )
 
 
+def test_error_with_multiple_offset():
+    # Generate a mock recording with 5 channels and 1 second duration
+    recording = generate_recording(num_channels=5, durations=[1.0])
+    # Rename channels to specific identifiers for clarity in error messages
+    recording = recording.rename_channels(new_channel_ids=["a", "b", "c", "d", "e"])
+    # Set different offsets for the channels
+    recording.set_channel_offsets(offsets=[0, 0, 1, 1, 2])
+
+    # Create a mock NWBFile object
+    nwbfile = mock_NWBFile()
+
+    # Expected error message
+    expected_message_lines = [
+        "Recording extractors with heterogeneous offsets are not supported.",
+        "Multiple offsets were found per channel IDs:",
+        "  Offset 0: Channel IDs ['a', 'b']",
+        "  Offset 1: Channel IDs ['c', 'd']",
+        "  Offset 2: Channel IDs ['e']",
+    ]
+    expected_message = "\n".join(expected_message_lines)
+
+    # Use re.escape to escape any special regex characters in the expected message
+    expected_message_regex = re.escape(expected_message)
+
+    # Attempt to add electrical series to the NWB file
+    # Expecting a ValueError due to multiple offsets, matching the expected message
+    with pytest.raises(ValueError, match=expected_message_regex):
+        add_electrical_series_to_nwbfile(recording=recording, nwbfile=nwbfile)
+
+
 class TestAddElectricalSeriesChunking(unittest.TestCase):
     @classmethod
     def setUpClass(cls):