Fix bug in Intan interface where extra device is written

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Upcoming
 
-### Bug fixes
+### Bug Fixes
+* Fixed a bug where `IntanRecordingInterface` added two devices [PR #1059](https://github.com/catalystneuro/neuroconv/pull/1059)
 * Fix a bug in `add_sorting_to_nwbfile` where `unit_electrode_indices` was only propagated if `waveform_means` was passed [PR #1057](https://github.com/catalystneuro/neuroconv/pull/1057)
 
 ### Deprecations
@@ -15,7 +16,7 @@
 ### Improvements
 * Using ruff to enforce existence of public classes' docstrings [PR #1034](https://github.com/catalystneuro/neuroconv/pull/1034)
 * Separated tests that use external data by modality [PR #1049](https://github.com/catalystneuro/neuroconv/pull/1049)
-
+* Improved device metadata of `IntanRecordingInterface` by adding the type of controller used [PR #1059](https://github.com/catalystneuro/neuroconv/pull/1059)
 
 
 ## v0.6.1 (August 30, 2024)

src/neuroconv/datainterfaces/ecephys/intan/intandatainterface.py

@@ -1,12 +1,9 @@
-import warnings
-from typing import Optional
+from pathlib import Path
 
-from packaging.version import Version
 from pydantic import FilePath
 from pynwb.ecephys import ElectricalSeries
 
 from ..baserecordingextractorinterface import BaseRecordingExtractorInterface
-from ....tools import get_package_version
 from ....utils import get_schema_from_hdmf_class
 
 
@@ -31,7 +28,6 @@ def get_source_schema(cls) -> dict:
     def __init__(
         self,
         file_path: FilePath,
-        stream_id: Optional[str] = None,
         verbose: bool = True,
         es_key: str = "ElectricalSeries",
         ignore_integrity_checks: bool = False,
@@ -43,8 +39,7 @@ def __init__(
         ----------
         file_path : FilePathType
             Path to either a rhd or a rhs file
-        stream_id : str, optional
-            The stream of the data for spikeinterface, "0" by default.
+
         verbose : bool, default: True
             Verbose
         es_key : str, default: "ElectricalSeries"
@@ -53,35 +48,17 @@ def __init__(
             check performed is that timestamps are continuous. If False, an error will be raised if the check fails.
         """
 
-        if stream_id is not None:
-            warnings.warn(
-                "Use of the 'stream_id' parameter is deprecated and it will be removed after September 2024.",
-                DeprecationWarning,
-            )
-            self.stream_id = stream_id
-        else:
-            self.stream_id = "0"  # These are the amplifier channels or to the stream_name 'RHD2000 amplifier channel'
+        self.file_path = Path(file_path)
 
         init_kwargs = dict(
-            file_path=file_path,
+            file_path=self.file_path,
             stream_id=self.stream_id,
             verbose=verbose,
             es_key=es_key,
             all_annotations=True,
+            ignore_integrity_checks=ignore_integrity_checks,
         )
 
-        neo_version = get_package_version(name="neo")
-        spikeinterface_version = get_package_version(name="spikeinterface")
-        if neo_version < Version("0.13.1") or spikeinterface_version < Version("0.100.10"):
-            if ignore_integrity_checks:
-                warnings.warn(
-                    "The 'ignore_integrity_checks' parameter is not supported for neo versions < 0.13.1. "
-                    "or spikeinterface versions < 0.101.0.",
-                    UserWarning,
-                )
-        else:
-            init_kwargs["ignore_integrity_checks"] = ignore_integrity_checks
-
         super().__init__(**init_kwargs)
 
     def get_metadata_schema(self) -> dict:
@@ -96,14 +73,21 @@ def get_metadata(self) -> dict:
         ecephys_metadata = metadata["Ecephys"]
 
         # Add device
-        device = dict(
+
+        system = self.file_path.suffix  # .rhd or .rhs
+        device_description = {".rhd": "RHD Recording System", ".rhs": "RHS Stim/Recording System"}[system]
+
+        intan_device = dict(
             name="Intan",
-            description="Intan recording",
+            description=device_description,
             manufacturer="Intan",
         )
-        device_list = [device]
+        device_list = [intan_device]
         ecephys_metadata.update(Device=device_list)
 
+        electrode_group_metadata = ecephys_metadata["ElectrodeGroup"]
+        electrode_group_metadata[0]["device"] = intan_device["name"]
+
         # Add electrodes and electrode groups
         ecephys_metadata.update(
             ElectricalSeriesRaw=dict(name="ElectricalSeriesRaw", description="Raw acquisition traces."),

src/neuroconv/tools/spikeinterface/spikeinterface.py

@@ -177,7 +177,7 @@ def add_electrode_groups_to_nwbfile(recording: BaseRecording, nwbfile: pynwb.NWB
             device_name = group_metadata.get("device", defaults[0]["device"])
             if device_name not in nwbfile.devices:
                 new_device_metadata = dict(Ecephys=dict(Device=[dict(name=device_name)]))
-                add_devices(nwbfile=nwbfile, metadata=new_device_metadata)
+                add_devices_to_nwbfile(nwbfile=nwbfile, metadata=new_device_metadata)
                 warnings.warn(
                     f"Device '{device_name}' not detected in "
                     "attempted link to electrode group! Automatically generating."

tests/test_on_data/ecephys/test_raw_recordings.py

@@ -10,7 +10,6 @@
 
 from neuroconv import NWBConverter
 from neuroconv.datainterfaces import (
-    IntanRecordingInterface,
     Plexon2RecordingInterface,
 )
 
@@ -49,23 +48,6 @@ class TestEcephysRawRecordingsNwbConversions(unittest.TestCase):
         ),
     ]
 
-    # Intan multiple files format
-    parameterized_recording_list.append(
-        param(
-            data_interface=IntanRecordingInterface,
-            interface_kwargs=dict(file_path=str(DATA_PATH / "intan" / "intan_fpc_test_231117_052630/info.rhd")),
-            case_name="one-file-per-channel",
-        )
-    )
-
-    parameterized_recording_list.append(
-        param(
-            data_interface=IntanRecordingInterface,
-            interface_kwargs=dict(file_path=str(DATA_PATH / "intan" / "intan_fps_test_231117_052500/info.rhd")),
-            case_name="one-file-per-signal",
-        )
-    )
-
     @parameterized.expand(input=parameterized_recording_list, name_func=custom_name_func)
     def test_recording_extractor_to_nwb(self, data_interface, interface_kwargs, case_name=""):
         nwbfile_path = str(self.savedir / f"{data_interface.__name__}_{case_name}.nwb")

tests/test_on_data/ecephys/test_recording_interfaces.py

@@ -209,17 +209,22 @@ def check_run_conversion_with_backend(self, nwbfile_path: str, backend: Literal[
         pass
 
 
-class TestIntanRecordingInterface(RecordingExtractorInterfaceTestMixin):
+class TestIntanRecordingInterfaceRHS(RecordingExtractorInterfaceTestMixin):
+    data_interface_cls = IntanRecordingInterface
+    interface_kwargs = dict(file_path=ECEPHY_DATA_PATH / "intan" / "intan_rhs_test_1.rhs")
+
+
+class TestIntanRecordingInterfaceRHD(RecordingExtractorInterfaceTestMixin):
     data_interface_cls = IntanRecordingInterface
-    interface_kwargs = []
     save_directory = OUTPUT_PATH
 
     @pytest.fixture(
         params=[
-            dict(file_path=str(ECEPHY_DATA_PATH / "intan" / "intan_rhd_test_1.rhd")),
-            dict(file_path=str(ECEPHY_DATA_PATH / "intan" / "intan_rhs_test_1.rhs")),
+            dict(file_path=ECEPHY_DATA_PATH / "intan" / "intan_rhd_test_1.rhd"),
+            dict(file_path=ECEPHY_DATA_PATH / "intan" / "intan_fpc_test_231117_052630/info.rhd"),
+            dict(file_path=ECEPHY_DATA_PATH / "intan" / "intan_fps_test_231117_052500/info.rhd"),
         ],
-        ids=["rhd", "rhs"],
+        ids=["rhd", "one-file-per-channel", "one-file-per-signal"],
     )
     def setup_interface(self, request):
 
@@ -230,6 +235,18 @@ def setup_interface(self, request):
 
         return self.interface, self.test_name
 
+    def test_devices_written_correctly(self, setup_interface):
+
+        from pynwb.testing.mock.file import mock_NWBFile
+
+        nwbfile = mock_NWBFile()
+        self.interface.add_to_nwbfile(nwbfile=nwbfile)
+
+        nwbfile.devices["Intan"].name == "Intan"
+        len(nwbfile.devices) == 1
+
+        nwbfile.devices["Intan"].description == "RHD Recording System"
+
 
 @pytest.mark.skip(reason="This interface fails to load the necessary plugin sometimes.")
 class TestMaxOneRecordingInterface(RecordingExtractorInterfaceTestMixin):