Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 38 additions & 8 deletions neo/rawio/spikeglxrawio.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,17 +133,12 @@ def _parse_header(self):
stream_names = sorted(list(srates.keys()), key=lambda e: srates[e])[::-1]
nb_segment = np.unique([info["seg_index"] for info in self.signals_info_list]).size

self.signals_info_dict = {}
self.signals_info_dict = _build_signals_info_dict(self.signals_info_list)

# one unique block
self._buffer_descriptions = {0: {}}
self._stream_buffer_slice = {}
for info in self.signals_info_list:
seg_index, stream_name = info["seg_index"], info["stream_name"]
key = (seg_index, stream_name)
if key in self.signals_info_dict:
raise KeyError(f"key {key} is already in the signals_info_dict")
self.signals_info_dict[key] = info

for (seg_index, stream_name), info in self.signals_info_dict.items():
buffer_id = stream_name
block_index = 0

Expand Down Expand Up @@ -412,6 +407,41 @@ def scan_files(dirname):
return info_list


def _build_signals_info_dict(info_list):
"""
Re-index a flat list of info dicts into a dict keyed by (seg_index, stream_name).

Requires each info dict to already have "seg_index", "stream_name", and "meta_file",
populated by scan_files + _add_segment_order.

Raises ValueError on duplicate keys, naming both colliding .meta paths and
listing common causes so users can self-diagnose.
"""
signals_info_dict = {}
for info in info_list:
key = (info["seg_index"], info["stream_name"])
if key in signals_info_dict:
existing = signals_info_dict[key]
seg_index, stream_name = key
raise ValueError(
f"Two SpikeGLX file pairs resolve to the same stream "
f"'{stream_name}' in segment {seg_index}:\n"
f" 1) {existing['meta_file']}\n"
f" 2) {info['meta_file']}\n"
f"This can happen if:\n"
f" - Files were renamed on disk. Stream names come from the "
f"'fileName' field inside the .meta, not the filename on disk.\n"
f" - Recordings from different sessions are in the same folder "
f"with the same gate/trigger numbers.\n"
f" - Duplicate copies exist in subfolders (the reader scans "
f"recursively).\n"
f" - A third-party tool rewrote the .meta file with an incorrect "
f"'fileName' (for example, LF meta pointing to the AP binary)."
)
signals_info_dict[key] = info
return signals_info_dict


def _add_segment_order(info_list):
"""
Uses gate and trigger numbers to construct a segment index (`seg_index`) for each signal in `info_list`.
Expand Down
28 changes: 27 additions & 1 deletion neo/test/rawiotest/test_spikeglxrawio.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@

import unittest

from neo.rawio.spikeglxrawio import SpikeGLXRawIO
import pytest

from neo.rawio.spikeglxrawio import SpikeGLXRawIO, _build_signals_info_dict
from neo.test.rawiotest.common_rawio_test import BaseTestRawIO
import numpy as np

Expand Down Expand Up @@ -197,5 +199,29 @@ def test_t_start_reading(self):
)


def test_build_signals_info_dict_collision_raises_value_error():
info_a = {"seg_index": 0, "stream_name": "imec0.ap", "meta_file": "/x/first.meta"}
info_b = {"seg_index": 0, "stream_name": "imec0.ap", "meta_file": "/x/second.meta"}

expected_message = (
"Two SpikeGLX file pairs resolve to the same stream 'imec0.ap' in segment 0:\n"
" 1) /x/first.meta\n"
" 2) /x/second.meta\n"
"This can happen if:\n"
" - Files were renamed on disk. Stream names come from the 'fileName' field "
"inside the .meta, not the filename on disk.\n"
" - Recordings from different sessions are in the same folder with the same "
"gate/trigger numbers.\n"
" - Duplicate copies exist in subfolders (the reader scans recursively).\n"
" - A third-party tool rewrote the .meta file with an incorrect 'fileName' "
"(for example, LF meta pointing to the AP binary)."
)

with pytest.raises(ValueError) as exc_info:
_build_signals_info_dict([info_a, info_b])

assert str(exc_info.value) == expected_message


if __name__ == "__main__":
unittest.main()
Loading