diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py new file mode 100644 index 00000000..30164acc --- /dev/null +++ b/eegnb/experiments/Experiment.py @@ -0,0 +1,100 @@ +""" +Initial run of the Experiment Class Refactor base class + +Derived classes have to set a few things in major: +1. load_stimulus function : returns an array of stimuli +2. present_stimulus function : presents the stimuli and pushes eeg data back and forth as needed +Additional parameters can be set from the derived class as per the initializer + +""" + +class Experiment: + + def __init_(self, exp_name): + """ Anything that must be passed as a minimum for the experiment should be initialized here """ + + """ Dk if this overwrites the class variable or is worth doing + if we just assume they will overwrite """ + + 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(exp_name) + self.duration=120 + self.eeg:EEG=None + self.save_fn=None + self.n_trials=2010 + self.iti = 0.4 + self.soa = 0.3 + self.jitter = 0.2 + + def setup(self): + + self.record_duration = np.float32(self.duration) + self.markernames = [1, 2] + + # Setup Trial list -> Common in most (csv in Unicorn) + self.parameter = np.random.binomial(1, 0.5, n_trials) + self.trials = DataFrame(dict(parameter=parameter, timestamp=np.zeros(n_trials))) + + # Setup Graphics + mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) + + # Needs to be overwritten by specific experiment + self.stim = self.load_stimulus() + + # Show Instruction Screen + self.show_instructions(duration=duration) + + # Establish save function + if self.save_fn is None: # If no save_fn passed, generate a new unnamed save file + random_id = random.randint(1000,10000) + self.save_fn = generate_save_fn(eeg.device_name, experiement_id, random_id, random_id, "unnamed") + print( + f"No path for a save file was passed to the experiment. Saving data to {save_fn}" + ) + + def present(self): + """ Do the present operation for a bunch of experiments """ + + # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point + if eeg: + eeg.start(self.save_fn, duration=self.record_duration + 5) + + start = time() + + # Iterate through the events + for ii, trial in self.trials.iterrows(): + + # Intertrial interval + core.wait(self.iti + np.random.rand() * self.jitter) + + # Some form of presenting the stimulus - sometimes order changed in lower files like ssvep + self.present_stimulus(self.trials, ii, self.eeg, self.markernames) + + # Offset + mywin.flip() + if len(event.getKeys()) > 0 or (time() - start) > self.record_duration: + break + event.clearEvents() + + # Close the EEG stream + if eeg: + eeg.stop() + + + def show_instructions(self): + + self.instruction_text = self.instruction_text % duration + + # graphics + mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) + mywin.mouseVisible = False + + # Instructions + text = visual.TextStim(win=mywin, text=self.instruction_text, color=[-1, -1, -1]) + text.draw() + mywin.flip() + event.waitKeys(keyList="space") + + mywin.mouseVisible = True + mywin.close() \ No newline at end of file diff --git a/eegnb/experiments/Experiment_readme.txt b/eegnb/experiments/Experiment_readme.txt new file mode 100644 index 00000000..acaa44b0 --- /dev/null +++ b/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 diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index 3f9822bd..90897756 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -17,107 +17,40 @@ from eegnb import generate_save_fn from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE +from Experiment import Experiment -__title__ = "Visual N170" - -def present(duration=120, eeg: EEG=None, save_fn=None, - n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): +def load_stimulus(): + + load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) - record_duration = np.float32(duration) - markernames = [1, 2] - - # Setup trial list - image_type = np.random.binomial(1, 0.5, n_trials) - trials = DataFrame(dict(image_type=image_type, timestamp=np.zeros(n_trials))) - - def load_image(fn): - return visual.ImageStim(win=mywin, image=fn) - - # start the EEG stream, will delay 5 seconds to let signal settle - - # Setup graphics - mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - faces = list(map(load_image, glob(os.path.join(FACE_HOUSE, "faces", "*_3.jpg")))) houses = list(map(load_image, glob(os.path.join(FACE_HOUSE, "houses", "*.3.jpg")))) - stim = [houses, faces] - - # Show the instructions screen - show_instructions(duration) - - if eeg: - if save_fn is None: # If no save_fn passed, generate a new unnamed save file - random_id = random.randint(1000,10000) - save_fn = generate_save_fn(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 {save_fn}" - ) - eeg.start(save_fn, duration=record_duration + 5) - - # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point - start = time() - - # Iterate through the events - for ii, trial in trials.iterrows(): - # Inter trial interval - core.wait(iti + np.random.rand() * jitter) - - # Select and display image - label = trials["image_type"].iloc[ii] - image = choice(faces if label == 1 else houses) - image.draw() - - # Push sample - if eeg: - timestamp = time() - if eeg.backend == "muselsl": - marker = [markernames[label]] - else: - marker = markernames[label] - eeg.push_sample(marker=marker, timestamp=timestamp) - mywin.flip() - - # offset - core.wait(soa) - mywin.flip() - if len(event.getKeys()) > 0 or (time() - start) > record_duration: - break - - event.clearEvents() - - # Cleanup - if eeg: - eeg.stop() - - mywin.close() - - -def show_instructions(duration): - - instruction_text = """ - Welcome to the N170 experiment! - - Stay still, focus on the centre of the screen, and try not to blink. - - This block will run for %s seconds. - - Press spacebar to continue. + return [houses, faces] - """ - instruction_text = instruction_text % duration +def present_stimulus(trials, ii, eeg, markernames): + + label = trials["image_type"].iloc[ii] + image = choice(faces if label == 1 else houses) + image.draw() - # graphics - mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - - mywin.mouseVisible = False - - # Instructions - text = visual.TextStim(win=mywin, text=instruction_text, color=[-1, -1, -1]) - text.draw() - mywin.flip() - event.waitKeys(keyList="space") + # Push sample + if eeg: + timestamp = time() + if eeg.backend == "muselsl": + marker = [markernames[label]] + else: + marker = markernames[label] + eeg.push_sample(marker=marker, timestamp=timestamp) + + +if __name__ == "__main__": + + test = Experiment("Visual N170") + test.instruction_text = instruction_text + test.load_stimulus = load_stimulus + test.present_stimulus = present_stimulus + test.setup() + test.present() - mywin.mouseVisible = True - mywin.close() diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index b8176205..2d8e0c46 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -10,103 +10,39 @@ from eegnb import generate_save_fn from eegnb.stimuli import CAT_DOG -__title__ = "Visual P300" - - -def present(duration=120, eeg=None, save_fn=None): - n_trials = 2010 - iti = 0.4 - soa = 0.3 - jitter = 0.2 - record_duration = np.float32(duration) - markernames = [1, 2] - - # Setup trial list - image_type = np.random.binomial(1, 0.5, n_trials) - trials = DataFrame(dict(image_type=image_type, timestamp=np.zeros(n_trials))) - - def load_image(fn): - return visual.ImageStim(win=mywin, image=fn) +def load_stimulus(): + load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) # Setup graphics mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - targets = list(map(load_image, glob(os.path.join(CAT_DOG, "target-*.jpg")))) nontargets = list(map(load_image, glob(os.path.join(CAT_DOG, "nontarget-*.jpg")))) - stim = [nontargets, targets] - - # Show instructions - show_instructions(duration=duration) - - # start the EEG stream, will delay 5 seconds to let signal settle - if eeg: - if save_fn is None: # If no save_fn passed, generate a new unnamed save file - save_fn = generate_save_fn(eeg.device_name, "visual_p300", "unnamed") - print( - f"No path for a save file was passed to the experiment. Saving data to {save_fn}" - ) - eeg.start(save_fn, duration=record_duration) - - # Iterate through the events - start = time() - for ii, trial in trials.iterrows(): - # Inter trial interval - core.wait(iti + np.random.rand() * jitter) - - # Select and display image - label = trials["image_type"].iloc[ii] - image = choice(targets if label == 1 else nontargets) - image.draw() - - # Push sample - if eeg: - timestamp = time() - if eeg.backend == "muselsl": - marker = [markernames[label]] - else: - marker = markernames[label] - eeg.push_sample(marker=marker, timestamp=timestamp) - - mywin.flip() + + return [nontargets, targets] - # offset - core.wait(soa) - mywin.flip() - if len(event.getKeys()) > 0 or (time() - start) > record_duration: - break +def present_stimulus(trials, ii, eeg, markernames): - event.clearEvents() + label = trials["image_type"].iloc[ii] + image = choice(targets if label == 1 else nontargets) + image.draw() - # Cleanup + # Push sample if eeg: - eeg.stop() - mywin.close() - - -def show_instructions(duration): - - instruction_text = """ - Welcome to the P300 experiment! - - Stay still, focus on the centre of the screen, and try not to blink. - - This block will run for %s seconds. - - Press spacebar to continue. + timestamp = time() + if eeg.backend == "muselsl": + marker = [markernames[label]] + else: + marker = markernames[label] + eeg.push_sample(marker=marker, timestamp=timestamp) + +if __name__ == "__main__": + + test = Experiment("Visual P300") + test.instruction_text = instruction_text + test.load_stimulus = load_stimulus + test.present_stimulus = present_stimulus + test.setup() + test.present() - """ - instruction_text = instruction_text % duration - - # graphics - mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - - mywin.mouseVisible = False - # Instructions - text = visual.TextStim(win=mywin, text=instruction_text, color=[-1, -1, -1]) - text.draw() - mywin.flip() - event.waitKeys(keyList="space") - mywin.mouseVisible = True - mywin.close() diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index 8e6aa765..a3b480cc 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -8,24 +8,10 @@ from psychopy import visual, core, event from eegnb import generate_save_fn +from Experiment import Experiment -__title__ = "Visual SSVEP" - - -def present(duration=120, eeg=None, save_fn=None): - n_trials = 2010 - iti = 0.5 - soa = 3.0 - jitter = 0.2 - record_duration = np.float32(duration) - markernames = [1, 2] - - # Setup trial list - stim_freq = np.random.binomial(1, 0.5, n_trials) - trials = DataFrame(dict(stim_freq=stim_freq, timestamp=np.zeros(n_trials))) - - # Set up graphics - mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) +def load_stimulus(): + grating = visual.GratingStim(win=mywin, mask="circle", size=80, sf=0.2) grating_neg = visual.GratingStim( win=mywin, mask="circle", size=80, sf=0.2, phase=0.5 @@ -36,28 +22,7 @@ def present(duration=120, eeg=None, save_fn=None): # Generate the possible ssvep frequencies based on monitor refresh rate def get_possible_ssvep_freqs(frame_rate, stim_type="single"): - """Get possible SSVEP stimulation frequencies. - Utility function that returns the possible SSVEP stimulation - frequencies and on/off pattern based on screen refresh rate. - Args: - frame_rate (float): screen frame rate, in Hz - Keyword Args: - stim_type (str): type of stimulation - 'single'-> single graphic stimulus (the displayed object - appears and disappears in the background.) - 'reversal' -> pattern reversal stimulus (the displayed object - appears and is replaced by its opposite.) - Returns: - (dict): keys are stimulation frequencies (in Hz), and values are - lists of tuples, where each tuple is the number of (on, off) - periods of one stimulation cycle - For more info on stimulation patterns, see Section 2 of: - Danhua Zhu, Jordi Bieger, Gary Garcia Molina, and Ronald M. Aarts, - "A Survey of Stimulation Methods Used in SSVEP-Based BCIs," - Computational Intelligence and Neuroscience, vol. 2010, 12 pages, - 2010. - """ - + if stim_type == "single": max_period_nb = int(frame_rate / 6) periods = np.arange(max_period_nb) + 1 @@ -76,25 +41,7 @@ def get_possible_ssvep_freqs(frame_rate, stim_type="single"): return freqs def init_flicker_stim(frame_rate, cycle, soa): - """Initialize flickering stimulus. - Get parameters for a flickering stimulus, based on the screen refresh - rate and the desired stimulation cycle. - Args: - frame_rate (float): screen frame rate, in Hz - cycle (tuple or int): if tuple (on, off), represents the number of - 'on' periods and 'off' periods in one flickering cycle. This - supposes a "single graphic" stimulus, where the displayed object - appears and disappears in the background. - If int, represents the number of total periods in one cycle. - This supposes a "pattern reversal" stimulus, where the - displayed object appears and is replaced by its opposite. - soa (float): stimulus duration, in s - Returns: - (dict): dictionary with keys - 'cycle' -> tuple of (on, off) periods in a cycle - 'freq' -> stimulus frequency - 'n_cycles' -> number of cycles in one stimulus trial - """ + if isinstance(cycle, tuple): stim_freq = frame_rate / sum(cycle) n_cycles = int(soa * stim_freq) @@ -108,10 +55,6 @@ def init_flicker_stim(frame_rate, cycle, soa): # Set up stimuli frame_rate = np.round(mywin.getActualFrameRate()) # Frame rate, in Hz freqs = get_possible_ssvep_freqs(frame_rate, stim_type="reversal") - stim_patterns = [ - init_flicker_stim(frame_rate, 2, soa), - init_flicker_stim(frame_rate, 3, soa), - ] print( ( @@ -121,24 +64,35 @@ def init_flicker_stim(frame_rate, cycle, soa): ) ) - # Show the instructions screen - show_instructions(duration) + return [ + init_flicker_stim(frame_rate, 2, soa), + init_flicker_stim(frame_rate, 3, soa), + ] - # start the EEG stream, will delay 5 seconds to let signal settle - if eeg: - if save_fn is None: # If no save_fn passed, generate a new unnamed save file - save_fn = generate_save_fn(eeg.device_name, "visual_ssvep", "unnamed") - print( - f"No path for a save file was passed to the experiment. Saving data to {save_fn}" - ) - eeg.start(save_fn, duration=record_duration) +def present_stimulus(trials, ii, eeg, markernames): + pass + + +if __name__ == "__main__": + + test = Experiment("Visual SSVEP") + test.instruction_text = instruction_text + test.load_stimulus = load_stimulus + test.present_stimulus = present_stimulus + test.iti = 0.5 + test.soa = 3.0 + test.setup() + test.present() - # Iterate through trials - start = time() + +def present(duration=120, eeg=None, save_fn=None): + + for ii, trial in trials.iterrows(): # Intertrial interval core.wait(iti + np.random.rand() * jitter) + """ Unique """ # Select stimulus frequency ind = trials["stim_freq"].iloc[ii] @@ -162,6 +116,8 @@ def init_flicker_stim(frame_rate, cycle, soa): mywin.flip() grating_neg.setAutoDraw(False) + """ Unique ends """ + # offset mywin.flip() if len(event.getKeys()) > 0 or (time() - start) > record_duration: