nnsa package

Subpackages

Submodules

nnsa.config module

This module contains constants and defaults related to the entire nnsa project.

Module contents

Classes:

`AmplitudeEeg`([eng])	Class for emulating an amplitude-integrated (aEEG) signal from single channel continuous EEG.
`AmplitudeEegResult`(aeeg, fs, ...[, label, ...])	High-level interface for processing the results of aEEG computation as created by nnsa.AmplitudeEeg().
`Annotation`(onset, duration, text)	Universal annotation object.
`AnnotationSet`([annotations, label])	Holds a collection of annotations (e.g.
`BaseDataset`(*args[, label])	Base class with common functionalities for Datasets.
`BrainAgeResult`(brain_age, algorithm_parameters)	Result class for (ensemble of) brain age estimates.
`BrainAgeSinc`([which, detect_novelties])	Deep shared inception ensemble model for brain age estimation.
`BurstDetection`(**kwargs)	Class for burst detection.
`BurstDetectionResult`(bursts, ibis, ...[, ...])	High-level interface for processing the results of burst detection analysis as created by nnsa.BurstDetection().
`Butterworth`([tf, sos, fs])	Create a Butterworth filter.
`ClassWithParameters`(**kwargs)	Base class for a class that has parameters.
`CleanDetectorCnn`([multi_channel, chunk_length])	Interface for applying clean EEG detection using a trained Convolutional Neural Network.
`CoherenceGraph`(**kwargs)	Class for computing a connectivity graph of multi-channel EEG based on coherence.
`CoherenceGraphResult`(mean_coh, im_coh, ...)	High-level interface for processing the results of a coherence analysis as created by nnsa.CoherenceGraph().
`ConfusionMatrix`(y_true, y_pred[, class_mapping])	High-level interface for manipulating the results of the confusion matrix.
`DynamicCoupling`(**kwargs)	Class for computing the dynamic coupling between two signals.
`DynamicCouplingResult`(coupling, ...[, ...])	High-level interface for processing dynamic coupling as computed by nnsa.DynamicCoupling().
`EdfReader`(filepath)	Child of edfreadpy.EdfReader, which includes interfaces to load data into nnsa containers.
`EegDataset`(*args[, label])	High-level interface for processing (multichannel) EEG for neonatal signal analysis.
`Envelope`(**kwargs)	Class for computing the envelope of a real valued signal.
`EnvelopeResult`(envelope, ...[, ...])	High-level interface for processing the results of envelope computation as created by nnsa.Envelope().
`FeatureSetResult`(features, feature_labels, ...)	High-level interface for processing features sets with multiple features, channels, segments.
`FilterBase`([tf, sos, fs])	Base class for a filter object with common utilities.
`LineLength`(**kwargs)	Class for computing the Line Length (LL) feature.
`LineLengthResult`(line_length, ...[, ...])	High-level interface for processing the results of line length computation as created by nnsa.LineLength().
`MovingAverage`([tf, sos, fs])	Create a moving average (smoothening) filter.
`MultiScaleEntropy`(**kwargs)	Class for performing a multi-scale entropy analysis.
`MultiScaleEntropyResult`(mse, ...[, ...])	High-level interface for processing the results of multi-scale entropy analysis as created by nnsa.MultiScaleEntropy().
`MultifractalAnalysis`([eng])	Class for performing a multifractal analysis.
`MultifractalAnalysisResult`(cp, hmin, hmax, ...)	High-level interface for processing the results of multifractal analysis as created by nnsa.MultifractalAnalysis().
`NotchIIR`([tf, sos, fs])	Create an IIR notch (bandstop) filter.
`Parameters`	Class for collecting parameters.
`PowerAnalysis`(**kwargs)	Class for performing a power analysis of EEG signal.
`PowerAnalysisResult`(power, algorithm_parameters)	High-level interface for processing the results of a power analysis as created by nnsa.PowerAnalysis().
`Psd`(**kwargs)	Class for estimating the power spectral density of a signal using Welch's method.
`PsdResult`(power_density, df, ...[, ...])	High-level interface for power spectral density results as created by nnsa.Psd().psd().
`RemezFIR`([tf, sos, fs])	Design a minimax optimal filter using the Remez exchange algorithm.
`ResultBase`([algorithm_parameters, ...])	Base class for high-level interface for the results of feature extraction algortihms.
`SideObsp`([order, regularizer, gammas])	Class implementing SIgnal DEcomposition based on Oblique Subspace Projections.
`SignalStats`(**kwargs)	Class for computing a set of statistical parameters of a signal.
`SignalStatsResult`(stats, algorithm_parameters)	High-level interface for processing signal statistics as computed by nnsa.SignalStats().
`SleepStages`(**kwargs)	Sleep stage classification from text annotations.
`SleepStagesCnn`([num_classes, detect_novelties])	Sleep stage classification using a Convolutional Neural Network.
`SleepStagesCnnResult`(probabilities, ...[, ...])	High-level interface for manipulating sleep stage probabilities as predicted by nnsa.SleepStagesCnn().sleep_stages_cnn().
`SleepStagesResult`(annotation_set[, ...])	High-level interface for manipulating sleep stage annotations as created by nnsa.SleepStages().sleep_stages().
`SleepStagesRobust`(**kwargs)	Robust algorthmic pipleine for 4-class sleep staging for neoantes > 36 weeks PMA.
`SleepStagesRobustResult`(df, algorithm_parameters)	High-level interface for manipulating sleep stage predictions as predicted by nnsa.SleepStagesRobust().process().
`TimeSeries`(signal, fs[, label, unit, info, ...])	High-level interface for processing of any type of time series.
`WaveletCoherence`(**kwargs)	Class for computing wavelet coherence.
`WaveletCoherenceResult`(P, A, freqs, fs, wavelet)	High-level interface for processing wavelet coherence as computed by nnsa.WaveletCoherence().
`WaveletResult`(P, A, freqs, fs, wavelet[, ...])	High-level interface for processing wavelet maps.
`WinFIR`([tf, sos, fs])	FIR filter design using the window method.

Functions:

`assert_equal`(actual, desired)	Tests if two objects are equal (like numpy's assert_equal).
`clamp`(x[, neg_thres, pos_thres])	Apply a clampling function to the array x.
`combine_channels`(x[, preferred, min_nan_len])	Combines multiple channels of a signal into one, by cutting the signals into pieces and pasting them together in an optimal way.
`detect_anomalous_channels`(x, window[, fs, ...])	Simple method to detect anomalous high-frequency/amplitude channels by setting a threshold on the line length.
`format_time_axis`([time_scale, relative, ax])	Format x-axes (time in seconds) as h:mm:ss, or change to hours or minutes.
`get_filepaths`(directory, pattern[, ...])	Return a list with paths of files living in directory and with pattern in the filename.
`get_output_dir`(output_root[, create_unique])	Return a directory path for saving the output that a script generates.
`list_attrs`(obj)	Returns a list with names of public attributes of an object.
`list_methods`(obj)	Returns a list with names of public (callable) methods of an object.
`moving_envelope`(x, n[, fs, axis, max_nan_frac])
`moving_line_length`(x, n[, fs, axis, ...])
`moving_mad`(x, n[, axis, maintain_nan, ...])	Compute the moving (centered) median absolute deviation (MAD) of x, with window size n.
`moving_max`(x, n[, axis, maintain_nan, ...])	Compute the moving (centered) max of x, with window size n.
`moving_mean`(x, n[, axis, maintain_nan, ...])	Compute the moving (centered) mean of x, with window size n.
`moving_median`(x, n[, axis, maintain_nan, ...])	Compute the moving (centered) median of x, with window size n.
`moving_std`(x, n[, axis, maintain_nan, ...])	Compute the moving (centered) standard deviation of x, with window size n.
`print_exception_info`(e)	Print error message and traceback of error e which is handled/caught.
`print_object_summary`(obj)	Print object summary: class name, public attributes, public methods.
`read_annotation_set_KL`(filepath)	Helper function to read annotations from an EDF+ file as annotated by KL (using **).
`read_annotation_set_preterm`(df, filename[, ...])	Helper function to read sleep labels from a dataframe for the preterm dataset.
`read_result_from_file`(filepath[, verbose])	Read result from file as created by the save_to_file() method of the ResultBase class.
`stripboxplot`([x, y, hue, style, data, ...])	Plot a stripplot and overlay the box of a boxplot.

class nnsa.AmplitudeEeg(eng=None, **kwargs)[source]

Bases: ClassWithParameters

Class for emulating an amplitude-integrated (aEEG) signal from single channel continuous EEG.

Multiple different algorithms can be used, see self.default_parameters().

Main method: amplitude_eeg() or process().

Parameters:

eng (matlab.engine.matlabengine.MatlabEngine, optional) – Deprecated. MATLAB engine to use for calling MATLAB functions. The required paths must already have been initialized (see self.eng()). Only relevant if an algorithm that requires MATLAB is used. If None, a new MATLAB engine will be started (which takes some time). Defaults to None.
**kwargs (optional) – see nnsa.ClassWithParameters.

Examples

>>> fs = 256
>>> t = 1/fs*np.arange(100000)
>>> x = 4 * np.sin(t)
>>> AEEG = AmplitudeEeg(method='neat', dtype=np.float32)
>>> print(type(AEEG.parameters).__name__)
Parameters

>>> result = AEEG.wct(x, fs=fs, verbose=0)
>>> print(type(result).__name__)
AmplitudeEegResult

>>> result.aeeg
array([0.21567792, 0.21360768, 0.21153691, ..., 0.08440158, 0.0844016 ,
       0.08440162], dtype=float32)

Methods:

`amplitude_eeg`(eeg, fs[, label, verbose])	Compute the aEEG signal.
`default_parameters`()	Return the default parameters.
`process`(args, *kwargs)	Shortcut to main function.

Attributes:

eng

Return a MATLAB engine.

amplitude_eeg(eeg, fs, label='aEEG', verbose=0)[source]

Compute the aEEG signal.

Samples that are nan in the input will be nan in the output.

Parameters:

eeg (np.ndarray) – single channel EEG signal.
fs (float) – sample frequency of eeg in Hz.
label (str, optional) – label for the aEEG signal. Defaults to ‘aEEG’.
verbose (int, optional) – verbose level. Defaults to 0.

Returns:

(nnsa.AmplitudeEegResult) – object containing the aEEG.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

property eng

Return a MATLAB engine.

Returns:: (matlab.engine.matlabengine.MatlabEngine) – MATLAB engine.

process(*args, **kwargs)[source]: Shortcut to main function.

class nnsa.AmplitudeEegResult(aeeg, fs, algorithm_parameters, label=None, data_info=None, time_offset=0)[source]

Bases: ResultBase

High-level interface for processing the results of aEEG computation as created by nnsa.AmplitudeEeg().

Parameters:

aeeg (np.ndarray) – the aEEG signal with sample frequency fs in semilog-scale (amplitudes 0-10 linear and amplitude > 10 logarithmic).
fs (float) – sample frequency of aeeg.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
label (str, optional) – label of the aEEG signal. Defaults to ‘aEEG’.
data_info (str, optional) – see ResultBase.

Methods:

`extract_margins`([segment_length, fs, ...])	Extract upper and lower margin amplitudes by extracting the low and high percentiles in (overlapping) segments.
`plot`([plot_margins, ax])	Plot the envelope.
`to_time_series`([which, unit])	Create a TimeSeries object from the aEEG data.

Attributes:

num_segments

Return the number of segments (number of samples).

extract_margins(segment_length=None, fs=None, q_lower=10, q_upper=90, max_nan=0.5, correct_delay=False)[source]

Extract upper and lower margin amplitudes by extracting the low and high percentiles in (overlapping) segments.

Parameters:

segment_length (float, optional) – length of the segments/moving window to compute the margins in (in seconds). If None, a default value is taken based on the aEEG conversion method used. Defaults to None.
fs (float) – desired sample frequency of the returned margins in Hz. If None, a default value is taken based on the aEEG conversion method used. Defaults to None.
q_lower (float, optional) – percentile for the lower margin (0-100). Defaults to 10.
q_upper (float, optional) – percentile for the upper margin (0-100). Defaults to 90.
max_nan (float, optional) – maximum fraction (0-1) of nan values in a segment. If the fraction of nan samples exceeds max_nan, the margins are nan for that segment. Defaults to 0.5.
correct_delay (bool) – whether to correct for delay by padding.

Returns:

aeeg_lower (np.ndarray) – lower margin amplitudes with frequency fs_margins.
aeeg_upper (np.ndarray) – upper margin amplitudes with frequency fs_margins.
fs (float) – sample frequency of aeeg_lower and aeeg_upper in Hz.

property num_segments

Return the number of segments (number of samples).

Returns:: (int) – number of segments/samples.

plot(plot_margins=True, ax=None, **kwargs)[source]

Plot the envelope.

Parameters:

plot_margins (bool, optional) – if True, plots the upper and lower margin amplitudes as dark fat lines. If False, does not plot the margins. Defaults to True.
ax (plt.axes, optional) – axes to plot in. If None, will plot in a new figure. Defaults to None.
**kwargs (optional) – keyword arguments for nnsa.TimeSeries.plot().

to_time_series(which='bandwidth', unit='uV', **kwargs)[source]

Create a TimeSeries object from the aEEG data.

Parameters:

which (str, optional) –
which signal to export to a time series. Choose from:

’raw’ ‘LMA’ (lower margin amplitude), ‘UMA’ (upper margin amplitude), ‘bandwidth’ (UMA - LMA).

Defaults to ‘bandwidth’.
unit (str, optional) – unit of the aEEG data. Defaults to ‘uV’.
**kwargs (optional) – optional keyword arguments for the TimeSeries object.

Returns:

ts (nnsa.TimeSeries) – TimeSeries containing the aEEG data.

class nnsa.Annotation(onset, duration, text)[source]

Bases: object

Universal annotation object.

An Annotation object has four properties:: Annotation.onset (int, float): the onset time of the annotation. Annotation.duration (int, float): the duration of the annotation (same unit as onset). Annotation.text (str): the annotation text. Annotation.offset (int, float): the end time of the annotation (onset + duration).

Parameters:

onset (int, float) – the onset time of the annotation.
duration (int, float) – the duration of the annotation (same unit as onset).
text (str) – the annotation text.

Attributes:

`duration`	Return the duration of the annotation in seconds.
`offset`	Return the end time (onset + duration).
`onset`	Return the onset of the annotation in seconds.
`text`	Return the annotation text.

property duration

Return the duration of the annotation in seconds.

Returns:: (float) – the duration of the annotation in seconds.

property offset

Return the end time (onset + duration).

Clips duration such that it cannot be negative. (sometimes, duration could be set to -1 if there is no duration associated with the Annotation, in this case, make the offset the same as the onset by making the duration 0).

property onset

Return the onset of the annotation in seconds.

Returns:: (float) – the onset of the annotation in seconds.

property text

Return the annotation text.

Returns:: (str) – the annotation text.

class nnsa.AnnotationSet(annotations=None, label='annotations')[source]

Bases: object

Holds a collection of annotations (e.g. all annotations from one file/investigator).

Parameters:

annotations (iterable, optional) – iterable collection with Annotation objects (e.g. list, generator).
label (str, optional) – a label for this AnnotationSet (e.g. code/name of the observer).

Methods:

`add_postfix`(postfix[, inplace])	Add postfix to all annotations.
`add_prefix`(prefix[, inplace])	Add prefix to all annotations.
`append`(annotation[, inplace])	Safely append an annotation to the annotation set.
`clip_overlaps`([how, inplace])	Clip annotations if they overlap.
`compute_nan_durations`([total_duration, inplace])	Compute duration from inter-annotation interval for annotations with duration np.nan.
`count_neighbors`(text)	Count the number of times a certain text is a neighbour (in time) of a specified text.
`durations`()	Return the durations of all annotations in the set.
`end_time`()	Return the time at which the last annotation ends.
`extend`(annotations[, inplace])	Safely extend the list with annotation with another annotations from an iterable collection.
`extract_epoch`([begin, end])	Extract epoch specified by begin and end.
`fill_unlabeled_periods`([label_to_insert, ...])	Insert Annotation objects into the AnnotationSet to make the annotations continuous.
`filter`(patterns[, how, case_sensitive, inplace])	Filter the annotation by looking for matching patterns in the texts.
`filter_fun`(fun[, pass_text, inplace])	Filter the annotation keeping only annotations that yield True after evaluating fun on them.
`from_df`(df[, onset, duration, text])	Convert a DataFrame to an AnnotationSet.
`from_lists`(onsets, durations, texts, **kwargs)	Convert a DataFrame to an AnnotationSet.
`from_mask`(mask[, time, label_mapping, label])	Convert a mask with arbitrary number of classes to an annotation set.
`get_occurrence_count`()	Return the unqiue annotation texts and their number of occurences (sorted).
`get_overlapping_annotations`([begin, end, ...])	Return a sub set of the annotations with annotations that (partially) overlap with the specified time interval.
`get_text_times`(texts)	Return the start and end times of a specific annotation text.
`interpolate_artefacts`([artefact_texts, ...])	Interpolate artefact segments.
`merge_successive_texts`([inplace])	Merge successive annotations if there texts are the same.
`onsets`()	Return the onsets of all annotations in the set.
`print_all_annotations`()	Print all annotations at once using a pandas dataframe.
`print_annotation`(index[, print_header])	Print the onset, text and duration of an annotation specified by the index.
`remove`(patterns[, case_sensitive, inplace])	Filter the annotation by looking for matching patterns in the texts.
`replace_in_range`(replace_with, begin, end[, ...])	Replace the texts of annotations that overlap with the given range with the specified text.
`replace_text_by_index`(indices, new_text[, ...])	Replace the text of annotations with index in indices with new_text.
`replace_texts`(texts, new_text[, inplace])	Replace the text of all annotations with of one texts with new_text.
`shade_axis`([begin, end, labels_mapping, ...])	Shade the areas in the current axis corresponding to the annotations.
`sleep_stages`(**kwargs)	Convert the annotations to sleep stages and return the sleep stages as a SleepStagesResult object.
`sort`([inplace])	Sort the annotations based on onset.
`standardize_annotations`([inplace])	Convert the annotations to standard labels and return a new AnnotationSet with the standard labels.
`start_time`()	Return the onset time of the first annotation (in seconds).
`subtract_offset`(offset[, inplace])	Alias for self.subtract_time() for backwards compatibility reasons.
`subtract_time`(dt[, inplace])	Subtract a constant amount of time from the onset of each annotation in the annotation set.
`texts`()	Return the texts of all annotations in the set.
`to_dataframe`()	Return a DataFrame with the annotations.
`to_df`()	Alias for self.to_dataframe().

Attributes:

annotations

Return the list with annotations.

add_postfix(postfix, inplace=False)[source]

Add postfix to all annotations.

Parameters:

prefix (str) – string to add after each Annotation’s text.
inplace (bool, optional) – if True, modify the annotation in place. If False, make a copy and modify the copy.

Returns:

annotation_set (edfreadpy.AnnotationSet) – new AnnotationSet object with the modified annotations (if inplace is False).

add_prefix(prefix, inplace=False)[source]

Add prefix to all annotations.

Parameters:

prefix (str) – string to add in front of each Annotation’s text.
inplace (bool, optional) – if True, modify the annotation in place. If False, make a copy and modify the copy.

Returns:

annotation_set (edfreadpy.AnnotationSet) – new AnnotationSet object with the modified annotations (if inplace is False).

property annotations

Return the list with annotations.

Returns:: (list) – list with edfreadpy.Annotation objects.

append(annotation, inplace=False)[source]

Safely append an annotation to the annotation set.

Parameters:

annotation (edfreadpy.Annotation) – Annotation object.
inplace (bool, optional) – if True, appends the annotation in place. If False, a new AnnotationSet object with the appended annotation is returned. Defaults to False.

Returns:

annotation_set (edfreadpy.AnnotationSet) – new AnnotationSet object with the appended annotation (if inplace is False).

clip_overlaps(how='first', inplace=False)[source]

Clip annotations if they overlap.

Parameters:

how (str, optional) – specifies which annotation is clipped. Choose from ‘first’, ‘second’. For two overlapping annotations, either the first one can be shortered (‘first’) by decreasing its duration or the second one (‘second’) can be shortened by increasing its onset.
inplace (bool, optional) – if True, clips the annotations in place. If False, a new AnnotationSet object with the clipped annotations is returned. Defaults to False.

Returns:

as_out (nnsa.AnnotationSet) – AnnotationSet object with clipped annotations (returned only if inplace is False).

compute_nan_durations(total_duration=None, inplace=False)[source]

Compute duration from inter-annotation interval for annotations with duration np.nan.

Parameters:

total_duration (float) – total duration of the annotated recording (needed to determine the duration of the last annotation).
inplace (bool, optional) – apply inplcace (True) or return a new object (False). Defaults to False.

Returns:

annotations (nnsa.AnnotationSet) – new AnnotationSet with Annotations (if inplace is False).

count_neighbors(text)[source]

Count the number of times a certain text is a neighbour (in time) of a specified text.

Assumes that the annotations are sorted by onset.

Parameters:: text (str) – annotation text of which the neighbours should be counted.
Returns:: (pd.DataFrame) – dataframe with counts where the index represents the left neighbour and the column represents the right neighbour

durations()[source]

Return the durations of all annotations in the set.

Returns:: (np.ndarray) – the durations (in seconds) of all annotations in the set.

end_time()[source]

Return the time at which the last annotation ends.

Returns:: (float) – the time (in seconds) at which the last annotation ends.

extend(annotations, inplace=False)[source]

Safely extend the list with annotation with another annotations from an iterable collection.

Parameters:

annotations (collection) – iterable collection with Annotation objects (e.g. list, generator).
inplace (bool, optional) – if True, extends the annotations in place. If False, a new AnnotationSet object extended with the annotations is returned. Defaults to False.

Returns:

annotation_set (edfreadpy.AnnotationSet) – new AnnotationSet object with the appended annotations (if inplace is False).

extract_epoch(begin=None, end=None)[source]

Extract epoch specified by begin and end.

Parameters:

begin (float) – begin time of epoch (seconds).
end (float) – end time of epoch (seconds).

Returns:

(nnsa.AnnotationSet) – subset of the current AnnotationSet.

fill_unlabeled_periods(label_to_insert=None, begin=None, end=None, inplace=False)[source]

Insert Annotation objects into the AnnotationSet to make the annotations continuous.

An Annotation with label label_to_insert is inserted between two succesive annotations if they are not continuous, i.e. the start of the second annotation does not coincide with the end of the first one.

Parameters:

label_to_insert (str, optional) – the text that is used for the Annotations that are inserted in the unlabeled periods. If None, NO_LABEL is used. Defaults to None.
begin (flaot, optional) – begin time of period to fill. Cannot be greater than the first onset. If None, the first annotation in the set marks the beginning. Defaults to None.
end (float, optional) – end time of period to fill. Cannot be smaller than the end time of the last annotation. If None, the last annotation marks the end. Defaults to None.
inplace (bool, optional) – if True, fills the annotations in place. If False, a new AnnotationSet object with the filled annotations is returned. Defaults to False.

Returns:

annotation_set (nnsa.AnnotationSet) – new AnnotationSet object that has continuous labels (if inplace is False).

filter(patterns, how='keep', case_sensitive=True, inplace=False)[source]

Filter the annotation by looking for matching patterns in the texts.

Parameters:

patterns (str or list) – pattern(s) to find matches for.
how (str, optional) –
whether to keep or remove the annotations that match the patterns. Options:
- ’keep’
- ’remove’
Defaults to ‘keep’.
case_sensitive (bool, optional) – whether the pattern matching should be case sensitive. Defaults to True.
inplace (bool, optional) – if True, filters the annotations in place. If False, a new AnnotationSet object with the filtered annotations is returned. Defaults to False.

Returns:

as_out (edfreadpy.AnnotationSet) – new AnnotationSet with a subset of the annotations.

Examples

>>> annot_set = AnnotationSet()
>>> annot_set.append(Annotation(onset=10.25, duration=100, text='QS'), inplace=True)
>>> annot_set.append(Annotation(110.25, 200, 'AS'), inplace=True)
>>> annot_set.append(Annotation(500, 300, 'QS'), inplace=True)
>>> annot_set.append(Annotation(800, 10, 'AS'), inplace=True)
>>> annot_set.print_all_annotations()
    onset  duration text
0   10.25       100   QS
1  110.25       200   AS
2  500.00       300   QS
3  800.00        10   AS
>>> as_new = annot_set.filter('QS', how='keep')
>>> as_new.print_all_annotations()
    onset  duration text
0   10.25     100.0   QS
1  500.00     300.0   QS
>>> as_new = annot_set.filter('QS', how='remove')
>>> as_new.print_all_annotations()
    onset  duration text
0  110.25       200   AS
1  800.00        10   AS

filter_fun(fun, pass_text=True, inplace=False)[source]

Filter the annotation keeping only annotations that yield True after evaluating fun on them.

Parameters:

fun (function) – function that takes in a string or Annotation object (depending on pass_text) and returns True or False. Annotations that yield False are removed.
pass_text (bool) – if Ture, the annotation text (str) is passed to fun, otherwise the Annotation object is passed.
inplace (bool, optional) – if True, filters the annotations in place. If False, a new AnnotationSet object with the filtered annotations is returned. Defaults to False.

Returns:

as_out (edfreadpy.AnnotationSet) – new AnnotationSet with annotations that match the specified pattern (only retruned if inplace is False).

from_df(df, onset='onset', duration='duration', text='text', **kwargs)[source]

Convert a DataFrame to an AnnotationSet.

Parameters:

df (pd.DataFrame) – a DataFrame.
onset (str, optional) – column name in df that corresponds to the onset times (in seconds).
duration – (str, optional): column name in df that corresponds to the duration times (in seconds).
text – (str, optional): column name in df that corresponds to the annotation text.
**kwargs (optional) – optional keyword arguments for AnnotationSet().

Returns:

annotation_set (AnnotationSet) – AnnotationSet.

from_lists(onsets, durations, texts, **kwargs)[source]

Convert a DataFrame to an AnnotationSet.

Parameters:

onsets (list, np.ndarray) – list with the onset times (in seconds).
durations (list, np.ndarray) – list with the duration times (in seconds).
texts (list, np.ndarray) – list with the annotation texts.
**kwargs (optional) – optional keyword arguments for AnnotationSet().

Returns:

annotation_set (AnnotationSet) – AnnotationSet.

from_mask(mask, time=None, label_mapping=None, label=None)[source]

Convert a mask with arbitrary number of classes to an annotation set.

Parameters:

mask (np.ndarray) – 1D mask array containing discrete values.
time (np.ndarray) – time array corresponding to starttimes of mask. If None, the onset and durations in the returned AnnotationSet are in samples.
label_mapping (dict, optional) – dictionary mapping a value in the mask to a label (str). Those labels will become the texts in the annotation set. If None, the mask values will be the labels.
label (str, optional) – label for the annotation set. Defaults to None.

Returns:

annotation_set (AnnotationSet) – annotation set with onsets, durations and texts.

Examples

>>> time = np.arange(10)
>>> mask = [0, 0, 1, 1, 1, 0, 0, 2, 2, 2]
>>> AnnotationSet().from_mask(time, mask)
AnnotationSet with label: "None" and annotations:
   onset  duration text
0      0         2    0
1      2         3    1
2      5         2    0
3      7         3    2
>>> AnnotationSet().from_mask(time, mask, label_mapping={0: 'null', 1: 'one', 2: 'two'})
AnnotationSet with label: "None" and annotations:
   onset  duration  text
0      0         2  null
1      2         3   one
2      5         2  null
3      7         3   two

get_occurrence_count()[source]

Return the unqiue annotation texts and their number of occurences (sorted).

Returns:: sorted_text_count (dict) – dictionary that maps unique annotations in the annotation set to the frequency/counts of that annotation in the set.

get_overlapping_annotations(begin=-inf, end=inf, clip=False, start=None)[source]

Return a sub set of the annotations with annotations that (partially) overlap with the specified time interval.

Parameters:

begin (float) – begin time in seconds of the time interval.
end (float) – end time in seconds of the time interval.
clip (bool) – if True, clips the first and last annotations to the begin and end times. If False, keeps original lengths.
start (flaot) – overrides begin (for backwards compatibility).

Returns:

(nnsa.AnnotationSet) – subset of the current AnnotationSet.

get_text_times(texts)[source]

Return the start and end times of a specific annotation text.

Parameters:

texts (str or list) – annotation text(s) to get the start and end times for.

Returns:

begin_times (np.ndarray) – begin (onset) times.
end_times (np.ndarray) – end times.

interpolate_artefacts(artefact_texts=None, max_duration=180, inplace=False)[source]

Interpolate artefact segments.

Replaces an artefact annotation by the surrounding class labels if the artefact annotation is surrounded by equal class labels and if its duration is less than the specified max_duration.

Assumes the annotations are sorted by onset. Ultimately merges equal succesive labels.

Parameters:

artefact_texts (str or list, optional) – annotation texts indicating artefact segments, can be a single string or a list of strings when multiple texts indicate artefacts. If None, a default set of artefact-related annotation texts are used. Defaults to None.
max_duration (float, optional) – maximum duration (in seconds) that the artefact annotation can have to get replaced. Defaults to 180.
inplace (bool, optional) – if True, replaces the annotations in place. If False, a new AnnotationSet object with the replaced annotations is returned. Defaults to False.

Returns:

standardized_annotations (nnsa.AnnotationSet) – new AnnotationSet object with potentially less artefact annotations/labels (if inplace is False).

merge_successive_texts(inplace=False)[source]

Merge successive annotations if there texts are the same.

Parameters:: inplace (bool, optional) – if True, merges the annotations in place. If False, a new AnnotationSet object with the merged annotations is returned. Defaults to False.
Returns:: annotation_set (nnsa.AnnotationSet) – new AnnotationSet object with successive labels merged (if inplace is False).

onsets()[source]

Return the onsets of all annotations in the set.

Returns:: (np.ndarray) – the onsets (in seconds) of all annotations in the set.

print_all_annotations()[source]: Print all annotations at once using a pandas dataframe.

print_annotation(index, print_header=True)[source]

Print the onset, text and duration of an annotation specified by the index.

Parameters:

index (int) – the index of the anntation in self.annotations list to print.
print_header (bool, optional) – if True, a header will be printed, if False, not. Defaults to True.

remove(patterns, case_sensitive=True, inplace=False)[source]

Filter the annotation by looking for matching patterns in the texts.

Parameters:

patterns (str or list) – pattern(s) to find matches for.
inplace (bool, optional) – if True, removes the annotations in place. If False, a new AnnotationSet object without the removed annotations is returned. Defaults to False.

Returns:

as_out (edfreadpy.AnnotationSet) – new AnnotationSet without annotations that match the specified pattern (only retruned if inplace is False).

Examples

>>> annot_set = AnnotationSet()
>>> annot_set.append(Annotation(onset=10.25, duration=100, text='QS'), inplace=True)
>>> annot_set.append(Annotation(110.25, 200, 'AS'), inplace=True)
>>> annot_set.append(Annotation(500, 300, 'QS'), inplace=True)
>>> annot_set.append(Annotation(800, 10, 'AS'), inplace=True)
>>> annot_set.print_all_annotations()
    onset  duration text
0   10.25       100   QS
1  110.25       200   AS
2  500.00       300   QS
3  800.00        10   AS
>>> as_new = annot_set.remove('QS')
>>> as_new.print_all_annotations()
    onset  duration text
0  110.25       200   AS
1  800.00        10   AS

replace_in_range(replace_with, begin, end, resize=False, inplace=False)[source]

Replace the texts of annotations that overlap with the given range with the specified text.

Note: by default, replaces the entire annotation in case of any overlap, see resize parameter.

Parameters:

replace_with (str) – new text for annotations overlapping with the specified range.
begin (float) – begin of range.
end (float) – end of range.
resize (bool, optional) – if True, the annotations that partially overlap are resized so that they fit. If False, all annotations with any overlap are removed completely.
inplace (bool, optional) – if True, replace the annotations in place. If False, a new AnnotationSet object with the new annotations is returned. Defaults to False.

Returns:

as_out (nnsa.AnnotationSet) – new object where all annotations that overlap with the range are replaced seconds (if inplace is False).

Examples

>>> ans = AnnotationSet([Annotation(0, 20, '1'), Annotation(20, 50, '2'), Annotation(100, 110, '3')])
>>> ans.replace_in_range('NaN', begin=0, end=10)
AnnotationSet with label: "annotations" and annotations:
   onset  duration text
0      0        20  NaN
1     20        50    2
2    100       110    3
>>> ans.replace_in_range('NaN', begin=0, end=10, resize=True)
AnnotationSet with label: "annotations" and annotations:
   onset  duration text
0      0         0    1
1      0        10  NaN
2     10        10    1
3     20        50    2
4    100       110    3
>>> ans.replace_in_range('NaN', begin=0, end=30)
AnnotationSet with label: "annotations" and annotations:
   onset  duration text
0      0        20  NaN
1     20        50  NaN
2    100       110    3
>>> ans.replace_in_range('NaN', begin=0, end=30, resize=True)
AnnotationSet with label: "annotations" and annotations:
   onset  duration text
0      0         0    1
1      0        30  NaN
2     30        50    2
3    100       110    3
>>> ans.replace_in_range('NaN', begin=60, end=70)
AnnotationSet with label: "annotations" and annotations:
   onset  duration text
0      0        20    1
1     20        50  NaN
2    100       110    3
>>> ans.replace_in_range('NaN', begin=60, end=70, resize=True)
AnnotationSet with label: "annotations" and annotations:
   onset  duration text
0      0        20    1
1     20        40    2
2     60        10  NaN
3     70         0    2
4    100       110    3

replace_text_by_index(indices, new_text, inplace=False)[source]

Replace the text of annotations with index in indices with new_text.

Parameters:

indices (list) – list with sorted indices that correspond to the to be replaced annotations in the annotations list.
new_text (str) – new text for the annotations that are to be replaced.
inplace (bool, optional) – if True, replaces the annotation texts in place. If False, a new AnnotationSet object with the replaced annotation texts is returned. Defaults to False.

Returns:

replaced_annotations (nnsa.AnnotationSet) – new AnnotationSet object with the replaced annotation texts (if inplace is False).

replace_texts(texts, new_text, inplace=False)[source]

Replace the text of all annotations with of one texts with new_text.

Parameters:

texts (str or list) – text of annotation to replace or a list of text to replace.
new_text (str) – new text for the annotations that are to be replaced.
inplace (bool, optional) – if True, replaces the annotation texts in place. If False, a new AnnotationSet object with the replaced annotation texts is returned. Defaults to False.

Returns:

replaced_annotations (nnsa.AnnotationSet) – new AnnotationSet object with the replaced annotation texts (if inplace is False).

shade_axis(begin=None, end=None, labels_mapping=None, time_scale='seconds', **kwargs)[source]

Shade the areas in the current axis corresponding to the annotations.

Parameters:

begin (float, optional) – begin time (in same units as time_scale) to start shading. If None, the time of the first annotation is used. Defaults to None.
end (float, optional) – end time (in same units as time_scale) to end shading. If None, the end time of the last annotation is used. Defaults to None.
labels_mapping (dict) – a dictionary mapping each to be included annotation text to a color. If None, all unique annotations will be assigned a unique color. Defaults to None.
time_scale (str, optional) – the time scale to use. Choose from ‘seconds’, ‘minutes’, ‘hours’. Defaults to ‘seconds’.
**kwargs (optional) – optional keyword arguments for shade_axis().

sleep_stages(**kwargs)[source]

Convert the annotations to sleep stages and return the sleep stages as a SleepStagesResult object.

This is a wrapper that prepares the input for SleepStages.sleep_stages() and returns the result.

Parameters:: **kwargs (optional) – optional keyword arguments to overrule default parameters of the SleepStages class.
Returns:: result (nnsa.SleepStagesResult) – SleepStagesResult object containing annotations related to sleep stages.

sort(inplace=False)[source]

Sort the annotations based on onset.

Parameters:: inplace (bool, optional) – if True, sorts the annotations in place. If False, a new AnnotationSet object with the sorted annotations is returned. Defaults to False.
Returns:: sorted_annotations (edfreadpy.AnnotationSet) – new AnnotationSet object with the sorted annotations (if inplace is False).

standardize_annotations(inplace=False)[source]

Convert the annotations to standard labels and return a new AnnotationSet with the standard labels.

Parameters:: inplace (bool, optional) – if True, standardizes the annotations in place. If False, a new AnnotationSet object with the standardized annotations is returned. Defaults to False.
Returns:: standardized_annotations (nnsa.AnnotationSet) – new AnnotationSet object with standard annotations/labels (if inplace is False).

start_time()[source]

Return the onset time of the first annotation (in seconds).

Returns:: (float) – onset time of first annotation (in seconds).

subtract_offset(offset, inplace=False)[source]: Alias for self.subtract_time() for backwards compatibility reasons.

subtract_time(dt, inplace=False)[source]

Subtract a constant amount of time from the onset of each annotation in the annotation set.

May be used to perfectly align the annotations with the recorded signals, in case the recording of the signals did not start at the exact (whole) second defined as the starttime the fileheader. In EDF+ files, the EDF Annotation channel contains the (fractional) onset of each datarecord w.r.t. the starttime defined in the file header.

Parameters:

dt (float) – the amount of time to subtract from the annotation onsets (in seconds).
inplace (bool, optional) – if True, subtracts the offset from the annotations in place. If False, a new AnnotationSet object with the edited annotations is returned. Defaults to False.

Returns:

annotation_set (edfreadpy.AnnotationSet) – new AnnotationSet object with the edited annotations (if inplace is False).

texts()[source]

Return the texts of all annotations in the set.

Returns:: (np.ndarray) – the texts of all annotations in the set.

to_dataframe()[source]

Return a DataFrame with the annotations.

Returns:: (pd.DataFrame) – DataFrame object with annotations.

to_df()[source]: Alias for self.to_dataframe().

class nnsa.BaseDataset(*args, label=None)[source]

Bases: object

Base class with common functionalities for Datasets.

Parameters:

*args (optional) – optional positional arguments may be TimeSeries objects containing signals. Those will be added to the Dataset upon initialization. When a list is given, this list is expected to be a list of TimeSeries objects.
label (str) – optional label for dataset.

Methods:

`append`(data_object)	Append a data object to the Dataset (in place).
`asarray`([channels, return_channel_labels, ...])	Return an array containing (a copy of) multi-channel data with shape (num_channels, num_samples) if channels_last is False.
`astype`(*args[, inplace])
`average_signals`([channels])	Average the signals and return as a new TimeSeries.
`clear`()	Empty the dataset (inplace).
`combine_channels`([channels, label])	Combine the channels into one signal by cutting into segments and pasting the best channels together.
`compute_power_cwt`([freq_low, freq_high, ...])	Compute power in specified band for each signal in the dataset.
`demean`([channels, verbose, inplace])	Demean each time series in the dataset.
`envelope`([verbose])	Compute the envelope of the signals.
`extract_channels`([channels, make_copy])	Extract the specified channels and return them in a new dataset
`extract_default_channels`(**kwargs)	Extract the 9 default channels ([]).
`extract_epoch`([begin, end, channels])	Extract a piece of data in the specified interval.
`fill_nan`([value, inplace])	Fill NaN values in the signals.
`filter`(filter_obj[, channels, inplace, verbose])	Filter each of the signals in the Dataset by calling the filter() method on each TimeSeries objects.
`filter_saved_filter`(filter_name[, channels, ...])	Filter each of the signals in the dataset by calling the filter_saved_filter() method on each TimeSeries object.
`filtfilt`(filter_obj[, channels, inplace, ...])	Filter each of the signals in the Dataset by calling the filtfilt() method on each TimeSeries objects.
`from_data_dict`(data_dict, **kwargs)	Create a dataset based on data in a dict.
`get_non_artefact_masks`(segment_length[, ...])	For each signal get a mask indicating non artefact segments.
`interp`(t[, channels])	Linearly interpolate values.
`interp_nan`(args, *kwargs)
`interpolate_nan`([inplace, max_nan_length])	Linearly interpolate nan values.
`is_synchronized`([channels])	Check if all the TimeSeries in the Dataset are synchronized.
`isempty`()	Check if dataset is empty.
`line_length`([preprocess, verbose])	Compute line length feature.
`logical_and`(other)
`logical_or`(other)
`merge`(other[, inplace])	Merge other BaseDataset object(s) into one object.
`multi_scale_entropy`([preprocess, verbose])	Perform a multi-scale entropy (mse) analysis.
`multifractal_analysis`([verbose])	Perform a multifractal analysis (mfa).
`nan_to_num`([channels, inplace])	Apply numpy's nan_to_num function to each timeseries (e.g.
`normalize`([how, channels, inplace, verbose])	Normalize the data.
`notch_filt`([f0, channels, inplace, verbose])
`plot`(*args[, begin, end, channels, scale, ...])	Plot the data of specified channels for a specifiec time frame.
`power_analysis`([verbose])	Perform a Fourier based power analysis.
`read_hdf5`(filepath[, ds, begin, end])	Read the Dataset object to a .hdf5 file.
`remove`(channel[, inplace, verbose])	Remove a signal from the dataset.
`remove_artefact_channels`([inplace])	Remove channels that do not meet the quality criteria.
`remove_artefacts`([inplace])	Replace samples that are artefacts by np.nan
`remove_flatlines`([inplace, verbose])	Replace flatline samples by np.nan
`remove_neighborhood_artefacts`(*args[, ...])	Replace values by np.nan if samples in their neighborhood are nan.
`remove_outliers`(*args[, inplace, verbose])	Replace outlier samples by np.nan
`remove_values`(*args[, inplace, verbose])	Replace values by np.nan
`resample`(fs_new[, method, channels, ...])	Resample the signals in this Dataset to new sampling frequency fs_new.
`save_hdf5`(filepath[, overwrite])	Save the Dataset object to a .hdf5 file.
`segment`(segment_length[, overlap])	Return a segment generator that segments the signals in the dataset (into smaller time segments).
`set_time_offset`(time_offset[, inplace])	Set the same time offset to all time series.
`signal_stats`([verbose])	Compute signal statistics.
`squeeze`()	Returns the TimeSeries if only one TimeSeries in the Dataset.
`stepwise_moving_average`(*args[, avg_fun])
`stepwise_reduce`(*args[, channels, inplace])	Compute stepwise values for the signals in this Dataset.
`to_time_series`([make_copy])	If the Dataset contains only one TimeSeries, return this.
`transform`(fun[, channels, inplace])	Apply a custom function to each signal in the dataset.

Attributes:

`channel_labels`	Return a list of channel labels (labels of the time series objects).
`dtype`	Return the dtype if all TimeSeries have the same dtype.
`fs`	Return the sample frequency if all TimeSeries have the same sample frequency.
`info`	Return the info dict of first signal in dataset.
`shape`	Return the shape (n_channels, n_samples).
`time`	Return the time array if all TimeSeries have the same length, sample frequency and time_offset.
`time_offset`	Return the time offset if all TimeSeries have the same time offset.
`time_series`	Return the TimeSeries objects of the Dataset object.
`unit`	Return the unit if all TimeSeries have the same unit.

append(data_object)[source]

Append a data object to the Dataset (in place).

Valid data objects:: TimeSeries

Parameters:: data_object (various) – an nnsa data object. If data_object is a list, this function will recursively be called on each item in the list such that data_object may also be a list of nnsa data objects.

asarray(channels=None, return_channel_labels=False, channels_last=False)[source]

Return an array containing (a copy of) multi-channel data with shape (num_channels, num_samples) if channels_last is False.

Parameters:

channels (list, optional) – list with labels of the signals that are put into the output array. The order of the labels determines the order of the signals in the output array. I.e. the signal with label labels[i] is put in array(i, :). If None, all signals in the EegDataset are put in the output array. Defaults to None.
return_channel_labels (bool, optional) – if True, additionally outputs the labels that correspond with the rows of the created array. If False, only outputs the array. Default is False.
channels_last (bool) – if False, output shape is (num_channels, num_samples), if True, output shape is (num_samples, num_channels).

Returns:

array (np.ndarray) – array with shape (num_channels, num_samples) if channels_last == False, containing (copies of) the multi-channel data.
channels (list, optional) – the labels corresponding to the rows in the output array (only if output_channel_labels is True).

astype(*args, inplace=False, **kwargs)[source]

average_signals(channels=None, **kwargs)[source]

Average the signals and return as a new TimeSeries.

Raises an error if the signals are not synchronized.

Parameters:

channels (list, optional) – list with labels of the signals to be average. If None, all signals in the Dataset are averaged. Defaults to None.
**kwargs (optional) – keyword arguments for the creation of a the new TimeSeries object.

Returns:

(nnsa.TimeSeries) – new TimeSeries object containing the average signal.

property channel_labels

Return a list of channel labels (labels of the time series objects).

Returns:: (list) – list with channel labels.

clear()[source]: Empty the dataset (inplace).

combine_channels(channels=None, label=None, **kwargs)[source]

Combine the channels into one signal by cutting into segments and pasting the best channels together.

See nnsa.combine_channels().

Parameters:

channels (list, optional) – list with labels of the signals to be combined. If None, all signals in the Dataset are combined. Defaults to None.
label (str, optional) – a label for the combined time series.
**kwargs (optional) – keyword arguments for nnsa.combine_channels().

Returns:

ts (nnsa.TimeSeries) – TimeSeries object containing the combined signal.

compute_power_cwt(freq_low=None, freq_high=None, channels=None, inplace=False, **kwargs)[source]

Compute power in specified band for each signal in the dataset.

Parameters:

freq_low (float, optional) – lower frequency cutoff for bandpower. If None, takes lowest possible.
freq_high (float, optional) – upper frequency cutoff for bandpower. If None, takes highest possible.
channels (list, optional) – list with labels of the signals to apply to. If None, all signals in the Dataset are processed. Defaults to None.
inplace (bool, optional) – if True, replaces the signals in place by their power. If False, returns a new object.
**kwargs (optional) – optional keyword arguments for compute_power_cwt().

Returns:

ds (BaseDataset) – dataset with power of the signals (if inplace is False).

demean(channels=None, verbose=1, inplace=False, **kwargs)[source]: Demean each time series in the dataset.

property dtype

Return the dtype if all TimeSeries have the same dtype. Raises an error otherwise.

Returns:: dtype – dtype common to all signals.

envelope(verbose=1, **kwargs)[source]

Compute the envelope of the signals.

This is a wrapper that prepares the input for Envelope.envelope() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the Envelope class.

Returns:

result (nnsa.EnvelopeResult) – the result of the envelope computation.

Examples

To get the envelope data as a EegDataset: >>> signal = 4 * np.sin(np.arange(10000)) >>> ts = TimeSeries(signal=signal, fs=100, label=’EEG Cz’, unit=’uV’, info={‘source’: ‘random’}) >>> ds = BaseDataset([ts]) >>> envelope_result = ds.envelope(verbose=0) >>> ds_envelope = envelope_result.to_eeg_dataset() >>> print(type(ds_envelope).__name__) EegDataset

>>> print(ds_envelope.time_series['EEG Cz'].signal.mean())
3.9998155220627787

extract_channels(channels=None, make_copy=True)[source]

Extract the specified channels and return them in a new dataset

Parameters:

channels (list) – labels of channels to return.
make_copy (bool) – whether to make a copy of the TimeSeries objects or not. If False and a TimeSeries object in the output dataset is mutated, that TimeSeries in the input dataset is alse mutated.

Returns:

ds_out (Dataset) – new dataset with the requested channels.

Raises:

ValueError if channels do not exist. –

extract_default_channels(**kwargs)[source]

Extract the 9 default channels ([]).

Parameters:: **kwargs – for self.extract_channels().
Returns:: ds_out (Dataset) – new dataset with the requested channels.

extract_epoch(begin=None, end=None, channels=None, **kwargs)[source]

Extract a piece of data in the specified interval.

Parameters:

begin (float, optional) – start time in seconds (including time_offset) of the epoch to extract. If negative, begin is duration + begin (python-like indexing). If None, the beginning of the signals is taken. Defaults to None.
end (float, optional) – end time in seconds (including time_offset) of the epoch to extract. If negative, the end is duration + end (python-like indexing). If None, the end of the signal is taken. Must point to a time greater than begin. Defaults to None.
channels (list, optional) – list signal labels to extract an epoch from. If None, all channels are taken. Defaults to None.
**kwargs (dict, optional) – for self.__init__().

Returns:

ds_epoch (nnsa.BaseDataset) – a new Dataset with only the data of the specified epoch.

fill_nan(value=0, inplace=False, **kwargs)[source]

Fill NaN values in the signals.

Parameters:

value (float) – value to fill nans with.
inplace (bool) – whether to fill nans in place or create a new object.
**kwargs – for TimeSeries.fill_nan().

Returns:

ds (BaseDataset) – new Dataset object containing the new signals (only if inplace is False).

filter(filter_obj, channels=None, inplace=False, verbose=1, **kwargs)[source]

Filter each of the signals in the Dataset by calling the filter() method on each TimeSeries objects.

Parameters:

filter_obj (FilterBase-derived) – a child class from the nnsa.FilterBase class. The fs property does not need to be set, will be done automatically.
channels (list, optional) – list with labels of the signals that should be filtered. If None, all signals in the Dataset are filtered. Defaults to None.
inplace (bool, optional) – if True, the filtered signals will replace the original signals in this Dataset. If False (default), the function returns a new Dataset object containing the filtered signals, leaving the data in the current Dataset unchanged. Note that if only certain channels are filtered, the newly created Dataset object will only contain those filtered channels.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – keyword arguments for TimeSeries.filter().

Returns:

(BaseDataset) – new Dataset object containing the filtered signals (only if inplace is False).

filter_saved_filter(filter_name, channels=None, inplace=False, verbose=1, **kwargs)[source]

Filter each of the signals in the dataset by calling the filter_saved_filter() method on each TimeSeries object.

Parameters:

filter_name (str) – see nnsa.preprocessing.filter_saved_filter().
channels (list, optional) – list with labels of the signals that should be filtered. If None, all signals in the Dataset are filtered. Defaults to None.
inplace (bool, optional) – if True, the filtered signals will replace the original signals in this Dataset. If False (default), the function returns a new Dataset object containing the filtered signals, leaving the data in the current Dataset unchanged. Note that if only certain channels are filtered, the newly created Dataset object will only contain those filtered channels.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – keyword arguments for TimeSeries.filter_saved_filter().

Returns:

(BaseDataset) – new Dataset object containing the filtered signals (only if inplace is False).

filtfilt(filter_obj, channels=None, inplace=False, verbose=1, **kwargs)[source]

Filter each of the signals in the Dataset by calling the filtfilt() method on each TimeSeries objects.

Parameters:

filter_obj (FilterBase-derived) – a child class from the nnsa.FilterBase class. The fs property does not need to be set, will be done automatically.
channels (list, optional) – list with labels of the signals that should be filtered. If None, all signals in the Dataset are filtered. Defaults to None.
inplace (bool, optional) – if True, the filtered signals will replace the original signals in this Dataset. If False (default), the function returns a new Dataset object containing the filtered signals, leaving the data in the current Dataset unchanged. Note that if only certain channels are filtered, the newly created Dataset object will only contain those filtered channels.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – keyword arguments for TimeSeries.filtfilt().

Returns:

(BaseDataset) – new Dataset object containing the filtered signals (only if inplace is False).

from_data_dict(data_dict, **kwargs)[source]

Create a dataset based on data in a dict.

Parameters:

dict( (data_dict =) – data=data, # 2D array with shape (n_channels, n_samples) fs=fs, # sampling frequency (Hz) label=label, # Optional list of labels for each channel in data filepath=filepath, # Optional filepath where the data comes from t0=t0, # Optional time offset (time at t0) in seconds.
**kwargs – for self.__init__().
) –

property fs

Return the sample frequency if all TimeSeries have the same sample frequency. Raises an error otherwise.

Returns:: fs (flaot) – sample frequency common to all signals.

get_non_artefact_masks(segment_length, overlap=0, artefact_criteria=None)[source]

For each signal get a mask indicating non artefact segments.

Parameters:

segment_length (float) – segment length.
overlap (float, optional) – overlap between segments. Defaults to 0.
artefact_criteria (dict) – dict with artefact criteria for determining whether each segment is an artefact.

Returns:

ds_out (BaseDataset) – new dataset with non artefact masks for each signal.

property info

Return the info dict of first signal in dataset.

Returns:: info (dict) – info dict of first signal in dataset.

interp(t, channels=None, **kwargs)[source]

Linearly interpolate values.

Parameters:

t (np.ndarray) – time instances at which to evaluate the signals.
channels (list) – determines which and the order of the channels that are interpolated. If None, uses all available channels.
**kwargs – for TimeSeries.interp().

Returns:

data (np.ndarray) – 2D array with shape (n_channels, len(t)) containing the interpolated data.

interp_nan(*args, **kwargs)[source]

interpolate_nan(inplace=False, max_nan_length=None)[source]

Linearly interpolate nan values.

Parameters:

inplace (bool, optional) – interpolate inplace (True) or return a copy (False). Defaults to False.
max_nan_length (int or str, optional) – number of maximum allowable consecutive nan samples that we interpolate (in seconds). If ‘auto’, the average duration of a constant value in the signal is taken. If None, no limit is applied, i.e. all nans are interpolated. Defaults to None.

Returns:

ds (BaseDataset) – dataset with nans interpolated (if inplace is False).

is_synchronized(channels=None)[source]

Check if all the TimeSeries in the Dataset are synchronized.

Parameters:: channels (list, optional) – list with labels of signals to check. If None, all signals are checked. Defaults to None.
Returns:: (bool) – True if all TimeSeries are in sync, False if not.

isempty()[source]

Check if dataset is empty.

Returns:: (bool) – True if the Dataset is empty, False if not.

line_length(preprocess=False, verbose=1, **kwargs)[source]

Compute line length feature.

This is a wrapper that prepares the input for LineLength.line_length() and returns the result.

Parameters:

preprocess (bool, optional) – apply a default preprocessing routine on the data before feature computation (True) or not (False). Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the LineLength class.

Returns:

result (nnsa.LineLengthResult) – the result of the line length computation.

logical_and(other)[source]

logical_or(other)[source]

merge(other, inplace=False)[source]

Merge other BaseDataset object(s) into one object.

Parameters:

other (BaseDataset or list) – list with (compatible) BaseDataset objects. Must contain the same TimeSeries (with same labels and fs) as self.
inplace (bool, optional) – if True, merges the data inplace by adding it to the current object. If False, a new object is returned, leaving the original ones unchanged. Defaults to False.

Returns:

out (BaseDataset) – new BaseDataset object containing the merged data (if inplace is False).

multi_scale_entropy(preprocess=False, verbose=1, **kwargs)[source]

Perform a multi-scale entropy (mse) analysis.

This is a wrapper that prepares the input for MultiScaleEntropy.multi_scale_entropy() and returns the result.

Parameters:

preprocess (bool, optional) – apply a default preprocessing routine on the data before feature computation (True) or not (False). Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the MultiScaleEntropy class.

Returns:

result (nnsa.MultiScaleEntropyResult) – the result of the mse analysis.

multifractal_analysis(verbose=1, **kwargs)[source]

Perform a multifractal analysis (mfa).

This is a wrapper that prepares the input for MultifractalAnalysis.multifractal_analysis() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the MultifractalAnalysis class.

Returns:

result (nnsa.MultifractalAnalysisResult) – the result of the mfa analysis.

nan_to_num(channels=None, inplace=False, **kwargs)[source]

Apply numpy’s nan_to_num function to each timeseries (e.g. to replace nans by zeros).

Parameters:

channels (list) – see self.transform.
inplace (bool) – see self.transform.
**kwargs (optional) – kwargs for self.transform.

normalize(how='zscore', channels=None, inplace=False, verbose=0, **kwargs)[source]

Normalize the data.

Parameters:

how (str) – how to normalize. Choose from: - “zscore”
channels (list, optional) – which channels to normalize. If None, all channels are normalized.
inplace (bool, optional) – whether to normalize inplace (True) or not.
verbose (int, optional) – verbosity level.
**kwargs (dict) – for TimeSeries.normalize().

Returns:

(BaseDataset) – new Dataset object (only returned if inplace is False).

notch_filt(f0=50, channels=None, inplace=False, verbose=1, **kwargs)[source]

plot(*args, begin=None, end=None, channels=None, scale=None, relative_time=False, time_scale='seconds', add_offsets=False, ticklabels=True, subplots=False, color=None, legend=True, ax=None, **kwargs)[source]

Plot the data of specified channels for a specifiec time frame.

Plots in the current axis.

Parameters:

*args (optional) – optional arguments for the plt.plot() function.
begin (float, optional) – begin time in seconds (incl. offset). If None, plots from the first sample. Defaults to None.
end (float, optional) – end time in seconds or None to specify the end of the entire signal. Relative to the beginning of the loaded signal ignores any offset. Defaults to None.
channels (list, optional) – list of labels specifying the signals to plot. If None, all signals in the Dataset are plotted. Defaults to None.
relative_time (bool, optional) – if True, the time axis is relative to the start of the segment to plot. If False, the time axis will correspond to the time in the recording.
time_scale (str, optional) – the time scale to use. Choose from ‘seconds’, ‘minutes’, ‘hours’. Defaults to ‘seconds’.
add_offsets (bool, optional) – if True, adds y offsets, based on signal SD, to each of the channels for easier visualization. If False, does not apply any offsets. If a float, adds this offset to each channel.
ticklabels (bool, optional) – if add_offsets is True, set ticklabels to True to put the signal labels as yticks. Set to False to keep numeric y-labels.
subplots (bool, optional) – if True, plots each signal in a separate subplot. If False, plots all signals in ax`. Defaults to False.
color (list or dict, optional) – list or dictionary woth colors for each TimeSeries in the Dataset.
legend (bool, optional) – include legend (True) or not (False) if subplots is False. Defaults to True.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. If subplots is True, always plots in a new figure. Defaults to None.
**kwargs (optional) – optional keyword arguments for the plt.plot() function.

Returns:

ax (plt.axes) – axes instance.

power_analysis(verbose=1, **kwargs)[source]

Perform a Fourier based power analysis.

This is a wrapper that prepares the input for PowerAnalysis.power_analysis() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the PowerAnalysis class.

Returns:

result (nnsa.PowerAnalysisResult) – the result of the power analysis.

static read_hdf5(filepath, ds=None, begin=None, end=None, **kwargs)[source]

Read the Dataset object to a .hdf5 file.

Parameters:

filepath (str) – filepath to read.
begin (float, optional) – start second (wrt time offset).
end (float, optional) – end second (wrt time offset).
ds (optional) – Dataset object to append the data to. If None, creates a BaseDataset object.
**kwargs (optional) – keyword arguments for TimeSeries().

Returns:

ds (nnsa.BaseDataset) – Dataset object with data read from the file.

Examples

>>> signal = np.random.rand(8, 1000)
>>> ds = BaseDataset()
>>> ts_all = [TimeSeries(signal=signal[i], fs=10, label=i) for i in range(len(signal))]
>>> ds.append(ts_all)
>>> ds.save_hdf5('testfile.hdf5')
>>> ds_read = BaseDataset.read_hdf5('testfile.hdf5')
>>> ds_read_epoch = BaseDataset.read_hdf5('testfile.hdf5', begin=10, end=20)
>>> os.remove('testfile.hdf5')
>>> _ = [assert_equal(ts.signal, ts_read.signal) for ts, ts_read in zip (ds, ds_read)]
>>> _ = [assert_equal(ts.signal, ts_read.signal) for ts, ts_read in zip (ds.extract_epoch(begin=10, end=20), ds_read_epoch)]

remove(channel, inplace=False, verbose=1)[source]

Remove a signal from the dataset.

Parameters:

channel (str) – label of the channel to remove.
inplace (bool, optional) – If True, the channel is removed directly from the EegDataset object itself. If False, the function returns a copy of the original EegDataset object in which the specified channel is removed, leaving the data in the current EegDataset unchanged. Default is False.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(BaseDataset) – new Dataset object containing the same signals, except for the removed channel (only if inplace is False).

remove_artefact_channels(inplace=False, **kwargs)[source]

Remove channels that do not meet the quality criteria.

Parameters:

inplace (bool, optional) – If True, the channel is removed directly from the Dataset object itself. If False, the function returns a copy of the original Dataset object in which the artefacted channels are removed, leaving the data in the current Dataset unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for overruling default signal quality criteria (see nnsa.artefacts.artefact_detection.detect_artefact_signals()).

Returns:

(EegDataset) – new Dataset object containing the same signals, except for the removed channels (only if inplace is False).

remove_artefacts(inplace=False, **kwargs)[source]

Replace samples that are artefacts by np.nan

Parameters:

inplace (bool, optional) – If True, the samples are replaced directly from the Dataset object itself. If False, the function returns a copy of the original Dataset object in which the artefacted samples are replaced, leaving the data in the current Dataset unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for overruling default sample quality criteria (see nnsa.artefacts.artefact_detection.detect_artefact_samples()).

Returns:

(BaseDataset) – new Dataset object containing the same signals, but with artefacted samples changed to np.nan (only returned if inplace is False).

remove_flatlines(inplace=False, verbose=1, **kwargs)[source]

Replace flatline samples by np.nan

Parameters:

inplace (bool, optional) – If True, the samples are replaced directly from the Dataset object itself. If False, the function returns a copy of the original Dataset object in which the flatline samples are replaced, leaving the data in the current Dataset unchanged. Default is False.
verbose (int) – verbosity level.
**kwargs (optional) – optional keyword arguments for TimeSeries.remove_flatlines().

Returns:

(BaseDataset) – new Dataset object containing the same signals, but with flatline samples changed to np.nan (only returned if inplace is False).

remove_neighborhood_artefacts(*args, inplace=False, verbose=1, **kwargs)[source]

Replace values by np.nan if samples in their neighborhood are nan.

Parameters:

*args – arguments for TimeSeries.remove_neighborhood_artefacts().
inplace (bool, optional) – If True, the samples are replaced directly from the Dataset object itself. If False, the function returns a copy of the original Dataset object in which the samples are replaced, leaving the data in the current Dataset unchanged. Default is False.
verbose (int) – verbosity level.
**kwargs (optional) – optional keyword arguments for TimeSeries.remove_neighborhood_artefacts().

Returns:

(BaseDataset) – new Dataset object containing the same signals, but with specific samples changed to np.nan (only returned if inplace is False).

remove_outliers(*args, inplace=False, verbose=1, **kwargs)[source]

Replace outlier samples by np.nan

Parameters:

*args – arguments for TimeSeries.remove_outliers().
inplace (bool, optional) – If True, the samples are replaced directly from the Dataset object itself. If False, the function returns a copy of the original Dataset object in which the samples are replaced, leaving the data in the current Dataset unchanged. Default is False.
verbose (int) – verbosity level.
**kwargs (optional) – optional keyword arguments for TimeSeries.remove_outliers().

Returns:

(BaseDataset) – new Dataset object containing the same signals, but with anomaly samples changed to np.nan (only returned if inplace is False).

remove_values(*args, inplace=False, verbose=1, **kwargs)[source]

Replace values by np.nan

Parameters:

*args – arguments for TimeSeries.remove_values().
inplace (bool, optional) – If True, the samples are replaced directly from the Dataset object itself. If False, the function returns a copy of the original Dataset object in which the samples are replaced, leaving the data in the current Dataset unchanged. Default is False.
verbose (int) – verbosity level.
**kwargs (optional) – optional keyword arguments for TimeSeries.remove_values().

Returns:

(BaseDataset) – new Dataset object containing the same signals, but with specific samples changed to np.nan (only returned if inplace is False).

resample(fs_new, method='polyphase_filtering', channels=None, inplace=False, verbose=1, **kwargs)[source]

Resample the signals in this Dataset to new sampling frequency fs_new.

Parameters:

fs_new (float) – new sampling frequency to resample to.
method (str, optional) – see nnsa.TimeSeries.resample()
channels (list, optional) – list with labels of the signals to be resampled. If None, all signals in the Dataset are resampled. Defaults to None.
inplace (bool, optional) – if True, the resampled signals will replace the original signals in this Dataset. If False (default), the function returns a new Dataset object containing the resampled signals, leaving the data in the current Dataset unchanged. Note that if only certain channels are resampled, the newly created Dataset object will only contain those resampled channels.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – see nnsa.TimeSeries.resample()

Returns:

(BaseDataset) – new Dataset object containing the resampled signals (only if inplace is False).

save_hdf5(filepath, overwrite=False)[source]

Save the Dataset object to a .hdf5 file.

Parameters:

filepath (str) – filepath to save to.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists and mode is ‘w’. Defaults to False.

segment(segment_length, overlap=0)[source]

Return a segment generator that segments the signals in the dataset (into smaller time segments).

Only works is all signals have the same sample frequency.

Parameters:

segment_length (float) – length of a segment in seconds.
overlap (float, optional) – overlap between succesive segments in seconds. Defaults to 0.

Yields:

(BaseDataset) – Dataset object holding the signals of the next segment.

set_time_offset(time_offset, inplace=False)[source]

Set the same time offset to all time series.

Parameters:

time_offset (float) – time offset to set.
inplace (bool, optional) – whether to set in place (True) or create a new object (False). Defaults to False.

Returns:

ds_out (BaseDataset) – new dataset with time offset set to all time series (if inplace is True).

property shape

Return the shape (n_channels, n_samples).

Raises an error if not all signals have the same lengths.

signal_stats(verbose=1, **kwargs)[source]

Compute signal statistics.

This is a wrapper that prepares the input for SignalStats.signal_stats() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the SignalStats class.

Returns:

result (nnsa.SignalStatsResult) – object containing the signal stats.

squeeze()[source]: Returns the TimeSeries if only one TimeSeries in the Dataset. Otherwise returns the Dataset itself.

stepwise_moving_average(*args, avg_fun=<function nanmean>, **kwargs)[source]

stepwise_reduce(*args, channels=None, inplace=False, **kwargs)[source]

Compute stepwise values for the signals in this Dataset.

Parameters:

*args – inputs for nnsa.TimeSeries.stepwise_reduce()
channels (list, optional) – list with labels of the signals to apply to. If None, all signals in the Dataset are processed. Defaults to None.
inplace (bool, optional) – if True, the new signals will replace the original signals in this Dataset. If False (default), the function returns a new Dataset object containing the new signals, leaving the data in the current Dataset unchanged. Note that if only certain channels are processed, the newly created Dataset object will only contain those processed channels.
**kwargs (optional) – keyword arguments for nnsa.TimeSeries.stepwise_reduce()

Returns:

(BaseDataset) – new Dataset object containing the new signals (only if inplace is False).

property time

Return the time array if all TimeSeries have the same length, sample frequency and time_offset. Raises an error otherwise.

Returns:: time (np.ndaaray) – time array common to all signals.

property time_offset

Return the time offset if all TimeSeries have the same time offset. Raises an error otherwise.

Returns:: time_offset (flaot) – time offset in seconds common to all signals.

property time_series

Return the TimeSeries objects of the Dataset object.

Returns:: (dict) – the TimeSeries objects in the Dataset object, where the keys are the TimeSeries labels and the values the corresponding TimeSeries objects.

to_time_series(make_copy=False)[source]

If the Dataset contains only one TimeSeries, return this. Otherwise raise a ValueError.

Parameters:: make_copy (bool, optional) – if True, the returned TimeSeries is a copy of the TimeSeries in self. If False, no copy is made.
Returns:: ts (TimeSeries) – nnsa TimeSeries object.

transform(fun, channels=None, inplace=False)[source]

Apply a custom function to each signal in the dataset.

Parameters:

fun (function) – function that transform the data. Takes in a 1D array and returns a 1D array.
channels (list, optional) – list with labels of the signals to transform. If None, all signals in the Dataset are transformed. Defaults to None.
inplace (bool) – transform the data inplace (True) or return a new dataset instance (False).

Returns:

ds (BaseDataset) – dataset with the transformed data (if inplace is False).

property unit

Return the unit if all TimeSeries have the same unit. Raises an error otherwise.

Returns:: unit (str) – unit common to all signals.

class nnsa.BrainAgeResult(brain_age, algorithm_parameters, fs=None, is_novelty=None, novelty_scores=None, model_names=None, data_info=None, segment_start_times=None, segment_end_times=None, time_offset=0)[source]

Bases: ResultBase

Result class for (ensemble of) brain age estimates.

Parameters:

brain_age (np.ndarray) – array with shape (n_segment, n_models). If only one model is used, n_models can be 1.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
fs (float) – 1/segment length in Hz.
is_novelty (np.ndarray, None) – optional boolean array with shape (n_segments,) indicating which segmentes were classified as novelties (out-of-data).
novelty_scores (np.ndarray, None) – optional array with shape (n_segments,) containing novelty scores.
model_names (list) – optional list with names for models. Should have length n_models.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.

Attributes:

`num_segments`	Return the number of segments.
`time_axis`
`y_pred`

Methods:

`plot`([time_scale, color, ax, novelty_kwargs])	Simple plot of brain age.
`remove_unreliabilities`(aci_per_seg[, ...])	Return an array with FBA estimates for each segment and model, but without the segments containing (too much) artefacts and novelties.
`robust_fba`([aci_per_seg, verbose])	Robust FBA prediction by selecting segments without artefacts or novelties.
`to_df`()	Convert segment-wise results to pandas DataFrame.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(time_scale=None, color=None, ax=None, novelty_kwargs=None, **kwargs)[source]

Simple plot of brain age.

Parameters:

time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘hours’.
color (str) – color to plot in.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None
novelty_kwargs (dict) – kwargs for plt.scatter for indicating novelties.
kwargs – for plt.plot().

remove_unreliabilities(aci_per_seg, aci_thres=50, min_num_segments=3, how='remove', verbose=1)[source]

Return an array with FBA estimates for each segment and model, but without the segments containing (too much) artefacts and novelties.

Parameters:

aci_per_seg (np.ndarray) – array with length n_segments containing the artefact contamination index (%) for each segment. If None, does not check for artefacts (only novelties).
aci_thres (float) – threshold for ACI (%).
min_num_segments (int) – minimum number of consecutive reliable segments.
how (str) – what to do with unreliabilities. ‘remove’: removes the segments from the array. ‘nan’: replaces the unreliable segments with nan.
verbose (int) – verbosity level.

Returns:

y_pred_ens (np.ndarray) – array with shape (n_segments, n_models) containing selected FBA estimates (novelties, artefact segments removed).

robust_fba(aci_per_seg=None, verbose=1)[source]

Robust FBA prediction by selecting segments without artefacts or novelties.

Parameters:

aci_per_seg (np.ndarray) – array with length n_segments containing the artefact contamination index (%) for each segment. If None, only the novelties are used for robust FBA.
verbose (int) – verbosity level.

Returns:

data (pd.Series) – results (median, some quantiles, and the FBAs of the selected segments).

property time_axis

to_df()[source]: Convert segment-wise results to pandas DataFrame.

property y_pred

class nnsa.BrainAgeSinc(which='v2', detect_novelties=True, **kwargs)[source]

Bases: ClassWithParameters

Deep shared inception ensemble model for brain age estimation.

References

A. Ansari et al., “Brain age as an estimator of neurodevelopmental outcome: A deep learning approach for neonatal cot-side monitoring,” Jan. 2023, doi: 10.1101/2023.01.24.525361

Parameters:

which (str) – string specifying which model to use. Options: - ‘v1’: the one trained by Ansari and Pillay as available on (as of June 2022): https://github.com/amirans65/brainagemodel/tree/c948967d7a42e237fa6cc8991b696b662300f6e2/brainagemodel/pretrained - ‘v2’: the one retrained by Hermans including more older neonates.
**kwargs – see default_parameters().

Attributes:

`all_models`	Return the model.
`data_requirements`	Return a dictionary with requirements for the input of the CNN.
`normalization_parameters`	Return the normalization parameters.

Methods:

`brain_age_sinc`(eeg_ds[, verbose])	TODO
`default_parameters`()	Return the default parameters as a dictionary.
`preprocess`(eeg, fs[, axis, verbose])	Preprocess, normalize, segment.
`process`(eeg, fs[, batch_size, axis, verbose])	Predict brain age.

property all_models

Return the model.

Returns:: (nnsa.model or keras.model) – model object.

brain_age_sinc(eeg_ds, verbose=1)[source]

TODO

Parameters:

x (np.ndarray) – TODO
verbose (int, optional) – verbose level (0 or 1). Default to 1.

Returns:

(BrainAgeSincResult) – nnsa object containing the results of the CNN brain age predictor.

property data_requirements

Return a dictionary with requirements for the input of the CNN.

Returns:: data_requirements (dict) – a dictionary with requirements for the input of the CNN (channel order, fs, segment length).

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the model.

property normalization_parameters

Return the normalization parameters.

Returns:: (tuple) – normalization parameters.

preprocess(eeg, fs, axis=1, verbose=1)[source]

Preprocess, normalize, segment.

Parameters:: self.process(). (see) –
Returns:: x (np.ndarray) – array with shape (n_seg, n_time, n_channels).

process(eeg, fs, batch_size=None, axis=-1, verbose=1)[source]

Predict brain age.

Includes preprocessing (filtering, segmentation, normalization)

Note that the channel order in eeg must be according to self.data_requirements[‘channel_order’].

Parameters:

eeg (np.ndarray) – 2D-array with raw (multichannel) EEG data. The channel order must be the same as specified in self.data_requirements.
fs (float) – sampling frequency of EEG.
batch_size (int, None) – number of segments to proces at once. If None, processes all segments at once, but in case of memory issues (for long recordings), you may set this value to e.g. 7200 to process 7200 segments (1h of EEG) at a time.
axis (int) – the time axis of eeg.
verbose (int) – verbosity level.

Returns:

(BrainAgeResult) – nnsa object containing the results of the brain age predictor.

class nnsa.BurstDetection(**kwargs)[source]

Bases: ClassWithParameters

Class for burst detection.

Main method: burst_detection().

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(10, 100000)
>>> bd = BurstDetection()
>>> print(type(bd.parameters).__name__)
Parameters

>>> result = bd.burst_detection(x, fs=256, verbose=0)
>>> print(type(result).__name__)
BurstDetectionResult

>>> result.bursts[0, 20000]
0.0

Methods:

`burst_detection`(data_matrix, fs[, ...])	Perform burst detection on multichannel data.
`default_parameters`()	Return the default parameters.
`dibi_burst_detection`(data_matrix, fs[, verbose])	Burst detection method as proposed by Vladimir Matic.
`envelope_burst_detection`(data_matrix, fs[, ...])	Detect bursts using the signal envelope (signal energy).
`line_length_burst_detection`(data_matrix, fs)	Detect bursts using the line length as introduced by Koolen et al.
`nleo_burst_detection`(data_matrix, fs[, lc, ...])	Detect bursts using the non-linear energy operator (NLEO) as introduced by Palmu et al.
`otoole_burst_detection`(data_matrix, fs[, ...])	Burst detection method as implemented by O'Toole in MATLAB.

burst_detection(data_matrix, fs, channel_labels=None, verbose=1)[source]

Perform burst detection on multichannel data.

Parameters:

data_matrix (np.ndarray) – EEG data. See check_multichannel_data_matrix(). The data might be filtered before, depending on the method for burst detection. Most methods, do their own specific filtering. See the documentation and code of the specific methods.
fs (float) – sample frequency of the EEG signals.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.BurstDetectionResult) – object containing the burst detection result per channel.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

static dibi_burst_detection(data_matrix, fs, verbose=1)[source]

Burst detection method as proposed by Vladimir Matic.

See detect_dibi() for details.

Parameters:

data_matrix (np.ndarray) – unfiltered EEG data, see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

bursts (np.ndarray) – array with dimensions (1, time) containing 1s at locations of bursts and 0s at locations of non-bursts.
ibis (np.ndarray) – array with dimensions (1, time) containing 1s at locations of inter-burst-intervals (IBIs) and 0s at locations of non-IBIs.

static envelope_burst_detection(data_matrix, fs, lc=0.5, hc=32, notch=50, amplitude_thr_high=30, channel_thr_high=2, min_ibi_dur=1, max_burst_dur=20, amplitude_thr_low=30, channel_thr_low=None)[source]

Detect bursts using the signal envelope (signal energy).

Implementation of the algorithm described by Jennekens et al. 2011.

Notes

The default parameters are the optimal values according to the papaer by Jennekens et al. 2011.

References

Jennekens W , Ruijs LS , Lommen CML , Niemarkt HJ , Pasman JW , van Kranen–Mastenbroek VM , et al. Automatic burst detection for the EEG of the preterm infant. Physiol Meas 2011;32(10):1623–37 .

Parameters:

data_matrix (np.ndarray) – unfiltered EEG data (in uV), see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
lc (float, optional) – low cut-off filter frequency (LC) in Hz. Defaults to 0.5.
hc (float, optional) – high cut-off filter frequency (HC) in Hz. Defaults to 32.
notch (float, optional) – notch filter frequency in Hz. Defaults to 50.
amplitude_thr_high (float, optional) – amplitude-threshold-high in uV (ATH). Sample points with an envelope value equal to or above ATH are considered high-voltage activity. Defaults to 30.
channel_thr_high (int, optional) – channel-threshold-high (CTH). If the number of channels with high voltage is >= CTH, the global EEG-activity is considered as high-voltage activity. Defaults to 2.
min_ibi_dur (float, optional) – the minimum separation between 2 bursts in seconds. If two detected periods of high-voltage activity appear within min_ibi_dur after each other it is assumed that they belong together and form one period. Defaults to 1.
max_burst_dur (float, optional) – the maximum duration of bursts in seconds. High-voltage periods >=max_burst_dur are classified as continuous patterns while periods < max_burst_dur are classified as bursts. Defaults to 20.
amplitude_thr_low (float, optional) – amplitude-threshold low in uV (ATL). If the envelope value is lower than the ATL, the corresponding sample is a candidate for IBI. Defaults to 30.
channel_thr_low (int, optional) – channel-threshold-low (CTL). The minimum number of channels with low voltage activity for the sample to be considered an IBI. If None, the total number of channels in the input data_matrix is used. Defaults to None.

Returns:

bursts (np.ndarray) – array with dimensions (1, time) containing 1s at locations of bursts and 0s at locations of non-bursts.
ibis (np.ndarray) – array with dimensions (1, time) containing 1s at locations of inter-burst-intervals (IBIs) and 0s at locations of non-IBIs.

static line_length_burst_detection(data_matrix, fs, F_1=0.85, F_2=0.4, threshold_window=None, min_ibi_dur=2, min_burst_amp=30, verbose=1, **line_length_kwargs)[source]

Detect bursts using the line length as introduced by Koolen et al.

Notes

The default parameters are the optimal values according to the paper by Koolen et al. 2014. In that paper, the algorithm was optimized for unipolar multi-channel EEG sampled at 250 Hz. If the given signals have a different sample frequency, the signals are resampled to 250 Hz.

The postprocessing that checks whether the high energy parts are valid bursts is a bit simpler than mentioned in the paper and also differs from the original Matlab code. In this code, high energy periods (candidates for bursts) are considered burst only if they contain a (normalized) line length value higher than some threshold (controlled by F_2 parameter) AND a high voltage ampltiude in any channel of the original signal (controlled by min_burst_amp parameter). More specifically for this first criterion: the median normalized line length (me_LL) in a high energy period must reach at least F_1*mean(me_LL) + F_2*std(me_LL) to be a burst (in Koolen et al., this is referred to as a ‘pronounced peak’).

References

Koolen, N. et al. Line length as a robust method to detect high-activity events: Automated burst detection in premature EEG recordings. Clinical Neurophysiology 125, 1985{1994 (2014).

Parameters:

data_matrix (np.ndarray) – (filtered) EEG data, see check_multichannel_data_matrix(). Koolen et al. originally used EEG signals bandpass filtered between 1 - 20 Hz.
fs (float) – sample frequency of the EEG signals. Must be 250 Hz or higher.
F_1 (float, optional) – scale factor for the burst detection (F in Eq. 4 of the paper). Defaults to 0.85.
F_2 (float, optional) – scale factor for the standard deviation (used to detect pronounced peaks in me_LL). Defaults to 0.40.
threshold_window (float, optional) – window for the adaptive threshold in seconds. The threshold is adapted in periods of threshold_window seconds, using a moving average of the given window size. If set to None, the threshold is fixed for the entire signal. Defaults to None.
min_ibi_dur (float, optional) – the minimum separation between 2 bursts in seconds. If two detected periods of high-voltage activity appear within min_ibi_dur after each other it is assumed that they belong together and form one burst period. Defaults to 2.
min_burst_amp (float, optional) – the minimum maximum EEG amplitude of a high energy period to be considered a burst. Defaults to 30.
verbose (int, optional) – verbose level. Defaults to 1.
**line_length_kwargs (optional) – keyword arguments with parameters for nnsa.LineLength(**line_length_kwargs). See nnsa.LineLength().

Returns:

bursts (np.ndarray) – array containing 1s at locations of bursts and 0s at locations of non-bursts.
ibis (np.ndarray) – array containing 1s at locations of inter-burst-intervals (IBIs) and 0s at locations of non-IBIs.
fs_LL (float) – sample frequency of corresponding to the output arrays.

static nleo_burst_detection(data_matrix, fs, lc=0.5, hc=10, max_ripple=1, min_attenuation=40, window_avg=1.5, window_baseline=60, sat_thr=1.5, min_duration=1, verbose=1)[source]

Detect bursts using the non-linear energy operator (NLEO) as introduced by Palmu et al.

Notes

The default parameters are the optimal values according to the papaer by Palmu et al. 2010. Here, the NLEO algorithm was optimized for the P3-P4 bipolar channel, using data sampled at 256 Hz. However, since the first step of the algorithm is bandpass filtering between 0.5-10 Hz, sampling frequencies >= 25 Hz are accepted. Then after filtering, the signal is upsampled to 256 Hz.

References

Palmu K , Stevenson N , Wikström S , Hellström-Westas L , Vanhatalo S , Palva JM . Optimization of an NLEO-based algorithm for automated detec- tion of spontaneous activity transients in early preterm EEG. Physiol Meas 2010;31(11):N85–93 .

Parameters:

data_matrix (np.ndarray) – unfiltered EEG data (in uV), see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
lc (float, optional) – low cut-off filter frequency (LC) in Hz. Defaults to 0.5.
hc (float, optional) – high cut-off filter frequency (HC) in Hz. Defaults to 10.
max_ripple (float, optional) – max ripple in dB for the low pass elliptic filter, see scipy.signal.ellip(). Note that this value was not mentioned in the paper. Defaults to 1.
min_attenuation (float, optional) – min attenuation in dB for the low pass elliptic filter, see scipy.signal.ellip(). Note that this value was not mentioned in the paper. Defaults to 40.
window_avg (float, optional) – window in seconds for averaging the absolute value of the NLEO output (g). Defaults to 1.5.
window_baseline (float, optional) – window in seconds for finding the baseline value (the minimum value in this window is the baseline value, which is subtracted). Defaults to 60.
sat_thr (float, optional) – threshold on x_nleo (in uV^2) for detection of bursts. Defaults to 1.5.
min_duration (float, optional) – minimum duration of bursts in seconds. Defaults to 1.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

bursts (np.ndarray) – array containing 1s at locations of bursts and 0s at locations of non-bursts.
ibis (np.ndarray) – array containing 1s at locations of inter-burst-intervals (IBIs) and 0s at locations of non-IBIs.
fs (float) – sample frequency of corresponding to the output arrays.

static otoole_burst_detection(data_matrix, fs, verbose=1)[source]

Burst detection method as implemented by O’Toole in MATLAB.

Notes

This method was developed for the following bipolar montgage: F3-C3, F4-C4, C3-O1, C4-O2, C3-T3, C4-T4, Cz-C3, and C4-Cz.

References

JM O’ Toole, GB Boylan, RO Lloyd, RM Goulding, S Vanhatalo, and NJ Stevenson, “Detecting Bursts in the EEG of Very and Extremely Premature Infants Using a Multi-Feature Approach”, Medical Engineering and Physics, vol. 45, pp. 42–50, 2017. DOI:10.1016/j.medengphy.2017.04.003

https://github.com/otoolej/burst_detector

Notes

The algorithm was developed with EEG data using the bipolar montage F4-C4, C4-O2, F3- C3, C3-O1, T4-C4, C4-Cz, Cz-C3, and C3-T3. Therefore, the input of this algorithm should be EEG data from one of these bipolar channels.

The parameters of this algorithm are hardcoded in the bd_parameters.m file (in the nnsa/matlab/burst_detector directory).

Parameters:

data_matrix (np.ndarray) – unfiltered EEG data (bipolar), see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
verbose (int, optional) – verbose level. Defaults to 1.
TODO – pass matlab engine as kwarg.

Returns:

bursts (np.ndarray) – array with the same size as data_matrix containing 1s at locations of bursts and 0s at locations of non-bursts.
ibis (np.ndarray) – array with the same size as data_matrix containing 1s at locations of inter-burst-intervals (IBIs) and 0s at locations of non-IBIs.

class nnsa.BurstDetectionResult(bursts, ibis, algorithm_parameters, fs, nan_mask=None, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None)[source]

Bases: ResultBase

High-level interface for processing the results of burst detection analysis as created by nnsa.BurstDetection().

Parameters:

bursts (np.ndarray) – array with dimensions (channels, time) containing 1s or True at locations of bursts and 0s or False at locations of non-bursts. May also contain np.nan for indicating missing values.
ibis (np.ndarray) – array with dimensions (channels, time) containing 1s or True at locations of inter-burst-intervals (IBIs) and 0s or False at locations of non-IBIs. May also contain np.nan for indicating missing values.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
fs (float) – sample frequency corresponding to the bursts and ibis arrays.
nan_mask (np.array, optional) – boolean array with True at locations of missing values and False at locations without missing values. If not None, this nan_mask is applied to bursts and ibis to set missing values to np.nan. This way, the bursts and ibis arrays can be saved as boolean arrays, even if there are missing values (memory efficient). Must have the same shape as bursts and ibis. Defaults to None.
channel_labels (list of str, optional) – labels of the channels corresponding to the channel dimensions of the arrays. If None, default labels will be created. Defaults to None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.

Methods:

`aggregate_bursts`([min_channels_frac, ...])	Combine the burst detection of all channels to get an aggregate, global brust detection array.
`aggregate_ibis`([min_channels_frac, ...])	Combine the IBI detection of all channels to get an aggregate, global IBI detection array.
`class_numbers`()	Return a 1D array with 2s at locations of bursts, 1s at IBIs and np.nan elsewhere.
`compute_event_durations`(detected, fs)	Compute the durations of events in detected.
`compute_event_occurrences`(detected)	Compute the number of occurences in detected.
`compute_event_percentage`(detected)	Compute the time-percentage of detected events.
`compute_features`(bursts, ibis, channel_labels)	Compute burst/IBI features for each channel.
`compute_global_features`(**kwargs)	Compute global features.
`compute_global_features_per_sleep_stage`(...)	Compute global features in specific sleep stages.
`extract_epoch`([begin, end, inplace])	TODO :param begin: in seconds.
`extract_global_features`([...])	Extract features that characterize the entire recording.
`plot`()	Plot the occurence of bursts/ibis as a function of time.
`shade_axis`(*args[, channel])	Shade the current axis based on bursts and IBIs.
`to_aggregate_result`()	Compute the aggregate result by combining the burst detection of all channels.
`to_annotation_set`([channel])	Convert to AnnotationSet object.

Attributes:

`burst_id`	Return the class number for bursts.
`burst_label`	Return the class label for bursts.
`class_labels`	Return the class labels corresponding to self.class_numbers().
`duration`	Return the total duration of the recorded samples.
`ibi_id`	Return the class number for ibis.
`ibi_label`	Return the class label for ibis.
`num_segments`	Return the number of segments (number of samples).

aggregate_bursts(min_channels_frac=0.25, min_channels_elong_frac=0.125)[source]

Combine the burst detection of all channels to get an aggregate, global brust detection array.

Parameters:

min_channels_frac (int, optional) – minimum fraction of channels that must detect a burst in order to consider it a global burst (after aggregation). See aggregate_channel_events(). Defaults to 2/8.
min_channels_elong_frac (int, optional) – minimum fraction of channels that must detect a burst when elongating the globally detected bursts. See aggregate_channel_events(). Defaults to 1/8.

Returns:

bursts (np.ndarray); array with dimensions corresponding to (1, time) – time instants where bursts occur.

aggregate_ibis(min_channels_frac=1, min_channels_elong_frac=1)[source]

Combine the IBI detection of all channels to get an aggregate, global IBI detection array.

Parameters:

min_channels_frac (int, optional) – minimum fraction of channels that must detect an IBI in order to consider it a global IBI (after aggregation). See aggregate_channel_events(). Defaults to 1.
min_channels_elong_frac (int, optional) – minimum fraction of channels that must detect an IBI when elongating the globally detected IBIs. See aggregate_channel_events(). Defaults to 1.

Returns:

ibis (np.ndarray); array with dimensions corresponding to (1, time) – time instants where IBIs occur.

property burst_id

Return the class number for bursts.

Returns:: (int) – class number for bursts.

property burst_label

Return the class label for bursts.

Returns:: (str) – class label for bursts.

property class_labels

Return the class labels corresponding to self.class_numbers().

Returns:: class_labels (dict) – dictionary that maps a class number to a class label.

class_numbers()[source]

Return a 1D array with 2s at locations of bursts, 1s at IBIs and np.nan elsewhere.

Returns:: class_numbers (np.ndarray) – 1D array with same shape as self.bursts and self.ibis. 2s in the array correspond to bursts, 1s to IBIs.

static compute_event_durations(detected, fs)[source]

Compute the durations of events in detected.

Ignores incomplete events (events surrounded by nans).

Parameters:

detected (np.ndarray) – 2D array with 1s (detected) and 0s (not-detected). Dimensions correspond to (channels, time).
fs (float) – sample frequency of detected in Hz.

Returns:

out (tuple) – tuple of arrays with the durations of the detected events in seconds (per channel).

Examples

>>> BurstDetectionResult.compute_event_durations(np.array([[1, 1, 1, 0, 0, 1, 1, 0]]), fs=1)
(array([2.]),)
>>> BurstDetectionResult.compute_event_durations(np.array([[0, 1, 1, 1, 0, 1, 1, 0]]), fs=1)
(array([3., 2.]),)
>>> BurstDetectionResult.compute_event_durations(np.array([[1, 1, 1, 0, 0, np.nan, np.nan]]), fs=1)
(array([0]),)
>>> BurstDetectionResult.compute_event_durations(np.array([[np.nan, 1, 1, 0, 0, np.nan, np.nan]]), fs=1)
(array([0]),)
>>> BurstDetectionResult.compute_event_durations(np.array([[np.nan, 0, 1, 1, 1, 0]]), fs=1)
(array([3.]),)
>>> BurstDetectionResult.compute_event_durations(np.array([[np.nan, 0, 1, np.nan, 1, 0]]), fs=1)
(array([0]),)

static compute_event_occurrences(detected)[source]

Compute the number of occurences in detected.

Counts the number of onsets.

Handles np.nan values by substituting them by zeros using np.nan_to_num().

Parameters:: detected (np.ndarray) – array with 1s (detected) and 0s (not-detected). The last dimension corresponds to the time dimension. E.g. bursts array.
Returns:: num_events (int or np.ndarray) – the number of detected events (per channel).

static compute_event_percentage(detected)[source]

Compute the time-percentage of detected events.

Ignores nan values.

Parameters:: detected (np.ndarray) – array with 1s (detected) and 0s (not-detected). The last dimension corresponds to the time dimension. E.g. bursts array.
Returns:: (float or np.ndarray) – percentage of detected events (per channel).

compute_features(bursts, ibis, channel_labels, postfix=None)[source]

Compute burst/IBI features for each channel.

Parameters:

bursts (np.ndarray) – array with dimensions (channels, time) containing 1s at locations of bursts and 0s at locations of non-bursts. May also contain np.nan for indicating missing values.
ibis (np.ndarray) – array with dimensions (channels, time) containing 1sat locations of inter-burst-intervals (IBIs) and 0s at locations of non-IBIs. May also contain np.nan for indicating missing values.
channel_labels (list) – list with the same length as bursts and ibis containing the labels of the channels (the first dimension of the arrays). May also be a string in case of only one channel.
postfix (string, optional) – optional postfix to add to the feature name in the output dictionary.

Returns:

features (dict) – dictionary of features values, hashed by feature name. Each channel is treated as a separate feature.

compute_global_features(**kwargs)[source]

Compute global features.

Parameters:: **kwargs (optional) – keyword arguments for pd.DataFrame.
Returns:: df (pd.DataFrame) – dataframe with one row, and feature values in columns.

compute_global_features_per_sleep_stage(sleep_stages_result, sleep_labels=None, line_length_result=None, **kwargs)[source]

Compute global features in specific sleep stages.

Parameters:

sleep_stages_result –
sleep_labels –
line_length_result (nnse.LineLengthResult, optional) – if given, the features in the 10 most suppressed period are computed additionally.
**kwargs (optional) – kwargs for self.compute_global_features().

Returns:

property duration

Return the total duration of the recorded samples.

Ignores nans.

Returns:: (float) – duration in seconds.

extract_epoch(begin=0, end=None, inplace=False)[source]

TODO :param begin: in seconds. :param end: in seconds. :param inplace:

Returns:

extract_global_features(aggregate_channels=True, sleep_stages=None, line_length=None)[source]

Extract features that characterize the entire recording.

Parameters:

aggregate_channels (bool, optional) – if True, returns features after aggragating the channels (AGG), if False, returns features for each channel. Defaults to True.
sleep_stages (nnsa.SleepStagesResult, optional) – object containing sleep stages result. Used to report global feature values per sleep stage. If None, all data is used (no distinction is made for sleep stages). Defaults to None.
line_length (nnsa.LineLengthResult, optional) – object containing line length result. If given, additional features are computed in the most suppressed period (10 minutes) according to the suppression curve, derived from the line length. Defaults to None.

Returns:

global_features (dict) – dictionary of features values, hashed by feature name.

property ibi_id

Return the class number for ibis.

Returns:: (int) – class number for ibis.

property ibi_label

Return the class label for ibis.

Returns:: (str) – class label for ibis.

property num_segments

Return the number of segments (number of samples).

Returns:: (int) – number of segments/samples.

plot()[source]

Plot the occurence of bursts/ibis as a function of time.

TODO Option to plot one channel only or to plot the aggregate.

shade_axis(*args, channel=None, **kwargs)[source]

Shade the current axis based on bursts and IBIs.

Wrapper that converts this class to an AnnotationSet and calls shade_axis() from AnnatationSet.

Parameters:

*args (optional) – see nnsa.AnnotationSet.shade_axis().
channel (str, optional) – see self.to_annotation_set().
**kwargs (optional) – see nnsa.AnnotationSet.shade_axis().

to_aggregate_result()[source]

Compute the aggregate result by combining the burst detection of all channels.

Returns:: (nnsa.BurstDetectionResult) – a new burst detection result object, with only one channel (AGG), which is an aggregate of all channels.

to_annotation_set(channel=None)[source]

Convert to AnnotationSet object.

Parameters:: channel (str) – the channel label for which to return an AnnotationSet. If None, an aggregate burst detection, combining all channels will be converted to an AnnotationSet. Defaults to None.
Returns:: (nnsa.AnnotationSet) – AnnotationSet containing self.burst_label and self.ibi_label annotations.

class nnsa.Butterworth(tf=None, sos=None, fs=None, **kwargs)[source]

Bases: FilterBase

Create a Butterworth filter.

Parameters:: FilterBase. (see) –

Examples

>>> fs = 2000
>>> t = np.linspace(0, 1.0, fs+1)
>>> xlow = np.sin(2 * np.pi * 5 * t)  # 5 Hz
>>> xhigh = np.sin(2 * np.pi * 250 * t)  # 250 Hz
>>> x = xlow + xhigh
>>> filt = Butterworth(fs=fs, passband=120, stopband=140)
>>> y = filt.filtfilt(x)
>>> print(np.abs(y - xlow).max())
0.004803688903207593

Methods:

default_parameters()

Return the default parameters.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

class nnsa.ClassWithParameters(**kwargs)[source]

Bases: object

Base class for a class that has parameters.

Parameters:: **kwargs (optional) – optional keyword arguments for overruling default parameters. For a description of the possible parameters, see the default_parameters() method.

Methods:

default_parameters()

Return the default parameters as a dictionary.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (dict or Parameters) – a default set of parameters for the object.

class nnsa.CleanDetectorCnn(multi_channel=True, chunk_length=3600, **kwargs)[source]

Bases: ClassWithParameters

Interface for applying clean EEG detection using a trained Convolutional Neural Network.

References

T. Hermans et al., “A multi-task and multi-channel convolutional neural network for semi-supervised neonatal artefact detection,” Journal of Neural Engineering, vol. 20, no. 2, p. 26013, Mar. 2023, doi: 10.1088/1741-2552/acbc4b. https://pubmed.ncbi.nlm.nih.gov/36791462/

Parameters:

multi_channel (bool) – if True, use the multi-channel model, which requires a specific montage and order of channels (see self.data_requirements). If False, use the single-channel model (no specific montage required).
chunk_length (int, float) – number of seconds of EEG to process at once (EEG will be divided in chunks with maximal length chunk_size, predicted and then merged together). It is to prevent memory problems. If the RAM is large enough, you can set chunk_length to None to not process into chunks.

Examples

# TODO Load some raw eeg data, referenced to Cz.

# Initiate artefact detector object. cd = CleanDetectorCnn()

# Use the predict method to predict the clean mask for the eeg array. clean_mask, clean_prob = cd.predict(eeg, fs=fs))

Attributes:

`data_requirements`	Return some data requirements.
`fs_output`	Return sampling frequency of output.
`is_clean_detector`	Returns: (bool): True if the model is a clean EEG detector (instead of an artefact detector).
`model`	Return keras model.
`model_settings`	Returns dict with settings.
`normpars`	Returns tuple with mean, sd and channel labels.

Methods:

`default_parameters`()	Return the default parameters as a dictionary.
`predict`(eeg, fs[, preprocess, detect_flats, ...])	Detect clean parts in Cz-referenced EEG using the CNN model.
`preprocess_eeg`(eeg, fs[, axis, verbose])	Preprocess EEG data.

property data_requirements: Return some data requirements.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (dict or Parameters) – a default set of parameters for the object.

property fs_output: Return sampling frequency of output.

property is_clean_detector: Returns: (bool): True if the model is a clean EEG detector (instead of an artefact detector).

property model: Return keras model.

property model_settings: Returns dict with settings.

property normpars: Returns tuple with mean, sd and channel labels.

predict(eeg, fs, preprocess=None, detect_flats=False, detect_peaks=False, verbose=1, **predict_kwargs)[source]

Detect clean parts in Cz-referenced EEG using the CNN model.

Parameters:

eeg (np.ndarray) – multichannel EEG referenced to Cz. Array with shape (n_time, n_channels). If using the multi-channel model, the order of the channels should be: [‘Fp1’, ‘Fp2’, ‘C3’, ‘C4’, ‘T3’, ‘T4’, ‘O1’, ‘O2’].
fs (flaot) – sampling frequency of eeg in Hz.
preprocess (bool) – specify whether the EEG needs to be preprocessed (filtered, resampled). Set to True if eeg is raw data (but note that it should still be referenced to Cz). If not specified, preprocessing will be done if fs is not 128, otherwise not.
detect_flats (bool) – if True, computes moving std in short windows and if its below a threshold, the sample is marked as artefact (since the CNN might not catch this).
detect_peaks (bool) – if True, computes moving max abs amplitude in short windows and if its above a threshold, the sample is marked as artefact (since the CNN might not catch this).
verbose (int) – verbosity level.
**predict_kwargs – kwargs for self.model.predict().

Returns:

mask (np.ndarray) – array with shape (n_time, n_channels) containing 1 at locations of clean EEG and 0 at artefacts.
prob (np.ndarray) – array with shape (n_samples, n_channels, n_classes) containing probabilies for artefact and clean. Note that n_samples depends on the output frequency (it is not the same as n_time of eeg).

static preprocess_eeg(eeg, fs, axis=0, verbose=1)[source]

Preprocess EEG data.

Parameters:

eeg (np.ndarray) – EEG data.
fs (float, int) – sampling frequency of eeg in Hz.
axis (int) – time axis. of eeg array.
verbose (int) – verbose level.

Returns:

eeg_out (np.ndarray) – preprocessed EEG data.
fs_out (int) – sampling frequency of preprocessed EEG.

class nnsa.CoherenceGraph(**kwargs)[source]

Bases: ClassWithParameters

Class for computing a connectivity graph of multi-channel EEG based on coherence.

References

This code was partly ported from Mario Lavanga and Ofelie de Wel’s MATLAB code.

@Article{LAVANGA2018,: author = {M Lavanga and O De Wel and A Caicedo and K Jansen and A Dereymaeker and G Naulaers and S Van Huffel}, title = {A brain-age model for preterm infants based on functional connectivity}, journal = {Physiological Measurement}, year = {2018}, volume = {39}, number = {4}, pages = {044006}, month = {apr}, abstract = {Objective: In this study, the development of EEG functional connectivity during early development has been investigated in order to provide a predictive age model for premature infants. Approach: The functional connectivity has been assessed via the coherency function (its imaginary part (ImCoh) and its mean squared magnitude (MSC)), the phase locking value () and the Hilbert–Schimdt dependence (HSD) in a dataset of 30 patients, partially described and employed in previous studies (Koolen et al 2016 Neuroscience 322 298–307; Lavanga et al 2017 Complexity 2017 1–13). Infants’ post-menstrual age (PMA) ranges from 27 to 42 weeks. The topology of the EEG couplings has been investigated via graph-theory indices. Main results: Results show a sharp decrease in ImCoh indices in θ, (4–8) Hz and α, (8–16) Hz bands and MSC in β, (16–32) Hz band with maturation, while a more modest positive correlation with PMA is found for HSD, and MSC in , θ, α bands. The best performances for the PMA prediction were mean absolute error equal to 1.51 weeks and adjusted coefficient of determination equal to 0.8. Significance: The reported findings suggest a segregation of the cortex connectivity, which favours a diffused tasks architecture on the brain scalp. In summary, the results indicate that the neonates’ brain development can be described via lagged-interaction network features.}, doi = {10.1088/1361-6579/aabac4}, file = {:LAVANGA2018 - A Brain Age Model for Preterm Infants Based on Functional Connectivity.pdf:PDF}, publisher = {{IOP} Publishing}, url = {https://doi.org/10.1088%2F1361-6579%2Faabac4},

}

Main method: coherence_graph()

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(8, 10000)
>>> cg = CoherenceGraph()
>>> print(type(cg.parameters).__name__)
Parameters

>>> cg_result = cg.coherence_graph(x, fs=250, verbose=0)
>>> print(type(cg_result).__name__)
CoherenceGraphResult

>>> cg_result.mean_coh['theta'][1, 0, 0]
0.07487598638814418

Methods:

`coherence_graph`(args, *kwargs)	Alias for self.process().
`default_parameters`()	Return the default parameters.
`process`(data_matrix, fs[, channel_labels, ...])	Compute the coherence graph of multichannel data as designed by Mario Lavanga and Ofelie De Wel.

coherence_graph(*args, **kwargs)[source]: Alias for self.process().

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

process(data_matrix, fs, channel_labels=None, verbose=1)[source]

Compute the coherence graph of multichannel data as designed by Mario Lavanga and Ofelie De Wel.

Code was ported from MATLAB and constsist of 6 steps: 1) Segmentation 2) Optional filtering 3) Artefact exclusion 4) PSD estimation 5) Computation of coherence in specified frequency bands per segment 6) Averaging the coherence in the specified frequency bands

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.CoherenceGraphResult) – object containing the adjacency matrices per frequency band per segment,
where the channels are the nodes.

class nnsa.CoherenceGraphResult(mean_coh, im_coh, algorithm_parameters, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing the results of a coherence analysis as created by nnsa.CoherenceGraph().

Parameters:

mean_coh (dict) – dictionary with frequency band labels as keys and (channels, channels, segments) arrays with channel-wise absolute squared coherence as values.
im_coh (dict) – dictionary with frequency band labels as keys and (channels, channels, segments) arrays with imaginary part of the channel-wise coherence as values.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
channel_labels (list of str, optional) – labels of the channels corresponding to the rows of the values in power. If None, default labels will be created. Default is None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.array, optional) – see ResultBase.
segment_end_times (np.array, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Methods:

`compute_average_coherence`(graph_name)	Compute average coherence value across channels per segment.
`compute_average_path_lengths`(graph_name, ...)	Compute average path lengths of the graphs.
`compute_global_features`(**kwargs)	Compute global features.
`extract_global_features`([sleep_stages, ...])	Extract features that characterize the entire recording.
`plot`(which)	Plot abs(mean_coh), abs(im_coh) and binary im_coh graph (averaged over segments) in the current figure.

Attributes:

num_segments

Return the number of segments.

compute_average_coherence(graph_name)[source]

Compute average coherence value across channels per segment.

Parameters:: graph_name (str) – see self._get_graphs().
Returns:: averages (dict) – average coherence values of graphs per frequency band per segment. The items in the dict correspond to the frequency bands. The arrays in the dict are 1D with the dimension corresponding to the segments.

compute_average_path_lengths(graph_name, graph_type)[source]

Compute average path lengths of the graphs.

Parameters:

graph_name (str) – name of the graphs to analyze: ‘mean_coh’ or ‘im_coh’.
graph_type (str) – see compute_average_path_length().

Returns:

path_lengths (dict) – average path lengths of graphs per frequency band per segment. The items in the dict correspond to the frequency bands. The arrays in the dict are 1D with the dimension corresponding to the segments.

compute_global_features(**kwargs)[source]

Compute global features.

Parameters:: **kwargs (optional) – keyword arguments for pd.DataFrame.
Returns:: df (pd.DataFrame) – dataframe with one row, and feature values in columns.

extract_global_features(sleep_stages=None, aggregate_segments=<function aggregate_segment_features>)[source]

Extract features that characterize the entire recording.

Parameters:

sleep_stages (nnsa.SleepStagesResult) – object containing sleep stages result. Used to report global feature values per sleep stage. If None, no sleep stage dependent feature values are returned. Defaults to None.
aggregate_segments (function, optional) – function that takes an array of segment features as input and returns one aggregate value. Note: this function may be feeded np.nan values. Defaults to nnsa.aggregate_segment_features.

Returns:

global_features (dict) – dictionary containing the feature name and value pairs.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(which)[source]

Plot abs(mean_coh), abs(im_coh) and binary im_coh graph (averaged over segments) in the current figure.

Parameters:: which (str) – which result to plot. Should be one of the keys in self.mean_coh and self.im_coh dicts.

class nnsa.ConfusionMatrix(y_true, y_pred, class_mapping=None)[source]

Bases: object

High-level interface for manipulating the results of the confusion matrix.

Samples with np.nan values in y_true or y_pred are ignored upon initialization of the confusion matrix.

Upon initializing a ConfusionMatrix object, the confusion matrix is computed. Subsequently, by calling methods on the object, all sorts of metrics derived from the confusion matrix can be quickly computed (e.g. accuracy, sensitivity, specificity). This works for binary classification as well as multiclass classification. In the case of multiclass classification, a one-vs-all method is applied if neccessary.

A nice overview of most of thos metrics can be found on Wikipedia: https://en.wikipedia.org/wiki/Confusion_matrix

Parameters:

y_true (np.ndarray) – 1D array with true class numbers 0:N-1, with N the number of classes.
y_pred (np.ndarray) – 1D array with predicted class numbers 0:N-1.
class_mapping (dict, optional) – dictionary that maps a class label to a class number. If None, default class labels are created for the class numbers. Defaults to None.

Raises:

ValueError – if length of y_true and y_pred are not equal.
ValueError – if not all detected class numbers are linked to a label in the specified class_mapping.

Methods:

`accuracy`()	Compute accuracy from the confusion matrix.
`average_accuracy`([class_label])	Compute average accuracy from the confusion matrix (which is fairer than accuracy for imbalanced data sets).
`f1`([class_label])
`false_negatives`([class_label])	Return the number of false negatives.
`false_positives`([class_label])	Return the number of false positives.
`kappa`()	Compute Cohen's kappa, i.e., inter-rater agreement (accuracy), corrected for chance.
`precision`([class_label])	Compute precision from confusion matrix.
`scores_general`()	Return some general evaluation metrics (independent of which class is the positive outcome).
`scores_per_class`()	Return some evaluation metrics per class (considering one class as the positive outcome at a time).
`sensitivity`([class_label])	Compute sensitivity from confusion matrix.
`specificity`([class_label])	Compute specificity from confusion matrix.
`support`([class_label])	Return the number of samples classified as the specified class_label in y_true.
`true_negatives`([class_label])	Return the number of true negatives.
`true_positives`([class_label])	Return the number of true positives.

accuracy()[source]

Compute accuracy from the confusion matrix.

Returns:: (float) – accuracy.

average_accuracy(class_label=None)[source]

Compute average accuracy from the confusion matrix (which is fairer than accuracy for imbalanced data sets).

Returns:: (float) – average accuracy.

f1(class_label=None)[source]

false_negatives(class_label=None)[source]

Return the number of false negatives.

Args:

class_label (str, optional): class label that represents the positive outcome.
If None, the last class is taken

.

Defaults to None.

Returns:: (int): the number of false negatives.

false_positives(class_label=None)[source]

Return the number of false positives.

Parameters:: class_label (str, optional) – class label that represents the positive outcome. If None, the last class is taken. Defaults to None.
Returns:: (int) – the number of false positives.

kappa()[source]

Compute Cohen’s kappa, i.e., inter-rater agreement (accuracy), corrected for chance.

Returns:: (float) – Cohen’s kappa.

precision(class_label=None)[source]

Compute precision from confusion matrix.

Parameters:: class_label (str, optional) – class label that represents the positive outcome. If None, the last class is taken. Defaults to None.
Returns:: (float) – precision.

scores_general()[source]

Return some general evaluation metrics (independent of which class is the positive outcome).

Returns:: scores (dict) – dictionary with scores.

scores_per_class()[source]

Return some evaluation metrics per class (considering one class as the positive outcome at a time).

Returns:: df (pd.DataFrame) – pandas DataFrame with scores.

sensitivity(class_label=None)[source]

Compute sensitivity from confusion matrix.

Parameters:: class_label (str, optional) – class label that represents the positive outcome. If None, the last class is taken. Defaults to None.
Returns:: (float) – sensitivity.

specificity(class_label=None)[source]

Compute specificity from confusion matrix.

Parameters:: class_label (str, optional) – class label that represents the positive outcome. If None, the last class is taken. Defaults to None.
Returns:: (float) – specificity.

support(class_label=None)[source]

Return the number of samples classified as the specified class_label in y_true.

Parameters:: class_label (str, optional) – class label to count the support of. If None, the last class is taken. Defaults to None.
Returns:: (int) – the support.

true_negatives(class_label=None)[source]

Return the number of true negatives.

Parameters:: class_label (str, optional) – class label that represents the positive outcome. If None, the last class is taken. Defaults to None.
Returns:: (int) – the number of true negatives.

true_positives(class_label=None)[source]

Return the number of true positives.

Parameters:: class_label (str, optional) – class label that represents the positive outcome. If None, the last class is taken. Defaults to None.
Returns:: (int) – the number of true positives.

class nnsa.DynamicCoupling(**kwargs)[source]

Bases: ClassWithParameters

Class for computing the dynamic coupling between two signals.

Main mehtod: dynamic_coupling().

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> t = np.linspace(0, 10, 200)
>>> x = np.sin(t)
>>> y = np.sin(t)
>>> dc = DynamicCoupling(segmentation={'segment_length': 50, 'overlap': 40}, method='correlation')
>>> result = dc.dynamic_coupling(x, y, fs=1, verbose=0)
>>> print(type(result).__name__)
DynamicCouplingResult

>>> result.coupling.shape
(16,)

>>> np.mean(result.coupling)
1.0

Methods:

`coherence_dynamic_coupling`(x, y, fs[, ...])	Compute coherence between x and y in a windowed way.
`correlation_dynamic_coupling`(x, y, fs[, ...])	Compute correlation between x and y in a windowed way.
`default_parameters`()	Return the default parameters.
`dynamic_coupling`(args, *kwargs)	Alias for self.process().
`process`(x, y, fs[, label, verbose])	Compute dynamic coupling between x and y.
`tf_dynamic_coupling`(x, y, fs[, f_low, ...])	Compute transfer function gain between x and y in a windowed way.
`wbtf_dynamic_coupling`(x, y, fs[, verbose])	Compute coherence between x and y in a windowed way.

coherence_dynamic_coupling(x, y, fs, f_low=0, f_high=None, verbose=1, window='hann', nperseg=None, noverlap=None, nfft=None, detrend='constant', **kwargs)[source]

Compute coherence between x and y in a windowed way.

Parameters:

x (np.ndarray) – 1D signal array.
y (np.ndarray) – 1D signal array.
fs (float) – sample frequency of x and y.
f_low (float, optional) – frequency specifying the low edge frequency of the band in which to compute the average coherence. Defaults to 0.
f_high (float, optional) – frequency specifying the high edge frequency of the band in which to compute the average coherence. If None, this will be the Nyqvist frequency, i.e. fs/2. Defaults to None.
verbose (int, optional) – verbose level. Defaults to 1.
other (See documentation of scipy.signal.coherence function for) –
**kwargs – for scipy.signal.coherence().

Returns:

coupling (np.ndarray) – 1D array with coupling values.
coupling_surrogates (np.ndarray) – coupling values for each surrogate.

correlation_dynamic_coupling(x, y, fs, delays=None, abs_corr=True, verbose=1)[source]

Compute correlation between x and y in a windowed way.

Parameters:

x (np.ndarray) – 1D signal array.
y (np.ndarray) – 1D signal array.
fs (float) – sample frequency of x and y.
delays (np.ndarray, optional) – delays to try to get maximum correlation. If None, no delay is applied. Defaults to None.
abs_corr (bool, optional) – return the absolute value of the correlation (True) or not (False). Defaults to True.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

coupling (np.ndarray) – 1D array with coupling values.
coupling_surrogates (np.ndarray) – coupling values for each surrogate.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

dynamic_coupling(*args, **kwargs)[source]: Alias for self.process().

process(x, y, fs, label=None, verbose=2)[source]

Compute dynamic coupling between x and y.

Parameters:

x (array-like) – 1D signal array.
y (array-like) – 1D signal array.
fs (float) – sample frequency of x and y (x and y must have equal fs).
label (str, optional) – label for the coupling output (e.g. ‘EEG-NIRS’). If None, the method parameter is used as label. Defaults to None.
verbose (int, optional) – verbose level. Set to 1 for a progress bar, 2 for also printing parameters.

Returns:

(nnsa.DynamicCouplingResult) – object containing the result.

tf_dynamic_coupling(x, y, fs, f_low=0, f_high=None, verbose=1, window='hann', nperseg=None, noverlap=None, nfft=None, detrend='constant', **kwargs)[source]

Compute transfer function gain between x and y in a windowed way.

Parameters:

x (np.ndarray) – 1D signal array.
y (np.ndarray) – 1D signal array.
fs (float) – sample frequency of x and y.
f_low (float, optional) – frequency specifying the low edge frequency of the band in which to compute the average coherence. Defaults to 0.
f_high (float, optional) – frequency specifying the high edge frequency of the band in which to compute the average coherence. If None, this will be the Nyqvist frequency, i.e. fs/2. Defaults to None.
verbose (int, optional) – verbose level. Defaults to 1.
other (See documentation of scipy.signal.csd function for) –
**kwargs – for compute_transfer_function().

Returns:

coupling (np.ndarray) – 1D array with coupling values.
coupling_surrogates (np.ndarray) – coupling values for each surrogate.

wbtf_dynamic_coupling(x, y, fs, verbose=1)[source]

Compute coherence between x and y in a windowed way.

Parameters:

x (np.ndarray) – 1D signal array.
y (np.ndarray) – 1D signal array.
fs (float) – sample frequency of x and y.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

coupling (np.ndarray) – 1D array with coupling values.
coupling_surrogates (np.ndarray) – coupling values for each surrogate.

class nnsa.DynamicCouplingResult(coupling, algorithm_parameters, coupling_surrogates=None, label='coupling', data_info=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing dynamic coupling as computed by nnsa.DynamicCoupling().

Parameters:

coupling (np.ndarray) – 1D array with coupling values with size (n_segments,).
algorithm_parameters (nnsa.Parameters) – see ResultBase.
coupling_surrogates (np.ndarray, optional) – optional surrogate values for the coupling. Can be a 1D array, with shape (n_surrogates,) or a 2D array with shape (n_surrogates, n_segments). If not specified, some methods using surrogate data will raise errors. Defaults to None.
label (str, optional) – label for the coupling output (e.g. ‘EEG-NIRS’). Defaults to ‘coupling’.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Methods:

`get_significance`(**kwargs)	Return a TimeSeries with the significance.
`plot`([time_scale, alpha, ax])	"
`plot_significance`(**kwargs)	Plot the significance (if coupling surrogates is available).
`to_time_series`(**kwargs)	Convert result to TimeSeries object.

Attributes:

num_segments

Return the number of segments.

get_significance(**kwargs)[source]

Return a TimeSeries with the significance.

Significance is defined as the fraction of surrogates that have lower coupling values as the real value. Therefore, if the significance is close to 1, the computed coupling can be deemed significant.

Parameters:: **kwargs (optional) – keyword arguments for TimeSeries.
Returns:: ts (TimeSeries) – TimeSeries object containing the significance.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(time_scale=None, alpha=0.05, ax=None, **kwargs)[source]

” Plot the coupling (and the surrogate coupling values if available) in the current axes.

Parameters:

time_scale (str) – see format_time_axis().
alpha (float) – significance value for surrogates.
ax (plt.Axes) – axes to plot in.
**kwargs (optional) – keyword arguments for plt.plot() for plotting the coupling values.

plot_significance(**kwargs)[source]

Plot the significance (if coupling surrogates is available).

Parameters:: **kwargs (optional) – keyword arguments for TimeSeries.plot().

to_time_series(**kwargs)[source]

Convert result to TimeSeries object.

Parameters:: **kwargs (optional) – keyword arguments for TimeSeries.
Returns:: ts (TimeSeries) – TimeSeries object containing the coupling.

class nnsa.EdfReader(filepath)[source]

Bases: EdfReader

Child of edfreadpy.EdfReader, which includes interfaces to load data into nnsa containers.

Parameters:: filepath (str) – path to the EDF(+) file to read (see edfreadpy.EdfReader).

Methods:

`read_bp_dataset`([begin, end, patterns])	Read all blood-pressure signals into a Dataset object.
`read_eeg_dataset`([channels, begin, end, ...])	Read EEG data from the EDF file as TimeSeries object per EEG channel and return them collectively in an EegDataset object.
`read_rso2_dataset`(**kwargs)	Read all NIRS-related signals into a Dataset object.
`read_spo2_dataset`(**kwargs)	Read all arterial oxygen-related signals into a Dataset object.
`read_temperature`(**kwargs)	Read temperature signal into a TimeSeries object.
`read_time_series`(channel[, begin, end, ...])	Return time series object for given channel.

read_bp_dataset(begin=0, end=None, patterns=None, **kwargs)[source]

Read all blood-pressure signals into a Dataset object.

Parameters:

begin (float, optional) – begin time (in seconds) of the signal to read. By default reading starts at the beginning of the signal (i.e. at 0 seconds).
end (float, optional) – end time (in seconds) of the signal to read. By default reading ends at the end of the signal.
patterns (list) – optional list of patterns to look for in the EDF signal labels. If None, a default list of patterns will be used.
**kwargs (optional) – optional additional keyword arguments for self.read_time_series.

Returns:

(nnsa.BaseDataset or list) – nnsa BaseDataset containing the signals from the EDF file. In case of EDF+D file and output == ‘all’, a list with BaseDataset objects is returned.

read_eeg_dataset(channels=None, begin=0, end=None, exclude_channels=None, discontinuous_mode='longest', verbose=0, **kwargs)[source]

Read EEG data from the EDF file as TimeSeries object per EEG channel and return them collectively in an EegDataset object.

Parameters:

channels (list, optional) – list of signal labels, specifying which channels to read. If None, all EEG channels are read. Default is None.
begin (float, optional) – begin time (in seconds) of the signal to read. By default reading starts at the beginning of the signal (i.e. at 0 seconds).
end (float, optional) – end time (in seconds) of the signal to read. By default reading ends at the end of the signal.
exclude_channels (list, optional) – list of signal labels, specifying which EEG channels not to read. If None, no EEG channels are excluded. Default is None.
discontinuous_mode (str, optional) – how to handle discontinuous data (EDF+D). If ‘longest’, return the longest continuous segment. If ‘all’, return a list of all continuous segments. If ‘fill’ or ‘full’, merge all sements filling up the gaps with nan. If ‘ignore’, the discontinuous signal is returned as if it is continuous. See also self._handle_discontinuous_signal(). Defaults to ‘longest’.
verbose (int) – verbosity level. If 1, a progress bar is shown.
**kwargs (optional) – optional keyword arguments for specifying TimeSeries parameters.

Returns:

(nnsa.EegDataset or list) – nnsa EegDataset containing the EEG signals from the EDF file. In case of EDF+D file and output == ‘all’, a list with EegDataset objects is returned.

read_rso2_dataset(**kwargs)[source]

Read all NIRS-related signals into a Dataset object.

Parameters:: kwargs – optional keyword arguments for self.read_time_series().
Returns:: (nnsa.OxygenDataset) – Dataset object with all NIRS signals.

read_spo2_dataset(**kwargs)[source]

Read all arterial oxygen-related signals into a Dataset object.

Parameters:: kwargs – optional keyword arguments for TimeSeries.
Returns:: (nnsa.OxygenDataset) – Dataset object with all SaO2 signals.

read_temperature(**kwargs)[source]

Read temperature signal into a TimeSeries object.

Raises an error if none or more than 1 temprature signal is found.

Parameters:: kwargs – optional keyword arguments for self.read_time_series().
Returns:: (nnsa.TimeSeries) – TimeSeries with temparature signal.

read_time_series(channel, begin=0, end=None, discontinuous_mode='longest', efficiency='speed', **kwargs)[source]

Return time series object for given channel.

Parameters:

channel (int or str) – the channel specifying the signal to read. May either be an integer specifying the order in which the signal appeard in the EDF file (i.e. index), or a string specifying the signal label.
begin (float, optional) – begin time (in seconds) of the signal to read. By default reading starts at the beginning of the signal (i.e. at 0 seconds).
end (float, optional) – end time (in seconds) of the signal to read. By default reading ends at the end of the signal.
discontinuous_mode (str, optional) – how to handle discontinuous data (EDF+D). If ‘longest’, return the longest continuous segment. If ‘all’, return a list of all continuous segments. If ‘fill’ or ‘full’, merge all sements filling up the gaps with nan. If ‘ignore’, the distontinuous signal is returned as if it is continuous. See also self._handle_discontinuous_signal(). Defaults to ‘longest’.
efficiency (str, optional) – the algorithm to use for reading: ‘speed’ uses an algorithm optimized for speed when loading a large portion of the signal (see _read_digital_data_max_speed), ‘memory’ uses an algorithm that requires the least amount of memory (see _read_digital_data_min_memory). Note that the ‘memory’ option may be faster than the ‘speed’ option when reading only a small part of the signal. However, when reading multiple times from the same file (e.g. read multiple signals), ‘speed’ is probably fastest, even when reading only small parts, since this algorithm stores the raw data of the entire file in memory the first time it’s called.
**kwargs (optional) – optional keyword arguments for the TimeSeries object.

Returns:

(nnsa.TimeSeries or list) – TimeSeries object that holds the (specified part of the) data of the specified channel. In case of EDF+D file and output == ‘all’, a list with TimeSeries objects is returned.

class nnsa.EegDataset(*args, label=None)[source]

Bases: BaseDataset

High-level interface for processing (multichannel) EEG for neonatal signal analysis.

Methods:

`amplitude_eeg`([channel_1, channel_2, verbose])	Compute amplitude-integrated EEG (aEEG) from single bipolar channel continuous EEG.
`as_brain_rt`([bp_filter])	Preprocess the data to make it similar to the view as in BrainRT.
`brain_age_rf`([verbose])	Compute the brain age from the EEG using the random forest approach.
`brain_age_sinc`([verbose])	Compute the brain age from the EEG using the deep Sinc network.
`burst_detection`([create_bipolar_channels, ...])	Perform burst detection on all channels in the dataset.
`clamp`([threshold, channels, inplace, verbose])	Clamp EEG values by applying a clamping function to reduce dynamic range.
`coherence_graph`([preprocess, verbose])	Compute a connectivity graph, where the weights of the edges are based on the coherence between channels.
`compute_features`([verbose, chunk_size])	Compute EEG feature set.
`create_bipolar_channel`(channel_1, channel_2)	Create a bipolar channel by subtracting channel_2 from channel_1, i.e. bipolar channel = channel_1 - channel_2.
`create_bipolar_channels`(channels_1, channels_2)	Create a set of bipolar channels by subtracting channels_2 from channels_1.
`detect_anomalous_channels`([window, ...])	Simple method to detect anomalous high-frequency/amplitude channels by setting a threshold on the line length.
`detect_artefacts`([multi_channel_cnn, verbose])	Helper function to quickly detect artefacts using a combination of methods (see code).
`detect_artefacts_amplitude`(threshold_high[, ...])	Detect locations of artefacts by settings thresholds on the running mean/maximum of the absolute amplitude.
`detect_artefacts_amplitude_kaupilla`([...])	Detect locations of artefacts by settings thresholds on the amplitude as proposed by Kaupilla et al. 2018.
`detect_artefacts_cnn`([preprocess, ...])	Detect locations of artefacts based on clean EEG detection with CNN.
`detect_artefacts_kota`([notch_filter, bp_filter])	Simple method to detect small and large amplitude artefacts.
`detect_artefacts_method`(how, **kwargs)	Shortcut to any of the artefact removal functions.
`detect_artefacts_rfc`([pma, preprocess, verbose])	Detect locations of artefacts based on a sample supervised random forest classifier from the artefact_detection package.
`filter`(*args[, remove_mean])	Calls the parent function, but with remove_mean set to True by default.
`filter_saved_filter`(*args[, remove_mean])	Calls the parent function, but with remove_mean set to True by default.
`filtfilt`(*args[, remove_mean])	Calls the parent function, but with remove_mean set to True by default.
`from_array`(eeg, fs[, channel_labels])	Create an EegDataset from a 2D array (channels x time).
`get_channel`(channel[, make_copy])	Return a channel as TimeSeries.
`get_segments`(segment_length[, overlap, ...])	Segment the EEG data and return the segmented data as a 3D array.
`kirubin_features`([reference_channel, ...])	Compute Kirubin's feature set.
`mean`([channels, label])	Return the nanmean of the specified channels as a TimeSeries.
`median`([channels, label])	Return the nanmedian of the specified channels as a TimeSeries.
`notch_filt`(*args[, remove_mean])	Calls the parent function, but with remove_mean set to True by default.
`plot`([begin, end, channels, color, scale, ...])	Plot the EEG data of specified channels for a specifiec time frame.
`plot_mask`(mask[, time_scale, color, ...])	Plot a shaded mask over an EEG plot.
`power_analysis`([verbose])	Calls the paretn method, but sets some common frequency bands for EEG.
`read_begin_end_time`(filepath)	Read the start and end time (wrt time offset) of the EEG data in a HDF5 file.
`read_edf`(filepath[, begin, end])	Read an EegDataset object from an .edf file.
`read_hdf5`(filepath[, begin, end])	Read an EegDataset object from an .hdf5 file.
`read_mat`(filepath[, begin, end, loadmat_kwargs])	Read an EegDataset object from a .mat file.
`read_pickle`(filepath[, begin, end])	Read an EegDataset object from a .pkl file.
`read_set`(filepath[, begin, end, loadmat_kwargs])	Read an EegDataset object from a .set file (from Matlab EEGLAB).
`reference`([reference_channel, ...])	Reference all EEG signals to reference_channel, i.e. subtract the signal of the reference_channel from all other EEG channels.
`remove_artefact_channels`([inplace])	Remove channels that do not meet the quality criteria.
`remove_artefacts_amplitude`([notch_filter, ...])	Remove artefacts as detected by self.detect_artefacts_amplitude().
`remove_artefacts_derivative`([threshold, inplace])	Remove artefacts based on a threshold on the derivative of the filtered signal.
`remove_artefacts_kota`([notch_filter, ...])	Remove artefacts as detected by self.detect_artefacts_kota().
`remove_artefacts_mask`(mask_ds[, inplace])	Insert nans where boolean mask is True.
`remove_artefacts_method`(how, **kwargs)	Shortcut to any of the artefact removal functions.
`remove_artefacts_rfc`(pma[, preprocess, inplace])	Remove artefacts based on a sample supervised random forest classifier from the artefact_detection package.
`save_csv`(filepath[, overwrite])	Save the EegDataset data to a .csv (comma-separated value) file.
`save_hdf5`(filepath[, mode, overwrite])	Save the EegDataset data to a .hdf5 file.
`save_mat`(filepath[, overwrite])	Save the EegDataset data to a .mat file.
`save_pickle`(filepath[, overwrite])	Save the EegDataset data to a .pkl (pickle) file.
`sleep_stages_cnn`([preprocess, remove_flats, ...])	Automatic sleep stage classification using a Convolutional Neural Network.
`sleep_stages_robust`([verbose])	Automatic robust sleep stage classification.
`substitute_bad_channels`(af_mask[, inplace])	Find segments with bad channels in the EEG and substitute by the mean of the good channels.

Attributes:

signal

Return the EEG data as a numpy array (channels, time).

amplitude_eeg(channel_1='EEG C3', channel_2='EEG C4', verbose=1, **kwargs)[source]

Compute amplitude-integrated EEG (aEEG) from single bipolar channel continuous EEG.

Use on raw EEG data.

Parameters:

channel_1 (str, optional) – the label of the first channel. Defaults to ‘EEG C3’.
channel_2 (str, optional) – the label of the second channel (will be subtracted from the first channel). Defaults to ‘EEG C4’.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the AmplitudeEeg class.

Returns:

result (nnsa.AmplitudeEegResult) – object containing the aEEG.

as_brain_rt(bp_filter=True)[source]

Preprocess the data to make it similar to the view as in BrainRT.

Creates bipolar channels and filter.

Parameters:: bp_filter (bool) – if True, filter the data as in Brain RT (with a first order bandpass Butterworth).
Returns:: eeg_ds (nnsa.EegDataset) – preprocessed dataset (new object).

brain_age_rf(verbose=1, **kwargs)[source]

Compute the brain age from the EEG using the random forest approach.

Use on raw EEG data.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the BrainAge class.

Returns:

result (BrainAgeResult) – object containing the brain age.

brain_age_sinc(verbose=1, **kwargs)[source]

Compute the brain age from the EEG using the deep Sinc network.

Use on raw EEG data.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the BrainAgeSinc class.

Returns:

result (BrainAgeResult) – object containing the brain age.

burst_detection(create_bipolar_channels=True, verbose=1, **kwargs)[source]

Perform burst detection on all channels in the dataset.

This is a wrapper that prepares the input for BurstDetection.burst_detection() and returns the result.

Parameters:

create_bipolar_channels (bool, optional) – if True, automatically create bipolar channels according to the method used. If False, run the algorithm on the channels as currently in the object. Defaults to True.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the BurstDetection class.

Returns:

result (nnsa.BurstDetectionResult) – the result of the burst detection.

clamp(threshold=250, channels=None, inplace=False, verbose=1)[source]

Clamp EEG values by applying a clamping function to reduce dynamic range.

Parameters:

threshold (float) – threshold for the clamping function.
inplace (bool, optional) – whether to apply inplace (True) or not.
verbose (int, optional) – verbosity level.

Returns:

ds_out (EegDataset) – new Dataset object (only returned if inplace is False).

coherence_graph(preprocess=False, verbose=1, **kwargs)[source]

Compute a connectivity graph, where the weights of the edges are based on the coherence between channels.

This is a wrapper that prepares the input for CoherenceGraph.coherence_graph() and returns the result.

Parameters:

preprocess (bool, optional) – apply a default preprocessing routine on the data before feature computation (True) or not (False). Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the CoherenceGraph class.

Returns:

result (nnsa.CoherenceGraphResult) – an object containing the resulting adjacency matrices of the graphs.

compute_features(verbose=1, chunk_size=None, **kwargs)[source]

Compute EEG feature set.

Parameters:

kwargs – for EegFeatures().
verbose (int) – verbosity level.
chunk_size (int) – number of time samples to process at once. If None, a suitable chunk_size is chosen automatically.

Returns:

result (FeatureSetResult) – feature set result.

create_bipolar_channel(channel_1, channel_2, label=None)[source]

Create a bipolar channel by subtracting channel_2 from channel_1, i.e. bipolar channel = channel_1 - channel_2.

Parameters:

channel_1 (str) – the label of the first channel.
channel_2 (str) – the label of the second channel (will be subtracted from the first channel).
label (str, optional) – label for the new bipolar channel. If None, a label will automatically be created from channel_1 and channel_2. Defaults to None.

Returns:

ts_bipolar (nnsa.TimeSeries) – TimeSeries object holding the bipolar channel data.

create_bipolar_channels(channels_1, channels_2, labels=None, missing_mode='error')[source]

Create a set of bipolar channels by subtracting channels_2 from channels_1.

channels_new[i] = channels_1[i] - channels_2[i]

Parameters:

channels_1 (list) – the labels of the first channels.
channels_2 (list) – the labels of the second channels (will be subtracted from the first channel). Must have the same length as channels_1.
labels (str, optional) – label for the new bipolar channels. Must have the same length as channels_1. If None, labels will automatically be created from channels_1 and channels_2. Defaults to None.
missing_mode (str, optional) – what to do if a specified channel is missing from the dataset. Choose from: - ‘error’: raise a ValueError if a channel is specified that is not in the EEG dataset. - ‘ignore’: ignore the requested bipolar channels that could not be created. - ‘warn’: ignore, but display a warning.

Returns:

ds_bipolar (nnsa.EegDataset) – EegDataset object holding the bipolar channels data.

detect_anomalous_channels(window=3, std_factor=8, p_trim=0.25, verbose=0)[source]

Simple method to detect anomalous high-frequency/amplitude channels by setting a threshold on the line length.

The log line length of the (1-p_strip)*n_channels with lowest line lengts is used to determine mean and std and if any channel exceeds mean + `std_factor`*std, then it is flagged as artefact. This is done per window.

Parameters:

window (float) – the window length (seconds) in which to compute line length.
std_factor (float) – the factor for std determining the threshold (mean+`std_factor*std).
p_trim (float) – the relative amount of channels to strip before computing the mean and std. The channels with the highest line lengths are stripped/excluded.
verbose (int) – verbosity level.

detect_artefacts(multi_channel_cnn=True, verbose=1)[source]

Helper function to quickly detect artefacts using a combination of methods (see code).

Apply to raw EEG.

Parameters:

multi_channel_cnn (bool) – whether to use the multi-channel CNN (True) or not (False).
verbose (int) – verbosity level.

Returns:

af_ds (EegDataset) – EegDataset with artefact masks consisting of boolean values: True at locations of artefacts, False otherwise.

detect_artefacts_amplitude(threshold_high, threshold_low=0, window=1, how='max')[source]

Detect locations of artefacts by settings thresholds on the running mean/maximum of the absolute amplitude.

Returns a boolean mask where True values indicate locations where x < threshold_low or x > threshold_high, where x is the moving average/maximum of the absolute signal values in a window of window seconds.

Parameters:

threshold_high (float) – threshold in uV.
threshold_low (float) – threshold in uV.
window (float) – window for the moving average in seconds.
how (str) – whether to take the moving ‘mean’ or ‘max’.

Returns:

af_ds (EegDataset) – EegDataset with artefact masks consisting of boolean values: True at locations of artefacts, False otherwise.

detect_artefacts_amplitude_kaupilla(notch_filter=True, bp_filter=True)[source]

Detect locations of artefacts by settings thresholds on the amplitude as proposed by Kaupilla et al. 2018.

To apply to raw EEG data: set notch_filter and bp_filter to True.

Parameters:

notch_filter (bool) – specify whether the EEG needs to be filtered. Set to True if eeg is raw data.
bp_filter (bool) – specify whether the EEG needs to be filtered. Set to True if eeg is raw data.

Returns:

af_ds (EegDataset) – EegDataset with artefact masks consisting of boolean values: True at locations of artefacts, False otherwise.

detect_artefacts_cnn(preprocess=None, detect_flats=True, detect_peaks=True, verbose=1, **kwargs)[source]

Detect locations of artefacts based on clean EEG detection with CNN.

EEG data must be referenced to Cz. If Cz is in self, referencing will be done automatically. If Cz is not present, referencing will be ignored (assuming the EEG data is already referenced).

Apply to raw EEG data and set preprocess_eeg to True.

Parameters:

preprocess (bool) – specify whether the EEG needs to be preprocessed (filtered, resampled). Set to True if eeg is raw data. If not specified, preprocessing will be done if fs is not as required by the model, otherwise not.
detect_flats (bool) – if True, computes moving std in short windows and if its below a threshold, the sample is marked as artefact (since the CNN might not catch this).
detect_peaks (bool) – if True, computes moving max abs amplitude in short windows and if its above a threshold, the sample is marked as artefact (since the CNN might not catch this).
verbose (int) – verbosity level.
**kwargs (dict) – for CleanDetectorCnn().

Returns:

af_ds (EegDataset) – EegDataset with artefact masks consisting of boolean values: True at locations of artefacts, False otherwise.

detect_artefacts_kota(notch_filter=True, bp_filter=True)[source]

Simple method to detect small and large amplitude artefacts.

Apply to raw data referenced to Cz with notch_filter and bp_filter set to True.

https://doi.org/10.1016/j.pediatrneurol.2021.06.001

detect_artefacts_method(how, **kwargs)[source]: Shortcut to any of the artefact removal functions.

detect_artefacts_rfc(pma=None, preprocess=None, verbose=1, **kwargs)[source]

Detect locations of artefacts based on a sample supervised random forest classifier from the artefact_detection package.

EEG data must be referenced to Cz. If Cz is in self, referencing will be done automatically. If Cz is not present, referencing will be ignored (assuming the EEG data is already referenced).

To apply to raw EEG data: set preprocess_eeg to True.

See also

RfcArtefactDetector, self.detect_artefacts_rfc().

Parameters:

pma (np.ndarray) – PMA of the neonate at time of recording.
preprocess (bool) – specify whether the EEG needs to be preprocessed (filtered, resampled). Set to True if eeg is raw data (but it should still be referenced to Cz). If not specified, preprocessing will be done if fs is not 128, otherwise not.
inplace (bool) – whether to remove in place or not.
**kwargs (dict) – for RfcArtefactDetector().

Returns:

ds_out (EegDataset) – EegDataset where the artefacts are replaced by nans.

save_csv(filepath, overwrite=False, **kwargs)[source]

Save the EegDataset data to a .csv (comma-separated value) file.

This is slow, as it used the pandas functionality.

Parameters:

filepath (str) – filepath to save to.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists. Defaults to False.
**kwargs (dict, optional) – for pandas.DataFrame.to_csv().

save_hdf5(filepath, mode='w', overwrite=False)[source]

Save the EegDataset data to a .hdf5 file.

Parameters:

filepath (str) – filepath to save to.
mode (str, optional) – ‘w’ for write mode or ‘a’ for append mode. Defaults to ‘w’.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists. Defaults to False.

save_mat(filepath, overwrite=False)[source]

Save the EegDataset data to a .mat file.

Parameters:

filepath (str) – filepath to save to.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists. Defaults to False.

save_pickle(filepath, overwrite=False, **kwargs)[source]

Save the EegDataset data to a .pkl (pickle) file.

Parameters:

filepath (str) – filepath to save to.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists. Defaults to False.
**kwargs (dict, optional) – for nnsa.pickle_save().

property signal

Return the EEG data as a numpy array (channels, time).

Returns:: (np.ndarray) – EEG data matrix.

sleep_stages_cnn(preprocess=None, remove_flats=True, remove_baseline_wander=False, stepsize=None, verbose=1, preprocess_kwargs=None, **kwargs)[source]

Automatic sleep stage classification using a Convolutional Neural Network.

This is a wrapper that prepares the input for SleepStagesCnn.sleep_stages_cnn() and returns the result.

Parameters:

preprocess (bool, optional) – if True, the EEG data in ds will be processed by filtering and resampling correspondingly. If False, the data in ds will not be preprocessed (use this when the ds already contains correspondingly prepreocessed data). If None, preprocessing will be done if the sampling frequency of the data does not match the required fs by the model.
remove_flats (bool, optional) – if True, removes flatlines.
remove_baseline_wander (bool) – if True, removes excessive baseline wander by applying a HP filter. This was not part of original preprocessing during training, and most EEGs don’t need this (in the original preprocessing there is already a HP filter, but milder). If not needed, don’t do this additional baseline removal.
stepsize (float) – stepsize (in seconds) for sleep stage prediction. If None, stepsize is equal to segment length (no overlap).
verbose (int, optional) – verbose level (0 or 1). Defaults to 1.
preprocess_kwargs (dict, optional) – dict with optional keyword arguments for SleepStagesCnn.preprocess_recording(). Defaults to None.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the SleepStagesCnn class.

Returns:

result (nnsa.SleepStagesCnnResult) – nnsa object containing the results of the CNN sleep stage classification.

sleep_stages_robust(verbose=1, **kwargs)[source]

Automatic robust sleep stage classification.

This is a wrapper that prepares the input for SleepStagesRobust.process() and returns the result.

Args:

verbose (int, optional): verbose level (0 or 1).
Defaults to 1.

**kwargs (optional): optional keyword arguments to the SleepStagesRobust class.

Returns:: result (nnsa.SleepStagesCnnResult) – nnsa object containing the results.

substitute_bad_channels(af_mask, inplace=False, **kwargs)[source]

Find segments with bad channels in the EEG and substitute by the mean of the good channels.

Parameters:

af_mask (np.ndarray or EegDataset) – boolean mask for the EEG array data with True at locations of artefacts. Shape should be (n_channels, n_time). An incompatible shape will be transposed if that makes it compatible, or raises an error otherwise.
inplace (bool) – whether to apply inplace (True) or not (False).
kwargs – for substitute_bad_channels.

Returns:

ds_out (EegDataset) – cleaned EegDataset with bad channels substituted.

class nnsa.Envelope(**kwargs)[source]

Bases: ClassWithParameters

Class for computing the envelope of a real valued signal.

Main method: envelope().

Parameters:: nnsa.ClassWithParameters (see) –

Examples

>>> fs = 256
>>> t = 1/fs*np.arange(8*100000).reshape(8, 100000)
>>> x = 4 * np.sin(t)
>>> env = Envelope(method='hilbert')
>>> print(type(env.parameters).__name__)
Parameters

>>> result = env.envelope(x, fs=fs, verbose=0)
>>> print(type(result).__name__)
EnvelopeResult

>>> print(result.envelope[2, 2222])
3.98601719157083

>>> print(result.envelope.mean())
3.997973665767074

Methods:

`default_parameters`()	Return the default parameters.
`envelope`(data_matrix, fs[, channel_labels, ...])	Perform the envelope computation on multichannel data.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

envelope(data_matrix, fs, channel_labels=None, verbose=1)[source]

Perform the envelope computation on multichannel data.

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix().
fs (float) – sample frequency of the signals in data_matrix.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.EnvelopeResult) – object containing the signal envelopes per channel.

