Utilities#

Utility functions for the core module.

Functions#

neurodent.core.utils.convert_units_to_multiplier(current_units: str, target_units: str = 'µV') → float[source]#

Convert between different voltage units and return the multiplication factor.

This function calculates the conversion factor needed to transform values from one voltage unit to another (e.g., from mV to µV).

Parameters:

  • current_units (str) – The current unit of the values. Must be one of: ‘µV’, ‘mV’, ‘V’, ‘nV’.

  • target_units (str, optional) – The target unit to convert to. Defaults to ‘µV’. Must be one of: ‘µV’, ‘mV’, ‘V’, ‘nV’.

Returns:

The multiplication factor to convert from current_units to target_units.

To convert values, multiply your data by this factor.

Return type:

float

Raises:

AssertionError – If current_units or target_units are not supported.

Examples

>>> convert_units_to_multiplier("mV", "µV")
1000.0
>>> convert_units_to_multiplier("V", "mV")
1000.0
>>> convert_units_to_multiplier("µV", "V")
1e-06
neurodent.core.utils.extract_mne_unit_info(raw_info: dict) → tuple[str | None, float | None][source]#

Extract unit information from MNE Raw info object.

Parameters:

raw_info (dict) – MNE Raw.info object containing channel information

Returns:

(unit_name, mult_to_uV) where unit_name

is the consistent unit across all channels and mult_to_uV is the conversion factor to µV

Return type:

tuple[str | None, float | None]

Raises:

ValueError – If channel units are inconsistent across channels

neurodent.core.utils.is_day(dt: datetime, sunrise=6, sunset=18)[source]#

Check if a datetime object is during the day.

Parameters:

  • dt (datetime) – Datetime object to check

  • sunrise (int, optional) – Sunrise hour (0-23). Defaults to 6.

  • sunset (int, optional) – Sunset hour (0-23). Defaults to 18.

Returns:

True if the datetime is during the day, False otherwise

Return type:

bool

Raises:

TypeError – If dt is not a datetime object

neurodent.core.utils.convert_colpath_to_rowpath(rowdir_path: str | Path, col_path: str | Path, gzip: bool = True, aspath: bool = True) → str | Path[source]#

Convert a ColMajor file path to its corresponding RowMajor file path.

This function transforms file paths from column-major format to row-major format, which is used when converting between different data storage layouts in NeuRodent.

Parameters:

  • rowdir_path (str | Path) – Directory path where the RowMajor file should be located.

  • col_path (str | Path) – Path to the ColMajor file to be converted. Must contain ‘ColMajor’ in the path.

  • gzip (bool, optional) – If True, append ‘.npy.gz’ extension. If False, append ‘.bin’. Defaults to True.

  • aspath (bool, optional) – If True, return as Path object. If False, return as string. Defaults to True.

Returns:

The converted RowMajor file path, either as string or Path object based on aspath parameter.

Return type:

str | Path

Raises:

ValueError – If ‘ColMajor’ is not found in col_path.

Examples

>>> convert_colpath_to_rowpath("/data/row/", "/data/col/file_ColMajor_001.bin")
PosixPath('/data/row/file_RowMajor_001.npy.gz')
>>> convert_colpath_to_rowpath("/data/row/", "/data/col/file_ColMajor_001.bin", gzip=False)
PosixPath('/data/row/file_RowMajor_001.bin')
>>> convert_colpath_to_rowpath("/data/row/", "/data/col/file_ColMajor_001.bin", aspath=False)
'/data/row/file_RowMajor_001.npy.gz'
neurodent.core.utils.filepath_to_index(filepath) → int[source]#

Extract the index number from a filepath.

This function extracts the last number found in a filepath after removing common suffixes and file extensions. For example, from “/path/to/data_ColMajor_001.bin” it returns 1.

Parameters:

filepath (str | Path) – Path to the file to extract index from.

Returns:

The extracted index number, or 0 if no number is found in the filename.

