This module provides a data class EEGData to work with EEG signal data for event-reaction-time delay experiments.
It works with two separate input datafiles, one storing the EEG signal itself as a 1D array, and one
describing event metadata as a 2D array, describing both the timepoints and the type of event in two columns.
To use this module for data analysis, only three steps are necessary,
-(1st) setup of the EEGData object, (2nd) event data extraction, and (3rd)
-data summary (which performs signal comparison).
-
# setting up the EEGData with some datafiles
-eeg = EEGData( eeg_path = "data/eeg.npy", event_path = "data/events.npy", sampling_frequency = 500 )
-
-# extract the events data
-data.extract( start_sec = -0.3 , stop_sec = 1 )
-
-# summarize and pair-wise compare event-signal types.
-data.summary(
- significance_level = 0.05,
- x_scale = 1000,
- y_scale = 10000,
- )
-
-
CLI
-
This module additionally offers a CLI to directly call the full analysis procedure from the terminal.
It works with two separate input datafiles, one storing the EEG signal itself as a 1D array, and one
describing event metadata as a 2D array, describing both the timepoints and the type of event in two columns.
-Supported file types are:
-- `npy`
-- `txt` ( space-separated for events datafiles )
-- `tsv`
-- `csv` (both `,` and `;` separated )
-
-### Example Usage
-To use this module for data analysis, only three steps are necessary,
-(1st) setup of the `EEGData` object, (2nd) event data extraction, and (3rd)
-data summary (which performs signal comparison).
-
-```
-# setting up the EEGData with some datafiles
-eeg = EEGData( eeg_path = "data/eeg.npy", event_path = "data/events.npy", sampling_frequency = 500 )
-
-# extract the events data
-data.extract( start_sec = -0.3 , stop_sec = 1 )
-
-# summarize and pair-wise compare event-signal types.
-data.summary(
- significance_level = 0.05,
- x_scale = 1000,
- y_scale = 10000,
- )
-```
-
-### CLI
-This module additionally offers a CLI to directly call the full analysis procedure from the terminal.
-
-```
-python3 EEGData.py \
- --eeg_path "./data/eeg.npy" \
- --event_path "./data/events.npy" \
- --sampling_frequency 500 \
- --p_value 0.05 \
- --start_sec -0.3 \
- --stop_sec 1.0 \
- --x_scale 1000 \
- --y_scale 10000 \
- --output "./test_output.pdf"
-```
-
"""
-import os
+import subprocess
+import inspect
+import os
import argparse
import numpy as np
import pandas as pd
@@ -165,7 +96,10 @@
CLI
self.read( signal_path = signal_path, event_path = event_path )
# now setup the frames for the events
- self.n_frames = len(self.signal)
+ self._n_frames = len(self.signal)
+
+ # this will setup self._events which is a
+ # dictionary of event identifiers : number of repeated measurements
self._set_n_events()
# setup a _data argument for the
@@ -181,6 +115,22 @@
CLI
self._start_sec = -0.5
self._stop_sec = 1
+ # and setup the extracted events in case
+ # only a subset are being extacted
+ self._extracted_events = None
+
+ # save a baseline for each event type which will be an
+ # np.ndarray to store the timepoints (or a subset thereof)
+ # before the signal onset. The time points will be sampled
+ # from the extracted timepoints...
+ self._baseline = None
+
+ # setup a dictionary to store the p-values of pair-wise comparison
+ # between either two signals or a signal with it's baseline.
+ # keys will be tuples of signal1, signal2, for baseline comparison
+ # signal1 = signal2...
+ self._pvalues = {}
+
def read( self, signal_path : str = None , event_path : str = None ) -> None:
"""
Read the provided data files and stores the
@@ -235,8 +185,7 @@
if event_type is None:
# get all events
- events_to_extract = self.n_events.keys()
+ events_to_extract = self._events.keys()
# extract each type from the loaded data
data = [
@@ -291,6 +240,7 @@
CLI
]
self._data = data
+ self._extracted_events = events_to_extract
return data
# check if there is a provided subset of events to extract
@@ -305,30 +255,16 @@
CLI
]
self._data = data
+ self._extracted_events = events_to_extract
return data
# now the part for extracting only a
# single event type data
-
- # first adjust the start and end to
- # match the sampling frequency
- start_frame = int( start_sec * self.sampling_frequency )
- stop_frame = int( stop_sec * self.sampling_frequency )
-
- # next generate a set of slices for the EEG data around the timepoints for
- # the events
- firing_slices = [
- slice( event[0]+start_frame, event[0]+stop_frame )
- for event in self.events
- if event[1] == event_type
- ]
-
- # now get the actual data of the event
- data = [ self.signal[ slice ] for slice in firing_slices ]
- data = np.array( data )
-
+ data = self._extract_window(start_sec, stop_sec, event_type)
self._data = data
+ self._extracted_events = event_type
+
# store start and stop sec values
# for later use in summary()
self._start_sec = start_sec
@@ -336,6 +272,138 @@
CLI
return data
+ def baseline( self, size : int or float = None ):
+ """
+ Generates a baseline distribution for EEG Signals,
+ using random sampling from pre-signal timepoints accross
+ replicates and events.
+
+ Note
+ ----
+ This requires that events have already been extacted!
+
+ Parameters
+ ----------
+ size : int or float
+ The number of random samples to draw. If `None` are provided (default)
+ the entire available pre-signal data is used. If an `integer` is provided
+ then the final baseline data contains exactly the given number of datapoints.
+ Alternatively, a `float` `0 < size <= 1` may be provided to specify a fraction
+ of data to sample from. E.g. `size = 0.2` would incorporate 20% of the available
+ datapoints into the baseline.
+
+ Returns
+ -------
+ baseline : np.ndarray
+ An np.ndarray of the randomly drawn samples.
+ """
+ start_sec, stop_sec = self._start_sec, self._stop_sec
+
+ # first get the time period before t=0, beginning at the starting time...
+ if isinstance( self._data, list ):
+ random_samples = [ self._extract_window( start_sec, 0, e ) for e in self._extracted_events ]
+ elif isinstance( self._data, np.ndarray ):
+ random_samples = [ self._extract_window( start_sec, 0, self._extracted_events ) ]
+ elif self._data is None:
+ raise Exception( f"No events data has been extracted yet! Make sure to run extract() before computing a baseline." )
+
+ # collapse the repeats into a single dataset
+ random_samples = [ np.reshape( i, i.size ) for i in random_samples ]
+
+ # now if there is a provided size we subsample
+ if size is not None:
+ if size <= 1:
+ random_samples = [ np.random.choice( i, size = size * i.size ) for i in random_samples ]
+ elif size > 1:
+ random_samples = [ np.random.choice( i, size = size ) for i in random_samples ]
+ else:
+ raise ValueError( f"size needs to be a fraction in [0,1] or an integer > 1 (got size = {size})" )
+
+ self._baseline = random_samples
+
+ # Alright, we currently have the entire sets of pre-timeframes for the baseline and we
+ # will use them as they are completely to use for the baseline comparison.
+ # With the code below we compare a sub-sampled versions thereof. Long story short,
+ # it works also pretty well with sub-sampled versions as well...
+ # import statsmodels.api as sm
+ # from matplotlib import colors
+
+ # fig, ax = plt.subplots( 2,3 )
+ # for r in random_samples:
+ # ax[0,0].plot( r )
+ # r1 = r.reshape( r.size )
+ # ax[0,1].hist( r1, bins = 50 )
+
+ # sm.qqplot( r1, ax = ax[0,2], alpha = 0.3, line = "s", color = list(colors.cnames.values())[ int(np.random.randint(low = 0, high = 10, size = 1))] )
+ # random_samples = [ np.random.choice( r.reshape(r.size), size = size, replace = False ) for r in random_samples ]
+
+ # for r in random_samples:
+ # ax[1,0].plot( r )
+ # r1 = np.reshape( r, r.size )
+ # ax[1,1].hist( r1, bins = 50 )
+
+ # sm.qqplot( r1, ax = ax[1,2], alpha = 0.3, line = "s", color = list(colors.cnames.values())[ int(np.random.randint(low = 0, high = 10, size = 1))] )
+ # # ax[1].hist( np.reshape( random_samples, random_samples.size) )
+ # plt.tight_layout()
+ # plt.show()
+
+
+ def pvalues( self, event1 : int, event2 : int = None ):
+ """
+ Gets the p-value np.ndarray for each signal timepoint from a comparison of
+ either two separate event types or one event with its baseline.
+
+ Parameters
+ ----------
+ event1 : int
+ The numeric event identifier of the (first) signal to get.
+ If `None` is provided, the entire dictionary of pvalues is returned.
+
+ event2 : int
+ The numeric event identifier of the (second) signal from the comparison to get.
+ If `None` is provided then the first signals comparison to it's baseline will be
+ returned (if baseline comparison was performed).
+
+ Returns
+ -------
+ pvalues : np.ndarray or dict
+ An np.ndarray of p-values from a given comparison.
+ """
+
+ if event1 is None:
+ return self._pvalues
+
+ if event2 is None:
+ key = (event1, event1)
+ else:
+ key = (event1, event2)
+ pvalues = self._pvalues.get( key, None )
+ return pvalues
+
+ @property
+ def events( self ):
+ """
+ Returns
+ -------
+ list
+ A list of all different event types from from the loaded metadata.
+ """
+ return list( self._events.keys() )
+
+ @property
+ def timeframe( self ):
+ """
+ Returns
+ -------
+ tuple
+ The used timeframe for event data extraction.
+ This consists of the pre-trigger and post-trigger
+ time offsets in seconds.
+ """
+ return ( self._start_sec, self._stop_sec )
+
+
+
def summary(self,
x_scale:float,
y_scale:float,
@@ -374,37 +442,48 @@
CLI
"""
# extract the event data if not yet done already
+ start_sec = kwargs.pop( "start_sec", self._start_sec )
+ stop_sec = kwargs.pop( "stop_sec", self._stop_sec )
if self._data is None:
- start_sec = kwargs.pop( "start_sec", self._start_sec )
- stop_sec = kwargs.pop( "stop_sec", self._stop_sec )
self.extract( start_sec = start_sec, stop_sec = stop_sec, **kwargs )
+ self.baseline()
data = list( self._data )
- signals = list(self.n_events.keys())
+ signals = list(self._events.keys())
n = len(data)
- start_sec, stop_sec = self._start_sec, self._stop_sec
-
# generate a new figure
- fig, ax = plt.subplots(n,n)
+ figsize = kwargs.pop( "figsize", ( 3*n,2*n ) )
+ fig, ax = plt.subplots(n,n, figsize = figsize )
+ # setup a baseline reference, either with the computed
+ # baselines or None ...
+ baseline = self._baseline if self._baseline is not None else [ None for i in range(n) ]
+
# now first plot the individual signals
# on their own on diagonal plots
for i in range(n):
# only the last subplot should make a legend
make_legend = i == n-1
- plot_signal(
+ p = plot_signal(
data[i],
self.sampling_frequency,
start_sec, stop_sec,
x_scale, y_scale,
+ baseline = baseline[i],
make_legend = make_legend,
+ significance_level = significance_level,
ax = ax[i,i] )
ax[i,i].set_title(f"Signal {signals[i]}")
+ # if we got a baseline to compare to we also want to
+ # store the resulting p-values
+ if p is not None:
+ self._pvalues[ (i,i) ] = p
+
# hide all "left-over" subplots from the layout
# i.e. hide the upper-right half of the figure...
for a in ax[ i, i+1: ]:
@@ -418,7 +497,7 @@
CLI
# only the last plot shall make a legend
make_legend = i == n-1 and j == i-1
- difference_plot(
+ p = difference_plot(
data[i],
data[j],
self.sampling_frequency,
@@ -430,6 +509,10 @@
CLI
)
ax[i,j].set_title(f"Signals: {signals[j]} vs {signals[i]}")
+ # we also want to store the resulting p-values of the
+ # signal comparison
+ self._pvalues[ ( signals[j],signals[i] ) ] = p
+
fig.tight_layout()
if output is None:
@@ -437,14 +520,47 @@
CLI
return fig
plt.savefig(output, bbox_inches = "tight" )
+
+ def _extract_window(self, start_sec, stop_sec, event_type):
+ """
+ Extracts a set of time-frame windows from the data
+ and returns them as a numpy ndarray.
+ """
+
+ # first adjust the start and end to
+ # match the sampling frequency
+ start_frame, stop_frame = self._adjust_timesteps(start_sec, stop_sec)
+
+ # next generate a set of slices for the EEG data around the timepoints for
+ # the events
+ firing_slices = [
+ slice( event[0]+start_frame, event[0]+stop_frame )
+ for event in self._events_data
+ if event[1] == event_type
+ ]
+
+ # now get the actual data of the event
+ data = [ self.signal[ slice ] for slice in firing_slices ]
+ data = np.array( data )
+ return data
+
+ def _adjust_timesteps(self, start_sec, stop_sec):
+ """
+ Adjusts time steps / time points with the used recording frequency,
+ to match the indices within the data.
+ """
+ start_frame = int( start_sec * self.sampling_frequency )
+ stop_frame = int( stop_sec * self.sampling_frequency )
+ return start_frame,stop_frame
+
def _set_n_events(self) -> None:
"""
Sets up a dictionary of the different event types
found in the events data.
"""
- event_types = {event[1] for event in self.events}
- self.n_events = {event_type: len([event for event in self.events if event[1] == event_type]) for event_type in event_types}
+ event_types = {event[1] for event in self._events_data}
+ self._events = {event_type: len([event for event in self._events_data if event[1] == event_type]) for event_type in event_types}
def _check_sanity(self, signal_path, event_path, sampling_frequency):
"""
@@ -546,16 +662,34 @@
CLI
--output "./test.png"
"""
- parser = argparse.ArgumentParser(prefix_chars='-')
+ descr1 = """
+
+-----------------------------------------------------
+▒█▀▀▀ ▒█▀▀▀ ▒█▀▀█ ▀▀█▀▀ █▀▀█ █▀▀█ █░░ ▒█░▄▀ ░▀░ ▀▀█▀▀
+▒█▀▀▀ ▒█▀▀▀ ▒█░▄▄ ░▒█░░ █░░█ █░░█ █░░ ▒█▀▄░ ▀█▀ ░░█░░
+▒█▄▄▄ ▒█▄▄▄ ▒█▄▄█ ░▒█░░ ▀▀▀▀ ▀▀▀▀ ▀▀▀ ▒█░▒█ ▀▀▀ ░░▀░░
+-----------------------------------------------------
+
+This script takes in two data files of EEG signal data and accompanying event-trigger metadata. It performs intra- and inter-signal type comparisons using pair-wise T-Tests over the time-series, highlighting significantly different stretches and producing a summary figure.
+ """
+ descr2 = f"""
+
+Input Data
+----------
+Accepted input file types are {supported_filetypes}. The EEG-signal datafile must specify a 1D array of measurements, while the trigger metadata file must specify a 2D array (2 columns) of trigger time points and event classifier labels (numerically encoded).
+ """
+
+ parser = argparse.ArgumentParser( prefix_chars = "-",
+ formatter_class=argparse.RawDescriptionHelpFormatter,description = descr1, epilog = descr2 )
parser.add_argument(
"--eeg_path", "--eeg",
- type=str, required=True,
+ type=str,
help = f"A file containing EEG signal data. Supported filetypes are {supported_filetypes}"
)
parser.add_argument(
"--event_path", "--event",
- type=str, required=True,
- help = "A file containing event metadata for the signal file. Supported filetypes are {supported_filetypes}"
+ type=str,
+ help = f"A file containing event metadata for the signal file. Supported filetypes are {supported_filetypes}"
)
parser.add_argument(
"--output", "-o",
@@ -564,7 +698,7 @@
CLI
)
parser.add_argument(
"--sampling_frequency", "--freq", "-f",
- type=float, required=True,
+ type=float,
help = "The frequency at which the EEG signal data was recorded (in Hertz)."
)
parser.add_argument(
@@ -574,14 +708,20 @@
CLI
)
parser.add_argument(
"--start_sec", "--start", "-s",
- type=float, required=True,
+ type=float,
help = "The upstream time-padding for event extraction (in seconds)."
)
parser.add_argument(
"--stop_sec", "--stop", "-e",
- type=float, required=True,
+ type=float,
help = "The downstream time-padding for event extraction (in seconds)."
)
+
+ parser.add_argument(
+ "--baseline", "-b",
+ type=bool, default = True,
+ help = "Perform baseline comparison for each event type using the same significance threshold as used for inter-signal comparisons. Will be performed by default."
+ )
parser.add_argument(
"--x_scale", "-x",
type=float, default = 1000,
@@ -592,24 +732,63 @@
CLI
type=float, default = 1000,
help = "A scaling factor for the signal-scale (y-values) from volts to some other unit. Default is 1000 (= millivolts)."
)
+
+ parser.add_argument(
+ "--viewer", "-i",
+ action="store_true",
+ default = False,
+ help = "Open the EEGToolKit Viewer GUI in a web browser."
+ )
args = parser.parse_args()
- # the main program (reading datafiles, extracting, and summarizing)
- data = EEGData(args.eeg_path, args.event_path, args.sampling_frequency)
- data.extract( args.start_sec, args.stop_sec )
- data.summary(
- significance_level = args.p_value,
- x_scale = args.x_scale,
- y_scale = args.y_scale,
- output = args.output
- )
-
- if args.output is not None:
- print( f"Output saved successfully to: '{args.output}'" )
+ # if the viewer is being called then we want to just open the
+ # viewer and nothing else
+ if args.viewer:
+ # first we need to get the relative location of the main.
+ # py file within the package.
+ directory = os.path.dirname(
+ inspect.getfile( plot_signal )
+ )
+ main_file = f"{directory}/main.py"
+
+ # then we call the web interface
+ print( "Starting the \033[94mEEGToolKit \033[96mViewer" )
+ subprocess.run( f"streamlit run {main_file}", shell = True )
+
+ else:
+
+ # the main program (reading datafiles, extracting, and summarizing)
+ data = EEGData(args.eeg_path, args.event_path, args.sampling_frequency)
+ data.extract( args.start_sec, args.stop_sec )
+ if args.baseline:
+ data.baseline()
+ data.summary(
+ significance_level = args.p_value,
+ x_scale = args.x_scale,
+ y_scale = args.y_scale,
+ output = args.output
+ )
+
+ if args.output is not None:
+ print( f"Output saved successfully to: '{args.output}'" )
if __name__ == "__main__":
- main()
+
+ test_mode = False
+ if not test_mode:
+ main()
+ else:
+ print( "Running in Test Mode" )
+
+ eeg = "./data/eeg.npy"
+ events = "./data/events.npy"
+
+ e = EEGData( eeg, events, 500 )
+ e.extract( -0.3, 1 )
+ e.baseline()
+ e.summary( 1000, 1000, output = "./test.pdf" )
+ plt.show()
@@ -657,16 +836,34 @@
Example Usage
--output "./test.png"
"""
- parser = argparse.ArgumentParser(prefix_chars='-')
+ descr1 = """
+
+-----------------------------------------------------
+▒█▀▀▀ ▒█▀▀▀ ▒█▀▀█ ▀▀█▀▀ █▀▀█ █▀▀█ █░░ ▒█░▄▀ ░▀░ ▀▀█▀▀
+▒█▀▀▀ ▒█▀▀▀ ▒█░▄▄ ░▒█░░ █░░█ █░░█ █░░ ▒█▀▄░ ▀█▀ ░░█░░
+▒█▄▄▄ ▒█▄▄▄ ▒█▄▄█ ░▒█░░ ▀▀▀▀ ▀▀▀▀ ▀▀▀ ▒█░▒█ ▀▀▀ ░░▀░░
+-----------------------------------------------------
+
+This script takes in two data files of EEG signal data and accompanying event-trigger metadata. It performs intra- and inter-signal type comparisons using pair-wise T-Tests over the time-series, highlighting significantly different stretches and producing a summary figure.
+ """
+ descr2 = f"""
+
+Input Data
+----------
+Accepted input file types are {supported_filetypes}. The EEG-signal datafile must specify a 1D array of measurements, while the trigger metadata file must specify a 2D array (2 columns) of trigger time points and event classifier labels (numerically encoded).
+ """
+
+ parser = argparse.ArgumentParser( prefix_chars = "-",
+ formatter_class=argparse.RawDescriptionHelpFormatter,description = descr1, epilog = descr2 )
parser.add_argument(
"--eeg_path", "--eeg",
- type=str, required=True,
+ type=str,
help = f"A file containing EEG signal data. Supported filetypes are {supported_filetypes}"
)
parser.add_argument(
"--event_path", "--event",
- type=str, required=True,
- help = "A file containing event metadata for the signal file. Supported filetypes are {supported_filetypes}"
+ type=str,
+ help = f"A file containing event metadata for the signal file. Supported filetypes are {supported_filetypes}"
)
parser.add_argument(
"--output", "-o",
@@ -675,7 +872,7 @@
Example Usage
)
parser.add_argument(
"--sampling_frequency", "--freq", "-f",
- type=float, required=True,
+ type=float,
help = "The frequency at which the EEG signal data was recorded (in Hertz)."
)
parser.add_argument(
@@ -685,14 +882,20 @@
Example Usage
)
parser.add_argument(
"--start_sec", "--start", "-s",
- type=float, required=True,
+ type=float,
help = "The upstream time-padding for event extraction (in seconds)."
)
parser.add_argument(
"--stop_sec", "--stop", "-e",
- type=float, required=True,
+ type=float,
help = "The downstream time-padding for event extraction (in seconds)."
)
+
+ parser.add_argument(
+ "--baseline", "-b",
+ type=bool, default = True,
+ help = "Perform baseline comparison for each event type using the same significance threshold as used for inter-signal comparisons. Will be performed by default."
+ )
parser.add_argument(
"--x_scale", "-x",
type=float, default = 1000,
@@ -703,21 +906,46 @@
Example Usage
type=float, default = 1000,
help = "A scaling factor for the signal-scale (y-values) from volts to some other unit. Default is 1000 (= millivolts)."
)
+
+ parser.add_argument(
+ "--viewer", "-i",
+ action="store_true",
+ default = False,
+ help = "Open the EEGToolKit Viewer GUI in a web browser."
+ )
args = parser.parse_args()
- # the main program (reading datafiles, extracting, and summarizing)
- data = EEGData(args.eeg_path, args.event_path, args.sampling_frequency)
- data.extract( args.start_sec, args.stop_sec )
- data.summary(
- significance_level = args.p_value,
- x_scale = args.x_scale,
- y_scale = args.y_scale,
- output = args.output
- )
-
- if args.output is not None:
- print( f"Output saved successfully to: '{args.output}'" )
+ # if the viewer is being called then we want to just open the
+ # viewer and nothing else
+ if args.viewer:
+ # first we need to get the relative location of the main.
+ # py file within the package.
+ directory = os.path.dirname(
+ inspect.getfile( plot_signal )
+ )
+ main_file = f"{directory}/main.py"
+
+ # then we call the web interface
+ print( "Starting the \033[94mEEGToolKit \033[96mViewer" )
+ subprocess.run( f"streamlit run {main_file}", shell = True )
+
+ else:
+
+ # the main program (reading datafiles, extracting, and summarizing)
+ data = EEGData(args.eeg_path, args.event_path, args.sampling_frequency)
+ data.extract( args.start_sec, args.stop_sec )
+ if args.baseline:
+ data.baseline()
+ data.summary(
+ significance_level = args.p_value,
+ x_scale = args.x_scale,
+ y_scale = args.y_scale,
+ output = args.output
+ )
+
+ if args.output is not None:
+ print( f"Output saved successfully to: '{args.output}'" )
@@ -807,7 +1035,10 @@
Parameters
self.read( signal_path = signal_path, event_path = event_path )
# now setup the frames for the events
- self.n_frames = len(self.signal)
+ self._n_frames = len(self.signal)
+
+ # this will setup self._events which is a
+ # dictionary of event identifiers : number of repeated measurements
self._set_n_events()
# setup a _data argument for the
@@ -823,6 +1054,22 @@
Parameters
self._start_sec = -0.5
self._stop_sec = 1
+ # and setup the extracted events in case
+ # only a subset are being extacted
+ self._extracted_events = None
+
+ # save a baseline for each event type which will be an
+ # np.ndarray to store the timepoints (or a subset thereof)
+ # before the signal onset. The time points will be sampled
+ # from the extracted timepoints...
+ self._baseline = None
+
+ # setup a dictionary to store the p-values of pair-wise comparison
+ # between either two signals or a signal with it's baseline.
+ # keys will be tuples of signal1, signal2, for baseline comparison
+ # signal1 = signal2...
+ self._pvalues = {}
+
def read( self, signal_path : str = None , event_path : str = None ) -> None:
"""
Read the provided data files and stores the
@@ -877,8 +1124,7 @@
if event_type is None:
# get all events
- events_to_extract = self.n_events.keys()
+ events_to_extract = self._events.keys()
# extract each type from the loaded data
data = [
@@ -933,6 +1179,7 @@
Parameters
]
self._data = data
+ self._extracted_events = events_to_extract
return data
# check if there is a provided subset of events to extract
@@ -947,30 +1194,16 @@
Parameters
]
self._data = data
+ self._extracted_events = events_to_extract
return data
# now the part for extracting only a
# single event type data
-
- # first adjust the start and end to
- # match the sampling frequency
- start_frame = int( start_sec * self.sampling_frequency )
- stop_frame = int( stop_sec * self.sampling_frequency )
-
- # next generate a set of slices for the EEG data around the timepoints for
- # the events
- firing_slices = [
- slice( event[0]+start_frame, event[0]+stop_frame )
- for event in self.events
- if event[1] == event_type
- ]
-
- # now get the actual data of the event
- data = [ self.signal[ slice ] for slice in firing_slices ]
- data = np.array( data )
-
+ data = self._extract_window(start_sec, stop_sec, event_type)
self._data = data
+ self._extracted_events = event_type
+
# store start and stop sec values
# for later use in summary()
self._start_sec = start_sec
@@ -978,6 +1211,138 @@
Parameters
return data
+ def baseline( self, size : int or float = None ):
+ """
+ Generates a baseline distribution for EEG Signals,
+ using random sampling from pre-signal timepoints accross
+ replicates and events.
+
+ Note
+ ----
+ This requires that events have already been extacted!
+
+ Parameters
+ ----------
+ size : int or float
+ The number of random samples to draw. If `None` are provided (default)
+ the entire available pre-signal data is used. If an `integer` is provided
+ then the final baseline data contains exactly the given number of datapoints.
+ Alternatively, a `float` `0 < size <= 1` may be provided to specify a fraction
+ of data to sample from. E.g. `size = 0.2` would incorporate 20% of the available
+ datapoints into the baseline.
+
+ Returns
+ -------
+ baseline : np.ndarray
+ An np.ndarray of the randomly drawn samples.
+ """
+ start_sec, stop_sec = self._start_sec, self._stop_sec
+
+ # first get the time period before t=0, beginning at the starting time...
+ if isinstance( self._data, list ):
+ random_samples = [ self._extract_window( start_sec, 0, e ) for e in self._extracted_events ]
+ elif isinstance( self._data, np.ndarray ):
+ random_samples = [ self._extract_window( start_sec, 0, self._extracted_events ) ]
+ elif self._data is None:
+ raise Exception( f"No events data has been extracted yet! Make sure to run extract() before computing a baseline." )
+
+ # collapse the repeats into a single dataset
+ random_samples = [ np.reshape( i, i.size ) for i in random_samples ]
+
+ # now if there is a provided size we subsample
+ if size is not None:
+ if size <= 1:
+ random_samples = [ np.random.choice( i, size = size * i.size ) for i in random_samples ]
+ elif size > 1:
+ random_samples = [ np.random.choice( i, size = size ) for i in random_samples ]
+ else:
+ raise ValueError( f"size needs to be a fraction in [0,1] or an integer > 1 (got size = {size})" )
+
+ self._baseline = random_samples
+
+ # Alright, we currently have the entire sets of pre-timeframes for the baseline and we
+ # will use them as they are completely to use for the baseline comparison.
+ # With the code below we compare a sub-sampled versions thereof. Long story short,
+ # it works also pretty well with sub-sampled versions as well...
+ # import statsmodels.api as sm
+ # from matplotlib import colors
+
+ # fig, ax = plt.subplots( 2,3 )
+ # for r in random_samples:
+ # ax[0,0].plot( r )
+ # r1 = r.reshape( r.size )
+ # ax[0,1].hist( r1, bins = 50 )
+
+ # sm.qqplot( r1, ax = ax[0,2], alpha = 0.3, line = "s", color = list(colors.cnames.values())[ int(np.random.randint(low = 0, high = 10, size = 1))] )
+ # random_samples = [ np.random.choice( r.reshape(r.size), size = size, replace = False ) for r in random_samples ]
+
+ # for r in random_samples:
+ # ax[1,0].plot( r )
+ # r1 = np.reshape( r, r.size )
+ # ax[1,1].hist( r1, bins = 50 )
+
+ # sm.qqplot( r1, ax = ax[1,2], alpha = 0.3, line = "s", color = list(colors.cnames.values())[ int(np.random.randint(low = 0, high = 10, size = 1))] )
+ # # ax[1].hist( np.reshape( random_samples, random_samples.size) )
+ # plt.tight_layout()
+ # plt.show()
+
+
+ def pvalues( self, event1 : int, event2 : int = None ):
+ """
+ Gets the p-value np.ndarray for each signal timepoint from a comparison of
+ either two separate event types or one event with its baseline.
+
+ Parameters
+ ----------
+ event1 : int
+ The numeric event identifier of the (first) signal to get.
+ If `None` is provided, the entire dictionary of pvalues is returned.
+
+ event2 : int
+ The numeric event identifier of the (second) signal from the comparison to get.
+ If `None` is provided then the first signals comparison to it's baseline will be
+ returned (if baseline comparison was performed).
+
+ Returns
+ -------
+ pvalues : np.ndarray or dict
+ An np.ndarray of p-values from a given comparison.
+ """
+
+ if event1 is None:
+ return self._pvalues
+
+ if event2 is None:
+ key = (event1, event1)
+ else:
+ key = (event1, event2)
+ pvalues = self._pvalues.get( key, None )
+ return pvalues
+
+ @property
+ def events( self ):
+ """
+ Returns
+ -------
+ list
+ A list of all different event types from from the loaded metadata.
+ """
+ return list( self._events.keys() )
+
+ @property
+ def timeframe( self ):
+ """
+ Returns
+ -------
+ tuple
+ The used timeframe for event data extraction.
+ This consists of the pre-trigger and post-trigger
+ time offsets in seconds.
+ """
+ return ( self._start_sec, self._stop_sec )
+
+
+
def summary(self,
x_scale:float,
y_scale:float,
@@ -1016,37 +1381,48 @@
Parameters
"""
# extract the event data if not yet done already
+ start_sec = kwargs.pop( "start_sec", self._start_sec )
+ stop_sec = kwargs.pop( "stop_sec", self._stop_sec )
if self._data is None:
- start_sec = kwargs.pop( "start_sec", self._start_sec )
- stop_sec = kwargs.pop( "stop_sec", self._stop_sec )
self.extract( start_sec = start_sec, stop_sec = stop_sec, **kwargs )
+ self.baseline()
data = list( self._data )
- signals = list(self.n_events.keys())
+ signals = list(self._events.keys())
n = len(data)
- start_sec, stop_sec = self._start_sec, self._stop_sec
-
# generate a new figure
- fig, ax = plt.subplots(n,n)
+ figsize = kwargs.pop( "figsize", ( 3*n,2*n ) )
+ fig, ax = plt.subplots(n,n, figsize = figsize )
+ # setup a baseline reference, either with the computed
+ # baselines or None ...
+ baseline = self._baseline if self._baseline is not None else [ None for i in range(n) ]
+
# now first plot the individual signals
# on their own on diagonal plots
for i in range(n):
# only the last subplot should make a legend
make_legend = i == n-1
- plot_signal(
+ p = plot_signal(
data[i],
self.sampling_frequency,
start_sec, stop_sec,
x_scale, y_scale,
+ baseline = baseline[i],
make_legend = make_legend,
+ significance_level = significance_level,
ax = ax[i,i] )
ax[i,i].set_title(f"Signal {signals[i]}")
+ # if we got a baseline to compare to we also want to
+ # store the resulting p-values
+ if p is not None:
+ self._pvalues[ (i,i) ] = p
+
# hide all "left-over" subplots from the layout
# i.e. hide the upper-right half of the figure...
for a in ax[ i, i+1: ]:
@@ -1060,7 +1436,7 @@
Parameters
# only the last plot shall make a legend
make_legend = i == n-1 and j == i-1
- difference_plot(
+ p = difference_plot(
data[i],
data[j],
self.sampling_frequency,
@@ -1072,6 +1448,10 @@
Parameters
)
ax[i,j].set_title(f"Signals: {signals[j]} vs {signals[i]}")
+ # we also want to store the resulting p-values of the
+ # signal comparison
+ self._pvalues[ ( signals[j],signals[i] ) ] = p
+
fig.tight_layout()
if output is None:
@@ -1079,14 +1459,47 @@
Parameters
return fig
plt.savefig(output, bbox_inches = "tight" )
+
+ def _extract_window(self, start_sec, stop_sec, event_type):
+ """
+ Extracts a set of time-frame windows from the data
+ and returns them as a numpy ndarray.
+ """
+
+ # first adjust the start and end to
+ # match the sampling frequency
+ start_frame, stop_frame = self._adjust_timesteps(start_sec, stop_sec)
+
+ # next generate a set of slices for the EEG data around the timepoints for
+ # the events
+ firing_slices = [
+ slice( event[0]+start_frame, event[0]+stop_frame )
+ for event in self._events_data
+ if event[1] == event_type
+ ]
+
+ # now get the actual data of the event
+ data = [ self.signal[ slice ] for slice in firing_slices ]
+ data = np.array( data )
+ return data
+
+ def _adjust_timesteps(self, start_sec, stop_sec):
+ """
+ Adjusts time steps / time points with the used recording frequency,
+ to match the indices within the data.
+ """
+ start_frame = int( start_sec * self.sampling_frequency )
+ stop_frame = int( stop_sec * self.sampling_frequency )
+ return start_frame,stop_frame
+
def _set_n_events(self) -> None:
"""
Sets up a dictionary of the different event types
found in the events data.
"""
- event_types = {event[1] for event in self.events}
- self.n_events = {event_type: len([event for event in self.events if event[1] == event_type]) for event_type in event_types}
+ event_types = {event[1] for event in self._events_data}
+ self._events = {event_type: len([event for event in self._events_data if event[1] == event_type]) for event_type in event_types}
def _check_sanity(self, signal_path, event_path, sampling_frequency):
"""
@@ -1170,8 +1583,166 @@
list
+A list of all different event types from from the loaded metadata.
+
+
+Expand source code
+
+
@property
+def events( self ):
+ """
+ Returns
+ -------
+ list
+ A list of all different event types from from the loaded metadata.
+ """
+ return list( self._events.keys() )
+
+
+
var timeframe
+
+
Returns
+
+
tuple
+
+
The used timeframe for event data extraction.
+This consists of the pre-trigger and post-trigger
+time offsets in seconds.
+
+
+
+Expand source code
+
+
@property
+def timeframe( self ):
+ """
+ Returns
+ -------
+ tuple
+ The used timeframe for event data extraction.
+ This consists of the pre-trigger and post-trigger
+ time offsets in seconds.
+ """
+ return ( self._start_sec, self._stop_sec )
+
+
+
Methods
+
+def baseline(self, size: int = None)
+
+
+
Generates a baseline distribution for EEG Signals,
+using random sampling from pre-signal timepoints accross
+replicates and events.
+
Note
+
This requires that events have already been extacted!
+
Parameters
+
+
size : int or float
+
The number of random samples to draw. If None are provided (default)
+the entire available pre-signal data is used. If an integer is provided
+then the final baseline data contains exactly the given number of datapoints.
+Alternatively, a float0 < size <= 1 may be provided to specify a fraction
+of data to sample from. E.g. size = 0.2 would incorporate 20% of the available
+datapoints into the baseline.
+
+
Returns
+
+
baseline : np.ndarray
+
An np.ndarray of the randomly drawn samples.
+
+
+
+Expand source code
+
+
def baseline( self, size : int or float = None ):
+ """
+ Generates a baseline distribution for EEG Signals,
+ using random sampling from pre-signal timepoints accross
+ replicates and events.
+
+ Note
+ ----
+ This requires that events have already been extacted!
+
+ Parameters
+ ----------
+ size : int or float
+ The number of random samples to draw. If `None` are provided (default)
+ the entire available pre-signal data is used. If an `integer` is provided
+ then the final baseline data contains exactly the given number of datapoints.
+ Alternatively, a `float` `0 < size <= 1` may be provided to specify a fraction
+ of data to sample from. E.g. `size = 0.2` would incorporate 20% of the available
+ datapoints into the baseline.
+
+ Returns
+ -------
+ baseline : np.ndarray
+ An np.ndarray of the randomly drawn samples.
+ """
+ start_sec, stop_sec = self._start_sec, self._stop_sec
+
+ # first get the time period before t=0, beginning at the starting time...
+ if isinstance( self._data, list ):
+ random_samples = [ self._extract_window( start_sec, 0, e ) for e in self._extracted_events ]
+ elif isinstance( self._data, np.ndarray ):
+ random_samples = [ self._extract_window( start_sec, 0, self._extracted_events ) ]
+ elif self._data is None:
+ raise Exception( f"No events data has been extracted yet! Make sure to run extract() before computing a baseline." )
+
+ # collapse the repeats into a single dataset
+ random_samples = [ np.reshape( i, i.size ) for i in random_samples ]
+
+ # now if there is a provided size we subsample
+ if size is not None:
+ if size <= 1:
+ random_samples = [ np.random.choice( i, size = size * i.size ) for i in random_samples ]
+ elif size > 1:
+ random_samples = [ np.random.choice( i, size = size ) for i in random_samples ]
+ else:
+ raise ValueError( f"size needs to be a fraction in [0,1] or an integer > 1 (got size = {size})" )
+
+ self._baseline = random_samples
+
+ # Alright, we currently have the entire sets of pre-timeframes for the baseline and we
+ # will use them as they are completely to use for the baseline comparison.
+ # With the code below we compare a sub-sampled versions thereof. Long story short,
+ # it works also pretty well with sub-sampled versions as well...
+ # import statsmodels.api as sm
+ # from matplotlib import colors
+
+ # fig, ax = plt.subplots( 2,3 )
+ # for r in random_samples:
+ # ax[0,0].plot( r )
+ # r1 = r.reshape( r.size )
+ # ax[0,1].hist( r1, bins = 50 )
+
+ # sm.qqplot( r1, ax = ax[0,2], alpha = 0.3, line = "s", color = list(colors.cnames.values())[ int(np.random.randint(low = 0, high = 10, size = 1))] )
+ # random_samples = [ np.random.choice( r.reshape(r.size), size = size, replace = False ) for r in random_samples ]
+
+ # for r in random_samples:
+ # ax[1,0].plot( r )
+ # r1 = np.reshape( r, r.size )
+ # ax[1,1].hist( r1, bins = 50 )
+
+ # sm.qqplot( r1, ax = ax[1,2], alpha = 0.3, line = "s", color = list(colors.cnames.values())[ int(np.random.randint(low = 0, high = 10, size = 1))] )
+ # # ax[1].hist( np.reshape( random_samples, random_samples.size) )
+ # plt.tight_layout()
+ # plt.show()
if event_type is None:
# get all events
- events_to_extract = self.n_events.keys()
+ events_to_extract = self._events.keys()
# extract each type from the loaded data
data = [
@@ -1262,6 +1833,7 @@
Returns
]
self._data = data
+ self._extracted_events = events_to_extract
return data
# check if there is a provided subset of events to extract
@@ -1276,30 +1848,16 @@
Returns
]
self._data = data
+ self._extracted_events = events_to_extract
return data
# now the part for extracting only a
# single event type data
-
- # first adjust the start and end to
- # match the sampling frequency
- start_frame = int( start_sec * self.sampling_frequency )
- stop_frame = int( stop_sec * self.sampling_frequency )
-
- # next generate a set of slices for the EEG data around the timepoints for
- # the events
- firing_slices = [
- slice( event[0]+start_frame, event[0]+stop_frame )
- for event in self.events
- if event[1] == event_type
- ]
-
- # now get the actual data of the event
- data = [ self.signal[ slice ] for slice in firing_slices ]
- data = np.array( data )
-
+ data = self._extract_window(start_sec, stop_sec, event_type)
self._data = data
+ self._extracted_events = event_type
+
# store start and stop sec values
# for later use in summary()
self._start_sec = start_sec
@@ -1308,6 +1866,64 @@
Returns
return data
+
+def pvalues(self, event1: int, event2: int = None)
+
+
+
Gets the p-value np.ndarray for each signal timepoint from a comparison of
+either two separate event types or one event with its baseline.
+
Parameters
+
+
event1 : int
+
The numeric event identifier of the (first) signal to get.
+If None is provided, the entire dictionary of pvalues is returned.
+
event2 : int
+
The numeric event identifier of the (second) signal from the comparison to get.
+If None is provided then the first signals comparison to it's baseline will be
+returned (if baseline comparison was performed).
+
+
Returns
+
+
pvalues : np.ndarray or dict
+
An np.ndarray of p-values from a given comparison.
+
+
+
+Expand source code
+
+
def pvalues( self, event1 : int, event2 : int = None ):
+ """
+ Gets the p-value np.ndarray for each signal timepoint from a comparison of
+ either two separate event types or one event with its baseline.
+
+ Parameters
+ ----------
+ event1 : int
+ The numeric event identifier of the (first) signal to get.
+ If `None` is provided, the entire dictionary of pvalues is returned.
+
+ event2 : int
+ The numeric event identifier of the (second) signal from the comparison to get.
+ If `None` is provided then the first signals comparison to it's baseline will be
+ returned (if baseline comparison was performed).
+
+ Returns
+ -------
+ pvalues : np.ndarray or dict
+ An np.ndarray of p-values from a given comparison.
+ """
+
+ if event1 is None:
+ return self._pvalues
+
+ if event2 is None:
+ key = (event1, event1)
+ else:
+ key = (event1, event2)
+ pvalues = self._pvalues.get( key, None )
+ return pvalues
events = self._read_datafile( event_path )
# now save
- self.events = events
+ self._events_data = events
@@ -1468,20 +2084,24 @@
Parameters
"""
# extract the event data if not yet done already
+ start_sec = kwargs.pop( "start_sec", self._start_sec )
+ stop_sec = kwargs.pop( "stop_sec", self._stop_sec )
if self._data is None:
- start_sec = kwargs.pop( "start_sec", self._start_sec )
- stop_sec = kwargs.pop( "stop_sec", self._stop_sec )
self.extract( start_sec = start_sec, stop_sec = stop_sec, **kwargs )
+ self.baseline()
data = list( self._data )
- signals = list(self.n_events.keys())
+ signals = list(self._events.keys())
n = len(data)
- start_sec, stop_sec = self._start_sec, self._stop_sec
-
# generate a new figure
- fig, ax = plt.subplots(n,n)
+ figsize = kwargs.pop( "figsize", ( 3*n,2*n ) )
+ fig, ax = plt.subplots(n,n, figsize = figsize )
+
+ # setup a baseline reference, either with the computed
+ # baselines or None ...
+ baseline = self._baseline if self._baseline is not None else [ None for i in range(n) ]
# now first plot the individual signals
# on their own on diagonal plots
@@ -1489,16 +2109,23 @@
Parameters
# only the last subplot should make a legend
make_legend = i == n-1
- plot_signal(
+ p = plot_signal(
data[i],
self.sampling_frequency,
start_sec, stop_sec,
x_scale, y_scale,
+ baseline = baseline[i],
make_legend = make_legend,
+ significance_level = significance_level,
ax = ax[i,i] )
ax[i,i].set_title(f"Signal {signals[i]}")
+ # if we got a baseline to compare to we also want to
+ # store the resulting p-values
+ if p is not None:
+ self._pvalues[ (i,i) ] = p
+
# hide all "left-over" subplots from the layout
# i.e. hide the upper-right half of the figure...
for a in ax[ i, i+1: ]:
@@ -1512,7 +2139,7 @@
Parameters
# only the last plot shall make a legend
make_legend = i == n-1 and j == i-1
- difference_plot(
+ p = difference_plot(
data[i],
data[j],
self.sampling_frequency,
@@ -1524,6 +2151,10 @@
Parameters
)
ax[i,j].set_title(f"Signals: {signals[j]} vs {signals[i]}")
+ # we also want to store the resulting p-values of the
+ # signal comparison
+ self._pvalues[ ( signals[j],signals[i] ) ] = p
+
fig.tight_layout()
if output is None:
@@ -1540,10 +2171,7 @@
Parameters
+ )
+ return pvalues
+
+def _shade_singificant_regions(ax, significance_level : float , pvalues : np.ndarray , xvalues : np.ndarray , ylim : tuple ):
+ """
+ Shades the background of regions (x-value ranges) where corresponding p-values
+ are below a given significance level.
+
+ Parameters
+ ----------
+ ax : plt.axes.Axes
+ An Axes object to plot to.
+ significance_level : float
+ The significance level to use.
+ pvalues : np.ndarray
+ An array of pvalues for each x-value.
+ xvalues : np.ndarray
+ An array of x-values.
+ ylim : tuple
+ A tuple of minimal and maximal y-values to shade.
+ """
+ ax.fill_between(
+ xvalues,
+ ylim[0],
+ ylim[1],
+ # now fill areas where the pvalues are
+ # below our significance_level
+ where = pvalues < significance_level,
+ facecolor = signif_shade_color,
+ alpha = signif_shade_alpha
+ )
+
+def _scale_xboundries(sampling_frequency, start_sec, stop_sec, x_scale):
+ """
+ Scales minimal and maximal x values from starting and final timestep values
+ given some scale and sampling frequency.
+ """
+ x_values = np.arange(
+ int( start_sec * sampling_frequency ),
+ int( stop_sec * sampling_frequency ),
+ dtype=int
+ )
+ x_values = x_values / sampling_frequency * x_scale
+ return x_values
+
+def _scale_yboundries(y_scale, max_y, min_y, pad = 1 ):
+ """
+ Scales minimal and maximal y values from some data
+ by a given scale and adds additional padding as a scalar factor
+ (pad = 1 means no padding).
+ """
+ max_y *= y_scale * pad
+ min_y *= y_scale * pad
+ return min_y, max_y
@@ -559,6 +662,13 @@
Parameters
E.g. y_scale = 1000 to adjust the signal-scale to millivolts.
make_legend : bool
Generates a legend for the plot.
+
+
Returns
+
+
pvalues : np.ndarray
+
+
The p-value from pair-wise T-Test comparison between the
+two signals at each signal position (timepoint).
@@ -598,7 +708,7 @@
Parameters
start_sec : float
The initial / low bound of the time-scale in seconds.
-
+
stop_sec : float
The final / upper bound of the time-scale in seconds.
@@ -612,6 +722,12 @@
Parameters
make_legend : bool
Generates a legend for the plot.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ The p-value from pair-wise T-Test comparison between the
+ two signals at each signal position (timepoint).
"""
# generate a new figure if no ax is given
@@ -620,22 +736,24 @@
Parameters
else:
fig = None
- _difference_plot_(ax,
- extracted_EEGData_1,
- extracted_EEGData_2,
- significance_level,
- sampling_frequency,
- start_sec,
- stop_sec,
- x_scale,
- y_scale,
- make_legend
- )
-
+ pvalues = _difference_plot_(ax,
+ extracted_EEGData_1,
+ extracted_EEGData_2,
+ significance_level,
+ sampling_frequency,
+ start_sec,
+ stop_sec,
+ x_scale,
+ y_scale,
+ make_legend
+ )
+
# we show the figure only if no ax was provided
# and thus no "bigger" figure is assembled elsewhere...
if fig is not None:
- plt.show()
+ plt.show()
+
+ return pvalues
Visualises a single EGG signal dataset from an m x n numpy ndarray
@@ -703,8 +821,22 @@
Parameters
y_scale : float
A scaling factor for the data's y-value range.
E.g. y_scale = 1000 to adjust the signal-scale to millivolts.
+
baseline : np.ndarray
+
An array of baseline data to compare the signal to using a pair-wise t-test.
+This will introduce shaded boxes for regions that show significant differences
+between the signal and the baseline.
+
significance_level : float
+
The significance threshold for which to accept a signal difference
+as significant. Default is 0.05.
make_legend : bool
Generates a legend for the plot.
+
+
Returns
+
+
pvalues : np.ndarray
+
+
The p-value from pair-wise T-Test comparison between the
+signal and the baseline (if provided) at each signal position (timepoint).
A scaling factor for the data's y-value range.
E.g. `y_scale = 1000` to adjust the signal-scale to millivolts.
+ baseline : np.ndarray
+ An array of baseline data to compare the signal to using a pair-wise t-test.
+ This will introduce shaded boxes for regions that show significant differences
+ between the signal and the baseline.
+
+ significance_level : float
+ The significance threshold for which to accept a signal difference
+ as significant. Default is `0.05`.
+
make_legend : bool
Generates a legend for the plot.
+ Returns
+ -----
+ pvalues : np.ndarray
+ The p-value from pair-wise T-Test comparison between the
+ signal and the baseline (if provided) at each signal position (timepoint).
"""
# generate a new figure if no ax is specified
@@ -758,19 +906,24 @@
Parameters
else:
fig = None
- _plot_(ax,
- extracted_EEGData,
- sampling_frequency,
- start_sec,
- stop_sec,
- x_scale,
- y_scale,
- make_legend )
-
+ pvalues = _plot_(ax,
+ extracted_EEGData,
+ sampling_frequency,
+ start_sec,
+ stop_sec,
+ x_scale,
+ y_scale,
+ baseline,
+ make_legend,
+ significance_level = significance_level
+ )
+
# we show the figure only if no ax was provided
# and thus no "bigger" figure is assembled elsewhere...
if fig is not None:
- plt.show()
Computes time-point-wise the mean and SEM values of EEG data stored in numpy ndarrays.
+
+
+Expand source code
+
+
"""
+Computes time-point-wise the mean and SEM values of EEG data stored in numpy ndarrays.
+"""
+
+import numpy as np
+from statsmodels.stats.weightstats import CompareMeans, DescrStatsW
+import matplotlib.pyplot as plt
+
+# for custom legend
+from matplotlib.lines import Line2D
+from matplotlib.patches import Patch
+
+# ----------------------------------------------------------------
+# Data Statistics
+# ----------------------------------------------------------------
+
+
+def mean(extracted_EEGData:np.ndarray) -> np.ndarray:
+ """
+ Computes the element-wise mean of an `m x n numpy ndarray`
+ with `m` repeated datasets and `n` entries per set along axis 1
+ (between repeated sets).
+
+ Parameters
+ ----------
+ extracted_EEGData : np.ndarray
+ An m x n numpy ndarray containing EEG data
+
+ Returns
+ -------
+ means : np.ndarray
+ An 1 x n numpy ndarray containing the element-wise means between all m repeated sets.
+ """
+ means = DescrStatsW( extracted_EEGData ).mean
+ return means
+
+def sem(extracted_EEGData:np.ndarray) -> np.ndarray:
+ """
+ Compute the standard error of the mean (SEM)
+ of an `m x n numpy ndarray` with `m` repeated
+ datasets and `n` entries per set along axis 1
+ (between repeated sets).
+
+ Parameters
+ ----------
+ extracted_EEGData : np.ndarray
+ An m x n numpy ndarray containing EEG data
+
+ Returns
+ -------
+ sems : np.ndarray
+ An 1 x n numpy ndarray containing the SEM between all m repeated sets.
+
+ """
+ sems = DescrStatsW( extracted_EEGData )
+ length = len( extracted_EEGData )
+ sems = sems.std / np.sqrt( length )
+ return sems
+
+
+
+def compare_signals(
+ extracted_EEGData_1 : np.ndarray,
+ extracted_EEGData_2 : np.ndarray
+ ) -> np.ndarray:
+ """
+ Compares two EEG signal datasets stored as identical `m x n` np ndarrays
+ with `m` repeated datasets and `n` entries per set, element-wise (along axis 1).
+ It performs T-Tests to check for siginificant differences between the two datasets
+ and returns the corresoponding p-values as a new array.
+
+ Parameters
+ ----------
+ extracted_EEGData_1 : np.ndarray
+ The first EEG signal dataset.
+
+ extracted_EEGData_2 : np.ndarray
+ The second EEG signal dataset.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ A 1 x n numpy ndarray of p-values for the
+ difference between Signals 1 and 2 at each position.
+ """
+
+ # compare the signal means
+ diff = CompareMeans.from_data(
+ extracted_EEGData_1,
+ extracted_EEGData_2
+ )
+
+ # perform t-test comparison
+ diff = diff.ttest_ind()
+
+ # get the p-values
+ pvalues = diff[1]
+ return pvalues
+
+# ----------------------------------------------------------------
+# Data Visualisation
+# ----------------------------------------------------------------
+
+# set up some default style settings
+
+# colorscheme
+signal_color = "gray"
+signal1_color = "xkcd:dark royal blue"
+signal2_color = "xkcd:watermelon"
+signif_shade_color = "xkcd:yellow tan"
+
+# opacities
+signal_alpha = 0.8
+sem_alpha = 0.2
+signif_shade_alpha = 0.5
+
+def plot_signal(
+ extracted_EEGData : np.ndarray,
+ sampling_frequency : float,
+ start_sec : float,
+ stop_sec : float,
+ x_scale = 10**3,
+ y_scale = 10**3,
+ baseline : np.ndarray = None,
+ significance_level:float = 0.05,
+ make_legend = False,
+ ax = None ) -> None:
+ """
+ Visualises a single EGG signal dataset from an `m x n numpy ndarray`
+ with `m` repeated datasets and `n` entries per set. It generates a solid
+ line for the time-point-wise mean and a shaded area of the corresponding SEM.
+
+ Parameters
+ ----------
+
+ extracted_EEGData : np.ndarray
+ An m x n numpy ndarray of repeated EEG data.
+
+ sampling_frequency : float
+ The frequency in which the EEG data was recorded in `Hertz`.
+
+ start_sec : float
+ The initial / low bound of the time-scale in seconds.
+
+ stop_sec : float
+ The final / upper bound of the time-scale in seconds.
+
+ x_scale : float
+ A scaling factor to adjust the data's x-value range.
+ E.g. `x_scale = 1000` to adjust the time-scale to milliseconds.
+
+ y_scale : float
+ A scaling factor for the data's y-value range.
+ E.g. `y_scale = 1000` to adjust the signal-scale to millivolts.
+
+ baseline : np.ndarray
+ An array of baseline data to compare the signal to using a pair-wise t-test.
+ This will introduce shaded boxes for regions that show significant differences
+ between the signal and the baseline.
+
+ significance_level : float
+ The significance threshold for which to accept a signal difference
+ as significant. Default is `0.05`.
+
+ make_legend : bool
+ Generates a legend for the plot.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ The p-value from pair-wise T-Test comparison between the
+ signal and the baseline (if provided) at each signal position (timepoint).
+ """
+
+ # generate a new figure if no ax is specified
+ if ax is None:
+ fig, ax = plt.subplots()
+ else:
+ fig = None
+
+ pvalues = _plot_(ax,
+ extracted_EEGData,
+ sampling_frequency,
+ start_sec,
+ stop_sec,
+ x_scale,
+ y_scale,
+ baseline,
+ make_legend,
+ significance_level = significance_level
+ )
+
+ # we show the figure only if no ax was provided
+ # and thus no "bigger" figure is assembled elsewhere...
+ if fig is not None:
+ plt.show()
+
+ return pvalues
+
+def difference_plot(extracted_EEGData_1:np.ndarray,
+ extracted_EEGData_2:np.ndarray,
+ sampling_frequency:float,
+ start_sec:float,
+ stop_sec:float,
+ significance_level:float = 0.05,
+ x_scale=10**3,
+ y_scale=10**3,
+ make_legend = False,
+ ax = None ) -> None:
+
+ """
+ Visualises the difference between two EEG signals and tests
+ time-point-wise the difference using T-Tests. Individual EEG
+ Signals are shown as mean-line with shaded SEM, and time-points
+ with significant differences are highlighted with overall-shading.
+
+ Parameters
+ ----------
+ extracted_EEGData_1 : np.ndarray
+ The first EEG signal dataset.
+
+ extracted_EEGData_2 : np.ndarray
+ The second EEG signal dataset.
+
+ sampling_frequency : float
+ The frequency in which the EEG data was recorded in `Hertz`.
+
+ significance_level : float
+ The significance threshold for which to accept a signal difference
+ as significant. Default is `0.05`.
+
+ start_sec : float
+ The initial / low bound of the time-scale in seconds.
+
+ stop_sec : float
+ The final / upper bound of the time-scale in seconds.
+
+ x_scale : float
+ A scaling factor to adjust the data's x-value range.
+ E.g. `x_scale = 1000` to adjust the time-scale to milliseconds.
+
+ y_scale : float
+ A scaling factor for the data's y-value range.
+ E.g. `y_scale = 1000` to adjust the signal-scale to millivolts.
+
+ make_legend : bool
+ Generates a legend for the plot.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ The p-value from pair-wise T-Test comparison between the
+ two signals at each signal position (timepoint).
+ """
+
+ # generate a new figure if no ax is given
+ if ax is None:
+ fig, ax = plt.subplots()
+ else:
+ fig = None
+
+ pvalues = _difference_plot_(ax,
+ extracted_EEGData_1,
+ extracted_EEGData_2,
+ significance_level,
+ sampling_frequency,
+ start_sec,
+ stop_sec,
+ x_scale,
+ y_scale,
+ make_legend
+ )
+
+ # we show the figure only if no ax was provided
+ # and thus no "bigger" figure is assembled elsewhere...
+ if fig is not None:
+ plt.show()
+
+ return pvalues
+
+def _plot_(
+ ax,
+ extracted_EEGData:np.ndarray,
+ sampling_frequency:float,
+ start_sec:float,
+ stop_sec:float,
+ x_scale=10**3,
+ y_scale=10**3,
+ baseline : np.ndarray = None,
+ make_legend = False,
+ **kwargs ) -> None:
+ """
+ Generates a mean signal line with shaded
+ SEM area from an m x n numpy ndarray.
+
+ Note
+ -----
+ This is the core for `plot_signal`
+ """
+ # compute mean and SEM for the signal
+ _mean = mean(extracted_EEGData)
+ _sem = sem(extracted_EEGData)
+
+ # now generate scaled xvalues
+ x_values = _scale_xboundries(sampling_frequency, start_sec, stop_sec, x_scale)
+
+ # now plot the signal's scaled mean line
+ signal = _mean * y_scale
+ ax.plot(x_values, signal, color = signal_color )
+
+ # and add the shading for SEM
+ lower = (_mean - _sem)
+ upper = (_mean + _sem)
+ lower *= y_scale
+ upper *= y_scale
+
+ ax.fill_between(
+ x_values,
+ lower,
+ upper,
+ color = signal_color,
+ edgecolor = None,
+ linewidth = 0,
+ alpha = sem_alpha
+
+ )
+
+ # if we have baseline data, we compare also to the baseline
+ # and shade siginificantly different regions...
+ pvalues = None
+ if baseline is not None:
+
+ pvalues = compare_signals( extracted_EEGData, baseline )
+
+ # generate the y-value boundries for the plot
+ # and scale the y-values to some user-defined range
+ # plus add a little padding
+ max_y = np.max( _mean )
+ min_y = np.min( _mean )
+ yvalues = _scale_yboundries( y_scale, max_y, min_y, pad = 1.2 )
+
+ # add the shaded fillings for significantly different
+ # timepoint areas
+ signif_level = kwargs.pop( "significance_level", 0.05 )
+ _shade_singificant_regions(
+ ax,
+ significance_level = signif_level,
+ pvalues = pvalues,
+ xvalues = x_values,
+ ylim = yvalues,
+ )
+
+
+
+ # and add a line for the start of the signal
+ ax.axvline( x=0, linewidth = 2, color = "black" )
+
+ # and some axes formatting...
+ ax.set_title("Average EEG Signal (Shaded area SEM)")
+ ax.set_ylabel("Signal\namplitude")
+ ax.set_xlabel("Time relative to event")
+
+ if make_legend:
+ handles = [
+ Line2D( [0], [0],
+ color = signal_color,
+ label = "Mean Signal"
+ ),
+ Patch(
+ facecolor = signal_color,
+ edgecolor = None, linewidth = 0,
+ alpha = sem_alpha,
+ label = "SEM"
+ )
+ ]
+ if baseline is not None:
+ handles.append(
+ Patch(
+ facecolor = signif_shade_color,
+ edgecolor = None, linewidth = 0,
+ alpha = signif_shade_alpha,
+ label = f"pvalue < {signif_level}"
+ )
+ )
+ # now add a custom legend
+ ax.legend( handles = handles,
+ bbox_to_anchor = (1, -0.5),
+ frameon = False
+ )
+ return pvalues
+
+def _difference_plot_(ax,
+ extracted_EEGData_1:np.ndarray,
+ extracted_EEGData_2:np.ndarray,
+ significance_level:float,
+ sampling_frequency:float,
+ start_sec:float,
+ stop_sec:float,
+ x_scale=10**3,
+ y_scale=10**3,
+ make_legend = False ) -> None:
+
+ """
+ Generates a plot of two EEG signals and their time-point-wise
+ difference using a T-Test.
+
+ Note
+ -----
+ This is the core of difference_plot
+ """
+
+ # compare the two EEG signals and generate a p-value
+ # from t-test position-wise comparisons...
+ pvalues = compare_signals(extracted_EEGData_1, extracted_EEGData_2)
+
+ # generate the mean lines for both signals
+ mean_1 = mean(extracted_EEGData_1)
+ mean_2 = mean(extracted_EEGData_2)
+
+ # generate the y-value boundries for the plot
+ max_y = np.max( [np.max(mean_1), np.max(mean_2)] )
+ min_y = np.min( [np.min(mean_1), np.min(mean_2)] )
+
+ # and scale the y-values to some user-defined range
+ # plus add a little padding
+ yvalues = _scale_yboundries( y_scale, max_y, min_y, pad = 1.2 )
+
+ # generate correspondingly scaled x-values
+ x_values = _scale_xboundries(sampling_frequency, start_sec, stop_sec, x_scale)
+
+ # add the shaded fillings for significantly different
+ # timepoint areas
+ _shade_singificant_regions(
+ ax,
+ significance_level = significance_level,
+ pvalues = pvalues,
+ xvalues = x_values,
+ ylim = yvalues,
+ )
+
+ # plot scaled signals 1 and 2
+ signal_1 = mean_1 * y_scale
+ signal_2 = mean_2 * y_scale
+
+ ax.plot(x_values, signal_1, color = signal1_color, alpha = signal_alpha )
+ ax.plot(x_values, signal_2, color = signal2_color, alpha = signal_alpha )
+
+ # plot a vertical line at the signal start
+ ax.axvline(x = 0, color = "black", linewidth = 2 )
+
+ # and some axes formatting...
+ ax.set_title("Average EEG Signal (Shaded area significant regions)")
+ ax.set_ylabel("Signal\namplitude")
+ ax.set_xlabel("Time relative to event")
+
+ if make_legend:
+ # now add a custom legend
+ ax.legend( handles = [
+ Line2D( [0], [0],
+ color = signal1_color,
+ label = "Mean horizontal Signal"
+ ),
+ Line2D( [0], [0],
+ color = signal2_color,
+ label = "Mean vertical Signal"
+ ),
+ Patch(
+ facecolor = signif_shade_color,
+ edgecolor = None, linewidth = 0,
+ alpha = signif_shade_alpha,
+ label = f"pvalue < {significance_level}"
+ )
+
+ ],
+ bbox_to_anchor = (1, -0.5),
+ frameon = False
+ )
+ return pvalues
+
+def _shade_singificant_regions(ax, significance_level : float , pvalues : np.ndarray , xvalues : np.ndarray , ylim : tuple ):
+ """
+ Shades the background of regions (x-value ranges) where corresponding p-values
+ are below a given significance level.
+
+ Parameters
+ ----------
+ ax : plt.axes.Axes
+ An Axes object to plot to.
+ significance_level : float
+ The significance level to use.
+ pvalues : np.ndarray
+ An array of pvalues for each x-value.
+ xvalues : np.ndarray
+ An array of x-values.
+ ylim : tuple
+ A tuple of minimal and maximal y-values to shade.
+ """
+ ax.fill_between(
+ xvalues,
+ ylim[0],
+ ylim[1],
+ # now fill areas where the pvalues are
+ # below our significance_level
+ where = pvalues < significance_level,
+ facecolor = signif_shade_color,
+ alpha = signif_shade_alpha
+ )
+
+def _scale_xboundries(sampling_frequency, start_sec, stop_sec, x_scale):
+ """
+ Scales minimal and maximal x values from starting and final timestep values
+ given some scale and sampling frequency.
+ """
+ x_values = np.arange(
+ int( start_sec * sampling_frequency ),
+ int( stop_sec * sampling_frequency ),
+ dtype=int
+ )
+ x_values = x_values / sampling_frequency * x_scale
+ return x_values
+
+def _scale_yboundries(y_scale, max_y, min_y, pad = 1 ):
+ """
+ Scales minimal and maximal y values from some data
+ by a given scale and adds additional padding as a scalar factor
+ (pad = 1 means no padding).
+ """
+ max_y *= y_scale * pad
+ min_y *= y_scale * pad
+ return min_y, max_y
Compares two EEG signal datasets stored as identical m x n np ndarrays
+with m repeated datasets and n entries per set, element-wise (along axis 1).
+It performs T-Tests to check for siginificant differences between the two datasets
+and returns the corresoponding p-values as a new array.
+
Parameters
+
+
extracted_EEGData_1 : np.ndarray
+
The first EEG signal dataset.
+
extracted_EEGData_2 : np.ndarray
+
The second EEG signal dataset.
+
+
Returns
+
+
pvalues : np.ndarray
+
A 1 x n numpy ndarray of p-values for the
+difference between Signals 1 and 2 at each position.
+
+
+
+Expand source code
+
+
def compare_signals(
+ extracted_EEGData_1 : np.ndarray,
+ extracted_EEGData_2 : np.ndarray
+ ) -> np.ndarray:
+ """
+ Compares two EEG signal datasets stored as identical `m x n` np ndarrays
+ with `m` repeated datasets and `n` entries per set, element-wise (along axis 1).
+ It performs T-Tests to check for siginificant differences between the two datasets
+ and returns the corresoponding p-values as a new array.
+
+ Parameters
+ ----------
+ extracted_EEGData_1 : np.ndarray
+ The first EEG signal dataset.
+
+ extracted_EEGData_2 : np.ndarray
+ The second EEG signal dataset.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ A 1 x n numpy ndarray of p-values for the
+ difference between Signals 1 and 2 at each position.
+ """
+
+ # compare the signal means
+ diff = CompareMeans.from_data(
+ extracted_EEGData_1,
+ extracted_EEGData_2
+ )
+
+ # perform t-test comparison
+ diff = diff.ttest_ind()
+
+ # get the p-values
+ pvalues = diff[1]
+ return pvalues
Visualises the difference between two EEG signals and tests
+time-point-wise the difference using T-Tests. Individual EEG
+Signals are shown as mean-line with shaded SEM, and time-points
+with significant differences are highlighted with overall-shading.
+
Parameters
+
+
extracted_EEGData_1 : np.ndarray
+
The first EEG signal dataset.
+
extracted_EEGData_2 : np.ndarray
+
The second EEG signal dataset.
+
sampling_frequency : float
+
The frequency in which the EEG data was recorded in Hertz.
+
significance_level : float
+
The significance threshold for which to accept a signal difference
+as significant. Default is 0.05.
+
start_sec : float
+
+
The initial / low bound of the time-scale in seconds.
+
stop_sec : float
+
The final / upper bound of the time-scale in seconds.
+
x_scale : float
+
A scaling factor to adjust the data's x-value range.
+E.g. x_scale = 1000 to adjust the time-scale to milliseconds.
+
y_scale : float
+
A scaling factor for the data's y-value range.
+E.g. y_scale = 1000 to adjust the signal-scale to millivolts.
+
make_legend : bool
+
Generates a legend for the plot.
+
+
Returns
+
+
pvalues : np.ndarray
+
+
The p-value from pair-wise T-Test comparison between the
+two signals at each signal position (timepoint).
+
+
+
+Expand source code
+
+
def difference_plot(extracted_EEGData_1:np.ndarray,
+ extracted_EEGData_2:np.ndarray,
+ sampling_frequency:float,
+ start_sec:float,
+ stop_sec:float,
+ significance_level:float = 0.05,
+ x_scale=10**3,
+ y_scale=10**3,
+ make_legend = False,
+ ax = None ) -> None:
+
+ """
+ Visualises the difference between two EEG signals and tests
+ time-point-wise the difference using T-Tests. Individual EEG
+ Signals are shown as mean-line with shaded SEM, and time-points
+ with significant differences are highlighted with overall-shading.
+
+ Parameters
+ ----------
+ extracted_EEGData_1 : np.ndarray
+ The first EEG signal dataset.
+
+ extracted_EEGData_2 : np.ndarray
+ The second EEG signal dataset.
+
+ sampling_frequency : float
+ The frequency in which the EEG data was recorded in `Hertz`.
+
+ significance_level : float
+ The significance threshold for which to accept a signal difference
+ as significant. Default is `0.05`.
+
+ start_sec : float
+ The initial / low bound of the time-scale in seconds.
+
+ stop_sec : float
+ The final / upper bound of the time-scale in seconds.
+
+ x_scale : float
+ A scaling factor to adjust the data's x-value range.
+ E.g. `x_scale = 1000` to adjust the time-scale to milliseconds.
+
+ y_scale : float
+ A scaling factor for the data's y-value range.
+ E.g. `y_scale = 1000` to adjust the signal-scale to millivolts.
+
+ make_legend : bool
+ Generates a legend for the plot.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ The p-value from pair-wise T-Test comparison between the
+ two signals at each signal position (timepoint).
+ """
+
+ # generate a new figure if no ax is given
+ if ax is None:
+ fig, ax = plt.subplots()
+ else:
+ fig = None
+
+ pvalues = _difference_plot_(ax,
+ extracted_EEGData_1,
+ extracted_EEGData_2,
+ significance_level,
+ sampling_frequency,
+ start_sec,
+ stop_sec,
+ x_scale,
+ y_scale,
+ make_legend
+ )
+
+ # we show the figure only if no ax was provided
+ # and thus no "bigger" figure is assembled elsewhere...
+ if fig is not None:
+ plt.show()
+
+ return pvalues
Computes the element-wise mean of an m x n numpy ndarray
+with m repeated datasets and n entries per set along axis 1
+(between repeated sets).
+
Parameters
+
+
extracted_EEGData : np.ndarray
+
An m x n numpy ndarray containing EEG data
+
+
Returns
+
+
means : np.ndarray
+
An 1 x n numpy ndarray containing the element-wise means between all m repeated sets.
+
+
+
+Expand source code
+
+
def mean(extracted_EEGData:np.ndarray) -> np.ndarray:
+ """
+ Computes the element-wise mean of an `m x n numpy ndarray`
+ with `m` repeated datasets and `n` entries per set along axis 1
+ (between repeated sets).
+
+ Parameters
+ ----------
+ extracted_EEGData : np.ndarray
+ An m x n numpy ndarray containing EEG data
+
+ Returns
+ -------
+ means : np.ndarray
+ An 1 x n numpy ndarray containing the element-wise means between all m repeated sets.
+ """
+ means = DescrStatsW( extracted_EEGData ).mean
+ return means
Visualises a single EGG signal dataset from an m x n numpy ndarray
+with m repeated datasets and n entries per set. It generates a solid
+line for the time-point-wise mean and a shaded area of the corresponding SEM.
+
Parameters
+
+
extracted_EEGData : np.ndarray
+
An m x n numpy ndarray of repeated EEG data.
+
sampling_frequency : float
+
The frequency in which the EEG data was recorded in Hertz.
+
start_sec : float
+
+
The initial / low bound of the time-scale in seconds.
+
stop_sec : float
+
The final / upper bound of the time-scale in seconds.
+
x_scale : float
+
A scaling factor to adjust the data's x-value range.
+E.g. x_scale = 1000 to adjust the time-scale to milliseconds.
+
y_scale : float
+
A scaling factor for the data's y-value range.
+E.g. y_scale = 1000 to adjust the signal-scale to millivolts.
+
baseline : np.ndarray
+
An array of baseline data to compare the signal to using a pair-wise t-test.
+This will introduce shaded boxes for regions that show significant differences
+between the signal and the baseline.
+
significance_level : float
+
The significance threshold for which to accept a signal difference
+as significant. Default is 0.05.
+
make_legend : bool
+
Generates a legend for the plot.
+
+
Returns
+
+
pvalues : np.ndarray
+
+
The p-value from pair-wise T-Test comparison between the
+signal and the baseline (if provided) at each signal position (timepoint).
+
+
+
+Expand source code
+
+
def plot_signal(
+ extracted_EEGData : np.ndarray,
+ sampling_frequency : float,
+ start_sec : float,
+ stop_sec : float,
+ x_scale = 10**3,
+ y_scale = 10**3,
+ baseline : np.ndarray = None,
+ significance_level:float = 0.05,
+ make_legend = False,
+ ax = None ) -> None:
+ """
+ Visualises a single EGG signal dataset from an `m x n numpy ndarray`
+ with `m` repeated datasets and `n` entries per set. It generates a solid
+ line for the time-point-wise mean and a shaded area of the corresponding SEM.
+
+ Parameters
+ ----------
+
+ extracted_EEGData : np.ndarray
+ An m x n numpy ndarray of repeated EEG data.
+
+ sampling_frequency : float
+ The frequency in which the EEG data was recorded in `Hertz`.
+
+ start_sec : float
+ The initial / low bound of the time-scale in seconds.
+
+ stop_sec : float
+ The final / upper bound of the time-scale in seconds.
+
+ x_scale : float
+ A scaling factor to adjust the data's x-value range.
+ E.g. `x_scale = 1000` to adjust the time-scale to milliseconds.
+
+ y_scale : float
+ A scaling factor for the data's y-value range.
+ E.g. `y_scale = 1000` to adjust the signal-scale to millivolts.
+
+ baseline : np.ndarray
+ An array of baseline data to compare the signal to using a pair-wise t-test.
+ This will introduce shaded boxes for regions that show significant differences
+ between the signal and the baseline.
+
+ significance_level : float
+ The significance threshold for which to accept a signal difference
+ as significant. Default is `0.05`.
+
+ make_legend : bool
+ Generates a legend for the plot.
+
+ Returns
+ -----
+ pvalues : np.ndarray
+ The p-value from pair-wise T-Test comparison between the
+ signal and the baseline (if provided) at each signal position (timepoint).
+ """
+
+ # generate a new figure if no ax is specified
+ if ax is None:
+ fig, ax = plt.subplots()
+ else:
+ fig = None
+
+ pvalues = _plot_(ax,
+ extracted_EEGData,
+ sampling_frequency,
+ start_sec,
+ stop_sec,
+ x_scale,
+ y_scale,
+ baseline,
+ make_legend,
+ significance_level = significance_level
+ )
+
+ # we show the figure only if no ax was provided
+ # and thus no "bigger" figure is assembled elsewhere...
+ if fig is not None:
+ plt.show()
+
+ return pvalues
Compute the standard error of the mean (SEM)
+of an m x n numpy ndarray with m repeated
+datasets and n entries per set along axis 1
+(between repeated sets).
+
Parameters
+
+
extracted_EEGData : np.ndarray
+
An m x n numpy ndarray containing EEG data
+
+
Returns
+
+
sems : np.ndarray
+
An 1 x n numpy ndarray containing the SEM between all m repeated sets.
+
+
+
+Expand source code
+
+
def sem(extracted_EEGData:np.ndarray) -> np.ndarray:
+ """
+ Compute the standard error of the mean (SEM)
+ of an `m x n numpy ndarray` with `m` repeated
+ datasets and `n` entries per set along axis 1
+ (between repeated sets).
+
+ Parameters
+ ----------
+ extracted_EEGData : np.ndarray
+ An m x n numpy ndarray containing EEG data
+
+ Returns
+ -------
+ sems : np.ndarray
+ An 1 x n numpy ndarray containing the SEM between all m repeated sets.
+
+ """
+ sems = DescrStatsW( extracted_EEGData )
+ length = len( extracted_EEGData )
+ sems = sems.std / np.sqrt( length )
+ return sems
+
+
+
+
+
\ No newline at end of file
diff --git a/docs/EEGToolkit/auxiliary.html b/docs/EEGToolkit/auxiliary.html
new file mode 100644
index 0000000..1f7dd30
--- /dev/null
+++ b/docs/EEGToolkit/auxiliary.html
@@ -0,0 +1,613 @@
+
+
+
+
+
+
+EEGToolkit.auxiliary API documentation
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Module EEGToolkit.auxiliary
+
+
+
Auxiliary functions to work within the streamlit environment
+
+
+Expand source code
+
+
"""
+Auxiliary functions to work within the streamlit environment
+"""
+
+from .EEGData import EEGData, supported_filetypes
+import streamlit as st
+from copy import deepcopy
+import os
+
+class Session:
+ """
+ A class to handle storing values into and getting them from the streamlit session state.
+ """
+ def __init__(self):
+
+ # a list of keys to include in the
+ # export session dictionary
+ self._to_export = set()
+
+ def add( self, key, value, export = True ):
+ """
+ Add a new item to the session but
+ can also change an existing item.
+
+ Parameters
+ ----------
+ key : str
+ The key of the new item
+ value
+ The value to store of the new item
+ export : bool
+ Wether or not to include the item in the exported session dictioanry.
+ """
+ st.session_state[ key ] = value
+
+ if export:
+ self._to_export.add( key )
+
+ def set( self, key, value, export = True ):
+ """
+ A synonym of `add`
+ """
+ self.add( key, value, export )
+
+ def remove( self, key ):
+ """
+ Remove an item from the session
+ """
+ try:
+ st.session_state.pop( key )
+ self._to_export.remove( key )
+ except:
+ pass
+
+ def get( self, key, default = None, rm = False, copy = False ):
+ """
+ Tries to get an item from the session
+ and returns the default if this fails.
+ If rm = True is set, the item will be
+ removed after getting.
+ If copy = True is set, it will return a deepcopy.
+ """
+ try:
+ value = st.session_state[ key ]
+ if copy:
+ value = deepcopy(value)
+ except:
+ value = default
+
+ if rm:
+ self.remove( key )
+
+ return value
+
+ def setup( self, key, value ):
+ """
+ Stores an item to the session only if it's not already present.
+ """
+ if not self.exists( key ):
+ self.add( key, value )
+
+ def exists( self, key ):
+ """
+ Checks if an item is in the session. Returns True if so.
+ """
+ verdict = key in st.session_state
+ return verdict
+
+ def hasvalue( self, key ):
+ """
+ Checks if an item is in the session and is also not None.
+ Returns True if both conditions are fulfilled.
+ """
+ exists = self.exists( key )
+ not_none = self.get( key, None ) is not None
+ verdict = exists and not_none
+ return verdict
+
+ def export( self ):
+ """
+ Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+ """
+ to_export = st.session_state[ self._to_export ]
+ return to_export
+
+
+
+class stEEGData( EEGData ):
+ """
+ A streamlit compatible version of EEGData
+ The primary difference here is that it replaces the original filepath variables
+ with filepath.name arguments to work with the string and not the UploadedFile objects.
+ """
+ def __init__( self, *args, **kwargs ):
+ super().__init__(*args, **kwargs)
+
+ def _filesuffix(self, filepath):
+ """
+ Returns the suffix from a filepath
+ """
+ suffix = os.path.basename( filepath.name )
+ suffix = suffix.split(".")[-1]
+ return suffix
+
+ def _csv_delimiter(self, filepath):
+ """
+ Checks if a csv file is , or ; delimited and returns the
+ correct delmiter to use...
+ """
+ # open the file and read
+ content = deepcopy(filepath).read().decode()
+
+ # check if a semicolon is present
+ # if so, we delimit at ;
+ has_semicolon = ";" in content
+ delimiter = ";" if has_semicolon else ","
+
+ return delimiter
+
+ def _check_sanity(self, signal_path, event_path, sampling_frequency):
+ """
+ Checks if valid data inputs were provided
+ """
+
+ # check if the datafiles conform to suppored filetypes
+ fname = os.path.basename(signal_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The signal datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ fname = os.path.basename(event_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The event datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ # check if the frequency is a positive number
+ if not sampling_frequency > 0:
+ raise ValueError( f"Sampling frequency {sampling_frequency} must be a positive value" )
+
+
+
+
+
+
+
+
+
+
Classes
+
+
+class Session
+
+
+
A class to handle storing values into and getting them from the streamlit session state.
+
+
+Expand source code
+
+
class Session:
+ """
+ A class to handle storing values into and getting them from the streamlit session state.
+ """
+ def __init__(self):
+
+ # a list of keys to include in the
+ # export session dictionary
+ self._to_export = set()
+
+ def add( self, key, value, export = True ):
+ """
+ Add a new item to the session but
+ can also change an existing item.
+
+ Parameters
+ ----------
+ key : str
+ The key of the new item
+ value
+ The value to store of the new item
+ export : bool
+ Wether or not to include the item in the exported session dictioanry.
+ """
+ st.session_state[ key ] = value
+
+ if export:
+ self._to_export.add( key )
+
+ def set( self, key, value, export = True ):
+ """
+ A synonym of `add`
+ """
+ self.add( key, value, export )
+
+ def remove( self, key ):
+ """
+ Remove an item from the session
+ """
+ try:
+ st.session_state.pop( key )
+ self._to_export.remove( key )
+ except:
+ pass
+
+ def get( self, key, default = None, rm = False, copy = False ):
+ """
+ Tries to get an item from the session
+ and returns the default if this fails.
+ If rm = True is set, the item will be
+ removed after getting.
+ If copy = True is set, it will return a deepcopy.
+ """
+ try:
+ value = st.session_state[ key ]
+ if copy:
+ value = deepcopy(value)
+ except:
+ value = default
+
+ if rm:
+ self.remove( key )
+
+ return value
+
+ def setup( self, key, value ):
+ """
+ Stores an item to the session only if it's not already present.
+ """
+ if not self.exists( key ):
+ self.add( key, value )
+
+ def exists( self, key ):
+ """
+ Checks if an item is in the session. Returns True if so.
+ """
+ verdict = key in st.session_state
+ return verdict
+
+ def hasvalue( self, key ):
+ """
+ Checks if an item is in the session and is also not None.
+ Returns True if both conditions are fulfilled.
+ """
+ exists = self.exists( key )
+ not_none = self.get( key, None ) is not None
+ verdict = exists and not_none
+ return verdict
+
+ def export( self ):
+ """
+ Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+ """
+ to_export = st.session_state[ self._to_export ]
+ return to_export
+
+
Methods
+
+
+def add(self, key, value, export=True)
+
+
+
Add a new item to the session but
+can also change an existing item.
+
Parameters
+
+
key : str
+
The key of the new item
+
value
+
The value to store of the new item
+
export : bool
+
Wether or not to include the item in the exported session dictioanry.
+
+
+
+Expand source code
+
+
def add( self, key, value, export = True ):
+ """
+ Add a new item to the session but
+ can also change an existing item.
+
+ Parameters
+ ----------
+ key : str
+ The key of the new item
+ value
+ The value to store of the new item
+ export : bool
+ Wether or not to include the item in the exported session dictioanry.
+ """
+ st.session_state[ key ] = value
+
+ if export:
+ self._to_export.add( key )
+
+
+
+def exists(self, key)
+
+
+
Checks if an item is in the session. Returns True if so.
+
+
+Expand source code
+
+
def exists( self, key ):
+ """
+ Checks if an item is in the session. Returns True if so.
+ """
+ verdict = key in st.session_state
+ return verdict
+
+
+
+def export(self)
+
+
+
Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+
+
+Expand source code
+
+
def export( self ):
+ """
+ Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+ """
+ to_export = st.session_state[ self._to_export ]
+ return to_export
Tries to get an item from the session
+and returns the default if this fails.
+If rm = True is set, the item will be
+removed after getting.
+If copy = True is set, it will return a deepcopy.
+
+
+Expand source code
+
+
def get( self, key, default = None, rm = False, copy = False ):
+ """
+ Tries to get an item from the session
+ and returns the default if this fails.
+ If rm = True is set, the item will be
+ removed after getting.
+ If copy = True is set, it will return a deepcopy.
+ """
+ try:
+ value = st.session_state[ key ]
+ if copy:
+ value = deepcopy(value)
+ except:
+ value = default
+
+ if rm:
+ self.remove( key )
+
+ return value
+
+
+
+def hasvalue(self, key)
+
+
+
Checks if an item is in the session and is also not None.
+Returns True if both conditions are fulfilled.
+
+
+Expand source code
+
+
def hasvalue( self, key ):
+ """
+ Checks if an item is in the session and is also not None.
+ Returns True if both conditions are fulfilled.
+ """
+ exists = self.exists( key )
+ not_none = self.get( key, None ) is not None
+ verdict = exists and not_none
+ return verdict
+
+
+
+def remove(self, key)
+
+
+
Remove an item from the session
+
+
+Expand source code
+
+
def remove( self, key ):
+ """
+ Remove an item from the session
+ """
+ try:
+ st.session_state.pop( key )
+ self._to_export.remove( key )
+ except:
+ pass
Stores an item to the session only if it's not already present.
+
+
+Expand source code
+
+
def setup( self, key, value ):
+ """
+ Stores an item to the session only if it's not already present.
+ """
+ if not self.exists( key ):
+ self.add( key, value )
+
+
+
+
+
+class stEEGData
+(*args, **kwargs)
+
+
+
A streamlit compatible version of EEGData
+The primary difference here is that it replaces the original filepath variables
+with filepath.name arguments to work with the string and not the UploadedFile objects.
+
+
+Expand source code
+
+
class stEEGData( EEGData ):
+ """
+ A streamlit compatible version of EEGData
+ The primary difference here is that it replaces the original filepath variables
+ with filepath.name arguments to work with the string and not the UploadedFile objects.
+ """
+ def __init__( self, *args, **kwargs ):
+ super().__init__(*args, **kwargs)
+
+ def _filesuffix(self, filepath):
+ """
+ Returns the suffix from a filepath
+ """
+ suffix = os.path.basename( filepath.name )
+ suffix = suffix.split(".")[-1]
+ return suffix
+
+ def _csv_delimiter(self, filepath):
+ """
+ Checks if a csv file is , or ; delimited and returns the
+ correct delmiter to use...
+ """
+ # open the file and read
+ content = deepcopy(filepath).read().decode()
+
+ # check if a semicolon is present
+ # if so, we delimit at ;
+ has_semicolon = ";" in content
+ delimiter = ";" if has_semicolon else ","
+
+ return delimiter
+
+ def _check_sanity(self, signal_path, event_path, sampling_frequency):
+ """
+ Checks if valid data inputs were provided
+ """
+
+ # check if the datafiles conform to suppored filetypes
+ fname = os.path.basename(signal_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The signal datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ fname = os.path.basename(event_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The event datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ # check if the frequency is a positive number
+ if not sampling_frequency > 0:
+ raise ValueError( f"Sampling frequency {sampling_frequency} must be a positive value" )
+
+
+
+
+
\ No newline at end of file
diff --git a/docs/EEGToolkit/auxiliary/auxiliary.html b/docs/EEGToolkit/auxiliary/auxiliary.html
new file mode 100644
index 0000000..ab5090d
--- /dev/null
+++ b/docs/EEGToolkit/auxiliary/auxiliary.html
@@ -0,0 +1,613 @@
+
+
+
+
+
+
+EEGToolkit.auxiliary.auxiliary API documentation
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Module EEGToolkit.auxiliary.auxiliary
+
+
+
Auxiliary functions to work within the streamlit environment
+
+
+Expand source code
+
+
"""
+Auxiliary functions to work within the streamlit environment
+"""
+
+from ..EEGData import EEGData, supported_filetypes
+import streamlit as st
+from copy import deepcopy
+import os
+
+class Session:
+ """
+ A class to handle storing values into and getting them from the streamlit session state.
+ """
+ def __init__(self):
+
+ # a list of keys to include in the
+ # export session dictionary
+ self._to_export = set()
+
+ def add( self, key, value, export = True ):
+ """
+ Add a new item to the session but
+ can also change an existing item.
+
+ Parameters
+ ----------
+ key : str
+ The key of the new item
+ value
+ The value to store of the new item
+ export : bool
+ Wether or not to include the item in the exported session dictioanry.
+ """
+ st.session_state[ key ] = value
+
+ if export:
+ self._to_export.add( key )
+
+ def set( self, key, value, export = True ):
+ """
+ A synonym of `add`
+ """
+ self.add( key, value, export )
+
+ def remove( self, key ):
+ """
+ Remove an item from the session
+ """
+ try:
+ st.session_state.pop( key )
+ self._to_export.remove( key )
+ except:
+ pass
+
+ def get( self, key, default = None, rm = False, copy = False ):
+ """
+ Tries to get an item from the session
+ and returns the default if this fails.
+ If rm = True is set, the item will be
+ removed after getting.
+ If copy = True is set, it will return a deepcopy.
+ """
+ try:
+ value = st.session_state[ key ]
+ if copy:
+ value = deepcopy(value)
+ except:
+ value = default
+
+ if rm:
+ self.remove( key )
+
+ return value
+
+ def setup( self, key, value ):
+ """
+ Stores an item to the session only if it's not already present.
+ """
+ if not self.exists( key ):
+ self.add( key, value )
+
+ def exists( self, key ):
+ """
+ Checks if an item is in the session. Returns True if so.
+ """
+ verdict = key in st.session_state
+ return verdict
+
+ def hasvalue( self, key ):
+ """
+ Checks if an item is in the session and is also not None.
+ Returns True if both conditions are fulfilled.
+ """
+ exists = self.exists( key )
+ not_none = self.get( key, None ) is not None
+ verdict = exists and not_none
+ return verdict
+
+ def export( self ):
+ """
+ Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+ """
+ to_export = st.session_state[ self._to_export ]
+ return to_export
+
+
+
+class stEEGData( EEGData ):
+ """
+ A streamlit compatible version of EEGData
+ The primary difference here is that it replaces the original filepath variables
+ with filepath.name arguments to work with the string and not the UploadedFile objects.
+ """
+ def __init__( self, *args, **kwargs ):
+ super().__init__(*args, **kwargs)
+
+ def _filesuffix(self, filepath):
+ """
+ Returns the suffix from a filepath
+ """
+ suffix = os.path.basename( filepath.name )
+ suffix = suffix.split(".")[-1]
+ return suffix
+
+ def _csv_delimiter(self, filepath):
+ """
+ Checks if a csv file is , or ; delimited and returns the
+ correct delmiter to use...
+ """
+ # open the file and read
+ content = deepcopy(filepath).read().decode()
+
+ # check if a semicolon is present
+ # if so, we delimit at ;
+ has_semicolon = ";" in content
+ delimiter = ";" if has_semicolon else ","
+
+ return delimiter
+
+ def _check_sanity(self, signal_path, event_path, sampling_frequency):
+ """
+ Checks if valid data inputs were provided
+ """
+
+ # check if the datafiles conform to suppored filetypes
+ fname = os.path.basename(signal_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The signal datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ fname = os.path.basename(event_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The event datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ # check if the frequency is a positive number
+ if not sampling_frequency > 0:
+ raise ValueError( f"Sampling frequency {sampling_frequency} must be a positive value" )
+
+
+
+
+
+
+
+
+
+
Classes
+
+
+class Session
+
+
+
A class to handle storing values into and getting them from the streamlit session state.
+
+
+Expand source code
+
+
class Session:
+ """
+ A class to handle storing values into and getting them from the streamlit session state.
+ """
+ def __init__(self):
+
+ # a list of keys to include in the
+ # export session dictionary
+ self._to_export = set()
+
+ def add( self, key, value, export = True ):
+ """
+ Add a new item to the session but
+ can also change an existing item.
+
+ Parameters
+ ----------
+ key : str
+ The key of the new item
+ value
+ The value to store of the new item
+ export : bool
+ Wether or not to include the item in the exported session dictioanry.
+ """
+ st.session_state[ key ] = value
+
+ if export:
+ self._to_export.add( key )
+
+ def set( self, key, value, export = True ):
+ """
+ A synonym of `add`
+ """
+ self.add( key, value, export )
+
+ def remove( self, key ):
+ """
+ Remove an item from the session
+ """
+ try:
+ st.session_state.pop( key )
+ self._to_export.remove( key )
+ except:
+ pass
+
+ def get( self, key, default = None, rm = False, copy = False ):
+ """
+ Tries to get an item from the session
+ and returns the default if this fails.
+ If rm = True is set, the item will be
+ removed after getting.
+ If copy = True is set, it will return a deepcopy.
+ """
+ try:
+ value = st.session_state[ key ]
+ if copy:
+ value = deepcopy(value)
+ except:
+ value = default
+
+ if rm:
+ self.remove( key )
+
+ return value
+
+ def setup( self, key, value ):
+ """
+ Stores an item to the session only if it's not already present.
+ """
+ if not self.exists( key ):
+ self.add( key, value )
+
+ def exists( self, key ):
+ """
+ Checks if an item is in the session. Returns True if so.
+ """
+ verdict = key in st.session_state
+ return verdict
+
+ def hasvalue( self, key ):
+ """
+ Checks if an item is in the session and is also not None.
+ Returns True if both conditions are fulfilled.
+ """
+ exists = self.exists( key )
+ not_none = self.get( key, None ) is not None
+ verdict = exists and not_none
+ return verdict
+
+ def export( self ):
+ """
+ Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+ """
+ to_export = st.session_state[ self._to_export ]
+ return to_export
+
+
Methods
+
+
+def add(self, key, value, export=True)
+
+
+
Add a new item to the session but
+can also change an existing item.
+
Parameters
+
+
key : str
+
The key of the new item
+
value
+
The value to store of the new item
+
export : bool
+
Wether or not to include the item in the exported session dictioanry.
+
+
+
+Expand source code
+
+
def add( self, key, value, export = True ):
+ """
+ Add a new item to the session but
+ can also change an existing item.
+
+ Parameters
+ ----------
+ key : str
+ The key of the new item
+ value
+ The value to store of the new item
+ export : bool
+ Wether or not to include the item in the exported session dictioanry.
+ """
+ st.session_state[ key ] = value
+
+ if export:
+ self._to_export.add( key )
+
+
+
+def exists(self, key)
+
+
+
Checks if an item is in the session. Returns True if so.
+
+
+Expand source code
+
+
def exists( self, key ):
+ """
+ Checks if an item is in the session. Returns True if so.
+ """
+ verdict = key in st.session_state
+ return verdict
+
+
+
+def export(self)
+
+
+
Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+
+
+Expand source code
+
+
def export( self ):
+ """
+ Exports a copy of the session state for better readablity wherein non-intelligible entries are removed.
+ """
+ to_export = st.session_state[ self._to_export ]
+ return to_export
Tries to get an item from the session
+and returns the default if this fails.
+If rm = True is set, the item will be
+removed after getting.
+If copy = True is set, it will return a deepcopy.
+
+
+Expand source code
+
+
def get( self, key, default = None, rm = False, copy = False ):
+ """
+ Tries to get an item from the session
+ and returns the default if this fails.
+ If rm = True is set, the item will be
+ removed after getting.
+ If copy = True is set, it will return a deepcopy.
+ """
+ try:
+ value = st.session_state[ key ]
+ if copy:
+ value = deepcopy(value)
+ except:
+ value = default
+
+ if rm:
+ self.remove( key )
+
+ return value
+
+
+
+def hasvalue(self, key)
+
+
+
Checks if an item is in the session and is also not None.
+Returns True if both conditions are fulfilled.
+
+
+Expand source code
+
+
def hasvalue( self, key ):
+ """
+ Checks if an item is in the session and is also not None.
+ Returns True if both conditions are fulfilled.
+ """
+ exists = self.exists( key )
+ not_none = self.get( key, None ) is not None
+ verdict = exists and not_none
+ return verdict
+
+
+
+def remove(self, key)
+
+
+
Remove an item from the session
+
+
+Expand source code
+
+
def remove( self, key ):
+ """
+ Remove an item from the session
+ """
+ try:
+ st.session_state.pop( key )
+ self._to_export.remove( key )
+ except:
+ pass
Stores an item to the session only if it's not already present.
+
+
+Expand source code
+
+
def setup( self, key, value ):
+ """
+ Stores an item to the session only if it's not already present.
+ """
+ if not self.exists( key ):
+ self.add( key, value )
+
+
+
+
+
+class stEEGData
+(*args, **kwargs)
+
+
+
A streamlit compatible version of EEGData
+The primary difference here is that it replaces the original filepath variables
+with filepath.name arguments to work with the string and not the UploadedFile objects.
+
+
+Expand source code
+
+
class stEEGData( EEGData ):
+ """
+ A streamlit compatible version of EEGData
+ The primary difference here is that it replaces the original filepath variables
+ with filepath.name arguments to work with the string and not the UploadedFile objects.
+ """
+ def __init__( self, *args, **kwargs ):
+ super().__init__(*args, **kwargs)
+
+ def _filesuffix(self, filepath):
+ """
+ Returns the suffix from a filepath
+ """
+ suffix = os.path.basename( filepath.name )
+ suffix = suffix.split(".")[-1]
+ return suffix
+
+ def _csv_delimiter(self, filepath):
+ """
+ Checks if a csv file is , or ; delimited and returns the
+ correct delmiter to use...
+ """
+ # open the file and read
+ content = deepcopy(filepath).read().decode()
+
+ # check if a semicolon is present
+ # if so, we delimit at ;
+ has_semicolon = ";" in content
+ delimiter = ";" if has_semicolon else ","
+
+ return delimiter
+
+ def _check_sanity(self, signal_path, event_path, sampling_frequency):
+ """
+ Checks if valid data inputs were provided
+ """
+
+ # check if the datafiles conform to suppored filetypes
+ fname = os.path.basename(signal_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The signal datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ fname = os.path.basename(event_path.name)
+ if not any( [ fname.endswith(suffix) for suffix in supported_filetypes ] ):
+ suffix = fname.split(".")[-1]
+ raise TypeError( f"The event datafile could not be interpreted ('.{suffix}'), only {supported_filetypes} files are supported!" )
+
+ # check if the frequency is a positive number
+ if not sampling_frequency > 0:
+ raise ValueError( f"Sampling frequency {sampling_frequency} must be a positive value" )
+
+
+
+
+
\ No newline at end of file
diff --git a/docs/EEGToolkit/index.html b/docs/EEGToolkit/index.html
index 1c01d39..e7728e6 100644
--- a/docs/EEGToolkit/index.html
+++ b/docs/EEGToolkit/index.html
@@ -38,7 +38,7 @@
Package EEGToolkit
Example Usage
To use this module for data analysis, only three steps are necessary,
-(1st) setup of the EEGToolkit.EEGData object, (2nd) event data extraction, and (3rd)
+(1st) setup of the EEGToolkit.EEGData object, (2nd) event data extraction, and (3rd)
data summary (which performs signal comparison).
# setting up the EEGData with some datafiles
eeg = EEGData( eeg_path = "data/eeg.npy", event_path = "data/events.npy", sampling_frequency = 500 )
@@ -108,14 +108,22 @@
This module provides a data class EEGToolkit.EEGData to work with EEG signal data for event-reaction-time delay experiments.
-It works with two separate input …
This script defines an interactive
+web-app for the EEGData package.
+
+
+Expand source code
+
+
"""
+This script defines an interactive
+web-app for the EEGData package.
+"""
+
+import streamlit as st
+from EEGToolkit.EEGData import supported_filetypes
+import pandas as pd
+from EEGToolkit.auxiliary import Session, stEEGData
+import os
+
+session = Session()
+
+# =================================================================
+# Header Section
+# =================================================================
+
+
+st.markdown(
+ """
+# EEGToolKit Viewer
+ """,
+ )
+
+# =================================================================
+# Data Input Section
+# =================================================================
+
+file_container = st.container()
+signal_col, events_col = file_container.columns( 2 )
+
+# add a signal datafile
+signal_file = signal_col.file_uploader(
+ "EEG Signal Data",
+ help = f"Upload here a datafile specifying a 1D array of EEG signal data.",
+ type = supported_filetypes
+ )
+if signal_file is not None:
+ session.add( "signal_filename", signal_file.name )
+ session.add( "signal_file", signal_file, export = False )
+
+# add an events metadata file
+events_file = events_col.file_uploader(
+ "Events Metadata",
+ help = f"Upload here a datafile specifying a 2-column array of events metadata for the signal data file.",
+ type = supported_filetypes
+ )
+if events_file is not None:
+ session.add( "events_filename", events_file.name )
+ session.add( "events_file", events_file, export = False )
+
+
+# =================================================================
+# Data Processing Section
+# =================================================================
+
+# In case anybody is wondering why the weird column setup here. The answer
+# is that streamlit does not (yet) support nested columns nor stretching objects
+# over multiple columns. Hence, one column, one object, if they should "appear"
+# in side by side, one above the other etc. layouts we need to have multiple
+# column blocks...
+
+controls = st.container()
+upper_ctrl_col1, upper_ctrl_col2 = controls.columns( (1, 3) )
+mid_ctrl_col1, mid_ctrl_col2, mid_ctrl_col3 = controls.columns( (2, 3, 3 ) )
+lower_ctrl_col1, lower_ctrl_col2, lower_ctrl_col3 = controls.columns( (2, 3, 3 ) )
+
+compute_baseline = upper_ctrl_col2.checkbox(
+ "Compare Baseline",
+ help = "Compute the baseline of a given signal type / event and compares the signal to the baseline through positition-wise T-Tests.",
+ value = True
+ )
+
+significance_level = upper_ctrl_col2.number_input(
+ "Significance Level",
+ help = "Choose a significance threshold level for signal-signal and signal-baseline comparisons",
+ min_value = 1e-5, max_value = 1.0, value = 0.05, step = 1e-3, format = "%e"
+ )
+session.add( "significance_level", significance_level )
+
+frequency = upper_ctrl_col2.number_input(
+ "Sampling Frequency",
+ help = "The sampling frequency in `Hertz` at which the signal was recorded",
+ min_value = 1, max_value = 100000, value = 500, step = 100, format = "%d",
+ )
+session.add( "frequency", frequency )
+
+
+start_sec = mid_ctrl_col2.number_input(
+ "Upstream Buffer",
+ help = "The time buffer pre-event to extract (in seconds).",
+ min_value = 1e-4, max_value = 10.0, value = 0.01, step = 0.001, format = "%e",
+ )
+session.add( "upstream_buffer", start_sec )
+
+stop_sec = mid_ctrl_col3.number_input(
+ "Downstream Buffer",
+ help = "The time buffer post-event to extract (in seconds).",
+ min_value = 1e-4, max_value = 10.0, value = 1.0, step = 0.001, format = "%e",
+ )
+session.add( "downstream_buffer", stop_sec )
+
+xscale = lower_ctrl_col2.number_input(
+ "Timeframe Scale",
+ help = "A scaling factor to adjust x-axis time scale. E.g. `1000` to adjust data from seconds to milliseconds.",
+ min_value = 1, max_value = 100000, value = 1000, step = 100, format = "%d",
+ )
+session.add( "timescale", xscale )
+
+yscale = lower_ctrl_col3.number_input(
+ "Signal Scale",
+ help = "A scaling factor to adjust y-axis signal scale. E.g. `1000` to adjust from volts to millivolts.",
+ min_value = 1, max_value = 100000, value = 1000, step = 100, format = "%d",
+ )
+session.add( "signalscale", yscale )
+
+# slightly ugly splitting of the text, but the only way to adjust the text to the column layout I want...
+mid_ctrl_col1.markdown( "Once your setup is done, press the `Compute` button below to " )
+lower_ctrl_col1.markdown( "commence data evaluation." )
+run_button = lower_ctrl_col1.button(
+ "Compute",
+ help = "Perform computations and output a figure",
+ )
+
+# =================================================================
+# Computational Section
+# =================================================================
+
+figure_container = st.container()
+if run_button:
+ if signal_file is None:
+ st.error( "No Signal File was provided so far!" )
+ st.stop()
+ if events_file is None:
+ st.error( "No Events File was provided so far!" )
+ st.stop()
+ eegdata = stEEGData(
+ signal_path = signal_file,
+ event_path = events_file,
+ sampling_frequency = frequency
+ )
+ eegdata.extract( -start_sec, stop_sec )
+
+ if compute_baseline:
+ eegdata.baseline()
+ fig = eegdata.summary(
+ x_scale = xscale,
+ y_scale = yscale,
+ significance_level = significance_level
+ )
+
+ figure_container.pyplot( fig )