class nnsa.EnvelopeResult(envelope, algorithm_parameters, fs, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None)[source]

Bases: ResultBase

High-level interface for processing the results of envelope computation as created by nnsa.Envelope().

Parameters:

envelope (np.ndarray) – array with shape (channels, segments) containing signal envelopes.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
fs (float) – sample frequency corresponding to the envelope array.
channel_labels (list of str, optional) – labels of the channels corresponding to the channel dimensions of envelope. If None, default labels will be created. Defaults to None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.

Methods:

`baseline_correction_min`([window_length, inplace])	Perform baseline correction of the envelope by subtracting the minimum value in a window before each sample.
`compute_features`(envs[, postfix, ...])	Compute features of an array of envelope values.
`extract_global_features`([max_amplitude, ...])	Extract features that characterize the entire recording.
`plot`([channel])	Plot the envelope in the current axis.
`to_base_dataset`([unit])	Create an BaseDataset object from the envelope data.
`to_eeg_dataset`([unit])	Create an EegDataset object from the envelope data.
`to_time_series`([channel, unit])	Create a TimeSeries object from the envelope data of the specified channel.

Attributes:

num_segments

Return the number of segments (number of samples).

baseline_correction_min(window_length=None, inplace=False)[source]

Perform baseline correction of the envelope by subtracting the minimum value in a window before each sample.