Return type:

int

Examples

>>> filepath_to_index("/path/to/data_ColMajor_001.bin")
1
>>> filepath_to_index("/path/to/data_2023_015_ColMajor.bin")
15
>>> filepath_to_index("/path/to/data_Meta_010.json")
10
neurodent.core.utils.parse_truncate(truncate: int | bool) → int[source]#

Parse the truncate parameter to determine how many characters to truncate.

If truncate is a boolean, returns 10 if True and 0 if False. If truncate is an integer, returns that integer value directly.

Parameters:

truncate (int | bool) – If bool, True=10 chars and False=0 chars. If int, specifies exact number of chars.

Returns:

Number of characters to truncate (0 means no truncation)

Return type:

int

Raises:

ValueError – If truncate is not a boolean or integer

neurodent.core.utils.get_feature_label(feature_name: str) → str[source]#

Convert a feature column name to a human-readable label.

Handles: - Base features: “rms” -> “RMS” - Banded features: “logpsdband_delta” -> “Log Band Power - Delta” - Baseline-subtracted: “logrms_nobase” -> “Log(RMS) - Baseline”

Parameters:

feature_name – Column name (e.g., “logpsdband_delta_nobase”)

Returns:

Human-readable label. Falls back to the original name if not found.

Examples

>>> get_feature_label("logpsdband_delta")
'Log Band Power (Delta)'
>>> get_feature_label("alphadelta")
'Alpha/Delta Ratio'
>>> get_feature_label("logrms_nobase")
'Log(RMS) - Baseline'
neurodent.core.utils.nanaverage(A: ndarray, weights: ndarray, axis: int = -1) → ndarray[source]#

Compute weighted average of an array, ignoring NaN values.

This function computes a weighted average along the specified axis while properly handling NaN values by masking them out of the calculation.

Parameters:

  • A (np.ndarray) – Input array containing the values to average.

  • weights (np.ndarray) – Array of weights corresponding to the values in A. Must be broadcastable with A along the specified axis.

  • axis (int, optional) – Axis along which to compute the average. Defaults to -1 (last axis).

Returns:

Weighted average with NaN values properly handled. If all values

along an axis are NaN, the result will be NaN for that position.

Return type:

np.ndarray

Examples

>>> import numpy as np
>>> A = np.array([[1.0, 2.0, np.nan], [4.0, np.nan, 6.0]])
>>> weights = np.array([1, 2, 1])
>>> nanaverage(A, weights, axis=1)
array([1.66666667, 5.        ])

Note

Be careful with zero or negative weights as they may produce unexpected results. The function uses numpy’s masked array functionality for robust NaN handling.

neurodent.core.utils.parse_str_to_animal(string: str, animal_param: tuple[int, str] | str | list[str] = (0, None)) → str[source]#

DEPRECATED: Use FileDiscoverer with {animal} placeholder in pattern instead.

Parses the filename of a binfolder to get the animal id.

Parameters:

  • string (str) – String to parse.

  • animal_param – Parameter specifying how to parse the animal ID: tuple[int, str]: (index, separator) for simple split and index. Not recommended for inconsistent naming conventions. str: regex pattern to extract ID. Most general use case. If multiple matches are found, returns the first match. list[str]: list of possible animal IDs to match against. Returns first match in list order, case-sensitive, ignoring empty strings.

Returns:

Animal id.

Return type:

str

Examples

# Tuple format: (index, separator) >>> parse_str_to_animal(“WT_A10_2023-01-01_data.bin”, (1, “_”)) ‘A10’ >>> parse_str_to_animal(“A10_WT_recording.bin”, (0, “_”)) ‘A10’

# Regex pattern format >>> parse_str_to_animal(“WT_A10_2023-01-01_data.bin”, r”A\d+”) ‘A10’ >>> parse_str_to_animal(“subject_123_data.bin”, r”\d+”) ‘123’

