Experiment Class Refactor (update to NeuroTechX#183), converting spec…

…ific experiments to subclasses (NeuroTechX#184)

* First commit

* Second commit

* Modifications

* Lol

* Lol

* Incorporated N170 and p300, looking good for a PR

* ssvep update

* Implementing subclasses instead of loose functions

* fix: fixed import (brainflow updated API)

* Playing around still

* Fixing import errors

* Adding abstractmethod decorators

* Still working on the import error

* Guess what's finally working

* Comments and naming ticks

* More comments

* Live coding demonstration

* ssvep adapted

* Adapting Auditory Oddball

* changing save_fn to self.save_fun

* This maybe the last big change

* utils file changed, changes work through cli

Co-authored-by: Erik Bjäreholt <erik@bjareho.lt>

eegnb/cli/utils.py

@@ -7,24 +7,26 @@
 
 from eegnb.devices.eeg import EEG
 
-from eegnb.experiments.visual_n170 import n170
-from eegnb.experiments.visual_p300 import p300
-from eegnb.experiments.visual_ssvep import ssvep
+from eegnb.experiments import VisualN170
+from eegnb.experiments import VisualP300
+from eegnb.experiments import VisualSSVEP
+from eegnb.experiments import AuditoryOddball
 from eegnb.experiments.visual_cueing import cueing
 from eegnb.experiments.visual_codeprose import codeprose
-from eegnb.experiments.auditory_oddball import aob, diaconescu
+from eegnb.experiments.auditory_oddball import diaconescu
 from eegnb.experiments.auditory_ssaep import ssaep, ssaep_onefreq
 
 
+# New Experiment Class structure has a different initilization, to be noted
 experiments = {
-    "visual-N170": n170,
-    "visual-P300": p300,
-    "visual-SSVEP": ssvep,
+    "visual-N170": VisualN170(),
+    "visual-P300": VisualP300(),
+    "visual-SSVEP": VisualSSVEP(),
     "visual-cue": cueing,
     "visual-codeprose": codeprose,
     "auditory-SSAEP orig": ssaep,
     "auditory-SSAEP onefreq": ssaep_onefreq,
-    "auditory-oddball orig": aob,
+    "auditory-oddball orig": AuditoryOddball(),
     "auditory-oddball diaconescu": diaconescu,
 }
 
@@ -42,7 +44,15 @@ def run_experiment(
 ):
     if experiment in experiments:
         module = experiments[experiment]
-        module.present(duration=record_duration, eeg=eeg_device, save_fn=save_fn)  # type: ignore
+
+        # Condition added for different run types of old and new experiment class structure
+        if experiment == "visual-N170" or experiment == "visual-P300" or experiment == "visual-SSVEP" or experiment == "auditory-oddball orig":
+            module.duration = record_duration
+            module.eeg = eeg_device
+            module.save_fn = save_fn
+            module.run()
+        else:
+            module.present(duration=record_duration, eeg=eeg_device, save_fn=save_fn)  # type: ignore
     else:
         print("\nError: Unknown experiment '{}'".format(experiment))
         print("\nExperiment can be one of:")

eegnb/experiments/Experiment.py

@@ -0,0 +1,168 @@
+""" 
+Initial run of the Experiment Class Refactor base class
+
+Specific experiments are implemented as sub classes that overload a load_stimulus and present_stimulus method
+
+Running each experiment:
+obj = VisualP300({parameters})
+obj.run()
+"""
+
+from abc import ABC, abstractmethod
+from psychopy import prefs
+#change the pref libraty to PTB and set the latency mode to high precision
+prefs.hardware['audioLib'] = 'PTB'
+prefs.hardware['audioLatencyMode'] = 3
+
+import os
+from time import time
+from glob import glob
+from random import choice
+from optparse import OptionParser
+import random
+
+import numpy as np
+from pandas import DataFrame
+from psychopy import visual, core, event
+
+from eegnb import generate_save_fn
+from eegnb.devices.eeg import EEG
+
+class BaseExperiment:
+
+    def __init__(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter):
+        """ Initializer for the Base Experiment Class """
+
+        self.exp_name = exp_name
+        self.instruction_text = """\nWelcome to the {} experiment!\nStay still, focus on the centre of the screen, and try not to blink. \nThis block will run for %s seconds.\n
+        Press spacebar to continue. \n""".format(self.exp_name)
+        self.duration = duration
+        self.eeg = eeg
+        self.save_fn = save_fn
+        self.n_trials = n_trials
+        self.iti = iti
+        self.soa = soa
+        self.jitter = jitter
+
+    @abstractmethod
+    def load_stimulus(self):
+        """ 
+        Method that loads the stimulus for the specific experiment, overwritten by the specific experiment
+        Returns the stimulus object in the form of [{stim1},{stim2},...]
+        Throws error if not overwritten in the specific experiment
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def present_stimulus(self, idx : int):
+        """
+        Method that presents the stimulus for the specific experiment, overwritten by the specific experiment
+        Displays the stimulus on the screen
+        Pushes EEG Sample if EEG is enabled
+        Throws error if not overwritten in the specific experiment
+
+        idx : Trial index for the current trial
+        """
+        raise NotImplementedError
+
+    def setup(self, instructions=True):
+
+        # Initializing the record duration and the marker names
+        self.record_duration = np.float32(self.duration)
+        self.markernames = [1, 2]
+
+        # Setting up the trial and parameter list
+        self.parameter = np.random.binomial(1, 0.5, self.n_trials)
+        self.trials = DataFrame(dict(parameter=self.parameter, timestamp=np.zeros(self.n_trials)))
+
+        # Setting up Graphics 
+        self.window = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) 
+
+        # Loading the stimulus from the specific experiment, throws an error if not overwritten in the specific experiment
+        self.stim = self.load_stimulus()
+
+        # Show Instruction Screen if not skipped by the user
+        if instructions:
+            self.show_instructions()
+
+        # Checking for EEG to setup the EEG stream
+        if self.eeg:
+             # If no save_fn passed, generate a new unnamed save file
+            if self.save_fn is None:  
+                # Generating a random int for the filename
+                random_id = random.randint(1000,10000)
+                # Generating save function
+                self.save_fn = generate_save_fn(self.eeg.device_name, "visual_n170", random_id, random_id, "unnamed")
+                print(
+                    f"No path for a save file was passed to the experiment. Saving data to {self.save_fn}"
+                )
+
+    def show_instructions(self):
+        """ 
+        Method that shows the instructions for the specific Experiment
+        In the usual case it is not overwritten, the instruction text can be overwritten by the specific experiment
+        No parameters accepted, can be skipped through passing a False while running the Experiment
+        """
+
+        # Splitting instruction text into lines
+        self.instruction_text = self.instruction_text % self.duration
+
+        # Disabling the cursor during display of instructions
+        self.window.mouseVisible = False
+
+        # Displaying the instructions on the screen
+        text = visual.TextStim(win=self.window, text=self.instruction_text, color=[-1, -1, -1])
+        text.draw()
+        self.window.flip()
+
+        # Waiting for the user to press the spacebar to start the experiment
+        event.waitKeys(keyList="space")
+
+        # Enabling the cursor again
+        self.window.mouseVisible = True
+
+    def run(self, instructions=True):
+        """ Do the present operation for a bunch of experiments """
+
+        # Setup the experiment, alternatively could get rid of this line, something to think about
+        self.setup(instructions)
+
+        print("Wait for the EEG-stream to start...")
+
+        # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point
+        if self.eeg:
+            self.eeg.start(self.save_fn, duration=self.record_duration + 5)
+
+        print("EEG Stream started")
+
+        start = time()
+
+        # Iterate through the events
+        for ii, trial in self.trials.iterrows():
+
+            # Intertrial interval
+            core.wait(self.iti + np.random.rand() * self.jitter)
+
+            # Stimulus presentation overwritten by specific experiment
+            self.present_stimulus(ii, trial)
+
+            # Offset
+            core.wait(self.soa)
+            self.window.flip()
+
+            # Exiting the loop condition, looks ugly and needs to be fixed
+            if len(event.getKeys()) > 0 or (time() - start) > self.record_duration:
+                break
+
+            # Clearing the screen for the next trial    
+            event.clearEvents()
+
+        # Closing the EEG stream 
+        if self.eeg:
+            self.eeg.stop()
+
+        # Closing the window
+        self.window.close()
+
+
+

eegnb/experiments/Experiment_readme.txt

@@ -0,0 +1,13 @@
+
+
+Looking for a general implementation structure where base class implements and passes the following functions, 
+
+def load_stimulus() -> stim (some form of dd array)
+
+def present_stimulus() -> given trial details does specific thing for experiment 
+
+** Slight issue is that a lot of parameters will have to be passed which is not the best in practice
+
+Stuff that can be overwritten in general ...
+instruction_text
+parameter/trial

eegnb/experiments/__init__.py

@@ -0,0 +1,4 @@
+from .visual_n170.n170 import VisualN170
+from .visual_p300.p300 import VisualP300
+from .visual_ssvep.ssvep import VisualSSVEP
+from .auditory_oddball.aob import AuditoryOddball