See nnsa.baseline_correction_min().

Parameters:

window_length (int, optional) – the length of the window (in samples) to look for the baseline (minimum), see nnsa.baseline_correction_min(). If None, a window corresponding to 1 minute is used. Defaults to None.
inplace (bool, optional) – if True, the corrected values replace the envelope values in the current object. If False, a new EnvelopeResult object is returned with the baseline corrected envelope values. Defaults to False.

Returns:

envelope_corrected (nnsa.EnvelopeResult) – a new EnvelopeResult object with the baseline corrected envelope values (only if inplace is True).

compute_features(envs, postfix=None, concatenate_channels=True)[source]

Compute features of an array of envelope values.

Parameters:

envs (np.ndarray) – array with envelope values with dimensions (channels, time).
postfix (str, optional) – postfix for the keys in the output dictionary.
concatenate_channels (bool, optional) – if True, the channels are concatenated before computing the features. If False, the features are computed on each channel. Defaults to True.

Returns:

features (dict) – features describing the envelope distribution.

extract_global_features(max_amplitude=200, sleep_stages=None, concatenate_channels=True)[source]

Extract features that characterize the entire recording.

Parameters:

max_amplitude (float, optional) – the threshold for the envelope in uV. Values above this threshold are discarded in the analysis. Defaults to 200.
sleep_stages (nnsa.SleepStagesResult, optional) – object containing sleep stages result. Used to report global feature values per sleep stage. If None, no sleep stage dependent feature values are returned. Defaults to None.
concatenate_channels (bool, optional) – see self.compute_features.