# List format: possible IDs to match >>> parse_str_to_animal(“WT_A10_2023-01-01_data.bin”, [“A10”, “A11”, “A12”]) ‘A10’ >>> parse_str_to_animal(“WT_A10_data.bin”, [“B15”, “C20”]) # No match ValueError: No matching ID found in WT_A10_data.bin from possible IDs: [‘B15’, ‘C20’]

neurodent.core.utils.parse_str_to_day(string: str, sep: str | None = None, parse_params: dict | None = None, parse_mode: Literal['full', 'split', 'window', 'all'] = 'split', date_patterns: list[tuple[str, str]] | None = None) → datetime[source]#

DEPRECATED: Use FileDiscoverer with {session} placeholder in pattern instead.

Parses the filename of a binfolder to get the day.

Parameters:

  • string (str) – String to parse.

  • sep (str, optional) – Separator to split string by. If None, split by whitespace. Defaults to None.

  • parse_params (dict, optional) – Parameters to pass to dateutil.parser.parse. Defaults to {‘fuzzy’:True}.

  • parse_mode (Literal["full", "split", "window", "all"], optional) – Mode for parsing the string. Defaults to “split”. “full”: Try parsing the entire cleaned string only “split”: Try parsing individual tokens only “window”: Try parsing sliding windows of tokens (2-4 tokens) only “all”: Use all three approaches in the order “full”, “split”, “window

  • date_patterns (list[tuple[str, str]], optional) – List of (regex_pattern, strptime_format) tuples to try before falling back to token-based parsing. This allows users to specify exact formats to handle ambiguous cases like MM/DD/YYYY vs DD/MM/YYYY. Only used in “split” and “all” modes. Defaults to None (no regex patterns).

Returns:

Datetime object corresponding to the day of the binfolder.

Return type:

datetime

Raises:

  • ValueError – If no valid date token is found in the string.

  • TypeError – If date_patterns is not a list of tuples.

Examples

>>> # Handle ambiguous date formats with explicit patterns
>>> patterns = [(r'(19\d{2}|20\d{2})-(\d{1,2})-(\d{1,2})', '%Y-%m-%d')]
>>> parse_str_to_day('2001_2023-07-04_data', date_patterns=patterns)
datetime.datetime(2023, 7, 4, 0, 0)

>>> # European format pattern
>>> patterns = [(r'(\d{1,2})/(\d{1,2})/(19\d{2}|20\d{2})', '%d/%m/%Y')]
>>> parse_str_to_day('04/07/2023_data', date_patterns=patterns)
datetime.datetime(2023, 7, 4, 0, 0)  # July 4th, not April 7th

Note

When date_patterns is provided, users have full control over date interpretation. Without date_patterns, the function falls back to token-based parsing which may be ambiguous for formats like MM/DD/YYYY vs DD/MM/YYYY.

neurodent.core.utils.resolve_channel(channel_name: str) → str[source]#

Resolve a raw channel name to its canonical channel abbreviation by exact lookup.

Resolution is explicit and never inferred: (1) the (stripped) name is already a canonical abbreviation (neurodent.constants.CHANNEL_ABBREVS); (2) it is an exact key in neurodent.constants.CHANNEL_ABBREV_BY_RAW (the per-dataset raw name -> abbrev map). Anything else raises loudly — there is no fuzzy, substring, or number-based guessing.

Parameters:

channel_name (str) – Raw channel name from the data.

Returns:

Canonical channel abbreviation.

Return type:

str

Raises:

ValueError – When the name is not in the configured channel map. Configure the exact raw name under its abbreviation (channels in the samples config, or neurodent.set_channel_map()).

Examples

>>> resolve_channel("LMot")          # already canonical
'LMot'
>>> resolve_channel("L Motor Ctx")   # configured raw name -> abbrev
'LMot'
neurodent.core.utils.resolve_channels(names: list[str]) → list[str][source]#

Abbreviate a list of raw channel names via exact lookup.

