Skip to content
Merged
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
74 changes: 55 additions & 19 deletions neo/rawio/intanrawio.py
Original file line number Diff line number Diff line change
Expand Up @@ -530,6 +530,57 @@ def _demultiplex_digital_data(self, raw_digital_data, channel_ids, i_start, i_st
output[:, channel_index] = demultiplex_data[i_start:i_stop].flatten()

return output

def get_intan_timestamps(self, i_start=None, i_stop=None):
"""
Retrieves the sample indices from the Intan raw data within a specified range.

Note that sample indices are called timestamps in the Intan format but they are
in fact just sample indices. This function extracts the sample index timestamps
from Intan files, which represent relative time points in sample units (not absolute time).
These indices can be particularly useful when working with recordings that have discontinuities.

Parameters
----------
i_start : int, optional
The starting index from which to retrieve sample indices. If None, starts from 0.
i_stop : int, optional
The stopping index up to which to retrieve sample indices (exclusive).
If None, retrieves all available indices from i_start onward.

Returns
-------
timestamps : ndarray
The flattened array of sample indices within the specified range.

Notes
-----
- Sample indices can be converted to seconds by dividing by the sampling rate of the amplifier stream.
- The function automatically handles different file formats:
* header-attached: Timestamps are extracted directly from the timestamp field
* one-file-per-signal: Timestamps are read from the timestamp stream
* one-file-per-channel: Timestamps are read from the first channel in the timestamp stream
- When recordings have discontinuities (indicated by the `discontinuous_timestamps`
attribute being True), these indices allow for proper temporal alignment of the data.
"""
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would add a notes section here:

1 line to explain that samples can be converted to time through the sampling rate and maybe 1 line to say that this is relative time and not absolute time stamps
and then the 1-3 lines to explain that we handle each file format automatically (1 line if you think it is enough to just say we do it automatically or 3 lines if you want to explain how we do each format).

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed.

if i_start is None:
i_start = 0

# Get the timestamps based on file format
if self.file_format == "header-attached":
timestamps = self._raw_data["timestamp"]
elif self.file_format == "one-file-per-signal":
timestamps = self._raw_data["timestamp"]
elif self.file_format == "one-file-per-channel":
timestamps = self._raw_data["timestamp"][0]

# TODO if possible ensure that timestamps memmaps are always of correct shape to avoid memory copy here.
timestamps = timestamps.flatten() if timestamps.ndim > 1 else timestamps

if i_stop is None:
return timestamps[i_start:]
else:
return timestamps[i_start:i_stop]

def _assert_timestamp_continuity(self):
"""
Expand All @@ -545,26 +596,11 @@ def _assert_timestamp_continuity(self):
NeoReadWriteError
If timestamps are not continuous and `ignore_integrity_checks` is False.
The error message includes a table detailing the discontinuities found.

Notes
-----
The method extracts timestamps from the raw data based on the file format:

* **header-attached:** Timestamps are extracted from a 'timestamp' field in the raw data.
* **one-file-per-signal:** Timestamps are taken from the last stream.
* **one-file-per-channel:** Timestamps are retrieved from the first channel of the last stream.
"""
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do these notes make sense any more for the private method? Maybe you should move this to the get timestamps method now in a notes section.

# check timestamp continuity
if self.file_format == "header-attached":
timestamp = self._raw_data["timestamp"].flatten()

# timestamps are always last stream for headerless binary files
elif self.file_format == "one-file-per-signal":
timestamp = self._raw_data["timestamp"]
elif self.file_format == "one-file-per-channel":
timestamp = self._raw_data["timestamp"][0]
timestamps = self.get_intan_timestamps()

discontinuous_timestamps = np.diff(timestamp) != 1
discontinuous_timestamps = np.diff(timestamps) != 1
timestamps_are_not_contiguous = np.any(discontinuous_timestamps)
if timestamps_are_not_contiguous:
# Mark a flag that can be checked after parsing the header to see if the timestamps are continuous or not
Expand All @@ -582,8 +618,8 @@ def _assert_timestamp_continuity(self):

amplifier_sampling_rate = self._global_info["sampling_rate"]
for discontinuity_index in np.where(discontinuous_timestamps)[0]:
prev_ts = timestamp[discontinuity_index]
next_ts = timestamp[discontinuity_index + 1]
prev_ts = timestamps[discontinuity_index]
next_ts = timestamps[discontinuity_index + 1]
time_diff = (next_ts - prev_ts) / amplifier_sampling_rate

error_msg += (
Expand Down