Returns:

global_features (dict) – dictionary of features values, hashed by feature name.

property num_segments

Return the number of segments (number of samples).

Returns:: (int) – number of segments/samples.

plot(channel=None)[source]

Plot the envelope in the current axis.

Parameters:: channel (str or list, optional) – string or list of strings with label(s) of the channel(s) to plot. If None, all channels are plot in the same figure with a separate color. Defaults to None.

to_base_dataset(unit='uV', **kwargs)[source]

Create an BaseDataset object from the envelope data.

Parameters:

unit (str, optional) – unit of the envelope data. Defaults to ‘uV’.
**kwargs (optional) – optional keyword arguments for the TimeSeries object.

Returns:

ds (nnsa.BaseDataset) – BaseDataset containing the envelope data of the available channels.

to_eeg_dataset(unit='uV', **kwargs)[source]

Create an EegDataset object from the envelope data.

Parameters:

unit (str, optional) – unit of the envelope data. Defaults to ‘uV’.
**kwargs (optional) – optional keyword arguments for the TimeSeries object.

Returns:

ds (nnsa.EegDataset) – EegDataset containing the envelope data of the available channels.

to_time_series(channel=None, unit='uV', **kwargs)[source]

Create a TimeSeries object from the envelope data of the specified channel.

Parameters:

channel (str, optional) – the label of the channel which to convert to a TimeSeries. If None and the envelope contains only one channel, this channel is selected. Defaults to None.
unit (str, optional) – unit of the envelope data. Defaults to ‘uV’.
**kwargs (optional) – optional keyword arguments for the TimeSeries object.

Returns:

ts (nnsa.TimeSeries) – TimeSeries containing the envelope data of the specified channel.