Unmappable names are warned about loudly (and kept as-is so callers comparing channel sets still get a value) rather than silently swallowed.

Parameters:

names – List of raw channel name strings.

Returns:

List of canonical abbreviations (same length as input); an unmappable entry is returned unchanged after a warning.

neurodent.core.utils.normalize_value_from_aliases(value: str, alias_dict: dict[str, list[str]]) → str | None[source]#

Normalize a value to its canonical form using a value map.

Performs exact matching: the value must equal one of the accepted spellings listed for a canonical label. Used for normalizing standalone configuration values against an exact _MAP (e.g. SEX_MAP, GENOTYPE_MAP).

Parameters:

  • value – The raw value to normalize (e.g., "M", "female").

  • alias_dict – Dictionary of {canonical_key: [accepted spellings]}.

Returns:

The canonical key if value matches any spelling, or None if no match.

neurodent.core.utils.set_temp_directory(path: str | Path) → None[source]#

Set the temporary directory for NeuRodent operations.

This function configures the temporary directory used by NeuRodent for intermediate files and operations. The directory will be created if it doesn’t exist.

Parameters:

path (str | Path) – Path to the temporary directory. Will be created if it doesn’t exist.

Examples

>>> set_temp_directory("/tmp/neurodent_temp")
>>> set_temp_directory(Path.home() / "neurodent_workspace" / "temp")

Note

This function modifies the TMPDIR environment variable, which affects the behavior of other temporary file operations in the process.

neurodent.core.utils.get_temp_directory() → Path[source]#

Get the current temporary directory used by NeuRodent.

Returns:

Path object representing the current temporary directory.

Return type:

Path

Examples

>>> temp_dir = get_temp_directory()
>>> print(f"Current temp directory: {temp_dir}")
Current temp directory: /tmp/neurodent_temp
Raises:

KeyError – If TMPDIR environment variable is not set.

Delete a file if it exists, ignoring a missing file.

Used for self-healing cache deletion: a corrupt cache file is removed so it can be regenerated, and a concurrently-removed file is not an error.

Parameters:

path – Path to the file to delete.

neurodent.core.utils.is_si_recording_folder(path: str | Path) → bool[source]#

Return True if path looks like a SpikeInterface recording output folder.

Recognizes the two formats written by LongRecording.save() as well as folders written by NeuRodent’s own LongRecordingOrganizer.save_recording(). This is a safety gate so destructive overwrites only ever target folders we actually produced — never an arbitrary user directory.

A folder qualifies when it is a directory and any of the following hold:

  • Zarr: the folder ends in .zarr and contains zarr group metadata (.zattrs, .zmetadata, or zarr.json).

  • Binary: the folder contains SpikeInterface’s recognition marker si_folder.json (or binary.json).

  • NeuRodent: the folder contains our own sidecar (NEURODENT_SIDECAR_NAME).

Parameters:

path – Path to inspect.

Returns:

True if path is a recognized recording output folder.

Return type:

bool

neurodent.core.utils.safe_rmtree(path: str | Path, *, require_marker: bool = True) → None[source]#

Recursively delete a directory tree, refusing unrecognized targets.

A guarded counterpart to safe_unlink() for directories. By default it will only delete a directory that is_si_recording_folder() recognizes, so a mistyped or malicious path can never wipe an arbitrary data directory.

Parameters:

  • path – Directory to remove.

  • require_marker – When True (default), raise ValueError unless the target is a recognized SpikeInterface/NeuRodent recording folder.

Raises:

ValueError – If require_marker is True and the target is not a recognized recording folder.

neurodent.core.utils.atomic_output_path(final_path: str | Path)[source]#

Context manager yielding a temporary sibling path for an atomic write.

The caller writes to the yielded temporary path. On clean exit the temp file is atomically moved into place with os.replace(); on exception the temp file is removed and the original error re-raised. Because the temp file lives in the same directory as final_path (same filesystem), os.replace is atomic, so a crash mid-write can never leave a partial file at final_path.

