Adding pipelines for cli analysis (#202)

* started pipelines function * almost working simple function equivalents of nb scripts * fix: fixed import (brainflow updated API) * sqc fixes for unicorn (#176) * Ignore pushes * Trying to create a cli * Stepping through the problem * First commit * Fixing pause in signal quality check * Fixing Signal quality check problem * fix the technical debt * Save path done for automated saving pdf * I feel amazing * Almost through * Update eegnb/cli/__main__.py Co-authored-by: Erik Bjäreholt <erik@bjareho.lt> * Trying to create cli but it's being really painful * Extra word cli error * Changed example handling * Pain * Adding whole datapath * Finally fixed cli * hmm * Looking good * added hyperlink * Having some issues with detecting css and image deltetion * Just the css now * Fixed the css linking problem though it's a weird soln * Automated running, still fnames problem * Hahahah embedded images in html * Improving code * Okay now * Look at that * Almost there just the two figures now * Now * Added attrdict to do with cli error Co-authored-by: John Griffiths <j.davidgriffiths@gmail.com> Co-authored-by: Erik Bjäreholt <erik@bjareho.lt> Co-authored-by: John Griffiths <JohnGriffiths@users.noreply.github.com>
NeuroTechX · Oct 16, 2022 · 6714740 · 6714740
1 parent 368afa3
commit 6714740
Show file tree

Hide file tree

Showing 12 changed files with 641 additions and 41 deletions.
diff --git a/eegnb/analysis/analysis_report.html b/eegnb/analysis/analysis_report.html
@@ -0,0 +1,39 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <link href="styling.css" rel="stylesheet" />
+    <title>Analysis Report</title>
+  </head>
+  <body>
+    <div class="topnav">
+      <a href="#Description">Description</a>
+      <a href="#Raw Epoch">Raw Epoch</a>
+      <a href="#Stimulus Response">Stimulus Response</a>
+    </div>
+    <div id="Description">
+      <h1>Analysis Report</h1>
+      <p>
+        <b></b>Experiment Name: {} <br>
+        Subject Id: {} <br>
+        Session Id: {} <br>
+        EEG Device: {} <br>
+        Drop Percentage: {} <br> <br>
+        This is an analysis report for the experiment. <br> For more information about the experiment, please visit the <a href="https://neurotechx.github.io/eeg-notebooks/">documentation</a>
+      </p>
+    </div>
+    <div id="Raw Epoch">
+      <h2>Raw Epoch</h2>
+      <p>
+        The raw epoch is shown below. The raw epoch is the data that is recorded from the EEG headset. The raw epoch is then processed to remove noise and artifacts.
+      </p>
+      <img src="power_spectrum.png" alt="Raw Epoch" />
+    </div>
+    <div id="Stimulus Response">
+      <h2>Stimulus Response</h2>
+      <p>
+        The stimulus response is shown below. The stimulus response is the data that is recorded from the EEG headset after removing noise and artifacts.
+      </p>
+      <img src="erp_plot.png" alt="Stimulus Response" />
+    </div>
+  </body>
+</html>
diff --git a/eegnb/analysis/analysis_report.py b/eegnb/analysis/analysis_report.py
@@ -0,0 +1,95 @@
+
+# Generating html using Python
+
+from airium import Airium
+from typing import Dict
+import os
+import eegnb
+import base64
+
+a = Airium()
+
+def get_experiment_information(experiment:str):
+    analysis_save_path = os.path.join(os.path.dirname(eegnb.__file__), "analysis")
+    file_path = os.path.join(analysis_save_path, "experiment_descriptions")
+
+    with open(os.path.join(file_path, experiment + ".txt"), 'r') as f:
+        experiment_text = f.readlines()
+
+    return experiment_text
+
+def get_img_string(image_save_path):
+    """ Returns image as string to embed into the html report """
+    return base64.b64encode(open(image_save_path, "rb").read()).decode()
+
+def get_html(experimental_parameters: Dict):
+
+    # add variable to store the link
+    analysis_save_path = os.path.join(os.path.dirname(eegnb.__file__), "analysis")
+    css_path = os.path.join(analysis_save_path, "styling.css")
+    eeg_device, experiment, subject, session, example, drop_percentage, epochs_chosen = experimental_parameters.values()
+
+    erp_image_path = os.path.join(os.getcwd(), "erp_plot.png")
+    pos_image_path = os.path.join(os.getcwd(), "power_spectrum.png")
+
+    experiment_text = get_experiment_information(experiment)
+
+
+    """ Possibility of unique experiment text - decision to be made """
+    #experiment_text = ""
+    #with open('experiment_descriptions/{}.txt'.format(experiment), 'r') as f:
+     #   experiment_text = f.readlines()
+
+    a('<!DOCTYPE html>')
+    with a.html():
+        with a.head():
+            a.link(href=css_path, rel='stylesheet', type="text/css")
+            a.title(_t="Analysis Report")
+
+        with a.body():
+
+            # Navigation bar
+            with a.div(klass="topnav"):
+                a.a(_t="Description", href="#Description")
+                a.a(_t="Raw Epoch", href="#Raw Epoch")
+                a.a(_t="Stimulus Response", href="#Stimulus Response")
+
+            # Description
+            with a.div(id="Description"):
+                a.h1(_t="Analysis Report")
+                with a.p():
+                    a("Experiment Name: {} <br>".format(experiment))
+
+                    if example:
+                        a("Example File <br>")
+                    else:
+                        a("Subject Id: {} <br>".format(subject))
+                        a("Session Id: {} <br>".format(session))
+
+                    a("EEG Device: {} <br>".format(eeg_device))
+                    a('This is an analysis report for the experiment. <br> For more information about the experiment, please visit the <a href="https://neurotechx.github.io/eeg-notebooks/">documentation</a><br><br>')
+                    a("{}<br>".format(experiment_text[0]))
+                    a("{}<br>".format(experiment_text[1]))
+
+            # Raw Epoch
+            with a.div(id="Raw Epoch"):
+                a.h2(_t="Raw Epoch")
+                with a.p():
+                    a("The power spectrum of the raw epoch is displayed below. The raw epoch is then processed to remove noise and artifacts.")
+                a.img(src="data:image/png;base64, {}".format(get_img_string(pos_image_path)), alt="Raw Epoch")
+
+            # Stimulus Response
+            with a.div(id="Stimulus Response"):
+                a.h2(_t="Stimulus Response")
+                with a.p():
+                    a("The stimulus response is shown below. The stimulus response is the amplitude response at the specific timescales where the response to the stimulus can be detected. <br>")
+                    a("Epochs chosen: {} <br>".format(epochs_chosen))
+                    a("Drop Percentage: {} %<br> <br>".format(round(drop_percentage,2)))
+                a.img(src="data:image/png;base64, {}".format(get_img_string(erp_image_path)), alt="Stimulus Response")
+
+    # Delete the images
+    os.remove(erp_image_path)
+    os.remove(pos_image_path)
+
+    # Return the html
+    return str(a)
diff --git a/eegnb/analysis/experiment_descriptions/visual-N170.txt b/eegnb/analysis/experiment_descriptions/visual-N170.txt
@@ -0,0 +1,2 @@
+The N170 is a large negative event-related potential (ERP) component that occurs after the detection of faces, but not objects, scrambled faces, or other body parts such as hands. 
+In the experiment we aim to detect the N170 using faces and houses as our stimuli. 
diff --git a/eegnb/analysis/experiment_descriptions/visual-P300.txt b/eegnb/analysis/experiment_descriptions/visual-P300.txt
@@ -0,0 +1,2 @@
+The P300 is a positive event-related potential (ERP) that occurs around 300ms after perceiving a novel or unexpected stimulus. It is most commonly elicited through ‘oddball’ experimental paradigms, where a certain subtype of stimulus is presented rarely amidst a background of another more common type of stimulus. 
+In the experiment, we aimed to elicit P300 response using a visual oddball stimulation.
diff --git a/eegnb/analysis/pipelines.py b/eegnb/analysis/pipelines.py
@@ -0,0 +1,243 @@
+"""
+
+CLI Pipeline for Analysis of EEGNB Recorded Data
+
+To do: 
+1. Beautify analysis pdf
+2. Handle cli automated errors for report creation
+
+Usage:
+
+For Recorded Data:
+
+from eegnb.analysis.pipelines import create_analysis_report()
+create_analysis_report(experiment, eegdevice, subject, session, filepath)s
+
+For Example Datasets:
+
+from eegnb.analysis.pipelines import example_analysis_report()
+example_analysis_report()
+
+"""
+
+# Some standard pythonic imports
+import os
+from collections import OrderedDict
+import warnings
+import matplotlib.pyplot as plt
+from datetime import datetime
+import numpy as np
+from typing import Dict
+
+warnings.filterwarnings('ignore')
+
+# MNE functions
+from mne import Epochs,find_events, create_info
+from mne.io import RawArray
+
+# EEG-Notebooks functions
+from eegnb import generate_save_fn
+from eegnb.analysis.utils import load_data,plot_conditions, load_csv_as_raw, fix_musemissinglines
+from eegnb.analysis.analysis_report import get_html
+from eegnb.datasets import fetch_dataset
+from eegnb.devices.utils import EEG_INDICES, SAMPLE_FREQS
+from pathlib import Path
+
+DATA_DIR = os.path.join(os.path.expanduser("~/"), ".eegnb", "data")
+eegdevice, experiment_name, subject_id, session_nb, example_flag = None, None, None, None, False
+
+def load_eeg_data(experiment, subject=1, session=1, device_name='muse2016_bfn', tmin=-0.1, tmax=0.6, baseline=None, 
+                    reject={'eeg': 5e-5}, preload=True, verbose=1,
+                        picks=[0,1,2,3], event_id = OrderedDict(House=1,Face=2), fnames=None, example=False):
+    """
+    Loads EEG data from the specified experiment, subject, session, and device.
+    Returns the raw and epochs objects.
+
+    Procedure
+    1. Loads the data using file names and retrives if not already present
+    2. Epochs the data
+    3. Computes the ERP
+    4. Returns the raw and ERP objects
+
+    Parameters
+    ----------
+    experiment : Experiment Name
+    subject : Subject ID of performed experiment
+    session : Session ID of performed experiment
+    device_name : Device used for performed experiment
+    tmin : Start time of the epochs in seconds, relative to the time-locked event.
+    tmax : End time of the epochs in seconds, relative to the time-locked event.
+    baseline : Not very sure..?
+    reject : Rejection parameters for the epochs.
+    preload : If True, preload the epochs into memory.
+    verbose : If True, print out messages.
+    picks : Channels to include in the analysis.
+    event_id : Dictionary of event_id's for the epochs
+    fnames : File names of the experiment data, if not passed, example files are used
+    """
+
+    # If not using the example dataset, load the data from the specified experiment using load_csv_as_raw
+    if not example:
+
+        # Obataining the specific parameters to load the data into MNE object
+        sfreq = SAMPLE_FREQS[device_name]
+        ch_ind = EEG_INDICES[device_name]
+
+        # Generate file names if not passed
+        if fnames is None:
+            raw = load_data(subject_id=subject, session_nb=session, experiment=experiment, device_name=device_name, site="local", data_dir=os.path.join(os.path.expanduser('~/'),'.eegnb', 'data'))
+
+        else:
+            # Replace Ch names has arbitarily been set to None
+            if device_name in ["muse2016", "muse2", "museS"]:   
+                raw = load_csv_as_raw([fnames], sfreq=sfreq, ch_ind=ch_ind, aux_ind=[5], replace_ch_names=None, verbose=verbose)
+            else:
+                raw = load_csv_as_raw([fnames], sfreq=sfreq, ch_ind=ch_ind, replace_ch_names=None, verbose=verbose)
+
+            # Getting the subject and session
+            subject, session = fnames.split('_')[1], fnames.split('_')[2]
+
+    # If using the example dataset, load the data from the example dataset
+    else:
+        subject, session = 1, 1
+
+        # Loading Data
+        eegnb_data_path = os.path.join(os.path.expanduser('~/'),'.eegnb', 'data')
+        experiment_data_path = os.path.join(eegnb_data_path, experiment, 'eegnb_examples')
+
+        # If dataset hasn't been downloaded yet, download it
+        if not os.path.isdir(experiment_data_path):
+            fetch_dataset(data_dir=eegnb_data_path, experiment=experiment, site='eegnb_examples')
+
+        raw = load_data(1,1,
+                        experiment=experiment, site='eegnb_examples', device_name=device_name,
+                        data_dir = eegnb_data_path)
+
+    # Filtering the data under a certain frequency range
+    raw.filter(1,30, method='iir')
+
+    # Visualising the power spectrum
+    fig = raw.plot_psd(fmin=1, fmax=30, show=False)
+
+    # Saving the figure so it can be accessed by the pdf creation. Automatically deleted when added to the pdf.
+    plt.tight_layout()
+    plt.savefig("power_spectrum.png")
+    plt.show(block=False)
+    plt.pause(10)
+    plt.close()
+
+    # Epoching
+    # Create an array containing the timestamps and type of each stimulus (i.e. face or house)
+    events = find_events(raw)
+
+    # Create an MNE Epochs object representing all the epochs around stimulus presentation
+    epochs = Epochs(raw, events=events, event_id=event_id,
+                    tmin=tmin, tmax=tmax, baseline=baseline,
+                    reject=reject, preload=preload,
+                    verbose=verbose, picks=picks)
+
+    print('sample drop %: ', (1 - len(epochs.events)/len(events)) * 100)
+    print(len(epochs.events), 'events found')
+    print(epochs)
+
+    experimental_parameters = {"eeg_device": device_name, "experiment_name": experiment, "subject_id": subject, "session_nb": session, "example_flag": example, "drop_percent": (1 - len(epochs.events)/len(events)) * 100, "epochs_chosen": len(epochs.events)}
+
+    return epochs, experimental_parameters
+
+
+def make_erp_plot(epochs, experimental_parameters:Dict, conditions=OrderedDict(House=[1],Face=[2]), ci=97.5, n_boot=1000, title='',
+                   diff_waveform=None, #(1, 2))
+                   channel_order=[1,0,2,3]):
+    """
+    Plots the ERP for the specified conditions.
+
+    Parameters
+    ----------
+    epochs : MNE Epochs object
+    conditions : OrderedDict holding the conditions to plot
+    ci: confidence interval 
+    n_boot: number of bootstrap samples
+    title: title of the plot
+    diff_waveform: tuple of two integers indicating the channels to compare
+    channel_order: list of integers indicating the order of the channels to plot
+    """
+
+    fig, ax = plot_conditions(epochs, conditions=conditions,
+                            ci=97.5, n_boot=1000, title='',
+                            diff_waveform=None, #(1, 2))
+                            channel_order=[1,0,2,3]) # reordering of epochs.ch_names according to [[0,2],[1,3]] of subplot axes
+
+    # Autoscaling the y axis to a tight fit to the ERP
+    for i in [0,1,2,3]: ax[i].autoscale(tight=True)
+
+    # Saving the figure so it can be accessed by the pdf creation. Automatically deleted when added to the pdf.
+    # Makes sure that the axis labels are not cut out
+    plt.tight_layout()
+    plt.savefig("erp_plot.png")
+    plt.show(block=False)
+    plt.pause(10)
+    plt.close()
+
+    # Creating the pdf, needs to be discussed whether we want to call it here or seperately.
+    create_pdf(experimental_parameters)
+
+def create_pdf(experimental_parameters:Dict):
+    """Creates analysis report using the power spectrum and ERP plots that are saved in the directory"""
+
+    # Unpack the experimental parameters
+    eegdevice, experiment, subject, session, example, drop_percentage, epochs_chosen = experimental_parameters.values()
+
+    # Getting the directory where the report should be saved
+    save_dir = get_save_directory(experiment=experiment, eegdevice=eegdevice, subject=subject, session=session, example=example, label="analysis")
+
+    #get whole filepath
+    filepath = os.path.join(save_dir, 'analysis_report_{}.html'.format(datetime.now().strftime("%d-%m-%Y_%H-%M-%S")))
+
+    # Get the report 
+    report_html = get_html(experimental_parameters)
+
+    # Save html file
+    with open(filepath, 'w') as f:
+        f.write(report_html)
+
+    # Informing the user that the report has been saved
+    print('Analysis report saved to {}\n'.format(filepath))
+    print("Open the report by clicking the following link: {}{}".format("file:///", filepath))
+
+def get_save_directory(experiment, eegdevice, subject, session, example, label):
+    """ Returns save directory as a String for the analysis report """
+
+    if not example:
+        site='local'
+    else:
+        site='eegnb_examples'
+
+    # Getting the directory where the analysis report should be saved
+    save_path = os.path.join(os.path.expanduser("~/"),'.eegnb', label)
+    save_path = os.path.join(save_path, experiment, site, eegdevice, "subject{}".format(subject), "session{}".format(session))
+
+    # Creating the directory if it doesn't exist
+    if not os.path.isdir(save_path):
+        os.makedirs(save_path)
+
+    return save_path
+
+def create_analysis_report_(experiment, eegdevice, subject=None, session=None, data_path=None, bluemuse_file_fix=False):
+    """ Interface with the erp plot function, basically cli type instructions """
+
+    # Prompt user to enter options and then take inputs and do the necessary
+    epochs, experimental_parameters = load_eeg_data(experiment=experiment, subject=subject, session=session, device_name=eegdevice, example=False, fnames=data_path)
+    make_erp_plot(epochs, experimental_parameters)
+
+def example_analysis_report():
+    """ Example of how to use the analysis report function """
+
+    experiment = ["visual-N170", "visual-P300"]
+    experiment_choice = experiment[int(input("Choose an experiment: {} 0 or 1\n".format(experiment)))]
+
+    if experiment_choice == "visual-N170":
+        epochs, experimental_parameters = load_eeg_data(experiment_choice, example=True)
+        make_erp_plot(epochs, experimental_parameters)
+    else:
+        epochs, experimental_parameters = load_eeg_data('visual-P300', device_name='muse2016', event_id={'Non-Target': 1, 'Target': 2}, example=True)
+        make_erp_plot(epochs, experimental_parameters, conditions=OrderedDict(NonTarget=[1],Target=[2]))