class nnsa.FeatureSetResult(features, feature_labels, algorithm_parameters, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing features sets with multiple features, channels, segments.

Parameters:

features (np.ndarray) – 3D array with feature values per channel per segment (segments, channels, features).
feature_labels (list) – list with labels of the features, corresponding to the rows of features.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
channel_labels (list of str, optional) – labels of the channels corresponding to the second axis in features. If None, default labels will be created. Default is None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Attributes:

num_segments

Return the number of segments.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

class nnsa.FilterBase(tf=None, sos=None, fs=None, **kwargs)[source]

Bases: ClassWithParameters

Base class for a filter object with common utilities.

Child classes inheriting from this class need to implement at least one of _compute_sos() or _compute_tf(). Depending on which of these are implemented, the filtfilt() method will call the appropriate function to do the filtering (see filtfilt()).

Parameters:

fs (float, optional) – frequency of the signal to filter in Hz.
tf (tuple, optional) – tuple of arrays (b, a) containing the transfer function filter coefficients.
sos (tuple, optional) – sos filter coefficients.
**kwargs – see ClassWithParameters.

Methods:

`compute_frequency_response`(**kwargs)	Compute the frequency response of the filter.
`default_parameters`()	Return the default parameters as a dictionary.
`filter`(x[, replace_nan, how])	Filter the signal x by applying a digital filter.
`filtfilt`(x[, replace_nan, how])	Filter the signal x by applying a digital filter forward and backward to a signal for zero-phase filtering.
`plot_frequency_response`([ax, angle])	Plot the frequency response.
`plot_impulse_response`([ax, n])	Plot the impulse response.

Attributes:

`fs`	Sample frequency.
`sos`	Second-order filter coefficients.
`tf`	Transfer function coefficients.

compute_frequency_response(**kwargs)[source]

Compute the frequency response of the filter.

Returns:

w (np.ndarray) – the normalized frequencies at which h was computed, in radians/sample, see also scipy.signal.freqz().
h (np.ndarray) – the frequency response., see also scipy.signal.freqz().
**kwargs – optional keyword arguments for the scipy function.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (dict or Parameters) – a default set of parameters for the object.

filter(x, replace_nan=True, how='mean', **kwargs)[source]

Filter the signal x by applying a digital filter.

This is a wrapper of scipy.signal lfilter and sosfilt functions. See scipy.signal.lfilter() and scipy.signal.sosfilt() for more complete information.

Parameters:

x (np.ndarray) – signal array.
replace_nan (bool, optional) – if True, replaces nans in the input with zeros and places nans in the output at the locations of nans in the input. Especially for IIR filters, this prevents the all subsequent output to become nan if there is a nan sample in the input. Defaults to True.
how (float, string) – how to deal with nans. If a float, this will be the fill value. If ‘mean’, nans will be replaced by the mean value. If ‘interpolate’, nans will be interpolated linearly.
**kwargs (optional) – see scipy.signal.lfilter() and scipy.signal.sosfilt().

Returns:

y (np.ndarray) – the filtered output.

Raises:

NotImplementedError if neither self.sos or self.tf are defined. –

filtfilt(x, replace_nan=True, how='mean', **kwargs)[source]

Filter the signal x by applying a digital filter forward and backward to a signal for zero-phase filtering.

This is a wrapper of scipy.signal filtfilt functions. See scipy.signal.filtfilt() for more complete information.

Parameters:

x (np.ndarray) – signal array.
replace_nan (bool, optional) – if True, replaces nans in the input with zeros and places nans in the output at the locations of nans in the input. Especially for IIR filters, this prevents the all subsequent output to become nan if there is a nan sample in the input. Defaults to True.
how (float, string) – how to deal with nans. If a float, this will be the fill value. If ‘mean’, nans will be replaced by the mean value. If ‘zeros’, nans will be replaced by zeros. If ‘interpolate’, nans will be interpolated linearly.
**kwargs (optional) – see scipy.signal.sosfiltfilt().

Returns:

y (np.ndarray) – the filtered output.

Raises:

NotImplementedError if neither self.sos or self.tf are defined. –

property fs

Sample frequency.

Returns:: self._fs (float) – Sample frequency.
Raises:: AttributeError if self._fs is None. –

plot_frequency_response(ax=None, angle=True, **kwargs)[source]

Plot the frequency response.

Parameters:

ax – matplotlib axis object for plotting. If None, it plots in the current axis. Defaults to None.
angle (bool) – plot the phase angle or not.
**kwargs – optional arguments for self.compute_frequency_response()

plot_impulse_response(ax=None, n=50, **kwargs)[source]

Plot the impulse response.

Parameters:

ax – matplotlib axis object for plotting. If None, it plots in the current axis. Defaults to None.
n (int) – number of samples to plot the response for.
**kwargs – optional arguments for plt.stem().

property sos

Second-order filter coefficients.

Returns:: self._sos (np.ndarray) – see self._compute_sos().

property tf

Transfer function coefficients.

Returns:: self._tf (np.ndarray) – see self._compute_tf().

class nnsa.LineLength(**kwargs)[source]

Bases: ClassWithParameters

Class for computing the Line Length (LL) feature.

Reference: Line Length implemented as presented by Koolen et al.:

@Article{KOOLEN2014,: author = {Ninah Koolen and Katrien Jansen and Jan Vervisch and Vladimir Matic and Maarten De Vos and Gunnar Naulaers and Sabine Van Huffel}, title = {Line length as a robust method to detect high-activity events: Automated burst detection in premature {EEG} recordings}, journal = {Clinical Neurophysiology}, year = {2014}, volume = {125}, number = {10}, pages = {1985–1994}, month = {oct}, doi = {10.1016/j.clinph.2014.02.015}, publisher = {Elsevier {BV}},

}

Main method: line_length():

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(10, 10000)
>>> line_length = LineLength()
>>> print(type(line_length.parameters).__name__)
Parameters

>>> result = line_length.line_length(x, fs=25, verbose=0)
>>> print(type(result).__name__)
LineLengthResult

>>> result.line_length[2, 3]
0.005513363795332106

Methods:

`default_parameters`()	Return the default parameters.
`line_length`(data_matrix, fs[, ...])	Perform the computation of the Line Lenght feature on multichannel data.
`process`(args, *kwargs)	Alias of main method.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

line_length(data_matrix, fs, channel_labels=None, verbose=1)[source]

Perform the computation of the Line Lenght feature on multichannel data.

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix(). Koolen et al. originally used EEG signals bandpass filtered between 1 - 20 Hz.
fs (float) – sample frequency of the EEG signals. For the line length, the signals should typically be sampled at 250 Hz. If another frequency is used, a warning is raised.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.LineLengthResult) – object containing the LL result per segment per channel.

process(*args, **kwargs)[source]: Alias of main method.

class nnsa.LineLengthResult(line_length, algorithm_parameters, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None, **kwargs)[source]

Bases: ResultBase

High-level interface for processing the results of line length computation as created by nnsa.LineLength().

Parameters:

line_length (np.ndarray) – array with shape (channels, segments) containing the (normalized) line lengths.
channel_labels (list of str, optional) – labels of the channels corresponding to the channel dimensions of the arrays. If None, default labels will be created. Defaults to None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.
**kwargs – for ResultBase.

Methods:

`extract_channels`(channels[, inplace])
`extract_global_features`([...])	Extract features that characterize the entire recording.
`histogram_features`(**kwargs)	Compute the histogram features from the log LL distribution.
`plot`(*args[, channels, time_scale, ...])	Plot the line length as function of time (segments).
`to_suppression_curve`([window_length])	Compute the suppression curve, as derived from the line length (Dereymaeker et al. 2015).
`to_time_series`([channel])

Attributes:

`median_line_length`	Retrun the median of the line lenghts across channels.
`num_segments`	Return the number of segments.

extract_channels(channels, inplace=False)[source]

extract_global_features(aggregate_channels=<function aggregate_channel_features>, sleep_stages=None)[source]

Extract features that characterize the entire recording.

Parameters:

aggregate_channels (function or None, optional) – function that takes an array of channel features and returns one aggregate values. If None, the channels are not aggregated, i.e. the feature values each channel are returned per channel. Defaults to nnsa.aggregate_channel_features.
sleep_stages (nnsa.SleepStagesResult, optional) – object containing sleep stages result. Used to report global feature values per sleep stage. If None, no sleep stage dependent feature values are returned. Defaults to None.

Returns:

global_features (dict) – dictionary containing the feature name and value pairs.

histogram_features(**kwargs)[source]

Compute the histogram features from the log LL distribution.

Wrapper of histogram_features_lll() function.

Parameters:: **kwargs (optional) – keyword arguments for histogram_features_lll().
Returns:: see histogram_features_lll().

property median_line_length

Retrun the median of the line lenghts across channels.

Returns:: (np.ndarray) – median line length per segment..

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(*args, channels='median', time_scale='seconds', log_transform=False, ax=None, **kwargs)[source]

Plot the line length as function of time (segments).

Parameters:

args (*) – positional arguments for plt.plot.
channels (list of str, optional) – a list of labels that specify which channels to plot. Each channel gets its own color in the plot. if ‘median’, the median line length across channels is plotted. If None, all channels are plotted. Defaults to ‘median’.
time_scale (str, optional) – the time scale to use. Choose from ‘seconds’, ‘minutes’, ‘hours’. Defaults to ‘seconds’.
log_transform (bool, optional) – if True, do a log10 transform on the LL. Defaults to False.
ax (plt.Axes, optional) – axes to plot in. If None, plots in current axes.
**kwargs (optional) – keyword arguments for plt.plot.

to_suppression_curve(window_length=150, **kwargs)[source]

Compute the suppression curve, as derived from the line length (Dereymaeker et al. 2015).

Parameters:

window_length (float) – the window length in seconds for the computation of the suppression curve.
**kwargs (optional) – optional keyword arguments for SuppressionCurve().

Returns:

suppression (np.array) – suppression values, with a sample frequency 1/window_length.

to_time_series(channel='median', **kwargs)[source]

class nnsa.MovingAverage(tf=None, sos=None, fs=None, **kwargs)[source]

Bases: FilterBase

Create a moving average (smoothening) filter.

Parameters:: FilterBase. (see) –

Examples

>>> np.random.seed(43)
>>> N = 1000
>>> mean = 4.3
>>> x = mean + (np.random.rand(N) - 0.5)*abs(mean)
>>> filt = MovingAverage(numtaps=200)
>>> y = filt.filter(x)
# Original signal at idx 500.
>>> x[500]
5.322134095439112
# Moving averaged signal at idx 500.
>>> y[500]
4.340021494627845

Methods:

default_parameters()

Return the default parameters.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

class nnsa.MultiScaleEntropy(**kwargs)[source]

Bases: ClassWithParameters

Class for performing a multi-scale entropy analysis.

Reference: this code was ported from Mario Lavanga and Ofelie de Wel’s MATLAB code.

Main method: multi_scale_entropy()

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(2, 2500)
>>> mse = MultiScaleEntropy()
>>> assert_equal(mse.parameters, MultiScaleEntropy.default_parameters())
>>> print(type(mse.parameters).__name__)
Parameters

>>> mse_result = mse.multi_scale_entropy(x, fs=25, verbose=0)
>>> print(type(mse_result).__name__)
MultiScaleEntropyResult

>>> mse_result.mse[10, 1, 0]
1.0185695809945732

Methods:

`default_parameters`()	Return the default parameters as a dictionary.
`multi_scale_entropy`(data_matrix, fs[, ...])	Perform the multi-scale entropy (MSE) analysis on multichannel data.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

multi_scale_entropy(data_matrix, fs, channel_labels=None, verbose=1)[source]

Perform the multi-scale entropy (MSE) analysis on multichannel data.

Analysis pipeline ported from MATLAB code designed by Ofelie De Wel and Mario Lavanga.

Pipeline constsist of 4 steps: 1) Segmentation 2) Optional filtering 3) Artefact exclusion 4) MSE computation

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.MultiScaleEntropyResult) – object containing the MSE result per segment and per channel.

class nnsa.MultiScaleEntropyResult(mse, algorithm_parameters, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing the results of multi-scale entropy analysis as created by nnsa.MultiScaleEntropy().

Parameters:

mse (np.ndarray) – array with multi-scale entropy values with dimensions (scales, channels, segments). mse[i] contains the result of scale i+1.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
channel_labels (list of str, optional) – labels of the channels corresponding to the second axis in mse. If None, default labels will be created. Default is None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Attributes:

`average_mse`	Return the mse averaged over all segments.
`num_segments`	Return the number of segments.
`scales`	Return the scales of the mse analysis.

Methods:

`compute_average_slopes`([split])	Compute the average slope of the mse curve in the small and large scales per channel.
`compute_complexity_index`()	Compute the complexity index, i.e. the area under the mse curve, for each segment and each channel.
`extract_channels`(channels[, inplace])
`extract_global_features`(**kwargs)	Extract features that characterize the entire recording.
`maximum_mse`()	Return the maximum value of the multiscale entropy curve.
`plot`([channels, segments, plot_fit, ax])	Average the mse values for all segments and plot the entropy against scale for the specified channels.
`remove_channel`(channels[, inplace])	Remove one or more channels.

property average_mse

Return the mse averaged over all segments.

Returns:

self._average_mse (np.ndarray) – array with dimensions (scales, channels) containing the average sample
entropy values.

compute_average_slopes(split=5)[source]

Compute the average slope of the mse curve in the small and large scales per channel.

Parameters:

split (int, optional) – the scale that divides the scale into small scales (<=split) and large scales (>split). Defaults to 5.

Returns:

slope_small (np.ndarray) – array with dimensions (channels, segments) containing the average slope of the mse curve of scales 1 until split.
slope_large (np.ndarray) – array with dimensions (channels, segments) containing the average slope of the mse curve of scales split until the maximum available scale.

compute_complexity_index()[source]

Compute the complexity index, i.e. the area under the mse curve, for each segment and each channel.

Returns:: (np.ndarray) – array with dimensions (channels, segments) with the complexity indices.

extract_channels(channels, inplace=False)[source]

extract_global_features(**kwargs)[source]

Extract features that characterize the entire recording.

Parameters:: **kwargs (optional) – optional keyword arguments for local_to_global_features().
Returns:: global_features (dict) – see local_to_global_features().

maximum_mse()[source]

Return the maximum value of the multiscale entropy curve.

Returns:: (np.ndarray) – array with dimensions (channels, segments) with the maxima.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(channels=None, segments=None, plot_fit=False, ax=None)[source]

Average the mse values for all segments and plot the entropy against scale for the specified channels.

Parameters:

channels (list of str, optional) – a list of labels that specify which channels to plot. Each channel gets its own color in the plot. If None, all channels are plotted. Defaults to None.
segments (list, optional) – iterable with indices of the segments to plot. Each segment is plotted as a separate solid line of a color corresponding to the channel. If None, the average sample entropy values over all segments are plotted per scale. Defaults to None.
plot_fit (bool, optional) – if True, plot the polynomial approximation. Defaults to False.
ax (plt.Axes, optional) – axes to plot in. If None, plots in a new figure. Defaults to None.

remove_channel(channels, inplace=False)[source]

Remove one or more channels.

Parameters:

channels (str or list) – (list of) channel name(s) to remove.
inplace (bool, optional) – bool tos pecify whether it happens in place or not.

Returns:

new MultiScaleEntropyResult object if inplace is False.

property scales

Return the scales of the mse analysis.

Returns:: (np.ndarray) – array with scales corresponding to first axis of self.mse.

class nnsa.MultifractalAnalysis(eng=None, **kwargs)[source]

Bases: ClassWithParameters

Class for performing a multifractal analysis.

References

This code was partly ported from Mario Lavanga and Ofelie de Wel’s MATLAB code and calls some MATLAB functions from the Wavelet p-Leader and Bootstrap based MultiFractal analysis (WLBMF) MATLAB toolbox, described in: H. Wendt, P.Abry, and S.Jaffard, “Bootstrap for Multifractal Analysis” IEEE SignalProcessingMagazine, vol. 24, no. 4, pp. 38–48, 2007.

Main method: multifractal_analysis()

Notes

The following project claims to have implemented the MATLAB toolbox in Python (I did not check it out): https://github.com/omardrwch/mfanalysis/blob/master/mfanalysis/mfa.py

Parameters:

eng (matlab.engine.matlabengine.MatlabEngine, optional) – MATLAB engine to use for calling MATLAB functions. The required paths must already have been initialized (see self.eng()). If None, a new MATLAB engine will be started (which takes some time). Defaults to None.
**kwargs (optional) – see nnsa.ClassWithParameters.

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(2, 2500)
>>> mfa = MultifractalAnalysis()
>>> print(type(mfa.parameters).__name__)
Parameters

>>> mfa_result = mfa.multifractal_analysis(x, fs=25, verbose=0)
Starting MATLAB engine...
Setting up the path...
>>> print(type(mfa_result).__name__)
MultifractalAnalysisResult

>>> mfa_result.cp[0, 0, 0]
0.5654653145233981

Methods:

`default_parameters`()	Return the default parameters.
`mfa_time_series`(time_series, fs, fs_rr, j1, ...)	Compute multifractality features of a time series.
`multifractal_analysis`(data_matrix, fs[, ...])	Perform the multifractal (MF) analysis on multichannel data.

Attributes:

eng

Return a MATLAB engine.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

property eng

Return a MATLAB engine.

Returns:: (matlab.engine.matlabengine.MatlabEngine) – MATLAB engine.

mfa_time_series(time_series, fs, fs_rr, j1, j2, nq)[source]

Compute multifractality features of a time series.

Parameters:

time_series (np.ndarray) – 1D array with time series data to analyze.
fs (int) – sample frequency of the data in time_series.
fs_rr – (int): resampling frequency for the power in the delta frequency band.
j1 (int) – smallest scale analysis.
j2 (int) – largest scale analysis.
nq (int) – number of (positive) q values that will be used (see build_q_log()).

Returns:

zq (np.ndarray) – estimates of the partition function zeta(q).
hq (np.ndarray) – estimates of Hurst exponents h(q).
dq (np.ndarray) – estimates of singularity/multifractal spectrum D(q).
cp (np.ndarray) – estimates of the coefficients c_p from the Taylor expansion that estimate tau(q).
Element i-1 in the cp array corresponds to c_i.
hmin (float) – Hurst exponent at the left of the spectrum (at q=-5).
hmax (float) – Hurst exponent at the right of the spectrum (at q=5).

multifractal_analysis(data_matrix, fs, channel_labels=None, verbose=1)[source]

Perform the multifractal (MF) analysis on multichannel data.

Analysis pipeline ported from MATLAB code designed by Ofelie De Wel and Mario Lavanga.

Pipeline constsist of 4 steps: 1) Segmentation 2) Optional filtering 3) Artefact exclusion 4) MF feature computation

Reference:: Lavanga M., De Wel O., Caicedo A., Heremans E., Jansen K., Dereymaeker A., Naulaers G., Van Huffel S. (2017), Automatic quiet sleep detection based on multifractality in preterm neonates: effects of maturation. Proc. of the 39th Annual International Conference of the IEEE Engineering in Medicine and Biology Society of the IEEE (EMBC 2017) Jeju Island, South Korea, Jul. 2017, pp. 2010-2013

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.MultifractalAnalysisResult) – object containing the MF result per segment and per channel.

class nnsa.MultifractalAnalysisResult(cp, hmin, hmax, algorithm_parameters, q_functions=None, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing the results of multifractal analysis as created by nnsa.MultifractalAnalysis().

Parameters:

cp (np.ndarray) – estimates of the coefficients c_p from the Taylor expansion that estimate tau(q) with dimensions (p, channels, segments). Element i-1 in the cp array corresponds to c_i.
hmin (np.ndarray) – Hurst exponent at the left of the spectrum (at q=-5) with dimensions (channels, segments).
hmax (np.ndarray) – Hurst exponent at the right of the spectrum (at q=5) with dimensions (channels, segments).
algorithm_parameters (nnsa.Parameters) – see ResultBase.
q_functions (dict, optional) – dict with functions of q (e.g. ‘zq’, ‘hq’, ‘Dq’). Each value in the dict is an array with dimensions (q, channels, segments). Each array in the dict must have this same shape.
channel_labels (list of str, optional) – labels of the channels corresponding to the channel dimensions of the arrays. If None, default labels will be created. Default is None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Methods:

`extract_features`()	Return standard set of features extracted from the result of the MFA.
`extract_global_features`(**kwargs)	Extract features that characterize the entire recording.
`plot`([ax])	Plot the result.

Attributes:

`num_segments`	Return the number of segments.
`q`	Return array with q values.

extract_features()[source]

Return standard set of features extracted from the result of the MFA.

Returns:: (dict) – dictionary with features. Features are arrays with dimension (channels, segments).

extract_global_features(**kwargs)[source]

Extract features that characterize the entire recording.

Parameters:: **kwargs (optional) – optional keyword arguments for local_to_global_features().
Returns:: global_features (dict) – see local_to_global_features().

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(ax=None)[source]

Plot the result.

Parameters:: ax –

Returns:

property q

Return array with q values.

Returns:: (np.ndarray) – array with q values.

class nnsa.NotchIIR(tf=None, sos=None, fs=None, **kwargs)[source]

Bases: FilterBase

Create an IIR notch (bandstop) filter.

Parameters:: FilterBase. (see) –

Examples

>>> fs = 2000
>>> t = np.linspace(0, 1.0, fs+1)
>>> xlow = np.sin(2 * np.pi * 5 * t)  # 5 Hz
>>> xhigh = np.sin(2 * np.pi * 250 * t)  # 250 Hz
>>> x = xlow + xhigh
>>> filt = NotchIIR(fs=fs, f0=250)
>>> y = filt.filtfilt(x)
>>> print(y[10])
0.5425567000090112

Methods:

default_parameters()

Return the default parameters.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

class nnsa.Parameters[source]

Bases: dict

Class for collecting parameters.

Inherits from dict, so functionality is mostly identical to dict. Customized/added methods of Parameters are:

__repr__: returns a more readable representation of the set. add: mimics the update method of dict to add new entries. update: only accepts existing keys.

Examples

>>> d = {'a': 1, 'b': {'c': {'d': 1}, 'e': 1}}

>>> p = Parameters(**d)
>>> print(p)
Parameters object with:
       a: 1
       b: {'c': {'d': 1}, 'e': 1}

>>> p.update({'b': {'e': 2}})
>>> print(p)
Parameters object with:
       a: 1
       b: {'c': {'d': 1}, 'e': 2}

>>> p.update({'c': 2})
Traceback (most recent call last):
    ...
KeyError: "'c' not in collection with keys ['a', 'b']."

>>> p['c'] = 3

>>> p.add({'x': 3, 'y': {'z': 3}})

>>> dict(p)
{'a': 1, 'b': {'c': {'d': 1}, 'e': 2}, 'c': 3, 'x': 3, 'y': {'z': 3}}

Methods:

`add`([other])	Add or update keys, value pairs in the Parameters set (in place).
`update`([other])	Update key, value pairs in the Parameter set, accepting only existing keys.

add(other=None, **kwargs)[source]

Add or update keys, value pairs in the Parameters set (in place).

Parameters:

other (dict or iterable, optional) – dictionary or iterable of key, value pairs to add to the Parameters set.
**kwargs (optional) – keyword arguments to add (in which the keyword is the key).

update(other=None, **kwargs)[source]

Update key, value pairs in the Parameter set, accepting only existing keys.

Updates nested dictionaries/Parameters objects up to arbitrary levels. In the deeper levels, non-existing keys are accepted if they are from an ordinary dict. In contrast, updating a non-existent key in a deeper Parameters object will raise an error.

Parameters:

other (dict or iterable, optional) – dictionary or iterable of key, value pairs with which to update.
**kwargs (optional) – keyword arguments with which to update (in which the keyword is the key).

Raises:

KeyError – if key to update not in Parameter set.

class nnsa.PowerAnalysis(**kwargs)[source]

Bases: ClassWithParameters

Class for performing a power analysis of EEG signal.

References: this code was ported from Mario Lavanga and Ofelie de Wel’s MATLAB code.

Main method: power_analysis()

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(10, 10000)
>>> power_analysis = PowerAnalysis()
>>> assert_equal(power_analysis.parameters, PowerAnalysis.default_parameters())
>>> print(type(power_analysis.parameters).__name__)
Parameters

>>> power_analysis_result = power_analysis.power_analysis(x, fs=25, verbose=0)
>>> print(type(power_analysis_result).__name__)
PowerAnalysisResult

>>> power_analysis_result.power['delta_1'][2, 3]
0.01603280237682399

Methods:

`default_parameters`()	Return the default parameters.
`power_analysis`(data_matrix, fs[, ...])	Perform the power analysis on multichannel data as designed by Mario Lavanga and Ofelie De Wel.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

power_analysis(data_matrix, fs, channel_labels=None, unit=None, verbose=1)[source]

Perform the power analysis on multichannel data as designed by Mario Lavanga and Ofelie De Wel.

Code was ported from MATLAB and constsist of 5 steps: 1) Segmentation 2) Optional filtering 3) Artefact exclusion 4) PSD estimation 5) Computation of frequency power in specified frequency bands per segment

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix().
fs (float) – sample frequency of the EEG signals.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
unit (str, optional) – the physical unit of the signals in data_matrix (e.g. ‘uV’). If None no unit is passed to the returned object (PowerAnalysisResult). Defaults to None.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.PowerAnalysisResult) – object containing the powers of the frequency bands per segment and per channel.

class nnsa.PowerAnalysisResult(power, algorithm_parameters, channel_labels=None, data_info=None, unit=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing the results of a power analysis as created by nnsa.PowerAnalysis().

Parameters:

power (dict) – dictionary with frequency band labels as keys and (channels, segments) arrays with power as values, e.g. as returned by nnsa.PowerAnalysis.power_analysis()
algorithm_parameters (nnsa.Parameters) – see ResultBase.
channel_labels (list of str, optional) – labels of the channels corresponding to the rows of the values in power. If None, default labels will be created. Default is None.
data_info (str, optional) – see ResultBase.
unit (str, optional) – unit of the values in power.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Methods:

`compute_relative_power`()	Compute the relative power per channel per segment for each band.
`extract_channels`(channels[, inplace])
`extract_global_features`(**kwargs)	Extract features that characterize the entire recording.
`extract_histogram_features`([ignore_nan])	Compute the histogram features of each feature attribute.
`plot`([channels, frequency_bands, legend, ax])	Plot the power analysis.
`summary`([cell_width, n_decimals, ...])	Print a summary of the power analysis.

Attributes:

num_segments

Return the number of segments.

compute_relative_power()[source]

Compute the relative power per channel per segment for each band.

Returns:: rel_power (dict) – relative power per band.

extract_channels(channels, inplace=False)[source]

extract_global_features(**kwargs)[source]

Extract features that characterize the entire recording.

Parameters:: **kwargs (optional) – optional keyword arguments for local_to_global_features().
Returns:: global_features (dict) – see local_to_global_features().

extract_histogram_features(ignore_nan=True)[source]

Compute the histogram features of each feature attribute.

Parameters:: ignore_nan (bool, optional) – if True, ignore nan values. Defaults to True.
Returns:: out (dict) – dict with feature names as keys and feature values as values.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(channels=None, frequency_bands=None, legend=True, ax=None)[source]

Plot the power analysis.

Creates a subplot for each frequency band and plots the corresponding power of all (specified) channels.

Parameters:

channels (list of str, optional) – a list of labels that specify which channels to plot. If None, all channels are plotted. Defaults to None.
frequency_bands (list of str, optional) – a list of labels that specify which frequency bands to plot. If None, all bands are plotted. Defaults to None.
legend (bool, optional) – whether to plot the legend or not. Defaults to True.

Returns:

fig – Figure handle.
ax (list) – list with axes handles for each subplot.

summary(cell_width=10, n_decimals=2, frequency_bands=None)[source]

Print a summary of the power analysis.

Parameters:

cell_width (int, optional) – width of a cell in the table that is printed.
n_decimals (int, optional) – number of decimals to print.
frequency_bands (list, optional) – a list with labels for the frequency bands to print the summary of.

class nnsa.Psd(**kwargs)[source]

Bases: ClassWithParameters

Class for estimating the power spectral density of a signal using Welch’s method.

Main method: psd()

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(10000)
>>> psd = Psd()
>>> assert_equal(psd.parameters, Psd.default_parameters())
>>> print(type(psd.parameters).__name__)
Parameters

>>> psd_result = psd.psd(x, fs=250)
>>> print(type(psd_result).__name__)
PsdResult

>>> print(psd_result.power_density)
[7.20648796e-01 1.43280205e+00 1.40759024e+00 ... 2.83669937e-04
 2.81437262e-04 1.40343941e-04]

Methods:

`default_parameters`()	Return the default parameters as a dictionary.
`process`(args, *kwargs)	Alias for self.psd().
`psd`(x, fs[, unit, axis, verbose])	Estimate the power spectral density using Welch's method.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

process(*args, **kwargs)[source]: Alias for self.psd().

psd(x, fs, unit='a.u.', axis=-1, verbose=0, **kwargs)[source]

Estimate the power spectral density using Welch’s method.

Parameters:

x (np.ndarray) – signal data array.
fs (float) – sample frequency of the signal in Hz.
unit (str, optional) – the unit of the signal values. Defaults to ‘a.u.’
axis (int, optional) – axis along which the periodogram is computed. Defaults to -1.
verbose (int, optional) – verbose level. Defaults to 0.
**kwargs (optional) – for scipy.signal.welch().

Returns:

(nnsa.feature_extraction.PsdResult) – object containing the power spectral density estimate.

class nnsa.PsdResult(power_density, df, algorithm_parameters, data_info=None, unit=None)[source]

Bases: ResultBase

High-level interface for power spectral density results as created by nnsa.Psd().psd().

Parameters:

power_density (np.ndarray) – power spectral density for frequencies 0, df, df*2, …, df*(len(power_density)-1).
df (float) – frequency resolution of the power spectral density.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
data_info (str, optional) – see ResultBase.
unit (str, optional) – the unit corresponding to power.

Methods:

`compute_power`(f_low, f_high)	Compute power of frequency band spanned by f_low and f_high by numerically integrating the psd over this domain.
`normalize`([how, inplace])	Normalize the PSD result.
`plot`(*args[, in_db, cumulative, ...])	Plot the power spectrum density.

Attributes:

`frequencies`	Return the frequencies corresponding to the powers in power_density.
`num_segments`	Return the number of segments/samples.

compute_power(f_low, f_high)[source]

Compute power of frequency band spanned by f_low and f_high by numerically integrating the psd over this domain.

Parameters:

f_low (float) – lower limit of the frequency band.
f_high (float) – upper limit of the frequency band.

Returns:

power (float) – the frequency power spanned by f_low and f_high.

property frequencies

Return the frequencies corresponding to the powers in power_density.

Returns:: (np.ndarray) – array of same length as self.power_denisty containing the corresponding frequencies.

normalize(how='max', inplace=False)[source]

Normalize the PSD result.

Parameters:

how (str) – how to normalize. Choose from: - “max”: noramlize such that maximum power is 1.
inplace (bool) – whether to apply inplace or not.

Returns:

psd (PsdResult) – normalized Psd (only returend if inplace is False).

property num_segments

Return the number of segments/samples.

Returns:: (int) – number of segments/samples.

plot(*args, in_db=True, cumulative=False, title_postfix=None, ax=None, **kwargs)[source]

Plot the power spectrum density.

Parameters:

*args (optional) – optional positional arguments for plt.semilogy function.
in_db (bool) – whether to plot the PSD in decibels (True) or just power (False).
title_postfix (str, optional) – if not None, this will be added to the plot title. Defaults to None.
cumulative (bool) – if True plots the cumulative power by integrating over frequency.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in a new figure. Defaults to None.
**kwargs (optional) – optional keyword arguments for plt.semilogy function.

class nnsa.RemezFIR(tf=None, sos=None, fs=None, **kwargs)[source]

Bases: FilterBase

Design a minimax optimal filter using the Remez exchange algorithm.

Parameters:: FilterBase. (see) –

Examples

>>> fs = 2000
>>> t = np.linspace(0, 1.0, fs+1)
>>> xlow = np.sin(2 * np.pi * 5 * t)  # 5 Hz
>>> xhigh = np.sin(2 * np.pi * 250 * t)  # 250 Hz
>>> x = xlow + xhigh
>>> filt = RemezFIR(fs=fs, passband=120, stopband=140)
>>> y = filt.filtfilt(x)
>>> print(np.abs(y - xlow).max())
0.05586442486553467

Methods:

default_parameters()

Return the default parameters.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

class nnsa.ResultBase(algorithm_parameters=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None, time_offset=0)[source]

Bases: object

Base class for high-level interface for the results of feature extraction algortihms.

For each class in feature_extraction, a corresponding Result class is made to manipulate/visualize the results.

Parameters:

algorithm_parameters (nnsa.Parameters) – Parameters object of the corresponding feature_extraction class.
data_info (str, optional) – optional string with information about the data (e.g. source file, preprocessing).
segment_start_times (np.array, optional) – start times (in seconds) for the segments. If None, it is assumed that the segments form a continuous series and start at 0 s. The segment times are then derived from the segmentation parameters (if present in algorithm parameters, see self._get_segment_start_times()). Defaults to None.
segment_end_times (np.array, optional) – end times (in seconds) for the segments. If None, the segment times are derived from segment_start_times and the segment length parameter in the segmentation parameters (if present in algorithm parameters, see self._get_segment_end_times()). Defaults to None.
fs (float, optional) – sample frequency corresponding to the segments axis, i.e. 1/segment length. In some cases, the feature extracted is on the same time scale as the original data. In that case, it makes sense to pass a sample frequency instead of segment start and end times. If fs is given while segment_start_times is not and no segmentation parameters are found, the time array corresponding to the segments axis will be derived from fs, assuming the samples are continuous and starting at 0 s. Defaults to None.
time_offset (float, optional) – optional time offset in seconds for the time arrays. Useful to set the define the time array together with fs. This offset will also be added to the segment start times, even if the segment times are defined explicitly. Defaults to 0.

Notes

For the time array corresponding to the segments (if applicable), either specify

fs and time_offset (if the datapoints are equidstant, continuous and starting at time_offset); or
segment_start_times and fs (if the datapoints are not equidistant, but do correspond to the same length, (fs=1/segment_length); or
segment_start_times and segment_end_times (in all other cases. NOTE: do not set segment_start_times == segment_end_times, since this will be interpreted as a discontinuous signal, see self.is_discontinuous()).

Attributes:

`algorithm_parameters`	Return the parameters of the corresponding feature extraction algortihm.
`fs`	Return the sample freqeuncy of the segments.
`num_segments`	Return the number of segments/samples.
`segment_end_times`	Return the end times of the segments in seconds (starting at 0 s).
`segment_length`	Return the segment length in seconds (if the segment lengths is consistent/fixed).
`segment_start_times`	Return the start times of the segments in seconds (starting at 0 s).
`segment_times`	Compute the segments time array corresponding to the segment dimensions of the feature array.
`time`	Time array starting containing start times of the segments.
`time_offset`	Return the time offset in seconds for the time array.

Methods:

`compute_global_features`(**kwargs)	Compute global features.
`extract_epoch`([begin, end, inplace])	Extract part of the data that falls within begin - end.
`is_discontinuous`()	Check if the result is continuous in time.
`merge`(other[, inplace])	Merge other Result objects into one object.
`read_file`(filepath[, verbose])
`save_to_file`(filepath[, overwrite, verbose])	Save ResultBase-derived object to a file.
`to_file`(args, *kwargs)	Shortcut to save_to_file().

property algorithm_parameters

Return the parameters of the corresponding feature extraction algortihm.

Returns:: (nnsa.Parameters) – Parameters object of the corresponding feature_extraction class.

compute_global_features(**kwargs)[source]

Compute global features.

Parameters:: **kwargs (optional) – keyword arguments for pd.DataFrame.
Returns:: df (pd.DataFrame) – dataframe with one row, and feature values in columns.

extract_epoch(begin=None, end=None, inplace=False)[source]

Extract part of the data that falls within begin - end.

Parameters:

begin (float, optional) – begin time of epoch to extract in seconds. Defaults to 0.
end (float, optional) – end time of epoch to extract in seconds. If None, the extracted epoch contains all availabel data from begin on. Defaults to None.
inplace (bool, optional) – if True, extract the epoch inplace by removing data that falls outside the epoch in the current object. If False, a new object is returned, leaving the original one unchanged. Defaults to False.

Returns:

(ResultBase-derived) – new Result object containing only the extracted epoch (if inplace is False).

property fs

Return the sample freqeuncy of the segments.

Returns:: (float) – sample frequency (Hz).

is_discontinuous()[source]

Check if the result is continuous in time. I.e., the time/segment array does not include jumps in time.

Returns:: (bool) – True is the result is discontinuous. False if not.

merge(other, inplace=False)[source]

Merge other Result objects into one object.

A warning is given if the algorithm parameters of the to be merged results are not the same. The algorithm parameters and data info of the current object are taken.

Parameters:

other (list) – list with Result objects (all of the same type).
inplace (bool, optional) – if True, merges the data inplace by adding it to the current object. If False, a new object is returned, leaving the original ones unchanged. Defaults to False.

Returns:

(ResultBase-derived) – new Result object containing the merged result (if inplace is False).

property num_segments

Return the number of segments/samples.

Returns:: (int) – number of segments/samples.

static read_file(filepath, verbose=1)[source]

save_to_file(filepath, overwrite=False, verbose=1)[source]

Save ResultBase-derived object to a file.

Automatically selects the corresponding writer function for the detected file type. The created file can later be read with the function read_result_from_file().

Parameters:

filepath (str) – path to the file to which to save the ResultBase-derived object to. If no file extension is included in the filepath, the .hdf5 will be added by default and the result will be saved to an hdf5 file. To save as a different file, specify a compatible file extension (see SUPPORTED_RESULT_FILE_TYPES).
overwrite (bool) – if True, overwrites exisiting file if filepath exists. If False, raises an error if filepath already exists. Defaults to False.
verbose (int, optional) – verbose level.

property segment_end_times

Return the end times of the segments in seconds (starting at 0 s).

Returns:: (np.array) – end times (in seconds) for the segments.

property segment_length

Return the segment length in seconds (if the segment lengths is consistent/fixed).

Raises a ValueError if the segment length is not constant.

Returns:: segment_length (float) – segment length (in seconds).

property segment_start_times

Return the start times of the segments in seconds (starting at 0 s).

Returns:: (np.array) – start times (in seconds) for the segments.

property segment_times

Compute the segments time array corresponding to the segment dimensions of the feature array.

Returns:: segment_times (np.ndarray) – time (in seconds) array for the axis corresponding to segments.

property time

Time array starting containing start times of the segments.

Shortcut when result is sampled with self.fs.

Returns:: self.segment_start_times

property time_offset

Return the time offset in seconds for the time array.

Returns:: (float) – time offset in seconds.

to_file(*args, **kwargs)[source]: Shortcut to save_to_file().

class nnsa.SideObsp(order=None, regularizer='smooth', gammas=None)[source]

Bases: object

Class implementing SIgnal DEcomposition based on Oblique Subspace Projections.

Parameters:

order (int, optional) – order of the moving average model (number of taps). This is parameter p in Eq. 12. The order is equal for all inputs. This parameter can be set manually, or found automatically using cross-validation with the fit_hyper() method. Defaults to None.
regularizer (str or None, optional) – regulizer to use for the linear model. See LinReg() for options. If None, no regularization is applied. Defaults to ‘smooth’.
gammas (np.ndarray, optional) – regularization constant for each input. This is parameter gamma in Eq. 10. This parameter can be set manually, or found automatically using cross-validation with the fit_hyper() method. Not that the length of gammas must be equal to the number of input signals that will be used for the decomposition. Is not needed if regularizer is None. Defaults to None.

References

A. Caicedo, C. Varon, B. Hunyadi, M. Papademetriou, I. Tachtsidis, and S. V. Huffel, “Decomposition of Near-Infrared Spectroscopy Signals Using Oblique Subspace Projections: Applications in Brain Hemodynamic Monitoring,” Frontiers in Physiology, vol. 7, Nov. 2016, doi: 10.3389/fphys.2016.00515.

Examples

>>> n = 1024
>>> signal = generate_timeseries(n)
>>> noise = generate_timeseries(n)
>>> X = [signal, noise]
>>> y = signal + 2*noise + np.random.rand(n)
>>> y_hat = SideObsp(regularizer='smooth').fit_hyper(X, y, orders_all=[1, 2, 3, 4, 5]).fit_transform(X, y)
>>> y_signal, y_noise = y_hat[:, 0], y_hat[:, 1]
>>> y_denoise = y - y_noise

Methods:

`fit`(X, y[, k])	Fit the coefficient vector(s) h (impulse responses) for the kth input(s).
`fit_hyper`(X, y[, orders_all, gammas_all, ...])	Fit hyper parameters (order, gammas) using a grid search and k-fold cross validation.
`fit_transform`(X, y[, k])	Shortcut to do a fit and transform the data in one go.
`transform`(X[, k])	Decompose the output y into contributions for each input signal in X using fitted impulse responses.

Attributes:

`gammas`	Return the regularization constant for each input.
`h`	Return the impulse responses for each input (the impulses are flipped).
`order`	Return the model order.

fit(X, y, k=None)[source]

Fit the coefficient vector(s) h (impulse responses) for the kth input(s).

Parameters:

X (list or np.ndarray) – regressor matrix with shape (n_samples, n_inputs). Can also be a list with n_inputs arrays of length n_samples.
y (list or np.ndarray) – the signal to decompose. Must be a list or 1D array with length n_samples.
k (int or list, optional) – index specifying for which signals in X to to find the impulse responses that can be used to estimate the linear contribution of that signal to the output. If None, the impulse responses are computed for each input. Defaults to None.

Returns:

self – the object itself, but with fitted coefficients.

fit_hyper(X, y, orders_all=None, gammas_all=None, k_folds=10, verbose=2)[source]

Fit hyper parameters (order, gammas) using a grid search and k-fold cross validation.

Parameters:

X – see self.fit().
y – see self.fit().
orders_all (list or np.ndarray, optional) – list or 1D array with orders to try. The order is parameter p in Eq. 12. If None, a list with default values is used if self.order is None, otherwise the order will be kept fixed to the current value. Defaults to None.
gammas_all (list or np.ndarray, optional) – list or 1D array with regularization constants to try. The regularization constant is parameter gamma (lower-case) in Eq. 10. If None, a list with default values is used if self.gammas is None, otherwise gammas will be kept fixed to the current value(s). Defaults to None.
k_folds (int, optional) – the number of folds for k-fold cross-validation. Defaults to 10.
verbose (int) – verbosity level. If 1, shows a progress bar. If 2, also prints the results.

Returns:

self – the object itself, but with fitted hyper parameters.

fit_transform(X, y, k=None)[source]: Shortcut to do a fit and transform the data in one go.

property gammas: Return the regularization constant for each input.

property h

Return the impulse responses for each input (the impulses are flipped).

Returns:: (np.ndarray) – array with shape (order, num_inputs).

property order: Return the model order.

transform(X, k=None)[source]

Decompose the output y into contributions for each input signal in X using fitted impulse responses.

This function can only be called after the model has been fit. It will raise an error otherwise.

Parameters:

X – see self.fit().
k – see self.fit().

Returns:

y_hat (np.ndarray) – the estimated linear contribution of each requested input regressor on the output y. Has shape (n_samples, len(k)).

class nnsa.SignalStats(**kwargs)[source]

Bases: ClassWithParameters

Class for computing a set of statistical parameters of a signal.

Main method: signal_stats().

Parameters:: nnsa.ClassWithParameters (see) –

Examples

>>> fs = 256
>>> np.random.seed(43)
>>> x = np.random.normal(loc=5, scale=10, size=(8, fs*300))
>>> ss = SignalStats(artefact_criteria=None)
>>> print(type(ss.parameters).__name__)
Parameters

>>> result = ss.signal_stats(x, fs=fs, verbose=0)
>>> print(type(result).__name__)
SignalStatsResult

# Some stats of 3nd channel (index 2), 5th segment (index 4). >>> print(result.stats[‘std’][2, 4]) 9.96795466696633 >>> print(result.stats[‘skewness’][2, 4]) 0.02042382947932932 >>> print(result.stats[‘kurtosis’][2, 4]) 0.028582535331379777

# Average std across all segments and all channels. >>> print(result.stats[‘std’].mean()) 10.00292371308106

Methods:

`default_parameters`()	Return the default parameters as a dictionary.
`signal_stats`(data_matrix, fs[, ...])	Compute signal stats on multichannel data.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

signal_stats(data_matrix, fs, channel_labels=None, verbose=1)[source]

Compute signal stats on multichannel data.

Analysis pipeline ported from MATLAB code designed by Ofelie De Wel and Mario Lavanga.

Pipeline constsist of 4 steps: 1) Segmentation 2) Optional filtering 3) Artefact exclusion 4) Stats computation

