LongRecordingOrganizer#

class neurodent.loading.LongRecordingOrganizer(
item: str | Path | list[str] | tuple[str] | DiscoveredFile,
mode: Literal['si', 'mne', None] = 'si',
truncate: bool | int = False,
cache_policy: Literal['auto', 'always', 'force_regenerate'] = 'auto',
multiprocess_mode: Literal['dask', 'serial'] = 'serial',
extract_func: Callable[[...], si.BaseRecording] | Callable[[...], Raw] | str = None,
manual_datetimes: datetime | list[datetime] = None,
datetimes_are_start: bool = True,
recording: si.BaseRecording = None,
**kwargs,
)[source]#

Bases: LroLoadingMixin, LroTimestampsMixin, LroFragmentsMixin, LroQualityMixin, LroPersistenceMixin, LroMergeMixin

Construct a long recording from various file formats or an existing recording object.

Parameters:

  • item (str | Path | list[str] | DiscoveredFile | None) – Input data specification. - str/Path: Single file or directory path - list[str]: Multiple files to concatenate - DiscoveredFile: File(s) discovered by FileDiscoverer (single or multi-file) - None: Used when initializing from an existing recording object

  • mode (Literal['si', 'mne', None], optional) – Data loading mode. Defaults to ‘si’. - ‘si’: Use SpikeInterface extractors - ‘mne’: Use MNE-Python extractors (creates intermediate file) - None: No data loading (item must be None, recording must be provided)

  • truncate (bool | int, optional) – If True, truncate to first 10 files. If an integer, truncate to first n files. Defaults to False.

  • cache_policy (Literal['auto', 'always', 'force_regenerate'], optional) – Cache policy for intermediate files. Defaults to ‘auto’.

  • multiprocess_mode (Literal['dask', 'serial'], optional) – Whether to load files in parallel with Dask. Only has an effect with mode='si' and an item that is a list of paths. mode='mne' loads a list serially. Defaults to ‘serial’.

  • extract_func (Callable | str, optional) – Function to extract data. - If str: name of SpikeInterface or MNE extractor (e.g., ‘read_intan’, ‘read_raw_edf’) - If Callable: custom extraction function - If None: defaults to si.load for SI mode

  • manual_datetimes (datetime | list[datetime], optional) – Manually provided timestamps.

  • datetimes_are_start (bool, optional) – If True (default), manual_datetimes are start times.

  • recording (si.BaseRecording, optional) – Existing SpikeInterface recording object for in-memory initialization. Use this when creating LRO wrappers around split recordings.

  • **kwargs – Additional arguments passed to the data loading functions.

Variables:

  • LongRecording (si.BaseRecording) – The SpikeInterface recording object.

  • meta (RecordingMetadata) – Technical metadata (sampling rate, channels, etc.).

  • channel_names (list[str]) – List of channel names.

  • file_durations (list[float]) – Duration of each individual file in seconds.

  • cumulative_file_durations (list[float]) – Cumulative duration timestamps for file boundaries.

  • temppaths (list[str]) – Paths to temporary files created during processing.

  • bad_channel_names (list[str]) – List of channels identified as bad/noisy.

  • _is_in_memory (bool) – True if this LRO was created from an in-memory recording (via split()).

Raises:

ValueError – If no data files are found, if the folder contains mixed file types, or if manual time parameters are invalid.

__init__(
item: str | Path | list[str] | tuple[str] | DiscoveredFile,
mode: Literal['si', 'mne', None] = 'si',
truncate: bool | int = False,
cache_policy: Literal['auto', 'always', 'force_regenerate'] = 'auto',
multiprocess_mode: Literal['dask', 'serial'] = 'serial',
extract_func: Callable[[...], si.BaseRecording] | Callable[[...], Raw] | str = None,
manual_datetimes: datetime | list[datetime] = None,
datetimes_are_start: bool = True,
recording: si.BaseRecording = None,
**kwargs,
)[source]#
property display_name: str#

Short display name for logging, derived from the item.

get_num_fragments(fragment_len_s)[source]#
get_fragment(fragment_len_s, fragment_idx)[source]#
apply_lof_threshold(lof_threshold: float)#

Apply threshold to existing LOF scores to determine bad channels.

Parameters:

lof_threshold (float) – Threshold for determining bad channels.

cleanup_rec()#
compute_bad_channels(
lof_threshold: float | None = None,
force_recompute: bool = False,
lof_chunk_duration_s: float = 60,
)#

Compute bad channels using LOF analysis with unified score storage.

Parameters:

  • lof_threshold (float, optional) – Threshold for determining bad channels from LOF scores. If None, only computes/loads scores without setting bad_channel_names.

  • force_recompute (bool) – Whether to recompute LOF scores even if they exist.

  • lof_chunk_duration_s (float) – Duration in seconds of each chunk used for the pairwise-distance computation in LOF. Defaults to 60.