Parameters:

final_path – The destination path the content should end up at.

Yields:

Path – A temporary path in the same directory to write to.

Examples

>>> with atomic_output_path("out.bin") as tmp:  
...     data.tofile(tmp)
neurodent.core.utils.atomic_write_json(path: str | Path, obj: Any, *, indent: int = 2) → None[source]#

Atomically write obj to path as JSON.

Serializes to a temporary sibling file and atomically renames it into place, so an interrupted write never leaves a partial/corrupt JSON file at path.

Parameters:

  • path – Destination JSON file path.

  • obj – JSON-serializable object to write.

  • indent – Indentation passed to json.dump().

neurodent.core.utils.cache_fragments_to_zarr(np_fragments: ndarray, n_fragments: int, tmpdir: str | None = None, chunk_size: int | None = None) → tuple[str, zarr.Array][source]#

Cache numpy fragments array to zarr format for efficient memory management.

This function converts a numpy array of recording fragments to a zarr array stored in a temporary location. This allows better memory management and garbage collection by avoiding keeping large numpy arrays in memory for extended periods.

Parameters:

  • np_fragments (np.ndarray) – Numpy array of shape (n_fragments, n_samples, n_channels) containing the recording fragments to cache.

  • n_fragments (int) – Number of fragments to cache (allows for subset caching).

  • tmpdir (str, optional) – Directory path for temporary zarr storage. If None, uses get_temp_directory(). Defaults to None.

  • chunk_size (int, optional) – Number of fragments per zarr chunk along the first axis. Controls the read/write granularity when accessing the zarr array. Smaller values reduce memory overhead per chunk; larger values improve sequential throughput. When None, defaults to min(100, n_fragments).

Returns:

A tuple containing:

  • str: Path to the temporary zarr file

  • zarr.Array: The zarr array object for accessing cached data

Return type:

tuple[str, zarr.Array]

Raises:

  • ImportError – If zarr is not available

  • ValueError – If chunk_size is not None and is less than 1

neurodent.core.utils.stream_fragments_to_zarr(get_fragment_fn: Callable[[int], ndarray], n_fragments: int, fragment_shape: tuple, fragment_dtype: dtype, chunk_size: int, tmpdir: str | None = None) → str[source]#

Stream recording fragments to a zarr store in memory-bounded batches.

Unlike cache_fragments_to_zarr(), this function never holds more than chunk_size fragments in RAM at once. It calls get_fragment_fn one batch at a time, writes each batch to the zarr store, and immediately frees the batch buffer — so peak RAM is proportional to chunk_size rather than n_fragments.

Parameters:

  • get_fragment_fn (Callable[[int], np.ndarray]) – A callable that accepts a fragment index (0-based) and returns the corresponding fragment as a NumPy array of shape fragment_shape.

  • n_fragments (int) – Total number of fragments to stream.

  • fragment_shape (tuple) – Shape of a single fragment (e.g. (n_samples, n_channels)).

  • fragment_dtype (np.dtype) – Data-type of the fragment arrays.

  • chunk_size (int) – Number of fragments to buffer per batch. Must be >= 1. Larger values improve sequential write throughput; smaller values reduce peak RAM.

  • tmpdir (str, optional) – Directory for the temporary zarr file. If None, uses get_temp_directory().

Returns:

Path to the temporary zarr file on disk.

Return type:

str

Raises:

  • ValueError – If chunk_size < 1.

  • ImportError – If zarr is not available.

neurodent.core.utils.stream_recording_to_zarr(recording, n_fragments: int, n_samples_per_frag: int, n_frag_per_chunk: int, tmpdir: str | None = None) → str[source]#

Stream a SpikeInterface recording to a zarr store in memory-bounded batches.

Reads chunk-sized slices from recording.get_traces(), reshapes each chunk to (n_frags_in_chunk, n_samples_per_frag, n_channels), and writes it to a zarr store. Peak RAM is proportional to n_frag_per_chunk rather than n_fragments.