Parameters:

data_matrix (np.ndarray) – see check_multichannel_data_matrix().
fs (float) – sample frequency of the signals.
channel_labels (list of str, optional) – see check_multichannel_data_matrix().
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.SignalStatsResult) – object containing statistics per segment per channel.

class nnsa.SignalStatsResult(stats, algorithm_parameters, channel_labels=None, data_info=None, segment_start_times=None, segment_end_times=None, fs=None)[source]

Bases: ResultBase

High-level interface for processing signal statistics as computed by nnsa.SignalStats().

Parameters:

stats (dict) – dict where each entry is some statistic computed per channel per segment. The entries are matrices with size (n_channels, n_segments).
algorithm_parameters (nnsa.Parameters) – see ResultBase.
channel_labels (list of str, optional) – labels of the channels corresponding to the rows of the values in mse. If None, default labels will be created. Default is None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
fs (flaot, optional) – see ResultBase.

Attributes:

num_segments

Return the number of segments.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

class nnsa.SleepStages(**kwargs)[source]

Bases: ClassWithParameters

Sleep stage classification from text annotations.

main method: sleep_stages()

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> ss = SleepStages()
>>> assert_equal(ss.parameters, SleepStages.default_parameters())
>>> print(type(ss.parameters).__name__)
Parameters

>>> annotation_set = AnnotationSet([Annotation(0, 100, 'QS TA'), Annotation(139.2, 60.8, 'AS1')])
>>> ss_result = ss.sleep_stages(annotation_set)
>>> print(type(ss_result).__name__)
SleepStagesResult

>>> print(ss_result.proportion('QS'))
0.5

Methods:

`annotation_conversions`(annotation_set)	Convert each unique annotation in the annotation set to a standard annotation text and a sleep label.
`default_parameters`()	Return the default parameters as a dictionary.
`sleep_stages`(annotation_set[, verbose])	Infer the course of the sleep stages from the annotations.

annotation_conversions(annotation_set)[source]

Convert each unique annotation in the annotation set to a standard annotation text and a sleep label.

Can be used to see how individual annotations are converted to sleep labels.

Parameters:: annotation_set (nnsa.AnnotationSet) – text annotations that need to be converted to sleep stages.
Returns:: (pd.DataFrame) – DataFrame with columns ‘original_text’, ‘standard_text’ and ‘sleep_label’.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the sleep stages.

sleep_stages(annotation_set, verbose=1)[source]

Infer the course of the sleep stages from the annotations.

Sort the annotations based on onset.
Standardize the annotations, i.e., find a standard annotation text for each annotation.

3) Convert the standard texts to sleep stage labels (controlled by class_mapping and include_labels_with_artefacts parameters). 4) Fill unlabeled periods using no_label annotations. 5) Merge successive annotations if their label is the same. 6) Interpolate/replace artefact segments (controlled by interpolate_no_label_kwargs parameter). 7) Remove short sleep stages (controlled by remove_short_stages_kwargs parameter).

Parameters:

annotation_set (nnsa.AnnotationSet) – text annotations that need to be converted to sleep stages.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.SleepStagesResult) – SleepStagesResult object containing the annotations related to sleep stages.

class nnsa.SleepStagesCnn(num_classes=2, detect_novelties=True, **kwargs)[source]

Bases: ClassWithParameters

Sleep stage classification using a Convolutional Neural Network.

References: A. H. Ansari et al., “A convolutional neural network outperforming state-of-the-art sleep staging algorithms for both preterm and term infants” Journal of Neural Engineering, vol. 17, no. 1, p. 16028, Jan. 2020, doi: 10.1088/1741-2552/ab5469.

main method: sleep_stages_cnn()

Note

Tested with keras version 2.2.4 and tensorflow version 1.13.1 and with python version 3.6.

Parameters:

num_classes (int) – number of classes in sleep stage classification (2-class (2), or 4-class (4)).
**kwargs (optional) – optional keyword arguments for nnsa.ClassWithParameters.

Examples

>>> np.random.seed(0)
>>> x = np.random.rand(20, 900, 8)*400 - 200
>>> cnn = SleepStagesCnn()
>>> assert_equal(cnn.parameters, SleepStagesCnn.default_parameters())
>>> print(type(cnn.parameters).__name__)
Parameters

>>> cnn_result = cnn.sleep_stages_cnn(x, verbose=0)
>>> print(type(cnn_result).__name__)
SleepStagesCnnResult

>>> print(cnn_result.probabilities[:, 0])
[0.04391909 0.95608091]

Attributes:

`data_requirements`	Return a dictionary with requirements for the input of the CNN.
`keras_model`	Return the model.
`keras_model_activations`	Return the model for intermediate activations.
`keras_model_latent`	Return the model for latent features.
`model_dir`	Return the directory containing model files.
`model_path`	Return the filpath of the model.
`norm_param_path`	Return the filpath with the normalization parameters.
`normalization_parameters`	Return the normalization parameters.

Methods:

`default_parameters`()	Return the default parameters as a dictionary.
`predict_activations`(x[, verbose])
`preprocess_recording`(raw_eeg, raw_fs[, verbose])	Run the preprocessing routine (filtering, resampling) on an entire recording.
`process`(eeg, fs[, verbose])	Process raw EEG data (channel order should match self.data_requirements['channel_order']).
`sleep_stages_cnn`(x[, segment_start_times, ...])	Apply the sleep stage classification algortihm to the preprocessed (filtered/resampled/segmented) input x: 1) normalize input x, 2) predict the CNN output, 3) postprocess the output.

property data_requirements

Return a dictionary with requirements for the input of the CNN.

Returns:: data_requirements (dict) – a dictionary with requirements for the input of the CNN (e.g., channel order, fs, segment length).

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the model.

property keras_model

Return the model.

Returns:: (nnsa.model or keras.model) – model object.

property keras_model_activations

Return the model for intermediate activations.

Returns:: (nnsa.model or keras.model) – model object.

property keras_model_latent

Return the model for latent features.

Returns:: (nnsa.model or keras.model) – model object.

property model_dir

Return the directory containing model files.

Returns:: model_dir (str) – path to directory with model files.

property model_path

Return the filpath of the model.

Returns:: model_path (str) – filepath of the model.

property norm_param_path

Return the filpath with the normalization parameters.

Returns:: norm_param_path (str) – filepath of the pickle file with the normalization parameters. The file contains 2 arrays. The first array contains the means per channel and the second the stds per channel for normalization (see self._load_normalization_parameters).

property normalization_parameters

Return the normalization parameters.

Returns:: (tuple) – normalization parameters.

predict_activations(x, verbose=1)[source]

preprocess_recording(raw_eeg, raw_fs, verbose=1)[source]

Run the preprocessing routine (filtering, resampling) on an entire recording.

Parameters:

raw_eeg (np.ndarray) – raw EEG data to wct with shape (time, channels).
raw_fs (float) – sample frequency of raw EEG data.
verbose – (int): verbosity level.

Returns:

eeg (np.ndarray) – preprocessed EEG data.
fs (float) – sample frequency of preprocessed EEG data.

process(eeg, fs, verbose=1, **kwargs)[source]

Process raw EEG data (channel order should match self.data_requirements[‘channel_order’]).

Parameters:

eeg (np.ndarray) – 2D array with shape (time, channels) with raw (unfiltered) EEG data. The data should be referenced as required and in the channels should have the order as required in self.data_requirements.
fs (float) – sampling frequency of eeg (Hz).
verbose (int) – verbosity level.

sleep_stages_cnn(x, segment_start_times=None, segment_end_times=None, verbose=1)[source]

Apply the sleep stage classification algortihm to the preprocessed (filtered/resampled/segmented) input x: 1) normalize input x, 2) predict the CNN output, 3) postprocess the output.

Parameters:

x (np.ndarray) – (unnormalized) input array with shape corresponding to (segments, time, channels) with preprocessed (filtered and resampled) EEG data. Note that the time and channels dimension depend on the shape of the input layer in the CNN model. E.g. for the default model, shape of x must be (num_segments, 900, 8), corresponding to 30 second segments sampled at 30 Hz and 8 mono-polar EEG channels with reference Cz in the following order: Fp1, Fp2, C3, C4, T3, T4, O1, O2. The array may also be 4 dimensional with shape (segments, time, channels, 1), which is the shape for the CNN model. If x does not have this fourth dimension, it will be added by this preprocessing.
verbose (int, optional) – verbose level (0 or 1). Default to 1.

Returns:

(SleepStagesCnnResult) – nnsa object containing the results of the CNN sleep stage classification.

class nnsa.SleepStagesCnnResult(probabilities, class_labels, fs, algorithm_parameters, is_novelty=None, novelty_score=None, latent_features=None, data_info=None, segment_start_times=None, segment_end_times=None, time_offset=0)[source]

Bases: ResultBase

High-level interface for manipulating sleep stage probabilities as predicted by nnsa.SleepStagesCnn().sleep_stages_cnn().

Also note the method .to_sleep_stages_result(), which returns a more general object SleepStagesResult that provides a high level interface for manipulating sleep stages/labels in general. SleepStagesCnnResult class differs from SleepStagesResult, in that this class contains the result of a classifier, and provides methods to analyze the outcome of the classifier (e.g. compute score metrics, investigate probabilities etc.), whereas the SleepStagesResult class focusses on the analysis of the sleep stages (which may either be defined automatically or manually by a clinician).

Parameters:

probabilities (np.ndarray) – probabilties for a number of segments per class with shape (num_classes, num_segments).
class_labels (list of str) – class labels corresponding to the rows of self.probabilities.
fs (float) – 1/segment length in Hz.
algorithm_parameters (nnsa.Parameters) – see ResultBase.
is_novelty (np.ndarray, None) – optional boolean array with shape (n_segments,) indicating which segmentes were classified as novelties (out-of-data).
is_novelty – optional array with shape (n_segments,) containing novelty scores for each segment.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.

Attributes:

`class_mapping`	Return class mapping (=dictionary that maps a class label to a class number).
`df`
`num_segments`	Return the number of segments.
`y_pred`	Return the class predictions.

Methods:

`compute_global_features`(**kwargs)	Compute global features.
`confusion_matrix`(y_true)	Return a ConfusionMatrix object conatining the confusion matrix.
`convert_classes`(target_class_labels)	Convert to different, derivative classes.
`get_classes`([threshold, return_probs])	Convert the probabilities to class numbers.
`get_naive_labels`([threshold])	Return the sleep label for each class.
`get_probabilities`(class_label)	Return the classification probabilities per segment for a given class_label.
`get_segment_indices`(class_label)	Return the indices of the segments classified as class_label.
`plot`(args, *kwargs)
`plot_probabilities`([class_label, ...])	Plot the probility of the segments belonging to the secified class.
`plot_probability`(args, *kwargs)
`remove_artefacts`(af_mask[, fs_mask, t_mask, ...])	Remove artefacts by setting the probabilities of segment during which too many samples were artefacts to nan.
`robust_sleep_labels`(aci_per_chan, amp[, ...])	Robust sleep prediction by introducing more classes, like artefacts, movement, novelty, uncertain, transitional.
`to_annotation_set`([threshold])	Convert the CNN output to an AnnotationSet with the sleep labels.
`to_annotation_set_per_segment`([threshold])	Convert the CNN output to an AnnotationSet with the sleep labels creating one Annotation per segment.
`to_dataframe`(args, *kwargs)	Collect relevant results in a pandas dataframe where each row is a segment.
`to_df`(args, *kwargs)
`to_frame`(args, *kwargs)
`to_sleep_stages_result`([threshold, verbose])	Convert the probabilities to sleep stages and return the sleep stages as a SleepStagesResult object.

property class_mapping

Return class mapping (=dictionary that maps a class label to a class number).

Returns:: (dict) – dictionary that maps a class label to a class number.

compute_global_features(**kwargs)[source]

Compute global features.

Parameters:: **kwargs (optional) – keyword arguments for SleepStagesResult.compute_global_features().
Returns:: df (pd.DataFrame) – dataframe with one row, and feature values in columns.

confusion_matrix(y_true)[source]

Return a ConfusionMatrix object conatining the confusion matrix.

Parameters:: y_true (np.ndarray) – 1D array with true class numbers.
Returns:: confusion_matrix (nnsa.ConfusionMatrix) – object containing the confusion matrix.

convert_classes(target_class_labels)[source]

Convert to different, derivative classes.

Parameters:: target_class_labels (list) – list with new class labels.
Returns:: ss_out (SleepStagesCnn) – new object with probabilities corresponding to the target classes.

property df

get_classes(threshold=None, return_probs=False)[source]

Convert the probabilities to class numbers.

Parameters:

threshold (float, optional) – threshold of probability. If probability for the maximum class is lower than the threshold, the class number of that segment will be np.nan. If None, the clas with the maximum probability will be taken. Defaults to None.
return_probs (bool) – return probabilities as well (True).

Returns:

classes (np.ndarray) – 1D array with same length as self.probabilities with integers representing the classes as defined in self.class_labels, i.e. class i corresponds to self.class_labels[i].
return_probs (np.ndarray) – arrray with same length as classes with the maximum probabilities per segment.

get_naive_labels(threshold=None)[source]

Return the sleep label for each class.

Parameters:: threshold (float, optional) – threshold of probability. If probability for the maximum class is lower than the threshold, the class number of that segment will be np.nan. If None, the clas with the maximum probability will be taken. Defaults to None.
Returns:: labels (list) – list with same length as self.probabilities with the sleep labels (str).

get_probabilities(class_label)[source]

Return the classification probabilities per segment for a given class_label.

Parameters:: class_label (str) – the label of the class to get the probabilities from.
Returns:: (np.ndarray) – probabilities of each segment for the class_label class.

get_segment_indices(class_label)[source]

Return the indices of the segments classified as class_label.

Parameters:: class_label (str) – the label of the class to get the segment indices from.
Returns:: (np.ndarray) – indices of segments classified as class_label.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(*args, **kwargs)[source]

plot_probabilities(class_label=None, class_number=None, time_scale='hours', ax=None, novelty_color='C3', *args, **kwargs)[source]

Plot the probility of the segments belonging to the secified class.

Parameters:

class_label (str) – the label of the class to plot the probabilities of.
novelty_color – color for highlighting novelties (if available). If None, novelties are not highlighted.

plot_probability(*args, **kwargs)[source]

remove_artefacts(af_mask, fs_mask=None, t_mask=None, max_af_samples_frac=0.5, max_af_channels_frac=0.125, inplace=False)[source]

Remove artefacts by setting the probabilities of segment during which too many samples were artefacts to nan.

Parameters:

af_mask (np.ndarray) – array with shape (n_channels, n_samples).
fs_mask (float) – optional sampling frequency of af_mask. If you do not specify this, specify t_mask.
t_mask (np.ndarray) – array with shape (n_samples) containing the time vector for af_mask.
max_af_samples_frac (float) – if the fraction of artefacts in a channel segment is larger than this, the channel is considered artefact for that segment.
max_af_channels_frac (float) – if the fraction of artefacted channels is larger than this. the segment is considered artefact and the corresponding values in self.probabilities will be set to nan.
inplace (bool) – whether to remove artefacts inplcae (True) or not (False). If not inplace, returns the new object, otherwise does not return anything.

Returns:

ss_out (SleepStagesCnnResult) – copy of the object, but with artefacts set to nan (only if inplace is False).

robust_sleep_labels(aci_per_chan, amp, add_naive=False)[source]

Robust sleep prediction by introducing more classes, like artefacts, movement, novelty, uncertain, transitional.

Parameters:

aci_per_chan (np.ndarray) – 2D array with shape (n_segments, n_channels) containing the artefact contamination index (%) for each segment.
amp (np.ndarray) – 2D array with shape (n_segments, n_channels) containing amplitudes of each segment and channel.
add_naive (bool) – if True or ‘all’ adds the naive/raw sleep label to the robust label in brackets. If ‘missing’: adds the naive/raw sleep label to the robust label only for missing sleep stages. If None or False, does not include the naive label.

Returns:

robust_sleep_labels (list) – list with length n_segments containing robust labels.

to_annotation_set(threshold=None)[source]

Convert the CNN output to an AnnotationSet with the sleep labels.

Parameters:: threshold (float, optional) – theshold for probability, see self.get_classes(). Defaults to None.
Returns:: annotation_set (nnsa.AnnotationSet) – object containing the sleep stages as annotations.

to_annotation_set_per_segment(threshold=None)[source]

Convert the CNN output to an AnnotationSet with the sleep labels creating one Annotation per segment.

Parameters:: threshold (float, optional) – theshold for probability, see self.get_classes(). If None, takes class with maximum probability.
Returns:: annotation_set (nnsa.AnnotationSet) – object containing the sleep stages as annotations.

to_dataframe(*args, **kwargs)[source]

Collect relevant results in a pandas dataframe where each row is a segment.

Parameters:

*args –
**kwargs –

Returns:

df (pd.DataFrame) – pandas DataFrame with segment-wise results.

to_df(*args, **kwargs)[source]

to_frame(*args, **kwargs)[source]

to_sleep_stages_result(threshold=None, verbose=1, **kwargs)[source]

Convert the probabilities to sleep stages and return the sleep stages as a SleepStagesResult object.

This is a wrapper that prepares the input for SleepStages.sleep_stages() and returns the result.

Parameters:

threshold (float, optional) – theshold for probability, see self.get_classes(). Defaults to None.
verbose (int, optional) – verbosity level for creation of sleep stages.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the SleepStages class.

Returns:

result (nnsa.SleepStagesResult) – SleepStagesResult object containing annotations related to sleep stages.

property y_pred

Return the class predictions.

Returns:: (np.ndarray) – array with integer values representing classes.

class nnsa.SleepStagesResult(annotation_set, algorithm_parameters=None, class_mapping=None, data_info=None)[source]

Bases: ResultBase

High-level interface for manipulating sleep stage annotations as created by nnsa.SleepStages().sleep_stages().

Parameters:

annotation_set (nnsa.AnnotationSet) – AnnotationSet consisting of sleep labels. All labels must have a duration. Each unique label is considered a class.
algorithm_parameters (nnsa.Parameters, optional) – see ResultBase.
class_mapping (dict, optional) – dictionary that maps a class label (annotation) to a class number. If None, a default mapping is applied with class numbers ranging from 0 to N-1, with N the number of unique class labels. Defaults to None.
data_info (str, optional) – see ResultBase.

Attributes:

`annotation_set`
`class_labels`
`class_mapping`
`class_mapping_invert`	Invert class mapping.
`num_segments`	Return the number of segments/samples.
`total_duration`	Return the total duration of the annotated period in seconds.

Methods:

`class_numbers_to_labels`(class_numbers)	Convert class numbers to class labels.
`compute_global_features`(**kwargs)	Compute global features.
`count_annotations_per_class`([default_value])	Count the number of annotations per class.
`create_mask`(class_label, query_times[, ...])	Return a boolean mask for (any of the) class_label on the query time points.
`cycle_count`(class_label[, ignore_labels])	Count the number of times a sleep stage transitioned into a different one and back.
`extract_cleanest_epoch`(epoch_length[, dt, ...])	Extract the cleanest epoch (a window of epoch_length with least 'NL' coverage).
`extract_epoch`([begin, end, inplace])	Extract epoch specified by begin and end.
`extract_features`(class_label)	Extract some features from the sleep stage annotations.
`extract_global_features`([postfix])	Extract features that characterize the entire recording.
`get_classes`()	Convert the annotation labels to class numbers and return them as an array.
`inter_class_intervals`(class_label[, ...])	Compute the intervals between a specific sleep stages, specified by class_label.
`interpolate_artefacts`([artefact_texts, ...])	Interpolate artefact labels if they are short enough and surrounded by the same non-artefact label.
`interpolate_classes`(query_times[, kind])	Interpolate class numbers to the time points specified by query_times.
`mean_std_duration_per_class`([default_value])	Compute mean and standard deviation of the duration of each sleep stage class.
`median_duration`(class_label)	Return the median duration of the class_label annotations.
`median_iqr_duration_per_class`([default_value])	Compute median and inter quartile range of the duration of each sleep stage class.
`pieplot`(**kwargs)
`plot`(*args[, offset, time_scale, fill, ax])	Quick plot of the sleep stages as function of time.
`plot_hypnogram`([time_scale, order, ...])	Plot the hypnogram (sleep stages as function of time).
`proportion`(class_label[, ignore_labels, ...])	Return the proportion of segments classified as class_label.
`remove_corrupted_stages`([corrupt_labels, ...])	Remove the sleep stages that are considered corrupted or incomplete.
`remove_short_stages`([min_duration, inplace])	Remove sleep stages that last too short by replacing the labels by no_label (NL).
`segment_labels`(segment_start_times, ...[, ...])	Return an array of labels corresponding to segments with specified start and end times.
`to_dataframe`()	Return a dataframe of self.annotation_set.
`total_duration_per_class`([default_value])	Compute the total duration of each sleep stage class.

property annotation_set

property class_labels

property class_mapping

property class_mapping_invert

Invert class mapping.

Returns:: mapping (dict) – inverted class mapping.

class_numbers_to_labels(class_numbers)[source]

Convert class numbers to class labels.

Nans are convert to no_label labels.

Parameters:: class_numbers (list, np.ndarray) – class numbers to convert.
Returns:: labels (list) – class labels.

compute_global_features(**kwargs)[source]

Compute global features.

Parameters:: **kwargs (optional) – keyword arguments for pd.DataFrame.
Returns:: df (pd.DataFrame) – dataframe with one row, and feature values in columns.

count_annotations_per_class(default_value=0)[source]

Count the number of annotations per class.

Parameters:: default_value (float, optional) – default value if no annotations of a class are present. Set this e.g. to np.nan or zero. Defaults to 0.
Returns:: count_dict (dict) – dictionary with the counts per class.

create_mask(class_label, query_times, check_class_label=True, no_label_to_nan=False)[source]

Return a boolean mask for (any of the) class_label on the query time points.

Parameters:

class_label (str or list) – the label(s) of the class to get the mask from. If a list, True is returned if the label at the query point is in class_label.
query_times (np.ndarray) – query time points for the mask in seconds.
check_class_label (bool, optional) – check if the class labels that are passed exist in self. If check_class_label is True and a label does not occur in self, an error is raised. Defaults to True.
no_label_to_nan (bool, optional) – returns a mask with nans where the label is ‘no_label’. If False, returns False at locations where the sleep label is nan. Defaults to False.

Returns:

mask (np.ndarray) – boolean array with same shape as time specifying whether the sleep stage equals the class_label at the corresponding time point.

cycle_count(class_label, ignore_labels=None)[source]

Count the number of times a sleep stage transitioned into a different one and back.

Ignores no_label.

Assumes the annotations are sorted by onset.

Parameters:

class_label (str) – the label of the sleep stage for which to compute the cycle_count.
ignore_labels (str or list, optional) – specify a class label or list of class labels to ignore. These labels won’t be considered as valid transitions (e.g. artefacts, no_label). no_label segments will automatically be ignored. If None, a default list of labels to ignore will be used. Defaults to None.

Returns:

(int) – number of interruptions.

extract_cleanest_epoch(epoch_length, dt=1, inplace=False)[source]

Extract the cleanest epoch (a window of epoch_length with least ‘NL’ coverage).

Parameters:

epoch_length (float) – length of the epoch to extract (in seconds).
dt (float) – stepsize when searching for the cleanest epoch. Decrease for more accuracy, at the cost of computational time.
inplace (bool) – whether to extract inplace or not.

Returns:

(SleepStagesResult) – extracted epoch, if inplace is False.

extract_epoch(begin=None, end=None, inplace=False)[source]

Extract epoch specified by begin and end.

Parameters:

begin (float) – begin time of epoch (seconds).
end (float) – end time of epoch (seconds).
inplace (bool) – whether to apply the operation inplace (True) or not (False).

Returns:

sleep_stages_result (SleepStagesResult) – new object with extracted epoch (if inplace if False).

extract_features(class_label)[source]

Extract some features from the sleep stage annotations.

Sorts the annotations (inplace).

Parameters:: class_label (str) – the class label to extract features from.
Returns:: (dict) – dictionary with features.

extract_global_features(postfix=None)[source]

Extract features that characterize the entire recording.

Parameters:: postfix (str, optional) – postfix for the keys in the output dictionary. If None, no postfix is added. Defaults to None.
Returns:: (dict) – dictionary containing the feature name and value pairs.

get_classes()[source]

Convert the annotation labels to class numbers and return them as an array.

Returns:: (np.ndarray) – array with the class numbers corrsponding to the sleep stage annotations.

inter_class_intervals(class_label, ignore_labels=None)[source]

Compute the intervals between a specific sleep stages, specified by class_label.

Assumes the annotations are sorted by onset.

Parameters:

class_label (str) – the label of the sleep stage for which to compute the intervals.
ignore_labels (str or list, optional) – specify a class label or list of class labels to ignore. These labels won’t be considered as valid intervals (e.g. artefacts, no_label). If None, ignore nothing. Defaults to None.

Returns:

(np.ndarray) – array with all the intervals that were found.

interpolate_artefacts(artefact_texts=None, max_duration=180, inplace=False, **kwargs)[source]

Interpolate artefact labels if they are short enough and surrounded by the same non-artefact label.

See also

AnnotationSet.interpolate_artefacts().

Parameters:

artefact_texts (str or list, optional) – specify a class label or list of class labels to interpolate. These labels will count as artefact. If None, a default list of corrupt labels will be used. Defaults to None.
max_duration (float) – maximal duration (seconds) of the artefact label to replace them.
inplace (bool) – whether to do this inplace or not.

Returns:

sleep_stages_result (SleepStagesResult) – new object with sleep stages (if inplace is False).

interpolate_classes(query_times, kind='previous')[source]

Interpolate class numbers to the time points specified by query_times.

Parameters:

query_times (np.ndarray) – query time points to interpolate the annotations to.
kind (str, optional) – how to interpolate, see interpolate.interp1d(). Defaults to ‘previous’. This takes the value of the last known sample, which is typically the correct way to interpolate sleep stage annotations.

Returns:

interpolated_classes (np.ndarray) – array with interpolated class numbers at query points.

mean_std_duration_per_class(default_value=nan)[source]

Compute mean and standard deviation of the duration of each sleep stage class.

Parameters:

default_value (float, optional) – default value if no annotations of a class are present. Set this e.g. to np.nan or zero. Defaults to np.nan.

Returns:

out_dict_means (dict) – defaultdict with of mean duration per class.
out_dict_stds (dict) – defaultdict with std of the durations per class.

median_duration(class_label)[source]

Return the median duration of the class_label annotations.

Parameters:: class_label (str) – class label.
Returns:: median_duration (float) – median duration of class_label.

median_iqr_duration_per_class(default_value=nan)[source]

Compute median and inter quartile range of the duration of each sleep stage class.

Parameters:

default_value (float, optional) – default value if no annotations of a class are present. Set this e.g. to np.nan or zero. Defaults to np.nan.

Returns:

out_dict_medians (dict) – defaultdict with of median duration per class and default value np.nan.
out_dict_iqrs (dict) – defaultdict with IQR of the durations per class and default value np.nan.

property num_segments

Return the number of segments/samples.

Returns:: (int) – number of segments/samples.

pieplot(**kwargs)[source]

plot(*args, offset=0, time_scale='minutes', fill=False, ax=None, **kwargs)[source]

Quick plot of the sleep stages as function of time.

Parameters:

*args (optional) – optional arguments for the plt.plot() function.
offset (float) – offset to the y axis. Useful for plotting multiple hypnograms and plotting them slightly above each other.
ax (plt.Axes, optional) – axes to plot is. If None, plots in a new figure. Defaults to None.
time_scale (str, optional) – the time scale to use. Choose from ‘seconds’, ‘minutes’, ‘hours’. Defaults to ‘seconds’.
fill (bool) – whether to fill the hypnogram or not.
**kwargs (optional) – optional keyword arguments for the plt.plot() function.

plot_hypnogram(time_scale=None, order=None, missing_color='C7', add_legend=None, ax=None, **kwargs)[source]

Plot the hypnogram (sleep stages as function of time).

Parameters:

time_scale (str, optional) – the time scale to use. Choose from ‘seconds’, ‘minutes’, ‘hours’, None. If None or ‘timedelta’, plots with a datetime.timedelta format (hh:mm:ss).
order (list) – order from top-to-bottom that the sleep stages will appear.
missing_color (str, dict) – missing labels will be shaded using this color scheme.
add_legend (bool) – whether to add a legend for the shaded missing labels.
ax (plt.Axes) – matplotlib axis to plot is.
**kwargs (optional) – optional keyword arguments for the fillplot() function.

proportion(class_label, ignore_labels=None, default_value=0)[source]

Return the proportion of segments classified as class_label.

Parameters:

class_label (str) – the label of the class for which to compute the proportion.
ignore_labels (str or list, optional) – specify a class label or list of class labels to ignore. These labels won’t be considered in the denominator. If None, ignore no_label only. If empty list, ignore nothing. Defaults to None.
default_value (float, optional) – default value if no annotations of the class is present. Set this e.g. to np.nan or zero. Defaults to 0.

Returns:

(float) – proportion of segments classified as class_label.

remove_corrupted_stages(corrupt_labels=None, inplace=False)[source]

Remove the sleep stages that are considered corrupted or incomplete.

A sleep stage is considered complete if it has a clear onset and transition. Annotations with one of corrupt_labels are not counted as transitions and if present during a sleep stage, the sleep stage is considered corrupted (anything may have happened, e.g. get out of the sleep stage).

Parameters:

inplace (bool, optional) – if True, remove the annotations in place. If False, a new SleepStagesResult object with the new set of annotations is returned. Defaults to False.
corrupt_labels (str or list, optional) – specify a class label or list of class labels that do not count as valid sleep stages. These labels won’t be considered as valid transitions (e.g. artefacts, no_label). no_label segments will automatically be considered corrupt. If None, a default list of corrupt labels will be used. Defaults to None.

Returns:

sleep_stages_result (SleepStagesResult) – new object with only complete sleep stages (and no_label segments to fill the gaps).

remove_short_stages(min_duration=180, inplace=False)[source]

Remove sleep stages that last too short by replacing the labels by no_label (NL).

Keep only annotations that last > min_duration.

Parameters:

min_duration (float, optional) – minimum duration in seconds that a sleep stage should last to keep it. Defaults to 180.
inplace (bool, optional) – if True, remove the annotations in place. If False, a new SleepStagesResult object with the new set of annotations is returned. Defaults to False.

Returns:

sleep_stages_result (SleepStagesResult) – new object where all sleep stages last for at least min_duration seconds.

segment_labels(segment_start_times, segment_end_times, return_labels=False)[source]

Return an array of labels corresponding to segments with specified start and end times.

Segment i lasts from segment_start_times[i] until segment_end_times[i]. If segment i does not fall entirely in one annotation, the value of y_true[i] will be np.nan.

Parameters:

segment_start_times (list, np.ndarray) – array with start times of the segments (must be sorted).
segment_end_times (list, np.ndarray) – array with end times of the segments (must be sorted).
return_labels (bool, optional) – if False, return the class numbers. If True, return class labels.

Returns:

y_true (np.ndarray) – 1D array with class numbers (or labels) for each segment.

Examples

>>> texts = ['QS', 'NQS', 'QS', 'NQS', 'QS']
>>> onsets = [10, 40, 120, 125, 140]
>>> durations = list(np.diff(onsets)) + [30]
>>> annotations = [Annotation(text=t, onset=o, duration=d) for (t, o, d) in zip(texts, onsets, durations)]
>>> anset = AnnotationSet(annotations=annotations)
>>> ss = SleepStagesResult(annotation_set=anset, algorithm_parameters=dict())
>>> print(ss.to_dataframe())
   onset  duration text
0   10.0      30.0   QS
1   40.0      80.0  NQS
2  120.0       5.0   QS
3  125.0      15.0  NQS
4  140.0      30.0   QS
>>> segment_start_times = [0, 40, 120, 130, 150, 160]
>>> segment_end_times = [20, 80, 130, 140, 160, 180]
>>> labels = ss.segment_labels(segment_start_times, segment_end_times, return_labels=True)
>>> print(labels)
['NL' 'NQS' 'NL' 'NQS' 'QS' 'NL']

to_dataframe()[source]

Return a dataframe of self.annotation_set.

Returns:: (pd.DataFrame) – dataframe of self.annotation_set.

property total_duration

Return the total duration of the annotated period in seconds.

Returns:: (float) – total duration in seconds.

total_duration_per_class(default_value=0)[source]

Compute the total duration of each sleep stage class.

Parameters:: default_value (float, optional) – default value if no annotations of a class are present. Set this e.g. to np.nan or zero. Defaults to 0.
Returns:: (dict) – defaultdict of total durations per class and default value fill_value.

class nnsa.SleepStagesRobust(**kwargs)[source]

Bases: ClassWithParameters

Robust algorthmic pipleine for 4-class sleep staging for neoantes > 36 weeks PMA.

Attributes:

`clean_detector_cnn`	Return the SleepStagesCnn object.
`data_requirements`	Return a dictionary with requirements for the input of this algorithm.
`hmm_model`	Return the hidden-Markov model.
`segment_length`
`sleep_stages_cnn`	Return the SleepStagesCnn object.

Methods:

`default_parameters`()	Return the default parameters as a dictionary.
`process`(eeg, fs[, verbose])	Process raw EEG data (channel order should match self.data_requirements['channel_order']).

property clean_detector_cnn: Return the SleepStagesCnn object.