convert_file_with_mne_to_recording(
extract_func: Callable[[...], Raw],
intermediate: Literal['edf', 'bin'] = 'edf',
intermediate_name=None,
intermediate_dir=None,
cache_policy: Literal['auto', 'always', 'force_regenerate'] = 'auto',
**kwargs,
)#
convert_file_with_si_to_recording(
extract_func: Callable[[...], BaseRecording],
cache_policy: Literal['auto', 'always', 'force_regenerate'] = 'auto',
multiprocess_mode: Literal['dask', 'serial'] = 'serial',
**kwargs,
)#

Load this item through a SpikeInterface-style reader.

skip_unreadable_files (an lro_kwargs entry, default False) controls what happens when the reader raises on a single file. Left off, the exception propagates and the animal fails loudly, which is the long-standing behaviour. Turned on, the file becomes a 0-sample placeholder that later stages drop, and every skipped path is recorded on self.skipped_files so the loss is auditable rather than a line lost in a log.

Turning it on is not free, and the cost is not where it looks. A dropped file is removed from the timeline, and when per-file datetimes are chained from a single session anchor the clock then advances only over the files that survived, so every later file in that session is stamped early by the dropped duration. That is why this is opt-in rather than the default: converting a loud crash into a silently shifted timeline would be a worse failure than the crash.

convert_to_mne() → RawArray#

Convert this LongRecording object to an MNE RawArray.

Returns:

The converted MNE RawArray

Return type:

mne.io.RawArray

detect_and_load_data(
mode: Literal['si', 'mne', None] = 'si',
cache_policy: Literal['auto', 'always', 'force_regenerate'] = 'auto',
multiprocess_mode: Literal['dask', 'serial'] = 'serial',
extract_func: Callable[[...], BaseRecording] | Callable[[...], Raw] | str | None = None,
**kwargs,
)#

Load in recording based on mode.

Parameters#

mode{“si”, “mne”, None}

Backend to use for loading recordings.

cache_policy{“auto”, “always”, “force_regenerate”}

Caching strategy for loaded recordings.

multiprocess_mode{“dask”, “serial”}

Parallelism strategy.

extract_funccallable or str, optional

Function (or reference to one) used to load each discovered file into a recording object. When a string, resolved in this order:

  1. Short name — looked up in spikeinterface.extractors / spikeinterface (for mode="si") or mne.io (for mode="mne"). Example: "read_intan".

  2. Module path (contains :) — imports a function from an installed module. Example: "neurodent.readers:read_bin_csv_pair".

  3. File path (contains :, ends in .py) — loads a function directly from a Python file. Example: "/absolute/path/to/readers.py:my_func".

**kwargs

Forwarded to the backend loading method.

finalize_file_timestamps()#

Finalize file timestamps using manual times if provided, otherwise validate CSV times.

get_date_string() → str#

Get the string representation of the recording date (Start Time).

Returns:

Date string in format “%b-%d-%Y” (e.g. “Jan-21-2022”).

Return type:

str

Raises:

ValueError – If no timestamps are available in the recording.

get_datetime_fragment(fragment_len_s, fragment_idx)#

Get the datetime for a specific fragment using the timestamp mapper.

Parameters:

  • fragment_len_s (float) – Length of each fragment in seconds

  • fragment_idx (int) – Index of the fragment to get datetime for

Returns:

The datetime corresponding to the start of the fragment

Return type:

datetime

Raises:

ValueError – If timestamp mapper is not initialized (only available in ‘bin’ mode)

get_dur_fragment(fragment_len_s, fragment_idx)[source]#
get_lof_scores() → dict#

Get LOF scores with channel names.

Returns:

Dictionary mapping channel names to LOF scores.

Return type:

dict

classmethod load_recording(
folder: str | Path,
*,
strict: bool = False,
) → LongRecordingOrganizer#

Load a recording previously written by save_recording().

Reloads the SpikeInterface recording from folder and overlays the NeuRodent metadata sidecar to restore the full LRO state (timestamps, durations, units, bad channels, labels). This is the faithful round-trip counterpart to save_recording(), and gives base_folder_path a concrete reader.

Parameters:

  • folder (Union[str, Path]) – Path to a saved recording folder (zarr or binary).

  • strict (bool, optional) – If True, raise when the sidecar is missing or invalid. If False (default), fall back to a bare reload (traces only, no restored timestamps) and emit a warning. Defaults to False.

Returns:

The reloaded LRO.

Return type:

LongRecordingOrganizer

Raises:

  • ImportError – If SpikeInterface is not available.

  • FileNotFoundError – If folder is not a directory, or (with strict=True) the sidecar is missing.

  • ValueError – With strict=True, if the sidecar is present but invalid.

merge(other_lro)#

Merge another LRO into this one using si.concatenate_recordings.

