Merge branch 'main' into add_imro_tests

Author: h-mayorquin

probeinterface/io.py

@@ -920,7 +920,7 @@ def write_csv(file, probe):
 }
 
 
-# Map imDatPrb_pn to imDatPrb_type when the latter is missing
+# Map imDatPrb_pn (probe number) to imDatPrb_type (probe type) when the latter is missing
 probe_number_to_probe_type = {
     "PRB_1_4_0480_1": 0,
     "PRB_1_4_0480_1_C": 0,
@@ -957,41 +957,56 @@ def read_imro(file_path: Union[str, Path]) -> Probe:
 
 def _read_imro_string(imro_str: str, imDatPrb_pn: str = None) -> Probe:
     """
-    Low-level function to parse the imro table when presented as a string
-    
-    See this doc https://billkarsh.github.io/SpikeGLX/help/imroTables/
-    
+    Parse the IMRO table when presented as a string and create a Probe object.
+
+    Parameters
+    ----------
+    imro_str : str
+        IMRO table as a string.
+    imDatPrb_pn : str, optional
+        Probe number, by default None.
+
+    Returns
+    -------
+    Probe
+        A Probe object built from  the parsed IMRO table data.
+
+    See Also
+    --------
+    https://billkarsh.github.io/SpikeGLX/help/imroTables/
+
     """
-    headers, *parts, _ = imro_str.strip().split(")")
+    imro_table_header_str, *imro_table_values_list, _ = imro_str.strip().split(")")
 
-    header = tuple(map(int, headers[1:].split(',')))
-    if len(header) == 3:
+    imro_table_header = tuple(map(int, imro_table_header_str[1:].split(',')))
+    if len(imro_table_header) == 3:
         # In older versions of neuropixel arrays (phase 3A), imro tables were structured differently. 
-        probe_serial_number, probe_option, num_contact = header
+        probe_serial_number, probe_option, num_contact = imro_table_header
         imDatPrb_type = 'Phase3a'
-    elif len(header) == 2:
-        imDatPrb_type, num_contact = header
+    elif len(imro_table_header) == 2:
+        imDatPrb_type, num_contact = imro_table_header
     else:
-        raise ValueError(f'read_imro error, the header has a strange shape: {header}')
+        raise RuntimeError(f'read_imro error, the header has a strange length: {imro_table_header}')
+    
 
     if imDatPrb_type in [0, None]:
         imDatPrb_type = probe_number_to_probe_type[imDatPrb_pn] 
 
     probe_description = npx_probe[imDatPrb_type]
     fields = probe_description["fields_in_imro_table"]
-    contact_info = {k: [] for k in fields}
-    for i, part in enumerate(parts):
-        values = tuple(map(int, part[1:].split(' ')))
-        for k, v in zip(fields, values):
-            contact_info[k].append(v)
+    contact_info = {k: [] for k in fields}      
+    for field_values_str in imro_table_values_list: # Imro table values look like '(value, value, value, ... '
+        values = tuple(map(int, field_values_str[1:].split(' '))) 
+        # Split them by space to get (int('value'), int('value'), int('value'), ...)
+        for field, field_value in zip(fields, values):
+            contact_info[field].append(field_value)
 
     channel_ids = np.array(contact_info['channel_ids'])
-    probe_types_without_elec_ids_in_their_imro_table = (0, 1015, 1022, 1030, 1031, 1032, "Phase3a", 1100)
-    if imDatPrb_type in probe_types_without_elec_ids_in_their_imro_table:
+    if "elect_ids" in contact_info:
+        elec_ids = np.array(contact_info['elect_ids'])
+    else:        
         banks = np.array(contact_info['banks'])
         elec_ids = banks * 384 + channel_ids
-    else:
-        elec_ids = np.array(contact_info['elec_ids'])
 
     # compute position
     y_idx, x_idx = np.divmod(elec_ids, probe_description["ncol"])

tests/data/spikeglx/Day_3_g0_t0.imec1.ap.meta renamed to tests/data/spikeglx/NP1_saved_only_subset_of_channels.meta

@@ -11,18 +11,32 @@
 
 data_path = Path(__file__).absolute().parent.parent / "data" / "spikeglx"
 
+def test_parse_meta():
+    for meta_file in [
+        "doppio-checkerboard_t0.imec0.ap.meta",
+        "NP1_saved_only_subset_of_channels.meta",
+        "allan-longcol_g0_t0.imec0.ap.meta",
+    ]:
+        meta = parse_spikeglx_meta(data_path / meta_file)
+
+def test_get_saved_channel_indices_from_spikeglx_meta():
+    # all channel saved + 1 synchro
+    chan_inds = get_saved_channel_indices_from_spikeglx_meta(
+        data_path / "Noise_g0_t0.imec0.ap.meta"
+    )
+    assert chan_inds.size == 385
+
+    # example by Pierre Yger NP1.0 with 384 but only 151 channels are saved + 1 synchro
+    chan_inds = get_saved_channel_indices_from_spikeglx_meta(
+        data_path / "NP1_saved_only_subset_of_channels.meta"
+    )
+    assert chan_inds.size == 152
 
 def test_NP1():
     probe = read_spikeglx(data_path / "Noise_g0_t0.imec0.ap.meta")
     assert "1.0" in probe.annotations["name"]
 
 
-def test_NP2_4_shanks():
-    probe = read_spikeglx(data_path / "TEST_20210920_0_g0_t0.imec0.ap.meta")
-    assert probe.get_shank_count() == 4
-    assert "2.0" in probe.annotations["name"]
-
-
 def test_NP2_1_shanks():
     probe = read_spikeglx(data_path / "p2_g0_t0.imec0.ap.meta")
     assert "2.0" in probe.annotations["name"]
@@ -31,19 +45,72 @@ def test_NP2_1_shanks():
 
 def test_NP_phase3A():
     # Data provided by rtraghavan
-    probe = read_spikeglx(data_path / "NeuropixelPhase3A_file_g0_t0.imec.ap.meta")
-    assert "Phase3a" in probe.annotations["name"]
+    probe = read_spikeglx(data_path / "phase3a.imec.ap.meta")
+
+    assert probe.annotations["name"] == "Phase3a"
+    assert probe.annotations["manufacturer"] == "IMEC"
+    assert probe.annotations["probe_type"] == "Phase3a"
+
+    assert probe.ndim == 2
     assert probe.get_shank_count() == 1
+    assert probe.get_contact_count() == 384
+    
+    # Test contact geometry
+    contact_width = 12.0
+    contact_shape = "square"
 
+    assert np.all(probe.contact_shape_params == {"width": contact_width})
+    assert np.all(probe.contact_shapes == contact_shape)
 
 def test_NP2_4_shanks():
+    probe = read_spikeglx(data_path / "NP2_4_shanks.imec0.ap.meta")
+
+    assert probe.annotations["name"] == "Neuropixels 2.0 - Four Shank"
+    assert probe.annotations["manufacturer"] == "IMEC"
+    assert probe.annotations["probe_type"] == 24
+    
+    assert probe.ndim == 2
+    assert probe.get_shank_count() == 4
+    assert probe.get_contact_count() == 384
+    
+    # Test contact geometry
+    contact_width = 12.0
+    contact_shape = "square"
+
+    assert np.all(probe.contact_shape_params == {"width": contact_width})
+    assert np.all(probe.contact_shapes == contact_shape)
+
+    # This file does not save the channnels from 0 as the one above (NP2_4_shanks_g0_t0.imec0.ap.meta)
+    ypos = probe.contact_positions[:, 1]
+    assert np.min(ypos) == pytest.approx(0)
+
+
+def test_NP2_4_shanks_with_different_electrodes_saved():
     # Data provided by Jennifer Colonell
-    probe = read_spikeglx(data_path / "NP24_g0_t0.imec0.ap.meta")
-    assert "2.0" in probe.annotations["name"]
+    probe = read_spikeglx(data_path / "NP2_4_shanks_save_different_electrodes.imec0.ap.meta")
+
+    assert probe.annotations["name"] == "Neuropixels 2.0 - Four Shank"
+    assert probe.annotations["manufacturer"] == "IMEC"
+    assert probe.annotations["probe_type"] == 24
+    
+    assert probe.ndim == 2
     assert probe.get_shank_count() == 4
+    assert probe.get_contact_count() == 384
+    
+    # Test contact geometry
+    contact_width = 12.0
+    contact_shape = "square"
 
+    assert np.all(probe.contact_shape_params == {"width": contact_width})
+    assert np.all(probe.contact_shapes == contact_shape)
+    
+    # This file does not save the channnels from 0 as the one above (NP2_4_shanks_g0_t0.imec0.ap.meta)
+    ypos = probe.contact_positions[:, 1]
+    assert np.min(ypos) == pytest.approx(2880.0)
+    assert np.max(ypos) == pytest.approx(5745.0)
 
-def test_NP1_large_depth_sapn():
+        
+def test_NP1_large_depth_span():
     # Data provided by Tom Bugnon NP1 with large Depth span
     probe = read_spikeglx(data_path / "allan-longcol_g0_t0.imec0.ap.meta")
     assert "1.0" in probe.annotations["name"]
@@ -64,35 +131,11 @@ def test_NP1_other_example():
 
 def tes_NP1_384_channels():
     # example by Pierre Yger NP1.0 with 384 but only 151 channels are saved
-    probe = read_spikeglx(data_path / "Day_3_g0_t0.imec1.ap.meta")
+    probe = read_spikeglx(data_path / "NP1_saved_only_subset_of_channels.meta")
     assert probe.get_shank_count() == 1
     assert probe.get_contact_count() == 151
     assert 152 not in probe.contact_annotations["channel_ids"]
 
-
-def test_parse_meta():
-    for meta_file in [
-        "doppio-checkerboard_t0.imec0.ap.meta",
-        "Day_3_g0_t0.imec1.ap.meta",
-        "allan-longcol_g0_t0.imec0.ap.meta",
-    ]:
-        meta = parse_spikeglx_meta(data_path / meta_file)
-
-
-def test_get_saved_channel_indices_from_spikeglx_meta():
-    # all channel saved + 1 synchro
-    chan_inds = get_saved_channel_indices_from_spikeglx_meta(
-        data_path / "Noise_g0_t0.imec0.ap.meta"
-    )
-    assert chan_inds.size == 385
-
-    # example by Pierre Yger NP1.0 with 384 but only 151 channels are saved + 1 synchro
-    chan_inds = get_saved_channel_indices_from_spikeglx_meta(
-        data_path / "Day_3_g0_t0.imec1.ap.meta"
-    )
-    assert chan_inds.size == 152
-
-
 def test_NPH_long_staggered():
     # Data provided by Nate Dolensek
     probe = read_spikeglx(data_path / "non_human_primate_long_staggered.imec0.ap.meta")