Fix main_channel_id donor selection when merged units disagree

src/spikeinterface/core/sorting_tools.py

@@ -518,8 +518,11 @@ def set_properties_after_merging(
             elif (not same_property_values) and key == "main_channel_id":
                 # Main channel id is special. For now, if there is a disagreement, we take the value of the unit
                 # with the most spikes. TODO: overwrite this for analyzer if templates exist.
+                # np.argmax() on a dict_values view always returned index 0, ignoring the counts entirely
+                # (NumPy can't iterate it, no __getitem__). Look counts up by unit id instead of by position.
                 num_spikes_per_unit = sorting_pre_merge.count_num_spikes_per_unit(unit_ids=merge_group)
-                max_unit_index = np.argmax(num_spikes_per_unit.values())
+                spike_counts_in_merge_group_order = [num_spikes_per_unit[unit_id] for unit_id in merge_group]
+                max_unit_index = np.argmax(spike_counts_in_merge_group_order)
                 new_values[new_index] = merge_values[max_unit_index]
             else:
                 if parent_values.dtype.kind not in default_missing_values:

src/spikeinterface/core/tests/test_sorting_tools.py

@@ -274,6 +274,22 @@ def test_set_properties_after_merging():
     assert not is_merged_diff[sorting_diff_merged.id_to_index("c")]
 
 
+def test_set_properties_after_merging_main_channel_id_disagreement():
+    # unit "a" has the most spikes (4); the merged unit should therefore keep "a"'s
+    # main_channel_id, regardless of the order merge_unit_groups lists the units in.
+    times = np.array([0, 1, 2, 3, 10, 20])
+    labels = np.array(["a", "a", "a", "a", "b", "c"])
+    sorting = NumpySorting.from_samples_and_labels([times], [labels], 10_000.0, unit_ids=["a", "b", "c"])
+    sorting.set_property("main_channel_id", np.array(["chA", "chB", "chC"]))
+
+    # merge_unit_groups lists "b" before "a", the reverse of sorting.unit_ids' order
+    sorting_merged, _, _ = apply_merges_to_sorting(
+        sorting, [["b", "a"]], censor_ms=None, new_id_strategy="append", return_extra=True
+    )
+    merged_main_channel_id = sorting_merged.get_property("main_channel_id")[sorting_merged.id_to_index("merge0")]
+    assert merged_main_channel_id == "chA"
+
+
 def test_set_properties_after_splits():
     times = np.array([0, 10, 20, 30, 40])
     labels = np.array(["a", "b", "b", "c", "c"])

src/spikeinterface/core/tests/test_sortinganalyzer.py

@@ -6,6 +6,8 @@
 
 from spikeinterface.core import (
     generate_ground_truth_recording,
+    generate_recording,
+    NumpySorting,
     create_sorting_analyzer,
     load_sorting_analyzer,
     get_available_analyzer_extensions,
@@ -1068,6 +1070,36 @@ def test_main_channel_from_templates_sparse_recordingless():
     assert np.array_equal(recovered_main_channel_ids, expected_main_channel_ids)
 
 
+def test_merge_units_main_channel_id_disagreement():
+    """`SortingAnalyzer.merge_units()` must keep the donor unit's `main_channel_id` on disagreement,
+    matching `apply_merges_to_sorting` (see test_sorting_tools.py), even when `merge_unit_groups`
+    lists the units in the same relative order as `sorting.unit_ids` (not just the reversed order):
+    a defect that only manifests for out-of-order groups would still pass this case.
+    """
+    recording = generate_recording(num_channels=3, durations=[2.0], set_probe=True, seed=0)
+    recording = recording.rename_channels(new_channel_ids=["chA", "chB", "chC"])
+
+    # unit "b" has more spikes (5) than unit "a" (2); "a" is listed first in both `sorting.unit_ids`
+    # and `merge_unit_groups` below, so a positional (rather than spike-count) donor choice would
+    # silently pick "a" and still pass an order-reversed-only regression test.
+    times = np.array([0, 1, 100, 110, 120, 130, 140])
+    labels = np.array(["a", "a", "b", "b", "b", "b", "b"])
+    sorting = NumpySorting.from_samples_and_labels(
+        [times], [labels], recording.sampling_frequency, unit_ids=["a", "b", "c"]
+    )
+    sorting.set_property("main_channel_id", np.array(["chA", "chB", "chC"]))
+    sorting.register_recording(recording)
+
+    analyzer = create_sorting_analyzer(sorting, recording, format="memory", sparse=False)
+    merged_analyzer, new_unit_ids = analyzer.merge_units(
+        merge_unit_groups=[["a", "b"]], new_id_strategy="append", return_new_unit_ids=True
+    )
+    merged_main_channel_id = merged_analyzer.sorting.get_property("main_channel_id")[
+        merged_analyzer.sorting.id_to_index(new_unit_ids[0])
+    ]
+    assert merged_main_channel_id == "chB"
+
+
 if __name__ == "__main__":
     tmp_path = Path("test_SortingAnalyzer")
     dataset = get_dataset()