FragmentAnalyzer

class neurodent.core.FragmentAnalyzer

Bases: object

Static class for analyzing fragments of EEG data. All functions receive a (N x M) numpy array, where N is the number of samples, and M is the number of channels.

FEATURE_DEPENDENCIES = {'cohere': ['coherency'], 'imcoh': ['coherency'], 'logampvar': ['ampvar'], 'lognspike': ['nspike'], 'logpsdband': ['psdband'], 'logpsdfrac': ['psdfrac'], 'logpsdtotal': ['psdtotal'], 'logrms': ['rms'], 'psdband': ['psd'], 'psdfrac': ['psdband'], 'psdslope': ['psd'], 'psdtotal': ['psd'], 'zcohere': ['cohere'], 'zimcoh': ['imcoh'], 'zpcorr': ['pcorr']}

static compute_rms(rec: ndarray, **kwargs) → ndarray

Compute the root mean square of the signal.

static compute_logrms(rec: ndarray, precomputed_rms: ndarray | None = None, **kwargs) → ndarray

Compute the log of the root mean square of the signal.

static compute_ampvar(rec: ndarray, **kwargs) → ndarray

Compute the amplitude variance of the signal.

static compute_logampvar(rec: ndarray, precomputed_ampvar: ndarray | None = None, **kwargs) → ndarray

Compute the log of the amplitude variance of the signal.

static compute_psd(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, multitaper: bool = False, **kwargs) → ndarray

Compute the power spectral density of the signal.

static compute_psdband(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, bands: list[tuple[float, float]] = {'alpha': (8, 13), 'beta': (13, 25), 'delta': (1, 4), 'gamma': (25, 40), 'theta': (4, 8)}, multitaper: bool = False, precomputed_psd: tuple | None = None, **kwargs) → dict[str, ndarray]

Compute the power spectral density of the signal for each frequency band.

static compute_logpsdband(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, bands: list[tuple[float, float]] = {'alpha': (8, 13), 'beta': (13, 25), 'delta': (1, 4), 'gamma': (25, 40), 'theta': (4, 8)}, multitaper: bool = False, precomputed_psd: tuple | None = None, precomputed_psdband: dict | None = None, **kwargs) → dict[str, ndarray]

Compute the log of the power spectral density of the signal for each frequency band.

static compute_psdtotal(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, band: tuple[float, float] = (1, 40), multitaper: bool = False, precomputed_psd: tuple | None = None, **kwargs) → ndarray

Compute the total power spectral density of the signal.

static compute_logpsdtotal(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, band: tuple[float, float] = (1, 40), multitaper: bool = False, precomputed_psd: tuple | None = None, precomputed_psdtotal: ndarray | None = None, **kwargs) → ndarray

Compute the log of the total power spectral density of the signal.

static compute_psdfrac(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, bands: list[tuple[float, float]] = {'alpha': (8, 13), 'beta': (13, 25), 'delta': (1, 4), 'gamma': (25, 40), 'theta': (4, 8)}, total_band: tuple[float, float] = (1, 40), multitaper: bool = False, precomputed_psdband: dict | None = None, **kwargs) → dict[str, ndarray]

Compute the power spectral density of bands as a fraction of the total power.

static compute_logpsdfrac(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, bands: list[tuple[float, float]] = {'alpha': (8, 13), 'beta': (13, 25), 'delta': (1, 4), 'gamma': (25, 40), 'theta': (4, 8)}, total_band: tuple[float, float] = (1, 40), multitaper: bool = False, precomputed_psdfrac: dict | None = None, **kwargs) → dict[str, ndarray]

Compute the log of the power spectral density of bands as a fraction of the log total power.

static compute_psdslope(rec: ndarray, f_s: float, welch_bin_t: float = 1, notch_filter: bool = True, band: tuple[float, float] = (1, 40), multitaper: bool = False, precomputed_psd: tuple | None = None, **kwargs) → ndarray

Compute the slope of the power spectral density of the signal on a log-log scale.

static compute_coherency(rec: ndarray, f_s: float, freq_res: float = 1, mode: Literal['cwt_morlet', 'multitaper'] = 'multitaper', geomspace: bool = False, cwt_n_cycles_max: float = 7.0, mt_bandwidth: float = 4.0, downsamp_q: int = 4, epsilon: float = 0.01, **kwargs) → ndarray

Compute the complex coherency of the signal.

static compute_cohere(rec: ndarray, f_s: float, precomputed_coherency: dict | None = None, **kwargs) → ndarray

Compute the coherence of the signal.

static compute_zcohere(rec: ndarray, f_s: float, z_epsilon: float = 1e-06, precomputed_cohere=None, **kwargs) → dict[str, ndarray]

Compute the Fisher z-transformed coherence of the signal.

Parameters:

• rec – Input signal array

• f_s – Sampling frequency

• z_epsilon – Small value to prevent arctanh(1) = inf. Values are clipped to [-1+z_epsilon, 1-z_epsilon]

• **kwargs – Additional arguments passed to compute_cohere

static compute_imcoh(rec: ndarray, f_s: float, precomputed_coherency: dict | None = None, **kwargs) → dict[str, ndarray]

Compute the imaginary coherence of the signal.

static compute_zimcoh(rec: ndarray, f_s: float, z_epsilon: float = 1e-06, precomputed_imcoh: dict | None = None, **kwargs) → dict[str, ndarray]

Compute the Fisher z-transformed imaginary coherence of the signal.

static compute_pcorr(rec: ndarray, f_s: float, lower_triag: bool = False, **kwargs) → ndarray

Compute the Pearson correlation coefficient of the signal.

static compute_zpcorr(rec: ndarray, f_s: float, z_epsilon: float = 1e-06, precomputed_pcorr: ndarray | None = None, **kwargs) → ndarray

Compute the Fisher z-transformed Pearson correlation coefficient of the signal.

Parameters:

• rec – Input signal array

• f_s – Sampling frequency

• z_epsilon – Small value to prevent arctanh(1) = inf. Values are clipped to [-1+z_epsilon, 1-z_epsilon]

• **kwargs – Additional arguments passed to compute_pcorr

static compute_nspike(rec: ndarray, **kwargs)

Returns NaN array as placeholder. Compute and load in spikes with SpikeAnalysisResult

static compute_lognspike(rec: ndarray, precomputed_nspike: ndarray | None = None, **kwargs)

Returns log-transformed NaN array as placeholder. Compute and load in spikes with SpikeAnalysisResult

static process_fragment_with_dependencies(fragment_data: ndarray, f_s: int, features: List[str], kwargs: Dict[str, Any]) → Dict[str, Any]

Process a single fragment with efficient dependency resolution.

This is the enhanced replacement for _process_fragment_features_dask that automatically resolves feature dependencies and reuses intermediate calculations to avoid redundant computations (e.g., computing PSD once for multiple dependent features).

Parameters:

• fragment_data – Single fragment data with shape (n_samples, n_channels)

• f_s – Sampling frequency

• features – List of features to compute

• kwargs – Additional parameters for feature computation

Returns:

Dictionary of computed features for this fragment