property data_requirements

Return a dictionary with requirements for the input of this algorithm.

Returns:: data_requirements (dict) – a dictionary with requirements for the input of self.process()

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (nnsa.Parameters) – a default set of parameters for the model.

property hmm_model: Return the hidden-Markov model.

process(eeg, fs, verbose=2)[source]

Process raw EEG data (channel order should match self.data_requirements[‘channel_order’]).

Parameters:

eeg (np.ndarray) – 2D array with shape (time, channels) with raw (unfiltered) EEG data. The data should be referenced as required and in the channels should have the order as required in self.data_requirements.
fs (float) – sampling frequency of eeg (Hz). Should be at minimum 128 Hz.
verbose (int) – verbosity level.

property segment_length

property sleep_stages_cnn: Return the SleepStagesCnn object.

class nnsa.SleepStagesRobustResult(df, algorithm_parameters, name=0, sleep_label_column='sleep_label_robust', fs=None, data_info=None, segment_start_times=None, segment_end_times=None, time_offset=0)[source]

Bases: ResultBase

High-level interface for manipulating sleep stage predictions as predicted by nnsa.SleepStagesRobust().process().

Parameters:

df (pd.DataFrame) – data with information per segment, as created by nnsa.SleepStagesRobust().process().
algorithm_parameters (nnsa.Parameters) – see ResultBase.
name (str, int) – optional name for the result/recording.
fs (float) – optional 1/segment length in Hz.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.

Methods:

`compute_recording_composition`([...])	Recording composition (percentages of each class in 'sleep_label_robust').
`compute_sleep_composition`([in_percentage, ...])	Compute sleep composition features (i.e., the distribution of the labeled sleep into the sleep stages).
`compute_sleep_cyclicity`()	Compute number of sleep cycles per hour.
`compute_survival`([two_class, df_km])	Compute surivival analysis for bout-durations features.
`compute_unexpected_transitions`()	Compute the number of unexpected sleep transitions per hour.
`compute_usable_percentage`()	Return the percentage of usable data.
`get_hypnogram`([which])	Helper function to collect durations and ending observed for survivial analysis. :param which: how specific the hypnogram is: '1class': non-sleep vs sleep. '2class': non-sleep vs 2-class sleep. '4class': non-sleep vs 4-class sleep. :type which: bool.
`plot`([which, order, label_colors, ...])	"
`prepare_kaplan_meier`(two_class[, ...])	Helper function to collect durations and ending observed for survivial analysis.

Attributes:

`num_segments`	Return the number of segments.
`sleep_label_column`

compute_recording_composition(subclass_sleep=False, in_percentage=True)[source]

Recording composition (percentages of each class in ‘sleep_label_robust’).

Parameters:

subclass_sleep (bool) – if True, uses the subclasses for the sleep. If False, counts all sleep subclasses into one Sleep class.
in_percentage (bool) – if True, the composition is in percentage. If False, the proportions are returned (values between 0 and 1).

Returns:

features (pd.Series) – series with features.

compute_sleep_composition(in_percentage=True, usable_only=False)[source]

Compute sleep composition features (i.e., the distribution of the labeled sleep into the sleep stages).

Parameters:

in_percentage (bool) – if True, the composition is in percentage. If False, the proportions are returned (values between 0 and 1).
usable_only (bool) – whether to include sleep segments during “unusable” epochs (False) or not (True).

Returns:

features (pd.Series) – series with features.

compute_sleep_cyclicity()[source]

Compute number of sleep cycles per hour.

Returns:: features (pd.Series) – series with features.

compute_survival(two_class=False, df_km=None)[source]

Compute surivival analysis for bout-durations features.

Parameters:: two_class (bool) – whether to do a 2-class analysis (True) or 4-class (False).
Returns:: features (pd.Series) – series with features.

compute_unexpected_transitions()[source]

Compute the number of unexpected sleep transitions per hour. This only looks at usable parts and ignores any non-sleep or TS stages in between sleep stages.

Returns:: features (pd.Series) – series with features.

compute_usable_percentage()[source]

Return the percentage of usable data.

Returns:: usable_percentage (pd.Series) – % usable data.

get_hypnogram(which='4class')[source]

Helper function to collect durations and ending observed for survivial analysis. :param which: how specific the hypnogram is:

‘1class’: non-sleep vs sleep. ‘2class’: non-sleep vs 2-class sleep. ‘4class’: non-sleep vs 4-class sleep.

Returns:: labels (np.ndarray) – sequence of labels.

property num_segments

Return the number of segments.

Returns:: (int) – number of segments.

plot(which=None, order=None, label_colors=None, add_legend=True, ax=None, mask_alpha=0.7, plot_quality=True, mask_unusable=True)[source]: ” Helper function to quikly plot the hypnogram.

prepare_kaplan_meier(two_class, missing_not_observed=False)[source]

Helper function to collect durations and ending observed for survivial analysis.

Parameters:

two_class (bool) – whether to do a 2-class analysis (True) or 4-class (False).
missing_not_observed (bool) – if True, sets the ending_observed to False for sleep stages followed by a non-sleep (missing) label. If False, only sets the first and last one to not observed.

Returns:

df (pd.DataFrame) – dataframe with bout durations, ending observed the corresponding label.

property sleep_label_column

class nnsa.TimeSeries(signal, fs, label='no label', unit='a.u.', info=None, time_offset=0, check_label=False, check_unit=False, **kwargs)[source]

Bases: ClassWithParameters

High-level interface for processing of any type of time series.

Parameters:

signal (np.ndarray) – 1D array containing the signal.
fs (float) – sampling frequency of the signal array.
label (str, optional) – the label of the signal (if check_label is True, should be a valid standardized name, see check_label()).
unit (str, optional) – the unit of the signal (should be a valid standardized unit, see check_unit()). Defaults to ‘a.u’.
info (dict, optional) – dict with info about the signal. Dict can contain ‘source’, specifying the path to the source file of the signal. It may have an optional additional field ‘processing’, which is a list of strings specifying several processing steps that were applied to the original signal loaded from ‘source’. If None, the ‘source’ field in the dict is set to ‘UNKNOWN’. Defaults to None.
time_offset (float, optional) – optional time offset in seconds. Time array starts at this offset. Defaults to 0.
check_label (bool, optional) – if True, the passed label is checked whether it is valid. If False, no check is performed and any label may be given. Defaults to True.
check_unit (bool, optional) – if True, the passed unit is checked whether it is valid. If False, no check is performed and any unit may be given. Defaults to True.
**kwargs (optional) – optional keyword arguments for setting parameters.

Methods:

`amplitude_eeg`([verbose])	Compute amplitude-integrated EEG (aEEG) from single channel continuous EEG.
`as_dict`()	Return the TimeSeries data in a dictionary.
`astype`(dtype[, inplace])
`burst_detection`([verbose])	Perform burst detection.
`clamp`(threshold[, inplace])	Clamp values by applying a clamping function to reduce dynamic range.
`compute_power_cwt`([freq_low, freq_high, inplace])	Compute power in specified band.
`cwt`([verbose])	Apply a continuous wavelet transform on the signal.
`default_parameters`()	Return the default parameters as a dictionary.
`demean`([mean_fun, inplace])	Demean the time series.
`detrend`(args, *kwargs)
`detrend_poly`(order[, inplace])	Detrend the time series using a polynomial of a specific order.
`envelope`([verbose])	Compute envelope.
`extract_epoch`([begin, end, error_mode])	Extract a piece of data in the specified interval.
`fill_nan`(value[, inplace, verbose])	Replace nans in the signal by a constant value.
`filter`(filter_obj[, remove_mean, inplace, ...])	Filter the signal with the given 1-D digital filter.
`filter_saved_filter`(filter_name[, ...])	Filter the signal with a specified saved filter.
`filtfilt`(filter_obj[, remove_mean, inplace, ...])	Filter the signal with the given filter using zero-phase filtering (filtfilt).
`interp`(t, **kwargs)	Wrapper for np.interp, performing linear interpolation to get the signal values at times t.
`interp_nan`(args, *kwargs)
`interpolate_nan`([inplace, max_nan_length])	Linearly interpolate nan values.
`line_length`([verbose])	Compute line length feature.
`mean`([max_nan_frac])	Return the mean.
`median`([max_nan_frac])	Return the median.
`merge`(other[, inplace])	Merge other TimeSeries object(s) into one object based on their time arrays.
`moving_average`(args, *kwargs)	Helper-function to apply a moving average.
`moving_max`(window[, inplace])	Helper-function to apply a moving maximum.
`moving_mean`(window[, inplace])	Helper-function to apply a moving average.
`moving_median`(window[, inplace])	Helper-function to apply a moving average.
`multi_scale_entropy`([verbose])	Perform a multi-scale entropy (mse) analysis.
`multifractal_analysis`([verbose])	Perform a multifractal analysis (mfa).
`normalize`([how, inplace])	Normalize the signal.
`notch_filt`([f0, verbose, inplace])	Helper-function to apply a notch filter to remove e.g.
`plot`(*args[, begin, end, time_scale, ax])	Plot (part of) the signal.
`power_analysis`([verbose])	Perform a Fourier based power analysis.
`psd`([verbose])	Estimate power spectral density.
`read_hdf5`(filepath[, label, begin, end])	Read a TimeSeries object from a .hdf5 file.
`read_mat`(filepath, **kwargs)	Read a TimeSeries object from a .mat file.
`read_pickle`(filepath, **kwargs)	Read a TimeSeries object from a pickle file.
`reference`(reference_signal, reference_label)	Reference the signal to reference_channel by subtracting reference_signal from the signal.
`remove_artefacts`([inplace])	Replace samples that are artefacts by np.nan
`remove_flatlines`(n_flatline[, inplace])	Replace flatlines by np.nan
`remove_neighborhood_artefacts`(size[, inplace])	Replace values by np.nan if sample sin their neighborhood are nan.
`remove_outliers`(*args[, inplace])	Replace outliers by np.nan
`remove_values`(*args[, inplace])	Replace values by np.nan
`resample`(fs_new[, method, inplace, verbose])	Resample the signal data to new sampling frequency fs_new.
`save_hdf5`(filepath[, mode, overwrite])	Save the TimeSeries data to a .hdf5 file.
`save_mat`(filepath[, overwrite])	Save the TimeSeries data to a MATLAB-style .mat file.
`save_pickle`(filepath[, overwrite])	Save the TimeSeries data to a pickle file.
`segment`(segment_length[, overlap])	Segment the signal (into smaller time segments).
`signal_quality`()	Collect some characteristics of the signal related to signal quality.
`stepwise_moving_average`(*args[, avg_fun])
`stepwise_reduce`(window[, stepsize, ...])	Compute stepwise value.
`transform`(fun[, fs, label, unit, ...])	Apply a custom function to the signal.
`zscore`([ddof, nan_policy, inplace])	Normalize the signal by centering and unit standard deviation.

Attributes:

`dtype`	Return the effective dtype (should be consistent with the dtype in the parameters).
`fs`	Return the sampling frequency of the signal.
`info`	Return the info dictionary.
`label`	Return the label of the signal.
`signal`	Return the signal array.
`time`	Shortcut to time_array (return the time array corresponding to self.signal).
`time_array`	Return the time array corresponding to self.signal.
`unit`	Return the unit of the signal.

amplitude_eeg(verbose=1, **kwargs)[source]

Compute amplitude-integrated EEG (aEEG) from single channel continuous EEG.

This is a wrapper that prepares the input for AmplitudeEeg.wct() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the AmplitudeEeg class.

Returns:

result (nnsa.AmplitudeEegResult) – object containing the aEEG.

as_dict()[source]: Return the TimeSeries data in a dictionary.

astype(dtype, inplace=False)[source]

burst_detection(verbose=1, **kwargs)[source]

Perform burst detection.

Note: this should be used on unfiltered data.

This is a wrapper that prepares the input for BurstDetection.burst_detection() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the BurstDetection class.

Returns:

result (nnsa.BurstDetectionResult) – the result of the burst detection.

clamp(threshold, inplace=False)[source]

Clamp values by applying a clamping function to reduce dynamic range.

Parameters:

threshold (float) – threshold for the clamping function.
inplace (bool, optional) – whether to apply inplace (True) or not.

Returns:

ts (TimeSeries) – new Dataset object (only returned if inplace is False).

compute_power_cwt(freq_low=None, freq_high=None, inplace=False, **kwargs)[source]

Compute power in specified band.

Parameters:

freq_low (float, optional) – lower frequency cutoff for bandpower. If None, takes lowest possible.
freq_high (float, optional) – upper frequency cutoff for bandpower. If None, takes highest possible.
inplace (bool, optional) – if True, replaces the signal in place by its power. If False, returns a new object.
**kwargs (optional) – optional keyword arguments for compute_power_cwt().

Returns:

ts (TimeSeries) – TimeSeries with power of the signal (if inplace is False).

cwt(verbose=1, **kwargs)[source]

Apply a continuous wavelet transform on the signal.

Parameters:

verbose (int, optional) – verbosity level.
**kwargs (optional) – keyword arguments for Cwt().

Returns:

CwtResult object containing the result.

static default_parameters()[source]

Return the default parameters as a dictionary.

Returns:: (dict) – a default set of parameters.

demean(mean_fun=<function nanmean>, inplace=False)[source]

Demean the time series.

Parameters:

mean_fun (function) – function that takes in a 1D array and retruns the mean that will be subtracted. Defaults to np.nanmean. Other possibilities could be np.nanmedian, np.mean, …
inplace (bool, optional) – whether to apply the operation inplace or return a new object.

Returns:

ts (nnsa.TimeSeries) – object with demeaned signal (returned if inplace is False).

detrend(*args, **kwargs)[source]

detrend_poly(order, inplace=False)[source]

Detrend the time series using a polynomial of a specific order.

Parameters:

order (int) – degree of the polynomial.
inplace (bool, optional) – whether to apply the operation inplace or return a new object.

Returns:

ts (nnsa.TimeSeries) – object with detrended signal (returned if inplace is False).

property dtype: Return the effective dtype (should be consistent with the dtype in the parameters).

envelope(verbose=1, **kwargs)[source]

Compute envelope.

This is a wrapper that prepares the input for Envelope.envelope() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the Envelope class.

Returns:

result (nnsa.EnvelopeResult) – the result of the envelope computation.

Examples

To get the envelope data as a new TimeSeries object: >>> signal = 4 * np.sin(np.arange(10000)) >>> ts = TimeSeries(signal=signal, fs=100, label=’EEG Cz’, unit=’uV’, info={‘source’: ‘random’}) >>> envelope_result = ts.envelope(verbose=0) >>> ts_envelope = envelope_result.to_time_series() >>> print(type(ts_envelope).__name__) TimeSeries

>>> print(ts_envelope.signal.mean())
3.9998155216320725

extract_epoch(begin=None, end=None, error_mode='raise')[source]

Extract a piece of data in the specified interval.

Parameters:

begin (float, optional) – start time in seconds (including time_offset) of the epoch to extract. If negative, begin is duration + begin (python-like indexing). If None, the beginning of the signal is taken. Defaults to None.
end (float, optional) – end time in seconds (including time_offset) of the epoch to extract. If negative, end is duration + end (python-like indexing). If None, the end of the signal is taken. Must point to a time greater than begin. Defaults to None.
error_mode (str) – whether to ‘raise’ an error when no time overlap is found, or to just ‘warn’.

Returns:

ts_epoch (nnsa.TimeSeries) – a new TimeSeries object with only the data of the specified epoch.

fill_nan(value, inplace=False, verbose=0)[source]

Replace nans in the signal by a constant value.

Parameters:

value (float) – the value to fill with.
inplace (bool, optional) – if True, resamples inplace, else a new object is returned. Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(TimeSeries) – new TimeSeries object containing the new signal (if inplace is False).

filter(filter_obj, remove_mean=False, inplace=False, verbose=1, **kwargs)[source]

Filter the signal with the given 1-D digital filter.

Note: this introduces a delay in the filter signal. For zero-phase filtering use filtfilt.

Parameters:

filter_obj (FilterBase-derived) – a child class from the nnsa.FilterBase class. The fs property does not need to be set correctly, as it will be automatically (re)set in this function.
inplace (bool, optional) – if True, filters inplace, else a new object is returned. Defaults to False.
remove_mean (bool, optional) – remove the mean value after filtering (useful for EEG data).
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments passed on to the filter() function of filter_obj.

Returns:

(TimeSeries) – new TimeSeries object containing the filtered signal (if inplace is False).

filter_saved_filter(filter_name, remove_mean=False, inplace=False, verbose=1, **kwargs)[source]

Filter the signal with a specified saved filter.

This is just a wrapper of the nnsa.preprocessing.filter_saved_filter() function.

Parameters:

filter_name (str) – see nnsa.preprocessing.filter_saved_filter().
remove_mean (bool, optional) – remove the mean value after filtering (useful for EEG data).
inplace (bool, optional) – if True, resamples inplace, else a new object is returned. Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments passed on to nnsa.preprocessing.filter_saved_filter().

Returns:

(TimeSeries) – new TimeSeries object containing the filtered signal (if inplace is False).

filtfilt(filter_obj, remove_mean=False, inplace=False, verbose=1, **kwargs)[source]

Filter the signal with the given filter using zero-phase filtering (filtfilt).

Parameters:

filter_obj (FilterBase-derived) – a child class from the nnsa.FilterBase class. The fs property does not need to be set correctly, as it will be automatically (re)set in this function.
inplace (bool, optional) – if True, filters inplace, else a new object is returned. Defaults to False.
remove_mean (bool, optional) – remove the mean value after filtering (useful for EEG data).
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments passed on to the filtfilt() function of filter_obj.

Returns:

(TimeSeries) – new TimeSeries object containing the filtered signal (if inplace is False).

property fs

Return the sampling frequency of the signal.

Returns:: (float) – the sampling frequency of the signal.

property info

Return the info dictionary.

Returns:: (dict) – info dictionary.

interp(t, **kwargs)[source]

Wrapper for np.interp, performing linear interpolation to get the signal values at times t.

Parameters:

t (np.ndarray) – time instances at which to evaluate the signal.
**kwargs (optional) – optional keyword arguments for np.interp.

Returns:

y (np.ndarray) – 1D array with interpolated values.

interp_nan(*args, **kwargs)[source]

interpolate_nan(inplace=False, max_nan_length=None)[source]

Linearly interpolate nan values.

Parameters:

inplace (bool, optional) – interpolate inplace (True) or return a copy (False). Defaults to False.
max_nan_length (int or str, optional) – number of maximum allowable length of a nan segment that we interpolate (in seconds). If ‘auto’, the average duration of a constant value in the signal is taken. If None, no limit is applied, i.e. all nans are interpolated. Defaults to None.

Returns:

ts (TimeSeries) – time series with nans interpolated (if inplace is False).

property label

Return the label of the signal.

Returns:: (str) – the label of the signal.

line_length(verbose=1, **kwargs)[source]

Compute line length feature.

This is a wrapper that prepares the input for LineLength.line_length() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the LineLength class.

Returns:

result (nnsa.LineLengthResult) – the result of the line length computation.

mean(max_nan_frac=1)[source]: Return the mean.

median(max_nan_frac=1)[source]: Return the median.

merge(other, inplace=False)[source]

Merge other TimeSeries object(s) into one object based on their time arrays.

Fills gaps in between the time arrays with nan.

Parameters:

other (TimeSeries or list) – list with (compatible) TimeSeries objects.
inplace (bool, optional) – if True, merges the data inplace by adding it to the current object. If False, a new object is returned, leaving the original ones unchanged. Defaults to False.

Returns:

out (TimeSeries) – new TimeSeries object containing the merged time series (if inplace is False).

moving_average(*args, **kwargs)[source]

Helper-function to apply a moving average.

Parameters:

window (float) – window of the moving average (in seconds).
inplace (bool) – whether to apply the filter inplace or not.
**kwargs – for nnsa.moving_average()

Returns:

(TimeSeries) – filtered data (if inplace is True).

moving_max(window, inplace=False, **kwargs)[source]

Helper-function to apply a moving maximum.

Parameters:

window (float) – window of the moving maximum (in seconds).
inplace (bool) – whether to apply the filter inplace or not.
**kwargs – for nnsa.moving_max()

Returns:

(TimeSeries) – filtered data (if inplace is True).

moving_mean(window, inplace=False, **kwargs)[source]

Helper-function to apply a moving average.

Parameters:

window (float) – window of the moving average (in seconds).
inplace (bool) – whether to apply the filter inplace or not.
**kwargs – for nnsa.moving_average()

Returns:

(TimeSeries) – filtered data (if inplace is True).

moving_median(window, inplace=False, **kwargs)[source]

Helper-function to apply a moving average.

Parameters:

window (float) – window of the moving average (in seconds).
inplace (bool) – whether to apply the filter inplace or not.
**kwargs – for nnsa.moving_average()

Returns:

(TimeSeries) – filtered data (if inplace is True).

multi_scale_entropy(verbose=1, **kwargs)[source]

Perform a multi-scale entropy (mse) analysis.

This is a wrapper that prepares the input for MultiScaleEntropy.multi_scale_entropy() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the MultiScaleEntropy class.

Returns:

result (nnsa.MultiScaleEntropyResult) – the result of the mse analysis.

multifractal_analysis(verbose=1, **kwargs)[source]

Perform a multifractal analysis (mfa).

This is a wrapper that prepares the input for MultifractalAnalysis.multifractal_analysis() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for the MultifractalAnalysis class.

Returns:

result (nnsa.MultifractalAnalysisResult) – the result of the mfa analysis.

normalize(how='zscore', inplace=False, **kwargs)[source]

Normalize the signal.

Parameters:

how (str) – how to normalize. Choose from: - “zscore”
inplace (bool, optional) – whether to normalize inplace (True) or not.
**kwargs (dict) – for normalization functions. Depends on how (see code).

Returns:

(TimeSeries) – normalized signal (only returned if inplace is False).

notch_filt(f0=50, verbose=1, inplace=False, **kwargs)[source]

Helper-function to apply a notch filter to remove e.g. power line interference.

Parameters:

f0 (float or list) – frequencie(s) to suppress.
verbose (int) – verbosity level.
inplace (bool) – whether to apply the filter inplace or not.
**kwargs (optional) – for self.filtfilt().

Returns:

(TimeSeries) – filtered data (if inplace is True).

plot(*args, begin=None, end=None, time_scale='seconds', ax=None, **kwargs)[source]

Plot (part of) the signal.

Parameters:

*args (optional) – optional arguments for the plt.plot() function.
begin (float, optional) – begin time in seconds. If None, plots from the first sample. Default is None.
end (float, optional) – end time in seconds or None to specify the end of the entire signal. Default is None.
time_scale (str, optional) – the time scale to use. Choose from ‘seconds’, ‘minutes’, ‘hours’. Defaults to ‘seconds’.
ax (plt.Axes, optional) – axes to plot is. If None, plots in a new figure. Defaults to None.
**kwargs (optional) – optional keyword arguments for the plt.plot() function.

power_analysis(verbose=1, **kwargs)[source]

Perform a Fourier based power analysis.

This is a wrapper that prepares the input for PowerAnalysis.power_analysis() and returns the result.

Parameters:

verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the PowerAnalysis class.

Returns:

result (nnsa.PowerAnalysisResult) – the result of the power analysis.

psd(verbose=1, **kwargs)[source]

Estimate power spectral density.

This is a wrapper that prepares the input for Psd.psd() and returns the results.

Parameters:

verbose (int, optional) – verbose level. Defaults to 0.
**kwargs (optional) – optional keyword arguments to overrule default parameters of the Psd class.

Returns:

result (nnsa.PsdResult) – object containing the result of the psd.

static read_hdf5(filepath, label=None, begin=None, end=None, **kwargs)[source]

Read a TimeSeries object from a .hdf5 file.

Parameters:

filepath (str or opened file) – filepath to read or an open HDF5 file.
label (str, optional) – label of the hdf5 dataset to read. If None, reads the first TimeSeries that it can find in the file. Defaults to None.
begin (float, optional) – start second (wrt time offset).
end (float, optional) – end second (wrt time offset).
**kwargs (optional) – keyword arguments for TimeSeries().

Returns:

ts (nnsa.TimeSeries) – TimeSeries object with data read from the file.

Examples

>>> signal = np.random.rand(1000)
>>> ts = TimeSeries(signal=signal, fs=10, label='EEG Cz')
>>> ts.save_hdf5('testfile.hdf5')
>>> ts_read = TimeSeries.read_hdf5('testfile.hdf5')
>>> ts_read_epoch = TimeSeries.read_hdf5('testfile.hdf5', begin=10, end=20)
>>> os.remove('testfile.hdf5')
>>> assert_equal(ts.signal, ts_read.signal)
>>> assert_equal(ts.extract_epoch(begin=10, end=20).signal, ts_read_epoch.signal)

static read_mat(filepath, **kwargs)[source]

Read a TimeSeries object from a .mat file.

Parameters:

filepath (str) – filepath to read.
**kwargs (optional) – keyword arguments for TimeSeries().

Returns:

ts (nnsa.TimeSeries) – TimeSeries object with data read from the file.

Examples

>>> signal = np.random.rand(1000)
>>> ts = TimeSeries(signal=signal, fs=10, label='EEG Cz')
>>> ts.save_mat('testfile.mat')
>>> ts_read = TimeSeries.read_mat('testfile.mat')
>>> os.remove('testfile.mat')
>>> assert_equal(ts.signal, ts_read.signal)

static read_pickle(filepath, **kwargs)[source]

Read a TimeSeries object from a pickle file.

Parameters:

filepath (str) – filepath to read.
**kwargs (optional) – keyword arguments for TimeSeries().

Returns:

ts (nnsa.TimeSeries) – TimeSeries object with data read from the file.

Examples

>>> signal = np.random.rand(1000)
>>> ts = TimeSeries(signal=signal, fs=10, label='EEG Cz')
>>> ts.save_pickle('testfile.pkl')
>>> ts_read = TimeSeries.read_pickle('testfile.pkl')
>>> os.remove('testfile.pkl')
>>> assert_equal(ts.signal, ts_read.signal)

reference(reference_signal, reference_label, inplace=False, verbose=1)[source]

Reference the signal to reference_channel by subtracting reference_signal from the signal.

Parameters:

reference_signal (TimeSeries or np.ndarray) – 1D array containing the reference signal to subtract.
reference_label (str) – label of the reference signal.
inplace (bool, optional) – if True, referenced inplace, else a new object is returned. Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(TimeSeries) – new TimeSeries object containing the referenced signal.

remove_artefacts(inplace=False, **kwargs)[source]

Replace samples that are artefacts by np.nan

Parameters:

inplace (bool, optional) – If True, the samples are replaced directly from the TimeSeries object itself. If False, the function returns a copy of the original TimeSeries object in which the artefacted samples are replaced, leaving the data in the current TimeSeries unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for overruling default sample quality criteria (see nnsa.artefacts.artefact_detection.detect_artefact_samples()).

Returns:

(TimeSeries) – new TimeSeries object containing the same signal, but with artefacted samples changed to np.nan (only returned if inplace is False).

remove_flatlines(n_flatline, inplace=False, **kwargs)[source]

Replace flatlines by np.nan

Parameters:

n_flatline (float) – minimal number of seconds for a non-changing segment to be a flatline.
inplace (bool, optional) – If True, the samples are replaced directly from the TimeSeries object itself. If False, the function returns a copy of the original TimeSeries object in which the flatline samples are replaced, leaving the data in the current TimeSeries unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for remove_flatlines.

Returns:

(TimeSeries) – new TimeSeries object containing the same signal, but with flatline samples changed to np.nan (only returned if inplace is False).

remove_neighborhood_artefacts(size, inplace=False, **kwargs)[source]

Replace values by np.nan if sample sin their neighborhood are nan.

Parameters:

size (int) – size of the neighborhood (in seconds). E.g., if the neighborhood is 3, sample at t=t is an artefact if there is an artefact in the signal between t-1:t+1.
inplace (bool, optional) – If True, the samples are replaced directly from the TimeSeries object itself. If False, the function returns a copy of the original TimeSeries object in which the flatline samples are replaced, leaving the data in the current TimeSeries unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for remove_neighborhood_artefacts().

Returns:

(TimeSeries) – new TimeSeries object containing the same signal, but with specific samples changed to np.nan (only returned if inplace is False).

remove_outliers(*args, inplace=False, **kwargs)[source]

Replace outliers by np.nan

Parameters:

*args (float) – arguments for remove_outliers().
inplace (bool, optional) – If True, the samples are replaced directly from the TimeSeries object itself. If False, the function returns a copy of the original TimeSeries object in which the flatline samples are replaced, leaving the data in the current TimeSeries unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for remove_outliers().

Returns:

(TimeSeries) – new TimeSeries object containing the same signal, but with outlier samples changed to np.nan (only returned if inplace is False).

remove_values(*args, inplace=False, **kwargs)[source]

Replace values by np.nan

Parameters:

*args (float) – arguments for remove_values().
inplace (bool, optional) – If True, the samples are replaced directly from the TimeSeries object itself. If False, the function returns a copy of the original TimeSeries object in which the flatline samples are replaced, leaving the data in the current TimeSeries unchanged. Default is False.
**kwargs (optional) – optional keyword arguments for remove_values().

Returns:

(TimeSeries) – new TimeSeries object containing the same signal, but with specific samples changed to np.nan (only returned if inplace is False).

resample(fs_new, method='polyphase_filtering', inplace=False, verbose=1, **kwargs)[source]

Resample the signal data to new sampling frequency fs_new.

Deals with nans by interpolating them before resampling and putting nans back at locations near the artefacts.

Parameters:

fs_new (float) – new sampling frequency to resample to.
method (str, optional) –
the resampling method: If ‘polyphase_filtering’, the data will be resampled using polyphase filtering (see

scipy.signal.resample_poly).

If ‘interpolation’, the data will be resampled using interpolation (see scipy.interpolate.interp1d). Default is ‘polyphase_filtering’.
inplace (bool, optional) – if True, resamples inplace, else a new object is returned. Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.
**kwargs (optional) – optional keyword arguments for scipy’s resample/interpolation function that is applied (which depends on the method argument).

Returns:

(nnsa.TimeSeries) – TimeSeries object with the resampled signal.

save_hdf5(filepath, mode='w', overwrite=False)[source]

Save the TimeSeries data to a .hdf5 file.

Parameters:

filepath (str) – filepath to save to.
mode (str, optional) – ‘w’ for write mode or ‘a’ for append mode. Defaults to ‘w’.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists and mode is ‘w’. Defaults to False.

save_mat(filepath, overwrite=False)[source]

Save the TimeSeries data to a MATLAB-style .mat file.

Parameters:

filepath (str) – filepath to save to.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists. Defaults to False.

save_pickle(filepath, overwrite=False)[source]

Save the TimeSeries data to a pickle file.

Parameters:

filepath (str) – filepath to save to.
overwrite (bool, optional) – if True, overwrites existing files. If False, raises an error when filepath already exists. Defaults to False.

segment(segment_length, overlap=0)[source]

Segment the signal (into smaller time segments).

First segment starts at the start of the signal. Any left over samples at the end of the signal that cannot make up a segment of the specified length are ignored.

Parameters:

segment_length (float) – length of a segment in seconds.
overlap (float, optional) – overlap between succesive segments in seconds. Defaults to 0.

Yields:

(TimeSeries) – TimeSeries object holding the data of the next segment.

property signal

Return the signal array.

Returns:: (np.ndarray) – the (1D) signal array.

signal_quality()[source]

Collect some characteristics of the signal related to signal quality.

This is a wrapper of nnsa.artefacts.artefact_detection.signal_quality()

Returns:: (dict) – see nnsa.artefacts.artefact_detection.signal_quality()

stepwise_moving_average(*args, avg_fun=<function nanmean>, **kwargs)[source]

stepwise_reduce(window, stepsize=None, reduce_fun=<function nanmean>, max_nan=0.5, inplace=False, verbose=1)[source]

Compute stepwise value.

Parameters:

window (float) – length of the segments /moving window to compute the average in (in seconds).
stepsize (float) – stepsize of the moving average window in seconds. The output will have a sample frequency of 1/stepsize Hz. If None, stepsize will be eqaul to window.
reduce_fun (function, optional) – function to reduce a block a data. Must handle nan and accept an axis keyword. Defaults to np.nanmean.
max_nan (float, optional) – maximum fraction (0-1) of nan values in a window. If the fraction of nan samples exceeds max_nan, the mean is nan for that window. Defaults to 0.5.
inplace (bool, optional) – if True, operates inplace, else a new object is returned. Defaults to False.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

(nnsa.TimeSeries) – TimeSeries object with the new signal.

property time

Shortcut to time_array (return the time array corresponding to self.signal).

Returns:: (np.ndarray) – time array (in seconds).

property time_array

Return the time array corresponding to self.signal.

Returns:: (np.ndarray) – time array (in seconds).

transform(fun, fs=None, label=None, unit=None, maintain_dtype=True, inplace=False)[source]

Apply a custom function to the signal.

Parameters:

fun (function) – function that transform the data. Takes in a 1D array and returns a 1D array.
fs (float) – sampling frequency of the transformed signal. If not specified, assumes that the transformed signal has the same sampling frequency as the original and an extra check will be performed to verify that the signal length (number of samples) did not change.
label (str) – optional new label for the transformed signal. If not provided, will keep the original label.
unit (str) – optional new unit for the transformed signal. If not provided, will keep the original unit.
maintain_dtype (bool) – if True, the new signal is forced to have the same dtype as self.
inplace (bool) – transform the data inplace (True) or return a new TimeSeries instance (False).

Returns:

ts (TimeSeries) – TimeSeries with the transformed data (if inplace is False).

property unit

Return the unit of the signal.

Returns:: (str) – the unit of the signal.

zscore(ddof=0, nan_policy='omit', inplace=False)[source]

Normalize the signal by centering and unit standard deviation.

Parameters:

ddof (int, optional) – degrees of freedom correction in the calculation of the standard deviation.
nan_policy (str, optional) – {‘propagate’, ‘raise’, ‘omit’} defines how to handle when input contains nan. ‘propagate’ returns nan, ‘raise’ throws an error, ‘omit’ performs the calculations ignoring nan values.
inplace (bool) – transform the data inplace (True) or return a new TimeSeries instance (False).

Returns:

ts (TimeSeries) – TimeSeries with the transformed data (if inplace is False).

class nnsa.WaveletCoherence(**kwargs)[source]

Bases: ClassWithParameters

Class for computing wavelet coherence.

Main method: compute_wavelet_coherence() or process().

Parameters:: nnsa.ClassWithParameters. (see) –

Examples

>>> np.random.seed(43)
>>> N = 1024
>>> fs = 1
>>> t = np.arange(0, N, 1/fs)
>>> f_mutual = 0.1
>>> x = np.cos(2*np.pi*t*f_mutual) + np.random.rand(N)
>>> y = np.sin(2*np.pi*t*f_mutual) + np.random.rand(N)
>>> wc = WaveletCoherence(surrogates={'n_surrogates': 0})
>>> result = wc.wct(x, y, fs=1, verbose=0)
>>> print(type(result).__name__)
WaveletCoherenceResult

>>> result.P.shape
(76, 1024)

# Print most prominent coupled frequency (should be close to the mutual frequency). >>> result.freqs[np.argmax(np.nanmean(result.P, axis=1))].round(1) 0.1

Methods:

`default_parameters`()	Return the default parameters.
`pct`(x, y, z, fs[, time_offset, labels, verbose])	Compute partial wavelet coherence: coherence between x and y, while ignoring common correlation with z.
`wavelet_coherence`(x, y, fs[, time_offset, ...])	Compute wavelet-based coherence between x and y.
`wct`(args, *kwargs)	Alias for self.wavelet_coherence().

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

pct(x, y, z, fs, time_offset=0, labels=None, verbose=1)[source]

Compute partial wavelet coherence: coherence between x and y, while ignoring common correlation with z.

Args: x (array-like): 1D signal array. y (array_like): 1D signal array. Must have the same length and sample frequency as x. z (array_like or list): 1D signal array. Must have the same length and sample frequency as x.

Can also be a list of 1D signal arrays, in which case the confounding relation from each of these signals will be removed.

fs (float): sample frequency of x and y in Hz. time_offset (float, optional): time offset in seconds (i.e. time of the first sample).

Defaults to 0.

labels (tuple, optional): labels for the output, see WaveletCoherence.wct().: If labels is not given and TimeSeries are passed, infers the labels from these objects. Otherwise, if labels is None, uses default labels. Defaults to None.
verbose (int, optional): verbose level.: Defaults to 1.

Returns:: (nnsa.WaveletResult) – object containing the result.

wavelet_coherence(x, y, fs, time_offset=0, labels=None, verbose=1)[source]

Compute wavelet-based coherence between x and y.

Args: x (array-like): 1D signal array. y (array_like): 1D signal array. Must have the same length and sample frequency as x. fs (float): sample frequency of x and y in Hz. time_offset (float, optional): time offset in seconds (i.e. time of the first sample).

Defaults to 0.

labels (tuple, optional): labels for the output.: If labels is not given and TimeSeries are passed, infers the labels from these objects. Otherwise, if labels is None, uses default labels (‘x’, ‘y’). Defaults to None.
verbose (int, optional): verbose level.: Defaults to 1.

Returns:: (nnsa.WaveletResult) – object containing the result.

wct(*args, **kwargs)[source]: Alias for self.wavelet_coherence().

class nnsa.WaveletCoherenceResult(P, A, freqs, fs, wavelet, insidecoi=None, name=None, algorithm_parameters=None, significance=None, P_surrogates=None, signals=None, labels=None, extra=None, data_info=None, segment_start_times=None, segment_end_times=None, time_offset=0)[source]

Bases: WaveletResult

High-level interface for processing wavelet coherence as computed by nnsa.WaveletCoherence().

Alias for WaveletResult().