Parameters:

  • recording – A SpikeInterface BaseRecording object (may be a lazy wrapper such as a NotchFilterRecording).

  • n_fragments (int) – Total number of fragments to stream.

  • n_samples_per_frag (int) – Number of samples in each fragment.

  • n_frag_per_chunk (int) – Number of fragments to buffer per batch. Must be >= 1. Larger values improve sequential write throughput; smaller values reduce peak RAM.

  • tmpdir (str, optional) – Directory for the temporary zarr file. If None, uses get_temp_directory().

Returns:

Path to the temporary zarr file on disk.

Return type:

str

Raises:

  • ValueError – If n_frag_per_chunk < 1.

  • ImportError – If zarr is not available.

neurodent.core.utils.chunked_channel_distance_matrix(get_traces_fn: Callable[[int, int], ndarray], n_channels: int, n_samples: int, chunk_samples: int) → ndarray[source]#

Compute pairwise Euclidean distance matrix between channels in chunks.

Instead of loading the full (n_samples, n_channels) trace matrix at once, this function reads chunk_samples frames at a time and accumulates squared distances using the identity

||c_i - c_j||^2 = ||c_i||^2 + ||c_j||^2 - 2 * c_i · c_j

so that peak RAM is proportional to chunk_samples * n_channels rather than n_samples * n_channels.

Parameters:

  • get_traces_fn – fn(start_frame, end_frame) -> np.ndarray with shape (frames, n_channels). Typically recording.get_traces(start_frame=..., end_frame=..., return_scaled=True).

  • n_channels – Number of channels.

  • n_samples – Total number of samples in the recording.

  • chunk_samples – Number of samples to read per chunk.

Returns:

Symmetric (n_channels, n_channels) Euclidean distance matrix.

Return type:

np.ndarray

neurodent.core.utils.get_file_stem(filepath: str | Path) → str[source]#

Get the true stem for files, handling double extensions like .npy.gz.

neurodent.core.utils.nanmean_series_of_np(x: Series, axis: int = 0) → ndarray[source]#

Efficiently compute NaN-aware mean of a pandas Series containing numpy arrays.

This function is optimized for computing the mean across a Series where each element is a numpy array. It uses different strategies based on the size of the Series for optimal performance.

Parameters:

  • x (pd.Series) – Series containing numpy arrays as elements.

  • axis (int, optional) – Axis along which to compute the mean. Defaults to 0. - axis=0: Mean across the Series elements (most common) - axis=1: Mean within each array element

Returns:

Array containing the computed means with NaN values properly handled.

Return type:

np.ndarray

Examples

>>> import pandas as pd
>>> import numpy as np
>>> # Create a Series of numpy arrays
>>> arrays = [np.array([1.0, 2.0, np.nan]),
...           np.array([4.0, np.nan, 6.0]),
...           np.array([7.0, 8.0, 9.0])]
>>> series = pd.Series(arrays)
>>> nanmean_series_of_np(series)
array([4. , 5. , 7.5])
Performance Notes:

  • For Series with more than 1000 elements containing numpy arrays, uses np.stack() for better performance

  • Falls back to list conversion for smaller Series or mixed types

  • Handles shape mismatches gracefully by falling back to the slower method

neurodent.core.utils.log_transform(rec: ndarray, **kwargs) → ndarray[source]#

Log transform the signal

Parameters:

rec (np.ndarray) – The signal to log transform.

Returns:

ln(rec + 1)

Return type:

np.ndarray

neurodent.core.utils.sort_dataframe_by_plot_order(df: DataFrame, df_sort_order: dict | None = None) → DataFrame[source]#

Sort DataFrame columns according to predefined orders.

Parameters#

dfpd.DataFrame

DataFrame to sort

df_sort_orderdict

Dictionary mapping column names to the order of the values in the column.

Returns#

pd.DataFrame

Sorted DataFrame

Raises#

