FragmentAnalyzer#

class neurodent.core.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.

Return type:

ndarray

Parameters:

rec (ndarray)

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

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

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • precomputed_rms (ndarray | None)

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

Compute the amplitude variance of the signal.

Return type:

ndarray

Parameters:

rec (ndarray)

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

Compute the log of the amplitude variance of the signal.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • precomputed_ampvar (ndarray | None)

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.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • multitaper (bool)

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.

Return type:

dict[str, ndarray]

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • bands (list[tuple[float, float]])

  • multitaper (bool)

  • precomputed_psd (tuple | None)

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.

Return type:

dict[str, ndarray]

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • bands (list[tuple[float, float]])

  • multitaper (bool)

  • precomputed_psd (tuple | None)

  • precomputed_psdband (dict | None)

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.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • band (tuple[float, float])

  • multitaper (bool)

  • precomputed_psd (tuple | None)

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.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • band (tuple[float, float])

  • multitaper (bool)

  • precomputed_psd (tuple | None)

  • precomputed_psdtotal (ndarray | None)

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.

Return type:

dict[str, ndarray]

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • bands (list[tuple[float, float]])

  • total_band (tuple[float, float])

  • multitaper (bool)

  • precomputed_psdband (dict | None)

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.

Return type:

dict[str, ndarray]

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • bands (list[tuple[float, float]])

  • total_band (tuple[float, float])

  • multitaper (bool)

  • precomputed_psdfrac (dict | None)

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.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • welch_bin_t (float)

  • notch_filter (bool)

  • band (tuple[float, float])

  • multitaper (bool)

  • precomputed_psd (tuple | None)

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.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • freq_res (float)

  • mode (Literal['cwt_morlet', 'multitaper'])

  • geomspace (bool)

  • cwt_n_cycles_max (float)

  • mt_bandwidth (float)

  • downsamp_q (int)

  • epsilon (float)

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

Compute the coherence of the signal.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • precomputed_coherency (dict | None)

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 (ndarray) – Input signal array

  • f_s (float) – Sampling frequency

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

  • **kwargs – Additional arguments passed to compute_cohere

Return type:

dict[str, ndarray]

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.

Return type:

dict[str, ndarray]

Parameters:

  • rec (ndarray)

  • f_s (float)

  • precomputed_coherency (dict | None)

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.

Return type:

dict[str, ndarray]

Parameters:

  • rec (ndarray)

  • f_s (float)

  • z_epsilon (float)

  • precomputed_imcoh (dict | None)

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

Compute the Pearson correlation coefficient of the signal.

Return type:

ndarray

Parameters:

  • rec (ndarray)

  • f_s (float)

  • lower_triag (bool)

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 (ndarray) – Input signal array

  • f_s (float) – Sampling frequency

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

  • **kwargs – Additional arguments passed to compute_pcorr

  • precomputed_pcorr (ndarray | None)

Return type:

ndarray

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

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

Parameters:

rec (ndarray)

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

Parameters:

  • rec (ndarray)

  • precomputed_nspike (ndarray | None)

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 (ndarray) – Single fragment data with shape (n_samples, n_channels)

  • f_s (int) – Sampling frequency

  • features (List[str]) – List of features to compute

  • kwargs (Dict[str, Any]) – Additional parameters for feature computation

Return type:

Dict[str, Any]

Returns:

Dictionary of computed features for this fragment