Parameters:: WaveletResult. (see) –

class nnsa.WaveletResult(P, A, freqs, fs, wavelet, insidecoi=None, name=None, algorithm_parameters=None, significance=None, P_surrogates=None, signals=None, labels=None, extra=None, data_info=None, segment_start_times=None, segment_end_times=None, time_offset=0)[source]

Bases: ResultBase

High-level interface for processing wavelet maps. Can handle CWT, WCT, PCT.

Parameters:

P (np.ndarray) – 2D array with wavelet power or squared coherence values with size (n_freqs, n_time). E.g. for a CWT: P = 1/scales.reshape(-1, 1) * np.abs(W)**2
A (np.ndarray) – 2D array with wavelet phase angles with size (n_freqs, n_time). E.g. for a CWT: A = np.angle(W)
freqs (np.ndarray) – frequencies in Hz corresponding to the first axis of P and A.
fs (float) – sampling frequency (Hz).
wavelet (Mother, str) – Mother object or string specifying which wavelet was used (e.g. ‘Morlet(6)’).
insidecoi (np.ndarray, optional) – boolean array of shape (n_freqs, n_time) containing True values at the cone/mask of influence. True pixels correspond to regions affected by adge/artefact effects. Defaults to None.
name (str, optional) – name for the result, e.g. ‘WCT EEG-NIRS’.
algorithm_parameters (nnsa.Parameters, optional) – see ResultBase.
significance (np.ndarray, optional) – array with same shape as P containing significance values for P between 0 - 1 (0 not significant, 1 significant). If not specified, methods that require significance values will raise errors. Defaults to None.
P_surrogates (np.ndarray, optional) – array with same size as P, with mean surrogate values. Defaults to None.
signals (np.ndarray, optional) – array of shape (n_signals, n_time) containing an arbitrary number of signals that were used in the computation (for visualization purposes). Defaults to None.
extra (dict, optional) – dict with optional extra arrays. E.g. to save some other surrogate values.
labels (tuple, optional) – tuple of length n_signals, with a label for each element in signals, e.g. (‘EEG’, ‘NIRS’). Defaults to None.
data_info (str, optional) – see ResultBase.
segment_start_times (np.ndarray, optional) – see ResultBase.
segment_end_times (np.ndarray, optional) – see ResultBase.
time_offset (float, optional) – see ResultBase.

Attributes:

`W`	Return the complex valued wavelet coefficients.
`dj`	Return 2log-space scale spacing.
`dt`	Return sampling period.
`num_segments`	Return the number of segments/samples.
`scales`	Return the wavelet scales equivalent to self.freqs.
`wavelet`	Return wavelet Mother object.

Methods:

`compute_coi`()	Return mask for the cone of influence at edges.
`downsample`([ratio, how, which, inplace])	Downsample the spectra along the frequency (axis=0) or time (axis=1) axis.
`get_cum_freq_profile`([freq_low, freq_high, ...])	Return the frequency profile.
`get_downsampled`([which, phase_center, ...])
`get_freq_profile`([freq_low, freq_high, ...])	Return the frequency profile.
`get_phase_mask`([A, phase_center, phase_tol])	Return a boolean mask based on the phase.
`get_profile`(which[, freq_low, freq_high, ...])	Return the time or frequency profile.
`get_significance_mask`([significance, alpha])	Return mask where the data is significant.
`get_smoothed`(window, stepsize[, time_scale, ...])	Return smoothed values.
`get_time_profile`([freq_low, freq_high, ...])	Return the frequency profile.
`match_shapes`()	Match shapes of signals (truncate longest one) (inplace).
`plot`([freq_low, freq_high, freq_scale, ...])	Plot the results.
`plot_complex_freq_profile`([freq_low, ...])
`plot_cum_freq_profile`([freq_low, freq_high, ...])	Plot the cumulative frequency profile (is a 2d image).
`plot_freq_profile`([freq_low, freq_high, ...])	Plot the frequency profile.
`plot_phase`([freq_low, freq_high, ...])	Plot the phase.
`plot_phase_discrete`([freq_low, freq_high, ...])	Plot the phase.
`plot_scalogram`([freq_low, freq_high, ...])	Plot the scalogram (contour plot of the wavelet coherence as a function of frequency and time).
`plot_signals`([time_start, time_stop, ...])	Plot the original signals as function of time.
`plot_time_profile`([freq_low, freq_high, ...])	Plot the time profile (i.e.
`shade`(mask[, freq_low, freq_high, ...])	Plot filled contour mask to shade a part of the scalogram.
`to_complex`()	Return complex values C.
`update_coi`(nan_mask, **kwargs)	Update cone of influence by combining it with a MOI based on nan_mask (inplace).

property W: Return the complex valued wavelet coefficients.

compute_coi()[source]: Return mask for the cone of influence at edges.

property dj: Return 2log-space scale spacing.

downsample(ratio=2, how='decimate', which='time', inplace=False)[source]

Downsample the spectra along the frequency (axis=0) or time (axis=1) axis. Does also downsample the signals.

Parameters:

ratio (int) – downsampling ratio.
how (str, optional) – method for downsampling. Options: - ‘decimate’: take every ratio`th sample. - ‘mean’: collapse `ratio consecutive samples into the mean of those samples. - ‘median’: collapse ratio consecutive samples into the median of those samples. Defaults to ‘decimate’.
which (str, optional) – dimension which to downsample. Choose from: - ‘time’ or ‘t’: downsample time dimension. - ‘freq’ or ‘f’: downsample frequency dimension. - ‘both’: downsample both time and frequency dimensions. Defaults to ‘time.
inplace (bool, optional) – if True, operates inplace, else a new object is returned. Defaults to False.

Returns:

wcr (WaveletCoherenceResult) – new WaveletCoherenceResult object containing the filtered signal (if inplace is False).

property dt: Return sampling period.

get_cum_freq_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', which='mean', phase_center=None, phase_tol=0.7853981633974483, mask_coi=True)[source]

Return the frequency profile.

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
which (str, optional) – what to plot as a function of frequency. Choose from: - ‘mean’: mean of P. - ‘mean_sig’: mean significance. - ‘frac_sig’: fraction of significant P. - ‘frac_phase’: occurence fraction of the specified phase.
phase_tol (phase_center and) – select phase. See self.get_phase_mask. If None, include all phases.
mask_coi (bool, optional) – ignore samples that are in the COI. Defaults to True.

Returns:

freqs (np.ndarray) – frequencies.
freq_profile (np.ndarray) – time averaged value per frequency.

get_downsampled(which='mean', phase_center=None, phase_tol=0.7853981633974483, time_start=None, time_stop=None, time_win=21600, time_step=None, freq_low=None, freq_high=None, freq_scale='Hz', freq_win=None, freq_step=None, mask_coi=True, max_nan_frac=0.5)[source]

get_freq_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', how='mean', phase_center=None, phase_tol=0.7853981633974483, mask_coi=True, max_nan=None, **kwargs)[source]

Return the frequency profile.

Parameters:

self.get_profile(). (See) –

Returns:

freqs (np.ndarray) – frequencies in the specified scale.
freq_profile (np.ndarray) – time averaged value per frequency.

get_phase_mask(A=None, phase_center=None, phase_tol=0.7853981633974483)[source]

Return a boolean mask based on the phase.

Parameters:

A (np.ndarray, optional) – array with phases. If None, takes self.A.
phase_center (float, optional) – center phase.
phase_tol (float, optional) – tolerance around center phase. Phase will be included if larger than (phase_center - phase_tol) and smaller than (phase_center + phase_tol).

Returns:

phase_mask (np.ndarray) – boolean mask for A.

get_profile(which, freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', how='mean', phase_center=None, phase_tol=0.7853981633974483, mask_coi=True, max_nan=None)[source]

Return the time or frequency profile.

Parameters:

which (str) – ‘time’ or (‘freq’, ‘frequency’, ‘scale’). Determines which profile is returned.
freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
how (str, optional) – what to plot as a function of frequency. Choose from: - ‘mean’: mean of P. - ‘median’: median of P. - ‘mean_sig’: mean significance. - ‘median_sig’: median significance. - ‘frac_sig’: fraction of significant P. - ‘frac_phase’: occurence fraction of the specified phase.
phase_tol (phase_center and) – select phase. See self.get_phase_mask. If None, include all phases.
mask_coi (bool, optional) – ignore samples that are in the COI. Defaults to True.

Returns:

x (np.ndarray) – time or frequencies in the specified scale.
profile (np.ndarray) – time or frequency profile.

get_significance_mask(significance=None, alpha=0.05)[source]: Return mask where the data is significant.

get_smoothed(window, stepsize, time_scale='seconds', which='abs_sq', mask_coi=True, max_nan=1, phase_center=None, phase_tol=0.7853981633974483)[source]: Return smoothed values. Choose from abs_sq, abs, real, imag, frac_sig.

get_time_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', how='mean', phase_center=None, phase_tol=0.7853981633974483, mask_coi=True, max_nan=None, **kwargs)[source]

Return the frequency profile.

Parameters:

self.get_profile(). (See) –

Returns:

time (np.ndarray) – time in the specified scale.
time_profile (np.ndarray) – frequency averaged value per time.

match_shapes()[source]: Match shapes of signals (truncate longest one) (inplace).

property num_segments

Return the number of segments/samples.

Returns:: (int) – number of segments/samples.

plot(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', scalogram_kwargs=None, time_profile_kwargs=None, freq_profile_kwargs=None, signals_kwargs=None, mask_coi=True, fig_kwargs=None)[source]

Plot the results.

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
scalogram_kwargs (dict, optional) – keyword arguments for self.plot_scalogram(). Defaults to None.
time_profile_kwargs (dict, optional) – keyword arguments for self.plot_time_profile(). Defaults to None.
freq_profile_kwargs (dict, optional) – keyword arguments for self.plot_freq_profile(). Defaults to None.
signals_kwargs (dict, optional) – keyword arguments for self.plot_signals(). Defaults to None.
mask_coi (bool, optional) – mask samples that were are inside the cone of influence (COI). Defaults to True.

Returns:

axes handles foe each subplot.

plot_complex_freq_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', which='amp_phase', significant_only=False, ax=None, mask_coi=True, **kwargs)[source]

plot_cum_freq_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', which='mean', phase_center=None, phase_tol=0.7853981633974483, ax=None, mask_coi=True, **kwargs)[source]

Plot the cumulative frequency profile (is a 2d image).

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
which (str, optional) – what to plot as a function of frequency. Choose from: - ‘mean’: mean of P. - ‘mean_sig’: mean of significance. - ‘frac_sig’: fraction of significant P. - ‘frac_phase’: occurence fraction of the specified phase.
phase_tol (phase_center and) – select phase. See self.get_phase_mask. If None, include all phases.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
mask_coi (bool, optional) – ignore samples that are in the COI. Defaults to True.
**kwargs (optional) – optional keyword arguments for the plt.contourf function.

plot_freq_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', how='mean', phase_center=None, phase_tol=0.7853981633974483, ax=None, orientation='horizontal', mask_coi=True, **kwargs)[source]

Plot the frequency profile.

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
how (str, optional) – what to plot as a function of frequency. Choose from: - ‘mean’: mean of P. - ‘median’: median of P. - ‘frac_sig’: fraction of significant P. - ‘frac_phase’: occurence fraction of the specified phase.
phase_tol (phase_center and) – select phase. See self.get_phase_mask. If None, include all phases.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
orientation (str, optional) – whether to plot ‘horitontal’ (i.e. frequency on the x-axis), or ‘vertical’ (i.e. frequency on the y-axis). Defaults to ‘horizontal’.
mask_coi (bool, optional) – ignore samples that are in the COI. Defaults to True.
**kwargs (optional) – optional keyword arguments for the plt.contourf function.

plot_phase(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', mask_non_significant=True, alpha=0.05, add_colorbar=True, ax=None, cmap=None, **kwargs)[source]

Plot the phase.

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
cmap (list, optional) – colopmap.
**kwargs (optional) – optional keyword arguments for the plt.pcolormesh function.

plot_phase_discrete(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', plot_significance=True, significance_mode='contour', alpha=0.05, ax=None, mask_coi=True, mask_mode='shade', phases=None, colors=None, **kwargs)[source]

Plot the phase.

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
plot_surrogates (bool, optional) – plot mean profile of the surrogates (if surrogates available). Defaults to False.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
orientation (str, optional) – whether to plot ‘horitontal’ (i.e. frequency on the x-axis), or ‘vertical’ (i.e. frequency on the y-axis). Defaults to ‘horizontal’.
mask_coi (bool, optional) – ignore samples that are in the COI. Defaults to True.
phases (list, optional) – list with phases classes.
colors (list, optional) – list with colors corresponding to phases.
colors –
**kwargs (optional) – optional keyword arguments for the plt.pcolormesh function.

plot_scalogram(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', plot_significance=True, significance_mode='contour', alpha=0.05, ax=None, mask_coi=True, mask_mode='shade', add_colorbar=False, **kwargs)[source]

Plot the scalogram (contour plot of the wavelet coherence as a function of frequency and time).

If surrogate data is available, a significance threshold is computed using the significance level alpha and regions where the coherence exceeds the threshold are encircled with black contours.

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
plot_significance (bool, optional) – plot contour lines alng the significant regions (True) or not (False). Defaults to True.
significance_mode (str, optional) – how to display significance. Choose from: - ‘contour’: draws contour lines at the significance level. - ‘hide’: hides non-significant pixels (shades them darkgrey). Defaults to ‘contour’.
alpha (float, optional) – float between 0 and 1, defining the significance level when computing the significance threshold if surrogate data is available. Defaults to 0.05.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
mask_coi (bool, optional) – mask samples in the COI. Defaults to True.
mask_mode (str, optional) – choose to ‘remove’ or ‘shade’ nans. Defaults to ‘shade’.
add_colorbar (bool) – whether to add a colorbar to the plot (True) or not (False).
**kwargs (optional) – optional keyword arguments for the plt.contourf function.

plot_signals(time_start=None, time_stop=None, time_scale='seconds', normalize=True, smooth_window=None, ax=None, colors=None, **kwargs)[source]

Plot the original signals as function of time.

Parameters:

time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
normalize (bool, optional) – normalize signals (zero mean, unit SD) prior to plotting. Defaults to True.
smooth_window (float) – optional window for moving mean to smooth time signals (in seconds).
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
colors (optional) – Specify one color for all signals (str, array), or one color for each signals (list, tuple, dict).
**kwargs (optional) – optional keyword arguments for the plt.plot function.

plot_time_profile(freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', plot_surrogates=False, ax=None, mask_coi=True, **kwargs)[source]

Plot the time profile (i.e. coherence averaged over a range of frequencies).

Parameters:

freq_low (float, optional) – low frequency to include (in the unit of freq_scale). If None, the minimum available frequency is used. Defaults to None.
freq_high (float, optional) – high frequency to include (in the unit of freq_scale). If None, the maximum available frequency is used. Defaults to None.
freq_scale (str, optional) – frequency scale. Use ‘Hz’, ‘mHz’, ‘seconds’, ‘minutes’. Defaults to ‘Hz’.
time_start (float, optional) – start time to include (in the unit of time_scale). If None, the first available timestamp is used. Defaults to None.
time_stop (float, optional) – end time to include (in the unit of time_scale). If None, the last available timestamp is used. Defaults to None.
time_scale (str, optional) – time scale, see nnsa.data.plotting.convert_time_scale(). Defaults to ‘seconds’.
plot_surrogates (bool, optional) – plot mean profile of the surrogates (if surrogates available). Defaults to False.
ax (plt.Axes, optional) – axes object to plot in. If None, plots in the current axes. Defaults to None.
mask_coi (bool, optional) – ignore samples that are in the COI. Defaults to True.
**kwargs (optional) – optional keyword arguments for the plt.plot function.

property scales: Return the wavelet scales equivalent to self.freqs.

shade(mask, freq_low=None, freq_high=None, freq_scale='Hz', time_start=None, time_stop=None, time_scale='seconds', color='k', alpha=0.5, ax=None)[source]

Plot filled contour mask to shade a part of the scalogram.

Parameters:: mask (np.ndarray) – binary mask, where True values will be shaded.

to_complex()[source]

Return complex values C.

C = sqrt(P) * exp(Aj)

Returns:: C (np.ndarray) – complex values, came shape as self.P and self.A.

update_coi(nan_mask, **kwargs)[source]

Update cone of influence by combining it with a MOI based on nan_mask (inplace).

Parameters:

nan_mask – see compute_moi.
**kwargs – for compute_moi().

property wavelet: Return wavelet Mother object.

class nnsa.WinFIR(tf=None, sos=None, fs=None, **kwargs)[source]

Bases: FilterBase

FIR filter design using the window method.

Parameters:: FilterBase. (see) –

Examples

TODO

Methods:

default_parameters()

Return the default parameters.

static default_parameters()[source]

Return the default parameters.

Returns:: (nnsa.Parameters) – a default set of parameters for the object.

nnsa.assert_equal(actual, desired)[source]

Tests if two objects are equal (like numpy’s assert_equal).

Recursive algorithm. Base case is when the desired object is an object accepted by numpy’s assert_equal (scalars, lists, tuples, dictionaries, numpy arrays and None). In that case, assert_equal will be called to compare the objects. In the other cases, where the desired object is some custom, arbitrary object, the function will be called recursively on the objects’ attributes.

Parameters:

actual (object) – the object to check.
desired (object) – the expected object.

Returns:

nnsa.clamp(x, neg_thres=-inf, pos_thres=inf)[source]

Apply a clampling function to the array x.

References

L. Webb, M. Kauppila, J. A. Roberts, S. Vanhatalo, and N. J. Stevenson, “Automated detection of artefacts in neonatal EEG with residual neural networks,” Computer Methods and Programs in Biomedicine, vol. 208, p. 106194, Sep. 2021, doi: 10.1016/j.cmpb.2021.106194.

nnsa.combine_channels(x, preferred=None, min_nan_len=None)[source]

Combines multiple channels of a signal into one, by cutting the signals into pieces and pasting them together in an optimal way.

Segments the data based on the onsets and offsets of nans (disturbances). Then, for each segment the best channel is chosen, which is the channel with the least amount of NaNs. When multiple channels have the same amount of NaNs (e.g. zero), then preferred is used to prefer one signal over the others (see Args).

Parameters:

x (np.ndarray) – 2D array with dimensions (channels, time) containing multiple channels of a certain signal.
preferred (int or list or function, optional) – indices for the channel dimension of x that specify which channels are preferred. If an int, specifies that a single channel is preferred. If a list, specifies a priority list of channels. If None, a channel is preferred such that the amount of switches between channels is minimized (sort of). If a function, takes in the two signals as a (n_channels, n_time) array and returns the index of the preferred signal segment.
min_nan_len (int, optional) – minimum number of consecutive nan samples to consider it as a disturbance and hence to consider it as a location to switch to another channel. If None, takes 5% of the total lenght.

Returns:

x_combined (np.ndarray) – 1D array with dimension (time,) containing the combined signal.

Examples

>>> x = np.random.rand(3, 1000)
>>> x[0, :200] = np.nan
>>> x[1, 300:700] = np.nan
>>> x[1, 250:260] = np.nan
>>> x[0, 600:] = np.nan
>>> preferred = [1, 2, 0]
>>> x_combined = combine_channels(x, preferred=preferred)

# plt.figure() # ax = plt.subplot(211) # plt.plot(x.T) # plt.subplot(212, sharex=ax) # plt.plot(x.T) # plt.plot(x_combined, ‘–‘)

nnsa.detect_anomalous_channels(x, window, fs=1, std_factor=8, p_trim=0.25, shape_mode='error')[source]

Simple method to detect anomalous high-frequency/amplitude channels by setting a threshold on the line length.

The log line length of the (1-p_strip)*n_channels with lowest line lengts is used to determine mean and std and if any channel exceeds mean + `std_factor`*std, then it is flagged as artefact. This is done per window.

Parameters:

x (np.ndarray) – multichannel signal with dimensions (n_channels, n_samples).
window (float) – the window length (in seconds if fs is given, else in samples) in which to compute line length.
fs (float) – sampling frequency of x.
std_factor (float) – the factor for std determining the threshold (mean+`std_factor*std).
p_trim (float) – the relative amount of channels to strip before computing the mean and std. The channels with the highest line lengths are stripped/excluded.
shape_mode (str) – what to do with unexpected input shape (see check_eeg_is_wide()).

Returns:

mask_anomamly (np.ndarray) – boolean array with same shape as x containing True at locations of anomalies.

Examples

>>> # Create dummy signal with 8 channels and an anomaly somewhere in the third channel.
>>> rng = np.random.RandomState(43)
>>> x = rng.random((8, 1000))
>>> x[2, 301:600] *= 10
>>> # Detect anomaly
>>> mask = detect_anomalous_channels(x, window=20)
>>> print(np.where(np.any(mask, axis=1))[0])
[2]
>>> onsets, offsets = get_onsets_offsets(mask[2])
>>> print(onsets)
[301]
>>> print(offsets)
[600]

nnsa.format_time_axis(time_scale=None, relative=False, ax=None)[source]

Format x-axes (time in seconds) as h:mm:ss, or change to hours or minutes.

Parameters:

time_scale (str) – the time scale to convert to. Choose from ‘seconds’, ‘minutes’, ‘hours’ or None (which displays the time as h:mm:ss).
relative (bool) – if True, sets the first time point (xlim()[0]) to 0.
ax (plt.Axes) – axes to change. If None, takes the current axes.

nnsa.get_filepaths(directory, pattern, case_sensitive=False, subdirectories=False, raise_error=False)[source]

Return a list with paths of files living in directory and with pattern in the filename.

See fnmatch.fnmatch() for the pattern matching. Use * for wildcards.

The paths are absolute if directory is absolute and relative if directory is relative.

Parameters:

directory (str) – path to directory in which the files reside.
pattern (str) – pattern that the returned filenames must contain.
case_sensitive (bool, optional) – if True, pattern is case-sensitive. If False, pattern is not case-sensitive. Defaults to False.
subdirectories (bool) – if True, also look for files in subdirectories. If False, only look for files directly in the provided directory. Defaults to False.
raise_error (bool) – set to True to raise an error if there were no files found.

Returns:

matching_filepaths (list) – filepaths of files in directory that match pattern.

nnsa.get_output_dir(output_root, create_unique=False)[source]

Return a directory path for saving the output that a script generates.

A directory under output_dir is created using the path of the script that calls this function. The output directory is created such that a similar structure is maintained in output as the code that generate the output. E.g.: if a file nnsa/python/scripts/example.py calls this function, and the output is nnsa/output, then the returned directory path is nnsa/output/python/scripts/example.

If the directory does not exist, the directory is created.

Parameters:

output_root (str) – path to the output root directory. The output dir will be located under this root directory.
create_unique (bool, optional) – if True, creates a unique output directory with a name based on the current date and time. If False, does not create this additional unique directory. Defaults to False.

Returns:

dir_out (str) – path to a directory for saving outputs of the script that calls this function.

nnsa.list_attrs(obj)[source]

Returns a list with names of public attributes of an object.

Parameters:: obj (object) – the object of which to get the attributes of.
Returns:: (list of str) – list with strings containing the attribute names of the object.

nnsa.list_methods(obj)[source]

Returns a list with names of public (callable) methods of an object.

Parameters:: obj (object) – the object of which to get the methods of.
Returns:: (list of str) – list with strings containing the method names of the object.

nnsa.moving_envelope(x, n, fs=1, axis=-1, max_nan_frac=0.1)[source]

nnsa.moving_line_length(x, n, fs=1, axis=-1, max_nan_frac=0.1)[source]

nnsa.moving_mad(x, n, axis=-1, maintain_nan=True, max_nan_frac=1, k=1.4826)[source]

Compute the moving (centered) median absolute deviation (MAD) of x, with window size n.

The output at index i is the average of a window of x, with the center of the window located at x[i]. Mirrors the input at the borders to reduce boundary effects.

Parameters:

x (np.ndarray) – data array.
n (int) – window size/number of taps. If n=-1 or None, computes metric over the entire array.
axis (int, optional) – axis along which to compute the moving metric. Defaults to -1.
maintain_nan (bool, optional) – if True, output is nan at locations where x is nan. If False, the local metric is returned even if the current sample was a nan (if other samples in window were not nan). Defaults to True.
max_nan_frac (float) – fraction between 0 and 1 specifying how many nans (proportion) are maximally allowed in a local window. If more nans are in the window, the output is nan for the corresponding output sample.
k (float) – scaling factor for MAD. If k=1.4826, MAD ~ std for Gaussian data. See also https://en.wikipedia.org/wiki/Median_absolute_deviation.

Returns:

running_val (np.ndarray) – the moving MAD of x. Has same shape as x.

Examples

>>> x = np.array([[0,  0,  0,  1     ,  2     ,  3     ,  3,  2, 10, 10],
...               [0,  0,  0,  np.nan,  np.nan,  np.nan,  3,  2, 10, 10]])
>>> moving_mad(x, n=3, axis=-1)
array([[0.    , 0.    , 0.    , 1.4826, 1.4826, 0.    , 0.    , 1.4826,
        0.    , 0.    ],
       [0.    , 0.    , 0.    ,    nan,    nan,    nan, 0.7413, 1.4826,
        0.    , 0.    ]])
>>> moving_mad(x, n=3, axis=0, maintain_nan=False, max_nan_frac=1)
array([[0., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
       [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.]])
>>> moving_mad(x, n=-1)
array([[2.2239, 2.2239, 2.2239, 2.2239, 2.2239, 2.2239, 2.2239, 2.2239,
        2.2239, 2.2239],
       [2.9652, 2.9652, 2.9652,    nan,    nan,    nan, 2.9652, 2.9652,
        2.9652, 2.9652]])
>>> moving_mad(x, n=3, max_nan_frac=0)
array([[0.    , 0.    , 0.    , 1.4826, 1.4826, 0.    , 0.    , 1.4826,
        0.    , 0.    ],
       [0.    , 0.    ,    nan,    nan,    nan,    nan,    nan, 1.4826,
        0.    , 0.    ]])

nnsa.moving_max(x, n, axis=-1, maintain_nan=True, max_nan_frac=1)[source]

Compute the moving (centered) max of x, with window size n.

The output at index i is the average of a window of x, with the center of the window located at x[i]. Mirrors the input at the borders to reduce boundary effects.

Parameters:

x (np.ndarray) – data array.
n (int) – window size/number of taps. If n=-1 or None, computes metric over the entire array.
axis (int, optional) – axis along which to compute the moving metric. Defaults to -1.
maintain_nan (bool, optional) – if True, output is nan at locations where x is nan. If False, the local metric is returned even if the current sample was a nan (if other samples in window were not nan). Defaults to True.
max_nan_frac (float) – fraction between 0 and 1 specifying how many nans (proportion) are maximally allowed in a local window. If more nans are in the window, the output is nan for the corresponding output sample.

Returns:

running_val (np.ndarray) – the moving max of x. Has same shape as x.

Examples

>>> x = np.array([[0,  0,  0,  1     ,  2     ,  3     ,  3,  2, 10, 10],
...               [0,  0,  0,  np.nan,  np.nan,  np.nan,  3,  2, 10, 10]])
>>> moving_max(x, n=3, axis=-1)
array([[ 0.,  0.,  1.,  2.,  3.,  3.,  3., 10., 10., 10.],
       [ 0.,  0.,  0., nan, nan, nan,  3., 10., 10., 10.]])
>>> moving_max(x, n=3, axis=0, maintain_nan=False, max_nan_frac=1)
array([[ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  2., 10., 10.],
       [ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  2., 10., 10.]])
>>> moving_max(x, n=-1)
array([[10., 10., 10., 10., 10., 10., 10., 10., 10., 10.],
       [10., 10., 10., nan, nan, nan, 10., 10., 10., 10.]])
>>> moving_max(x, n=3, max_nan_frac=0)
array([[ 0.,  0.,  1.,  2.,  3.,  3.,  3., 10., 10., 10.],
       [ 0.,  0., nan, nan, nan, nan, nan, 10., 10., 10.]])

nnsa.moving_mean(x, n, axis=-1, maintain_nan=True, max_nan_frac=1)[source]

Compute the moving (centered) mean of x, with window size n.

The output at index i is the average of a window of x, with the center of the window located at x[i]. Mirrors the input at the borders to reduce boundary effects.

Parameters:

x (np.ndarray) – data array.
n (int) – window size/number of taps. If n=-1 or None, computes metric over the entire array.
axis (int, optional) – axis along which to compute the moving metric. Defaults to -1.
maintain_nan (bool, optional) – if True, output is nan at locations where x is nan. If False, the local metric is returned even if the current sample was a nan (if other samples in window were not nan). Defaults to True.
max_nan_frac (float) – fraction between 0 and 1 specifying how many nans (proportion) are maximally allowed in a local window. If more nans are in the window, the output is nan for the corresponding output sample.

Returns:

running_val (np.ndarray) – the moving mean of x. Has same shape as x.

Examples

>>> x = np.array([[0,  0,  0,  1     ,  2     ,  3     ,  3,  2, 10, 10],
...               [0,  0,  0,  np.nan,  np.nan,  np.nan,  3,  2, 10, 10]])
>>> moving_mean(x, n=3, axis=-1)
array([[ 0.        ,  0.        ,  0.33333333,  1.        ,  2.        ,
         2.66666667,  2.66666667,  5.        ,  7.33333333, 10.        ],
       [ 0.        ,  0.        ,  0.        ,         nan,         nan,
                nan,  2.5       ,  5.        ,  7.33333333, 10.        ]])
>>> moving_mean(x, n=3, axis=0, maintain_nan=False, max_nan_frac=1)
array([[ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  2., 10., 10.],
       [ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  2., 10., 10.]])
>>> moving_mean(x, n=-1)
array([[3.1       , 3.1       , 3.1       , 3.1       , 3.1       ,
        3.1       , 3.1       , 3.1       , 3.1       , 3.1       ],
       [3.57142857, 3.57142857, 3.57142857,        nan,        nan,
               nan, 3.57142857, 3.57142857, 3.57142857, 3.57142857]])
>>> moving_mean(x, n=3, max_nan_frac=0)
array([[ 0.        ,  0.        ,  0.33333333,  1.        ,  2.        ,
         2.66666667,  2.66666667,  5.        ,  7.33333333, 10.        ],
       [ 0.        ,  0.        ,         nan,         nan,         nan,
                nan,         nan,  5.        ,  7.33333333, 10.        ]])

nnsa.moving_median(x, n, axis=-1, maintain_nan=True, max_nan_frac=1)[source]

Compute the moving (centered) median of x, with window size n.

The output at index i is the average of a window of x, with the center of the window located at x[i]. Mirrors the input at the borders to reduce boundary effects.

Parameters:

x (np.ndarray) – data array.
n (int) – window size/number of taps. If n=-1 or None, computes metric over the entire array.
axis (int, optional) – axis along which to compute the moving metric. Defaults to -1.
maintain_nan (bool, optional) – if True, output is nan at locations where x is nan. If False, the local metric is returned even if the current sample was a nan (if other samples in window were not nan). Defaults to True.
max_nan_frac (float) – fraction between 0 and 1 specifying how many nans (proportion) are maximally allowed in a local window. If more nans are in the window, the output is nan for the corresponding output sample.

Returns:

running_val (np.ndarray) – the moving median of x. Has same shape as x.

Examples

>>> x = np.array([[0,  0,  0,  1     ,  2     ,  3     ,  3,  2, 10, 10],
...               [0,  0,  0,  np.nan,  np.nan,  np.nan,  3,  2, 10, 10]])
>>> moving_median(x, n=3, axis=-1)
array([[ 0. ,  0. ,  0. ,  1. ,  2. ,  3. ,  3. ,  3. , 10. , 10. ],
       [ 0. ,  0. ,  0. ,  nan,  nan,  nan,  2.5,  3. , 10. , 10. ]])
>>> moving_median(x, n=3, axis=0, maintain_nan=False, max_nan_frac=1)
array([[ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  2., 10., 10.],
       [ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  2., 10., 10.]])
>>> moving_median(x, n=-1)
array([[ 2.,  2.,  2.,  2.,  2.,  2.,  2.,  2.,  2.,  2.],
       [ 2.,  2.,  2., nan, nan, nan,  2.,  2.,  2.,  2.]])
>>> moving_median(x, n=3, max_nan_frac=0)
array([[ 0.,  0.,  0.,  1.,  2.,  3.,  3.,  3., 10., 10.],
       [ 0.,  0., nan, nan, nan, nan, nan,  3., 10., 10.]])

nnsa.moving_std(x, n, axis=-1, maintain_nan=True, max_nan_frac=1, ddof=0)[source]

Compute the moving (centered) standard deviation of x, with window size n.

The output at index i is the average of a window of x, with the center of the window located at x[i]. Mirrors the input at the borders to reduce boundary effects.

Parameters:

x (np.ndarray) – data array.
n (int) – window size/number of taps. If n=-1 or None, computes metric over the entire array.
axis (int, optional) – axis along which to compute the moving metric. Defaults to -1.
maintain_nan (bool, optional) – if True, output is nan at locations where x is nan. If False, the local metric is returned even if the current sample was a nan (if other samples in window were not nan). Defaults to True.
max_nan_frac (float) – fraction between 0 and 1 specifying how many nans (proportion) are maximally allowed in a local window. If more nans are in the window, the output is nan for the corresponding output sample.
ddof (int) – degrees of freedom for the denominator when computing the std (see np.std()).

Returns:

running_val (np.ndarray) – the moving std of x. Has same shape as x.

Examples

>>> x = np.array([[0,  0,  0,  1     ,  2     ,  3     ,  3,  2, 10, 10],
...               [0,  0,  0,  np.nan,  np.nan,  np.nan,  3,  2, 10, 10]])
>>> moving_std(x, n=3, axis=-1)
array([[0.        , 0.        , 0.47140452, 0.81649658, 0.81649658,
        0.47140452, 0.47140452, 3.55902608, 3.77123617, 0.        ],
       [0.        , 0.        , 0.        ,        nan,        nan,
               nan, 0.5       , 3.55902608, 3.77123617, 0.        ]])
>>> moving_std(x, n=3, axis=0, maintain_nan=False, max_nan_frac=1)
array([[0., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
       [0., 0., 0., 0., 0., 0., 0., 0., 0., 0.]])
>>> moving_std(x, n=-1)
array([[3.6180105 , 3.6180105 , 3.6180105 , 3.6180105 , 3.6180105 ,
        3.6180105 , 3.6180105 , 3.6180105 , 3.6180105 , 3.6180105 ],
       [4.20398256, 4.20398256, 4.20398256,        nan,        nan,
               nan, 4.20398256, 4.20398256, 4.20398256, 4.20398256]])
>>> moving_std(x, n=3, max_nan_frac=0)
array([[0.        , 0.        , 0.47140452, 0.81649658, 0.81649658,
        0.47140452, 0.47140452, 3.55902608, 3.77123617, 0.        ],
       [0.        , 0.        ,        nan,        nan,        nan,
               nan,        nan, 3.55902608, 3.77123617, 0.        ]])

nnsa.print_exception_info(e)[source]

Print error message and traceback of error e which is handled/caught.

Parameters:: e (Exception-derived) – exception to print.

Examples

>>> a = [1, 2, 3]
>>> try:
...     idx = a.index(0)
... except ValueError as ex:
...     print_exception_info(ex)
...     idx = None
>>> print(idx)
None

nnsa.print_object_summary(obj)[source]

Print object summary: class name, public attributes, public methods.

Parameters:: obj (object) – object to print the summary of.

nnsa.read_annotation_set_KL(filepath)[source]

Helper function to read annotations from an EDF+ file as annotated by KL (using **).

Parameters:: filepath (str) – filepath of the EDF+.
Returns:: annotation_set (nnsa.AnnotationSet) – object containing the annotations from KL.

nnsa.read_annotation_set_preterm(df, filename, include_dubious_qs=False, training_only=False)[source]

Helper function to read sleep labels from a dataframe for the preterm dataset.

Parameters:

df (pd.DataFrame) – dataframe with the contents of the xlsx file with the sleep labels (e.g. QS_labels_preterm_dataset.xlsx).
filename (str) – the EDF filename to extract the sleep labels of. This EDF filename should be in the dataframe as column ‘Filename’.
include_dubious_qs (bool, optional) – include the dubious QS segments (Dubious QS column in xlsx). Defaults to False.
training_only (bool, optional) – use only the QS segments used for training (Training column in xlsx). Defaults to False.

Returns:

annotation_set (nnsa.AnnotationSet) – object containing the annotations.

nnsa.read_result_from_file(filepath, verbose=1)[source]

Read result from file as created by the save_to_file() method of the ResultBase class.

Automatically calls the corresponding read function for the detected file type.

Parameters:

filepath (str) – path to the file containing the result as created by the save_to_file() method of the ResultBase class. The object returned by this function will be equal to the object that created the file. The file extension must be included in filepath.
verbose (int, optional) – verbose level. Defaults to 1.

Returns:

result (nnsa.ResultBase) – instance of a ResultBase-derived class containing the result.

nnsa.stripboxplot(x=None, y=None, hue=None, style=None, data=None, order=None, hue_order=None, orient=None, color=None, palette=None, markers=None, ax=None, mediansize=0.75, boxkwargs=None, stripkwargs=None, legendkwargs=None)[source]

Plot a stripplot and overlay the box of a boxplot.

Parameters:

inputs (Most) – Common inputs to seaborn’s boxplot and stripplot functions. See seaborn.boxplot and/or seaborn.stripplot.
style – column in data for which to use different markers in the stripplot.
markers – list or dict specifying the markers to be used for each unique marker category.
mediansize – length of the median line in the boxplot, as a fraction of the boxplot width.
boxkwargs – kwargs for seaborn’s boxplot function.
stripkwargs – kwargs for seaborn’s stripplot function.
legendkwargs – kwargs for matplotlib’s legend.

Returns:

ax – axes handle.