ValueError

If df_sort_order is not a valid dictionary or contains invalid categories

class neurodent.core.utils.Natural_Neighbor[source]#

Bases: object

Natural Neighbor algorithm implementation for finding natural neighbors in a dataset.

This class implements the Natural Neighbor algorithm which finds mutual neighbors in a dataset by iteratively expanding the neighborhood radius until convergence.

load(filename)[source]#

Load dataset from a CSV file, separating attributes and classes.

Parameters:

filename (str) – Path to the CSV file containing the dataset

read(data: ndarray)[source]#

Load data directly from a numpy array.

Parameters:

data (np.ndarray) – Input data array

read_distance_matrix(distance_matrix: ndarray)[source]#

Load a precomputed distance matrix for neighbor search.

When a distance matrix is provided, algorithm() uses argsort-based neighbor lookup instead of a KDTree, avoiding the need to hold the raw high-dimensional data in memory.

Parameters:

distance_matrix (np.ndarray) – Symmetric (n, n) distance matrix.

asserts()[source]#

Initialize data structures for the algorithm.

Sets up the necessary data structures including: - nan_edges as an empty set - knn, nan_num, and repeat dictionaries for each instance

count()[source]#

Count the number of instances that have no natural neighbors.

Returns:

Number of instances with zero natural neighbors

Return type:

int

findKNN(inst, r, tree)[source]#

Find the indices of the k nearest neighbors.

Parameters:

  • inst – Instance to find neighbors for

  • r (int) – Radius/parameter for neighbor search

  • tree – KDTree object for efficient neighbor search

Returns:

Array of neighbor indices (excluding the instance itself)

Return type:

np.ndarray

algorithm()[source]#

Execute the Natural Neighbor algorithm.

The algorithm iteratively expands the neighborhood radius until convergence, finding mutual neighbors between instances.

When a precomputed distance matrix is available (see read_distance_matrix()), neighbor lookup is performed via argsort instead of a KDTree, which avoids holding the raw high-dimensional data in memory.

Returns:

The final radius value when convergence is reached

Return type:

int

class neurodent.core.utils.TimestampMapper(file_end_datetimes: list[datetime], file_durations: list[float])[source]#

Bases: object

Map each fragment to its source file’s timestamp.

This class provides functionality to map data fragments back to their original file timestamps when data has been concatenated from multiple files with different recording times.

Variables:

  • file_end_datetimes (list[datetime]) – The end datetimes of each source file.

  • file_durations (list[float]) – The durations of each source file in seconds.

  • file_start_datetimes (list[datetime]) – Computed start datetimes of each file.

  • cumulative_durations (np.ndarray) – Cumulative sum of file durations.

Examples

>>> from datetime import datetime, timedelta
>>> # Set up files with known end times and durations
>>> end_times = [datetime(2023, 1, 1, 12, 0), datetime(2023, 1, 1, 13, 0)]
>>> durations = [3600.0, 1800.0]  # 1 hour, 30 minutes
>>> mapper = TimestampMapper(end_times, durations)
>>>
>>> # Get timestamp for fragment at index 2 with 60s fragments
>>> timestamp = mapper.get_fragment_timestamp(2, 60.0)
>>> print(timestamp)
2023-01-01 11:02:00
get_fragment_timestamp(fragment_idx: int, fragment_len_s: float) → datetime[source]#

Get the timestamp for a specific fragment based on its index and length.

Parameters:

  • fragment_idx (int) – The index of the fragment (0-based).

  • fragment_len_s (float) – The length of each fragment in seconds.

Returns:

The timestamp corresponding to the start of the specified fragment.

Return type:

datetime

Examples

>>> # Get timestamp for the 5th fragment (index 4) with 30-second fragments
>>> timestamp = mapper.get_fragment_timestamp(4, 30.0)
>>> # This returns the timestamp 2 minutes into the first file
neurodent.core.utils.validate_timestamps(timestamps: list[datetime], gap_threshold_seconds: float = 60) → list[datetime][source]#

