FragmentAnalyzer#

class neurodent.analysis.FragmentAnalyzer[source]#

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[source]#

Compute the root mean square of the signal.

static compute_logrms(
rec: ndarray,
precomputed_rms: ndarray | None = None,
**kwargs,
) → ndarray[source]#

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

static compute_ampvar(rec: ndarray, **kwargs) → ndarray[source]#

Compute the amplitude variance of the signal.

static compute_logampvar(
rec: ndarray,
precomputed_ampvar: ndarray | None = None,
**kwargs,
) → ndarray[source]#

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[source]#

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][source]#

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][source]#

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[source]#

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[source]#

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][source]#

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][source]#

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[source]#

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[source]#

Compute the complex coherency of the signal.

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

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][source]#

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][source]#

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][source]#

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

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

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[source]#

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)[source]#

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

static compute_lognspike(
rec: ndarray,
precomputed_nspike: ndarray | None = None,
**kwargs,
)[source]#

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][source]#

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