This creates a new concatenated recording from this LRO and the other LRO. The other LRO should represent a later time period to maintain temporal order.

Parameters:

other_lro (LongRecordingOrganizer) – The LRO to merge into this one

Raises:

  • ValueError – If LROs are incompatible (different channels, sampling rates, etc.)

  • ImportError – If SpikeInterface is not available

merge_many(other_lros)#

Merge several LROs into this one with a single flat concatenation.

Equivalent to calling merge() for each entry, but the recordings are concatenated once at the end instead of once per step, so the result is one level deep rather than N-1.

Why that matters: si.concatenate_recordings does not flatten, and merge passes its own output back in as the first operand, so folding N files nests N-1 deep. Sample lookup then walks the nesting, making a full pass O(N^2) (measured 2.75x slower at N=500), and past roughly 1000 files get_traces raises RecursionError against Python’s default limit. Pickling fails far earlier, at about N=125, which breaks any recording sent to a dask worker. The fold also leans left, putting the OLDEST file deepest, and analysis reads from frame 0 first, so the wall is hit immediately rather than partway through.

Per-step validation, channel renaming, 0-sample handling and metadata accumulation are all preserved; only the concatenation is deferred.

Parameters:

other_lros (list[LongRecordingOrganizer]) – LROs to merge, in temporal order.

Raises:

  • ValueError – If any LRO is incompatible with this one.

  • ImportError – If SpikeInterface is not available.

save_recording(
output_dir: str | Path,
format: Literal['zarr', 'binary'] = 'zarr',
overwrite: bool = False,
n_jobs: int = 1,
chunk_duration: str = '1s',
progress_bar: bool = True,
**kwargs,
) → Path#

Save this LRO’s recording to disk as a self-contained folder.

Delegates the recording write to SpikeInterface’s save() and additionally writes a NeuRodent metadata sidecar (NEURODENT_SIDECAR_NAME) inside the output folder so the full LRO state — timestamps, durations, units, bad channels, and labels — can be restored faithfully via load_recording().

The sidecar is written last (and atomically), so an interrupted save leaves a folder with no sidecar, which load_recording() detects and degrades on.

Parameters:

  • output_dir (Union[str, Path]) – Directory to save the recording to. For format="zarr" a .zarr suffix is appended if not already present.

  • format (Literal["zarr", "binary"], optional) – Save format. Defaults to “zarr”.

  • overwrite (bool, optional) – If False (default), raise FileExistsError when the target already exists. If True, the existing target is removed first — but only when it is a recognized SpikeInterface/NeuRodent recording folder; an unrecognized directory is never deleted (raises ValueError). Defaults to False.

  • n_jobs (int, optional) – Workers for SpikeInterface’s chunked disk write. Controls the write only. Feature computation is parallelized separately, by multiprocess_mode on the analysis functions. Defaults to 1.

  • chunk_duration (str, optional) – Chunk duration for processing. Defaults to “1s”.

  • progress_bar (bool, optional) – Show progress bar. Defaults to True.

  • **kwargs – Additional arguments passed to SI’s save().

Returns:

The actual output directory where the recording was saved (with the .zarr suffix applied for zarr format).

Return type:

Path

Raises:

  • ImportError – If SpikeInterface is not available.

  • ValueError – If there is no recording to save, or overwrite=True targets a directory that is not a recognized recording folder.

  • FileExistsError – If the target exists and overwrite=False, or the target exists but is not a directory.

save_to_edf(
filename: str | Path,
overwrite: bool = False,
)#

Save the recording to an EDF file via MNE.

Parameters:

  • filename (str | Path) – Path to save the EDF file to.

  • overwrite (bool) – Whether to overwrite if file exists.

split(
groups: dict[str, list[str]],
) → dict[str, LongRecordingOrganizer]#

Split the current recording into multiple in-memory LongRecordingOrganizer objects.

This creates lightweight LRO wrappers around channel-sliced views of the recording. No disk I/O is performed. Use save_recording() on individual results to save them to disk if needed.

Parameters:

groups (dict[str, list[str]]) – Dictionary mapping group names (e.g., ‘AnimalA’) to lists of channel names.

Returns:

Dictionary mapping group names to new LRO instances.

Return type:

dict[str, LongRecordingOrganizer]

Raises:

  • ValueError – If requested channels are not found in the recording.

  • ImportError – If SpikeInterface is not available.

Example

>>> lro = LongRecordingOrganizer("/path/to/data", mode="bin")
>>> splits = lro.split({"AnimalA": ["Ch1", "Ch2"], "AnimalB": ["Ch3", "Ch4"]})
>>> splits["AnimalA"].save_recording("/output/AnimalA", format="zarr")
__str__()[source]#

Return a string representation of critical long recording features.

__repr__()[source]#

Return a detailed string representation for debugging.