Validate that timestamps are in chronological order and check for large gaps.

Parameters:

  • timestamps (list[datetime]) – List of timestamps to validate

  • gap_threshold_seconds (float, optional) – Threshold in seconds for warning about large gaps. Defaults to 60.

Returns:

The validated timestamps in chronological order

Return type:

list[datetime]

Raises:

ValueError – If no valid timestamps are provided

neurodent.core.utils.should_use_cached_file(cache_path: str | Path, source_paths: list[str | Path], use_cached: Literal['auto', 'always', 'never', 'error'] = 'auto') → bool[source]#

Determine whether to use a cached intermediate file based on caching policy and file timestamps.

Parameters:

  • cache_path – Path to the cached intermediate file

  • source_paths – List of source file paths that the cache depends on

  • use_cached – Caching policy - “auto”: Use cached if exists and newer than all sources (default) - “always”: Always use cached if it exists - “never”: Never use cached (always regenerate) - “error”: Raise error if cached doesn’t exist

Returns:

True if cached file should be used, False if it should be regenerated

Return type:

bool

Raises:

  • FileNotFoundError – When use_cached=”error” and cache doesn’t exist

  • ValueError – For invalid use_cached values

neurodent.core.utils.get_cache_status_message(cache_path: str | Path, use_cached: bool) → str[source]#

Generate a descriptive message about cache usage for logging.

neurodent.core.utils.should_use_cache_unified(cache_path: str | Path, source_paths: list[str | Path], cache_policy: Literal['auto', 'always', 'force_regenerate']) → bool[source]#

Unified cache decision logic for all intermediate files.

Parameters:

  • cache_path – Path to the cache file

  • source_paths – List of source file paths to check timestamps against

  • cache_policy – Caching policy: - “auto”: Use cache if exists and newer than sources, regenerate with logging if missing/invalid - “always”: Use cache if exists, raise error if missing/invalid - “force_regenerate”: Always regenerate and overwrite existing cache

Returns:

True if cache should be used, False if should regenerate

Return type:

bool

Raises:

ValueError – If cache_policy is invalid

neurodent.core.utils.rename_mne_channels(mne_obj)[source]#

Rename an MNE object’s channels in place to canonical abbreviations.

Applies resolve_channel() (exact lookup) to every entry of mne_obj.info['ch_names']. Format-agnostic — works on any MNE object whose raw channel names are declared in CHANNEL_MAP.

Parameters:

mne_obj – An MNE object exposing info['ch_names'] (e.g. a RawArray).

Returns:

The same mne_obj, with channel names replaced by their canonical abbreviations.

neurodent.core.utils.slugify(value, allow_unicode=False)[source]#

Convert a string to a URL-friendly slug.

Converts to ASCII (unless allow_unicode is True), lowercases, removes non-alphanumeric characters (except hyphens and underscores), and converts spaces and repeated dashes to single dashes.

Drop-in replacement for django.utils.text.slugify using only the standard library.

Path-safety convention.

This is the canonical helper for converting display-friendly identifiers (animal IDs, animaldays, genotype strings) into filesystem-safe path components. Any code that constructs a Path or filename from one of these strings must route the value through slugify(...) directly, or through one of the path_safe_* accessors on WindowAnalysisResult, LazyWindowAnalysisResult, or FrequencyDomainSpikeAnalysisResult.

Display strings — which may contain /, ;, parens, spaces, etc. (e.g. the real arxrosa genotype Arx(F/y); Rosa(+/wt)) — are correct domain notation and stay unchanged on the public attributes (animal_id, animaldays, genotype). They’re the source of truth for what humans see in logs and plot labels. Only the path_safe_* accessors return the slugified form.

Parameters:

  • value – The string to slugify.

  • allow_unicode – If True, keep Unicode characters instead of transliterating to ASCII.

Returns:

A URL-safe slug string.

Return type:

str