From 1634e81277c1edb79183591d5b49fccdc553299d Mon Sep 17 00:00:00 2001 From: Parvfect Date: Mon, 30 May 2022 18:38:30 +0100 Subject: [PATCH 01/22] First commit --- eegnb/experiments/Experment.py | 79 ++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 eegnb/experiments/Experment.py diff --git a/eegnb/experiments/Experment.py b/eegnb/experiments/Experment.py new file mode 100644 index 00000000..872c32d4 --- /dev/null +++ b/eegnb/experiments/Experment.py @@ -0,0 +1,79 @@ + + + +class Experiment: + + def __init_(): + + + def present(duration=120, eeg: EEG=None, save_fn=None, n_trials=2010, exp_name=""): + """ Do the present operation for a bunch of experiments """ + + # Setup Trial list -> Common in most + parameter = np.random.binomial(1, 0.5, n_trials) + trials = DataFrame(dict(parameter=parameter, timestamp=np.zeros(n_trials))) + + # Setup Graphics + mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) + """ Does specific thing within Experiment Type to load stimulus data """ + + # Show Instruction Screen + show_instructions(duration=duration) + + # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point + 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 = time() + + # Iterate through the events + for ii, trial in trials.iterrows(): + + # Intertrial interval + core.wait(iti + np.random.rand() * jitter) + + # Some form of presenting the stimulus - sometimes order changed in lower files like ssvep + + # Push sample + if eeg: + timestamp = time() + if eeg.backend == "muselsl": + marker = [markernames[label]] + else: + marker = markernames[label] + eeg.push_sample(marker=marker, timestamp=timestamp) + + # Offset + mywin.flip() + if len(event.getKeys()) > 0 or (time() - start) > record_duration: + break + event.clearEvents() + + + # Close the EEG stream + if eeg: + eeg.stop() + + def show_instructions(instruction_text): + + 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() \ No newline at end of file From d48093adef7094815ea4625b4f6bb1f5c477db9a Mon Sep 17 00:00:00 2001 From: Parvfect Date: Mon, 30 May 2022 18:43:52 +0100 Subject: [PATCH 02/22] Second commit --- eegnb/experiments/Experment.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/eegnb/experiments/Experment.py b/eegnb/experiments/Experment.py index 872c32d4..51ddd18e 100644 --- a/eegnb/experiments/Experment.py +++ b/eegnb/experiments/Experment.py @@ -1,5 +1,20 @@ +""" +Right now the quesitons are simple + +1. Do we want to have as much code as possible in the master class, somewhat like a main function that works with generic types +being passed in? EG. Experiment class has all the code, specific functions, variables, events and stimuli are passed in that are +called according to the stage of the cycle + +2. Do we want to just have a common set of shared data members in the form of a data class as per Issue 76? + +3. Do we want to split the main piece of code into a lot of functions like settng up trials, graphics, etc? + +4. How different are the next experiments that are going to be incorporated be? Will they be able to stick to such a protocol? + +""" + class Experiment: From f05f1c4919c5f635242c0c8cb574e259ed84a0e8 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Sun, 5 Jun 2022 00:25:17 +0100 Subject: [PATCH 03/22] Modifications --- .../{Experment.py => Experiment.py} | 47 ++++++++++++------- eegnb/experiments/visual_n170/n170.py | 40 ++++++++++++++-- 2 files changed, 66 insertions(+), 21 deletions(-) rename eegnb/experiments/{Experment.py => Experiment.py} (68%) diff --git a/eegnb/experiments/Experment.py b/eegnb/experiments/Experiment.py similarity index 68% rename from eegnb/experiments/Experment.py rename to eegnb/experiments/Experiment.py index 51ddd18e..af32453f 100644 --- a/eegnb/experiments/Experment.py +++ b/eegnb/experiments/Experiment.py @@ -18,32 +18,46 @@ class Experiment: - def __init_(): - + def __init_(self): + + # Should have overwriting property for different experiments + 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) + + + def setup(self): - def present(duration=120, eeg: EEG=None, save_fn=None, n_trials=2010, exp_name=""): - """ Do the present operation for a bunch of experiments """ - - # Setup Trial list -> Common in most - parameter = np.random.binomial(1, 0.5, n_trials) - trials = DataFrame(dict(parameter=parameter, timestamp=np.zeros(n_trials))) + # 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) """ Does specific thing within Experiment Type to load stimulus data """ - + + # Needs to be overwritten by specific experiment + self.stim = self.load_stimulus() + # Show Instruction Screen - show_instructions(duration=duration) + self.show_instructions(duration=duration) - # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point - if eeg: - if save_fn is None: # If no save_fn passed, generate a new unnamed save file + # 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) - save_fn = generate_save_fn(eeg.device_name, "visual_n170", random_id, random_id, "unnamed") + 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}" ) - eeg.start(save_fn, duration=record_duration + 5) + + + + def present(self, duration=120, eeg: EEG=None, save_fn=None, n_trials=2010, exp_name=""): + """ 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(save_fn, duration=record_duration + 5) start = time() @@ -75,9 +89,10 @@ def present(duration=120, eeg: EEG=None, save_fn=None, n_trials=2010, exp_name=" if eeg: eeg.stop() + def show_instructions(instruction_text): - instruction_text = instruction_text % duration + self.instruction_text = self.instruction_text % duration # graphics mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index 3f9822bd..a12ca74d 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -18,9 +18,40 @@ from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE + + + __title__ = "Visual N170" +def load_stimulus(): + def load_image(fn): + return visual.ImageStim(win=mywin, image=fn) + 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] + return stim + +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. + + """ + + +if __name__ == "__main__": + + test = Experiment() + test.instruction_text = instruction_text + test.load_stimulus = load_stimulus + test.run() + + def present(duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): @@ -39,9 +70,6 @@ def load_image(fn): # 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) @@ -63,6 +91,7 @@ def load_image(fn): # 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) @@ -97,7 +126,7 @@ def load_image(fn): def show_instructions(duration): instruction_text = """ - Welcome to the N170 experiment! + Welcome to the {} experiment! Stay still, focus on the centre of the screen, and try not to blink. @@ -105,7 +134,8 @@ def show_instructions(duration): Press spacebar to continue. - """ + """.format(self.experiment_id) + instruction_text = instruction_text % duration # graphics From 38993e6a5cf7c0e9f4b35556be8c5cb70a117e73 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Sun, 5 Jun 2022 00:47:33 +0100 Subject: [PATCH 04/22] Lol --- eegnb/experiments/Experiment_readme.txt | 0 eegnb/experiments/visual_p300/p300.py | 23 +++++++++++++++++++++++ 2 files changed, 23 insertions(+) create mode 100644 eegnb/experiments/Experiment_readme.txt diff --git a/eegnb/experiments/Experiment_readme.txt b/eegnb/experiments/Experiment_readme.txt new file mode 100644 index 00000000..e69de29b diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index b8176205..7d6f5853 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -12,6 +12,29 @@ __title__ = "Visual P300" +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. + + """ + +def load_stimulus(): + pass + +def present_stimulus(): + pass + + +if __name__ == "__main__": + + test = Experiment() + + def present(duration=120, eeg=None, save_fn=None): n_trials = 2010 From ba01229a93040daf916ca3c61908811b180996bc Mon Sep 17 00:00:00 2001 From: Parvfect Date: Sun, 5 Jun 2022 00:47:55 +0100 Subject: [PATCH 05/22] Lol --- eegnb/experiments/Experiment_readme.txt | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/eegnb/experiments/Experiment_readme.txt b/eegnb/experiments/Experiment_readme.txt index e69de29b..acaa44b0 100644 --- a/eegnb/experiments/Experiment_readme.txt +++ 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 From 42d3f2634d01e25e5f5e2fd618c70284b055848c Mon Sep 17 00:00:00 2001 From: Parvfect Date: Thu, 9 Jun 2022 21:43:20 +0100 Subject: [PATCH 06/22] Incorporated N170 and p300, looking good for a PR --- eegnb/experiments/visual_n170/n170.py | 143 +++++--------------------- 1 file changed, 23 insertions(+), 120 deletions(-) diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index a12ca74d..90897756 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -17,137 +17,40 @@ from eegnb import generate_save_fn from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE - - - - -__title__ = "Visual N170" +from Experiment import Experiment def load_stimulus(): - def load_image(fn): - return visual.ImageStim(win=mywin, image=fn) + + load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) + 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] - return stim - -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] - """ +def present_stimulus(trials, ii, eeg, markernames): + + 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) + if __name__ == "__main__": - test = Experiment() + test = Experiment("Visual N170") test.instruction_text = instruction_text test.load_stimulus = load_stimulus - test.run() - - -def present(duration=120, eeg: 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) - - # 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) - - - # 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 {} 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. - - """.format(self.experiment_id) - - 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") + test.present_stimulus = present_stimulus + test.setup() + test.present() - mywin.mouseVisible = True - mywin.close() From 76c5036abc012074e770ea5fe3f0f0accd4f37a3 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Thu, 9 Jun 2022 21:51:33 +0100 Subject: [PATCH 07/22] ssvep update --- eegnb/experiments/Experiment.py | 69 ++++++------- eegnb/experiments/visual_p300/p300.py | 129 ++++-------------------- eegnb/experiments/visual_ssvep/ssvep.py | 104 ++++++------------- 3 files changed, 81 insertions(+), 221 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index af32453f..30164acc 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -1,39 +1,43 @@ +""" +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 """ -Right now the quesitons are simple - -1. Do we want to have as much code as possible in the master class, somewhat like a main function that works with generic types -being passed in? EG. Experiment class has all the code, specific functions, variables, events and stimuli are passed in that are -called according to the stage of the cycle - -2. Do we want to just have a common set of shared data members in the form of a data class as per Issue 76? - -3. Do we want to split the main piece of code into a lot of functions like settng up trials, graphics, etc? - -4. How different are the next experiments that are going to be incorporated be? Will they be able to stick to such a protocol? - -""" - class Experiment: - def __init_(self): + def __init_(self, exp_name): + """ Anything that must be passed as a minimum for the experiment should be initialized here """ - # Should have overwriting property for different experiments + """ 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) - """ Does specific thing within Experiment Type to load stimulus data """ # Needs to be overwritten by specific experiment self.stim = self.load_stimulus() @@ -49,58 +53,45 @@ def setup(self): f"No path for a save file was passed to the experiment. Saving data to {save_fn}" ) - - - def present(self, duration=120, eeg: EEG=None, save_fn=None, n_trials=2010, exp_name=""): + 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(save_fn, duration=record_duration + 5) + eeg.start(self.save_fn, duration=self.record_duration + 5) start = time() # Iterate through the events - for ii, trial in trials.iterrows(): + for ii, trial in self.trials.iterrows(): # Intertrial interval - core.wait(iti + np.random.rand() * jitter) + core.wait(self.iti + np.random.rand() * self.jitter) # Some form of presenting the stimulus - sometimes order changed in lower files like ssvep - - # Push sample - if eeg: - timestamp = time() - if eeg.backend == "muselsl": - marker = [markernames[label]] - else: - marker = markernames[label] - eeg.push_sample(marker=marker, timestamp=timestamp) + self.present_stimulus(self.trials, ii, self.eeg, self.markernames) # Offset mywin.flip() - if len(event.getKeys()) > 0 or (time() - start) > record_duration: + 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(instruction_text): + 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=instruction_text, color=[-1, -1, -1]) + text = visual.TextStim(win=mywin, text=self.instruction_text, color=[-1, -1, -1]) text.draw() mywin.flip() event.waitKeys(keyList="space") diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 7d6f5853..2d8e0c46 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -10,126 +10,39 @@ from eegnb import generate_save_fn from eegnb.stimuli import CAT_DOG -__title__ = "Visual P300" -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. - - """ - def load_stimulus(): - pass - -def present_stimulus(): - pass - - -if __name__ == "__main__": - - test = Experiment() - - - -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) - + 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() + timestamp = time() + if eeg.backend == "muselsl": + marker = [markernames[label]] + else: + marker = markernames[label] + eeg.push_sample(marker=marker, timestamp=timestamp) +if __name__ == "__main__": -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. + 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: From dd958d2390a86ec97bb7371288dcbd50f99c51d0 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Wed, 15 Jun 2022 13:04:37 +0100 Subject: [PATCH 08/22] Implementing subclasses instead of loose functions --- eegnb/experiments/Experiment.py | 47 +++--- eegnb/experiments/visual_n170/n170.py | 64 ++++---- eegnb/experiments/visual_p300/p300.py | 64 ++++---- eegnb/experiments/visual_ssvep/ssvep.py | 191 +++++++++--------------- 4 files changed, 152 insertions(+), 214 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index 30164acc..b282deff 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -1,32 +1,29 @@ """ 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 - +Specific experiments are implemented as sub classes that inherit a load_stimulus and present_stimulus method """ class Experiment: - def __init_(self, exp_name): + def __init_(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): """ 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.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 + 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 + def load_stimulus(self): + """ Needs to be overwritten by specific experiment """ + pass + def setup(self): self.record_duration = np.float32(self.duration) @@ -43,7 +40,7 @@ def setup(self): self.stim = self.load_stimulus() # Show Instruction Screen - self.show_instructions(duration=duration) + self.show_instructions() # Establish save function if self.save_fn is None: # If no save_fn passed, generate a new unnamed save file @@ -52,7 +49,11 @@ def setup(self): print( f"No path for a save file was passed to the experiment. Saving data to {save_fn}" ) - + + def present_stimulus(self): + """ Needs to be overwritten by specific experiment """ + pass + def present(self): """ Do the present operation for a bunch of experiments """ @@ -69,7 +70,7 @@ def present(self): 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) + self.present_stimulus() # Offset mywin.flip() @@ -84,7 +85,7 @@ def present(self): def show_instructions(self): - self.instruction_text = self.instruction_text % duration + self.instruction_text = self.instruction_text % self.duration # graphics mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index 90897756..30d35615 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -20,37 +20,35 @@ from Experiment import Experiment -def load_stimulus(): +class VisualN170(Experiment): + + def __init__(self, duration=120, eeg: EEG=None, save_fn=None, + n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): + + exp_name = "Visual N170" + super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) + + def load_stimulus(self): + + load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) + + 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")))) + + return [houses, faces] + + def present_stimulus(self): - load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) - - 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")))) - - return [houses, faces] - -def present_stimulus(trials, ii, eeg, markernames): - - 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) - - -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() - + label = self.trials["image_type"].iloc[ii] + image = choice(faces if label == 1 else houses) + image.draw() + + # Push sample + if self.eeg: + timestamp = time() + if self.eeg.backend == "muselsl": + marker = [self.markernames[label]] + else: + marker = self.markernames[label] + self.eeg.push_sample(marker=marker, timestamp=timestamp) + \ No newline at end of file diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 2d8e0c46..5cd51900 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -11,38 +11,34 @@ from eegnb.stimuli import CAT_DOG -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")))) +class VisualP300(Experiment): - return [nontargets, targets] - -def present_stimulus(trials, ii, eeg, markernames): - - 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) - -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() - - - + def __init__(self, duration=120, eeg: EEG=None, save_fn=None, + n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): + + exp_name = "Visual P300" + super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) + + def load_stimulus(self): + 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")))) + + return [nontargets, targets] + + def present_stimulus(self): + + label = self.trials["image_type"].iloc[ii] + image = choice(targets if label == 1 else nontargets) + image.draw() + + # Push sample + if self.eeg: + timestamp = time() + if self.eeg.backend == "muselsl": + marker = [self.markernames[label]] + else: + marker = self.markernames[label] + self.eeg.push_sample(marker=marker, timestamp=timestamp) diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index a3b480cc..14db038b 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -10,100 +10,86 @@ from eegnb import generate_save_fn from Experiment import Experiment -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 - ) - fixation = visual.GratingStim( - win=mywin, size=0.2, pos=[0, 0], sf=0.2, color=[1, 0, 0], autoDraw=True - ) - - # Generate the possible ssvep frequencies based on monitor refresh rate - def get_possible_ssvep_freqs(frame_rate, stim_type="single"): - if stim_type == "single": - max_period_nb = int(frame_rate / 6) - periods = np.arange(max_period_nb) + 1 - - if stim_type == "single": - freqs = dict() - for p1 in periods: - for p2 in periods: - f = frame_rate / (p1 + p2) - try: - freqs[f].append((p1, p2)) - except: - freqs[f] = [(p1, p2)] - elif stim_type == "reversal": - freqs = {frame_rate / p: [(p, p)] for p in periods[::-1]} - - return freqs - - def init_flicker_stim(frame_rate, cycle, soa): - - if isinstance(cycle, tuple): - stim_freq = frame_rate / sum(cycle) - n_cycles = int(soa * stim_freq) - else: - stim_freq = frame_rate / cycle - cycle = (cycle, cycle) - n_cycles = int(soa * stim_freq) / 2 - - return {"cycle": cycle, "freq": stim_freq, "n_cycles": n_cycles} - - # Set up stimuli - frame_rate = np.round(mywin.getActualFrameRate()) # Frame rate, in Hz - freqs = get_possible_ssvep_freqs(frame_rate, stim_type="reversal") - - print( - ( - "Flickering frequencies (Hz): {}\n".format( - [stim_patterns[0]["freq"], stim_patterns[1]["freq"]] - ) - ) - ) - return [ - init_flicker_stim(frame_rate, 2, soa), - init_flicker_stim(frame_rate, 3, soa), - ] +class VisualSSVEP(Experiment): -def present_stimulus(trials, ii, eeg, markernames): - pass + def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.5, soa = 3.0, jitter = 0.2): + + exp_name = "Visual SSVEP" + super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) + def load_stimulus(self): + + 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 + ) + fixation = visual.GratingStim( + win=mywin, size=0.2, pos=[0, 0], sf=0.2, color=[1, 0, 0], autoDraw=True + ) -if __name__ == "__main__": + # Generate the possible ssvep frequencies based on monitor refresh rate + def get_possible_ssvep_freqs(frame_rate, stim_type="single"): + if stim_type == "single": + max_period_nb = int(frame_rate / 6) + periods = np.arange(max_period_nb) + 1 + + if stim_type == "single": + freqs = dict() + for p1 in periods: + for p2 in periods: + f = frame_rate / (p1 + p2) + try: + freqs[f].append((p1, p2)) + except: + freqs[f] = [(p1, p2)] + elif stim_type == "reversal": + freqs = {frame_rate / p: [(p, p)] for p in periods[::-1]} + + return freqs + + def init_flicker_stim(frame_rate, cycle, soa): + + if isinstance(cycle, tuple): + stim_freq = frame_rate / sum(cycle) + n_cycles = int(soa * stim_freq) + else: + stim_freq = frame_rate / cycle + cycle = (cycle, cycle) + n_cycles = int(soa * stim_freq) / 2 + + return {"cycle": cycle, "freq": stim_freq, "n_cycles": n_cycles} - 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() + # Set up stimuli + frame_rate = np.round(mywin.getActualFrameRate()) # Frame rate, in Hz + freqs = get_possible_ssvep_freqs(frame_rate, stim_type="reversal") + print( + ( + "Flickering frequencies (Hz): {}\n".format( + [stim_patterns[0]["freq"], stim_patterns[1]["freq"]] + ) + ) + ) -def present(duration=120, eeg=None, save_fn=None): - - - for ii, trial in trials.iterrows(): - # Intertrial interval - core.wait(iti + np.random.rand() * jitter) + return [ + init_flicker_stim(frame_rate, 2, soa), + init_flicker_stim(frame_rate, 3, soa), + ] - """ Unique """ + def present_stimulus(self): + # Select stimulus frequency - ind = trials["stim_freq"].iloc[ii] + ind = self.trials["stim_freq"].iloc[ii] # Push sample if eeg: timestamp = time() - if eeg.backend == "muselsl": - marker = [markernames[ind]] + if self.eeg.backend == "muselsl": + marker = [self.markernames[ind]] else: - marker = markernames[ind] - eeg.push_sample(marker=marker, timestamp=timestamp) + marker = self.markernames[ind] + self.eeg.push_sample(marker=marker, timestamp=timestamp) # Present flickering stim for _ in range(int(stim_patterns[ind]["n_cycles"])): @@ -115,47 +101,4 @@ def present(duration=120, eeg=None, save_fn=None): for _ in range(stim_patterns[ind]["cycle"][1]): mywin.flip() grating_neg.setAutoDraw(False) - - """ Unique ends """ - - # offset - 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 SSVEP 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. - - Warning: This experiment contains flashing lights and may induce a seizure. Discretion is advised. - - """ - 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() + pass From f208e086a7302ffec069ed1475faabbf6b95f9d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Erik=20Bj=C3=A4reholt?= Date: Thu, 16 Jun 2022 16:24:10 +0200 Subject: [PATCH 09/22] fix: fixed import (brainflow updated API) --- eegnb/devices/eeg.py | 57 +++++++++++++++++++++--------------------- eegnb/devices/utils.py | 14 +++++------ 2 files changed, 36 insertions(+), 35 deletions(-) diff --git a/eegnb/devices/eeg.py b/eegnb/devices/eeg.py index 6b22a377..13fdb83a 100644 --- a/eegnb/devices/eeg.py +++ b/eegnb/devices/eeg.py @@ -14,11 +14,17 @@ import numpy as np import pandas as pd -from brainflow import BoardShim, BoardIds, BrainFlowInputParams +from brainflow.board_shim import BoardShim, BoardIds, BrainFlowInputParams from muselsl import stream, list_muses, record, constants as mlsl_cnsts from pylsl import StreamInfo, StreamOutlet, StreamInlet, resolve_byprop -from eegnb.devices.utils import get_openbci_usb, create_stim_array,SAMPLE_FREQS,EEG_INDICES,EEG_CHANNELS +from eegnb.devices.utils import ( + get_openbci_usb, + create_stim_array, + SAMPLE_FREQS, + EEG_INDICES, + EEG_CHANNELS, +) logger = logging.getLogger(__name__) @@ -39,18 +45,19 @@ "notion2", "freeeeg32", "crown", - "museS_bfn", # bfn = brainflow with native bluetooth; - "museS_bfb", # bfb = brainflow with BLED dongle bluetooth + "museS_bfn", # bfn = brainflow with native bluetooth; + "museS_bfb", # bfb = brainflow with BLED dongle bluetooth "muse2_bfn", "muse2_bfb", "muse2016_bfn", - "muse2016_bfb" + "muse2016_bfb", ] class EEG: device_name: str stream_started: bool = False + def __init__( self, device=None, @@ -85,8 +92,8 @@ def initialize_backend(self): self.timestamp_channel = BoardShim.get_timestamp_channel(self.brainflow_id) elif self.backend == "muselsl": self._init_muselsl() - self._muse_get_recent() # run this at initialization to get some - # stream metadata into the eeg class + self._muse_get_recent() # run this at initialization to get some + # stream metadata into the eeg class def _get_backend(self, device_name): if device_name in brainflow_devices: @@ -126,7 +133,7 @@ def _start_muse(self, duration): print("will save to file: %s" % self.save_fn) self.recording = Process(target=record, args=(duration, self.save_fn)) self.recording.start() - + time.sleep(5) self.stream_started = True self.push_sample([99], timestamp=time.time()) @@ -137,7 +144,7 @@ def _stop_muse(self): def _muse_push_sample(self, marker, timestamp): self.muse_StreamOutlet.push_sample(marker, timestamp) - def _muse_get_recent(self, n_samples: int=256, restart_inlet: bool=False): + def _muse_get_recent(self, n_samples: int = 256, restart_inlet: bool = False): if self._muse_recent_inlet and not restart_inlet: inlet = self._muse_recent_inlet else: @@ -157,9 +164,8 @@ def _muse_get_recent(self, n_samples: int=256, restart_inlet: bool=False): self.info = info self.n_chans = n_chans - timeout = (n_samples/sfreq)+0.5 - samples, timestamps = inlet.pull_chunk(timeout=timeout, - max_samples=n_samples) + timeout = (n_samples / sfreq) + 0.5 + samples, timestamps = inlet.pull_chunk(timeout=timeout, max_samples=n_samples) samples = np.array(samples) timestamps = np.array(timestamps) @@ -260,13 +266,13 @@ def _init_brainflow(self): elif self.device_name == "muse2_bfn": self.brainflow_id = BoardIds.MUSE_2_BOARD.value - + elif self.device_name == "muse2_bfb": self.brainflow_id = BoardIds.MUSE_2_BLED_BOARD.value elif self.device_name == "muse2016_bfn": self.brainflow_id = BoardIds.MUSE_2016_BOARD.value - + elif self.device_name == "muse2016_bfb": self.brainflow_id = BoardIds.MUSE_2016_BLED_BOARD.value @@ -291,11 +297,13 @@ def _start_brainflow(self): # only start stream if non exists if not self.stream_started: self.board.start_stream() - + self.stream_started = True # wait for signal to settle - if (self.device_name.find("cyton") != -1) or (self.device_name.find("ganglion") != -1): + if (self.device_name.find("cyton") != -1) or ( + self.device_name.find("ganglion") != -1 + ): # wait longer for openbci cyton / ganglion sleep(10) else: @@ -314,21 +322,19 @@ def _stop_brainflow(self): # Create a column for the stimuli to append to the EEG data stim_array = create_stim_array(timestamps, self.markers) - timestamps = timestamps[ ..., None ] - + timestamps = timestamps[..., None] + # Add an additional dimension so that shapes match total_data = np.append(timestamps, eeg_data, 1) # Append the stim array to data. - total_data = np.append(total_data, stim_array, 1) - + total_data = np.append(total_data, stim_array, 1) + # Subtract five seconds of settling time from beginning total_data = total_data[5 * self.sfreq :] data_df = pd.DataFrame(total_data, columns=["timestamps"] + ch_names + ["stim"]) data_df.to_csv(self.save_fn, index=False) - - def _brainflow_extract(self, data): """ Formats the data returned from brainflow to get @@ -357,14 +363,12 @@ def _brainflow_extract(self, data): eeg_data = data[:, BoardShim.get_eeg_channels(self.brainflow_id)] timestamps = data[:, BoardShim.get_timestamp_channel(self.brainflow_id)] - return ch_names,eeg_data,timestamps - + return ch_names, eeg_data, timestamps def _brainflow_push_sample(self, marker): last_timestamp = self.board.get_current_board_data(1)[self.timestamp_channel][0] self.markers.append([marker, last_timestamp]) - def _brainflow_get_recent(self, n_samples=256): # initialize brainflow if not set @@ -405,7 +409,6 @@ def start(self, fn, duration=None): elif self.backend == "muselsl": self._start_muse(duration) - def push_sample(self, marker, timestamp): """ Universal method for pushing a marker and its timestamp to store alongside the EEG data. @@ -445,6 +448,4 @@ def get_recent(self, n_samples: int = 256): sorted_cols = sorted(df.columns) df = df[sorted_cols] - return df - diff --git a/eegnb/devices/utils.py b/eegnb/devices/utils.py index 2c085340..8e01ad9a 100644 --- a/eegnb/devices/utils.py +++ b/eegnb/devices/utils.py @@ -3,14 +3,14 @@ import platform import serial -from brainflow import BoardShim, BoardIds +from brainflow.board_shim import BoardShim, BoardIds # Default channel names for the various EEG devices. EEG_CHANNELS = { - "muse2016": ['TP9', 'AF7', 'AF8', 'TP10', 'Right AUX'], - "muse2": ['TP9', 'AF7', 'AF8', 'TP10', 'Right AUX'], - "museS": ['TP9', 'AF7', 'AF8', 'TP10', 'Right AUX'], + "muse2016": ["TP9", "AF7", "AF8", "TP10", "Right AUX"], + "muse2": ["TP9", "AF7", "AF8", "TP10", "Right AUX"], + "museS": ["TP9", "AF7", "AF8", "TP10", "Right AUX"], "muse2016_bfn": BoardShim.get_eeg_names(BoardIds.MUSE_2016_BOARD.value), "muse2016_bfb": BoardShim.get_eeg_names(BoardIds.MUSE_2016_BLED_BOARD.value), "muse2_bfn": BoardShim.get_eeg_names(BoardIds.MUSE_2_BOARD.value), @@ -26,7 +26,7 @@ "notion1": BoardShim.get_eeg_names(BoardIds.NOTION_1_BOARD.value), "notion2": BoardShim.get_eeg_names(BoardIds.NOTION_2_BOARD.value), "crown": BoardShim.get_eeg_names(BoardIds.CROWN_BOARD.value), - "freeeeg32": [f'eeg_{i}' for i in range(0,32)], + "freeeeg32": [f"eeg_{i}" for i in range(0, 32)], } BRAINFLOW_CHANNELS = { @@ -62,9 +62,9 @@ "muse2016": 256, "muse2": 256, "museS": 256, - "muse2016_bfn": BoardShim.get_sampling_rate(BoardIds.MUSE_2016_BOARD.value), + "muse2016_bfn": BoardShim.get_sampling_rate(BoardIds.MUSE_2016_BOARD.value), "muse2016_bfb": BoardShim.get_sampling_rate(BoardIds.MUSE_2016_BLED_BOARD.value), - "muse2_bfn": BoardShim.get_sampling_rate(BoardIds.MUSE_2_BOARD.value), + "muse2_bfn": BoardShim.get_sampling_rate(BoardIds.MUSE_2_BOARD.value), "muse2_bfb": BoardShim.get_sampling_rate(BoardIds.MUSE_2_BLED_BOARD.value), "museS_bfn": BoardShim.get_sampling_rate(BoardIds.MUSE_S_BOARD.value), "museS_bfb": BoardShim.get_sampling_rate(BoardIds.MUSE_S_BLED_BOARD.value), From 5db55f3c930a0390d617d01634b150b89908d279 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Fri, 17 Jun 2022 12:03:48 +0100 Subject: [PATCH 10/22] Playing around still --- eegnb/experiments/visual_vep/vep.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/eegnb/experiments/visual_vep/vep.py b/eegnb/experiments/visual_vep/vep.py index 0da5d8fd..b31b32c3 100644 --- a/eegnb/experiments/visual_vep/vep.py +++ b/eegnb/experiments/visual_vep/vep.py @@ -4,7 +4,7 @@ from time import time, strftime, gmtime from optparse import OptionParser from pylsl import StreamInfo, StreamOutlet - +from Experiment import Experiment def present(duration=120): From efb1928847e9d4d54d338ee4245c2f388e691b8b Mon Sep 17 00:00:00 2001 From: Parvfect Date: Fri, 17 Jun 2022 21:55:43 +0100 Subject: [PATCH 11/22] Fixing import errors --- eegnb/experiments/visual_n170/n170.py | 2 +- eegnb/experiments/visual_p300/p300.py | 2 +- eegnb/experiments/visual_ssvep/ssvep.py | 2 +- eegnb/experiments/visual_vep/vep.py | 18 +++++++++++++++++- 4 files changed, 20 insertions(+), 4 deletions(-) diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index 30d35615..f4da0c2d 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -17,7 +17,7 @@ from eegnb import generate_save_fn from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE -from Experiment import Experiment +from eegnb.experiments.Experiment import Experiment class VisualN170(Experiment): diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 5cd51900..986aa62c 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -9,7 +9,7 @@ from eegnb import generate_save_fn from eegnb.stimuli import CAT_DOG - +from eegnb.experiments.Experiment import Experiment class VisualP300(Experiment): diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index 14db038b..e6a3415e 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -8,7 +8,7 @@ from psychopy import visual, core, event from eegnb import generate_save_fn -from Experiment import Experiment +from eegnb.experiments.Experiment import Experiment class VisualSSVEP(Experiment): diff --git a/eegnb/experiments/visual_vep/vep.py b/eegnb/experiments/visual_vep/vep.py index b31b32c3..f1370f94 100644 --- a/eegnb/experiments/visual_vep/vep.py +++ b/eegnb/experiments/visual_vep/vep.py @@ -4,7 +4,23 @@ from time import time, strftime, gmtime from optparse import OptionParser from pylsl import StreamInfo, StreamOutlet -from Experiment import Experiment +from eegnb.experiments.Experiment import Experiment + + +class VisualVEP(Experiment): + + def __init__(self, duration=120, eeg: EEG=None, save_fn=None, + n_trials = 2000, iti = 0.2, soa = 0.2, jitter = 0.1): + + exp_name = "Visual VEP" + super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) + + def load_stimulus(): + pass + + def present_stimulus(): + pass + def present(duration=120): From 7436444bcdbeca3cbae8affa697ff7aea1b6b625 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Fri, 17 Jun 2022 22:43:18 +0100 Subject: [PATCH 12/22] Adding abstractmethod decorators --- eegnb/experiments/Experiment.py | 15 +++++++++------ eegnb/experiments/visual_n170/n170.py | 4 ++-- eegnb/experiments/visual_p300/p300.py | 2 +- 3 files changed, 12 insertions(+), 9 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index b282deff..33aea592 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -4,6 +4,8 @@ Specific experiments are implemented as sub classes that inherit a load_stimulus and present_stimulus method """ +from abc import ABC, abstractmethod + class Experiment: def __init_(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): @@ -20,9 +22,14 @@ def __init_(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): self.soa = soa self.jitter = jitter + @abstractmethod def load_stimulus(self): """ Needs to be overwritten by specific experiment """ - pass + raise NotImplementedError + + @abstractmethod + def present_stimulus(self): + raise NotImplementedError def setup(self): @@ -49,11 +56,7 @@ def setup(self): print( f"No path for a save file was passed to the experiment. Saving data to {save_fn}" ) - - def present_stimulus(self): - """ Needs to be overwritten by specific experiment """ - pass - + def present(self): """ Do the present operation for a bunch of experiments """ diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index f4da0c2d..f08beaf2 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -17,10 +17,10 @@ from eegnb import generate_save_fn from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE -from eegnb.experiments.Experiment import Experiment +from eegnb.experiments import Experiment -class VisualN170(Experiment): +class VisualN170(Experiment.Experiment): def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 986aa62c..54710ee3 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -9,7 +9,7 @@ from eegnb import generate_save_fn from eegnb.stimuli import CAT_DOG -from eegnb.experiments.Experiment import Experiment +from eegnb.experiments.Experiment import Experiment as Experiment class VisualP300(Experiment): From ee7cca9ed27e49469af8bef36bb15f5eb1c95307 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Fri, 17 Jun 2022 22:57:42 +0100 Subject: [PATCH 13/22] Still working on the import error --- eegnb/experiments/visual_n170/n170.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index f08beaf2..f4da0c2d 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -17,10 +17,10 @@ from eegnb import generate_save_fn from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE -from eegnb.experiments import Experiment +from eegnb.experiments.Experiment import Experiment -class VisualN170(Experiment.Experiment): +class VisualN170(Experiment): def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): From d73abe080b58173edc41351b912a4137c8850152 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Mon, 27 Jun 2022 23:25:34 +0100 Subject: [PATCH 14/22] Guess what's finally working --- eegnb/experiments/Experiment.py | 81 +++++++++++++++++-------- eegnb/experiments/visual_n170/n170.py | 19 +++--- eegnb/experiments/visual_p300/p300.py | 19 +++--- eegnb/experiments/visual_ssvep/ssvep.py | 12 ++-- 4 files changed, 83 insertions(+), 48 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index 33aea592..4e986805 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -1,14 +1,37 @@ """ Initial run of the Experiment Class Refactor base class -Specific experiments are implemented as sub classes that inherit a load_stimulus and present_stimulus method +Specific experiments are implemented as sub classes that overload a load_stimulus and present_stimulus method + + +Running each experiment: +obj = VisualP300({parametrs}) +obj.present() """ 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 Experiment: - def __init_(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): + def __init__(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): """ Anything that must be passed as a minimum for the experiment should be initialized here """ self.exp_name = exp_name @@ -21,7 +44,7 @@ def __init_(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): self.iti = iti self.soa = soa self.jitter = jitter - + @abstractmethod def load_stimulus(self): """ Needs to be overwritten by specific experiment """ @@ -37,11 +60,11 @@ def setup(self): 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))) + self.parameter = np.random.binomial(1, 0.5, self.n_trials) + self.trials = DataFrame(dict(parameter=self.parameter, timestamp=np.zeros(self.n_trials))) # Setup Graphics - mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) + self.mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) # Needs to be overwritten by specific experiment self.stim = self.load_stimulus() @@ -49,20 +72,22 @@ def setup(self): # Show Instruction Screen self.show_instructions() - # 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}" - ) + if self.eeg: + if 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(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 present(self): """ Do the present operation for a bunch of experiments """ + + self.setup() # 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) + if self.eeg: + self.eeg.start(self.save_fn, duration=self.record_duration + 5) start = time() @@ -73,17 +98,21 @@ def present(self): 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.present_stimulus(ii) # Offset - mywin.flip() + core.wait(self.soa) + self.mywin.flip() if len(event.getKeys()) > 0 or (time() - start) > self.record_duration: break event.clearEvents() - + # Close the EEG stream - if eeg: - eeg.stop() + if self.eeg: + self.eeg.stop() + + # Close the window + self.mywin.close() def show_instructions(self): @@ -91,14 +120,14 @@ def show_instructions(self): self.instruction_text = self.instruction_text % self.duration # graphics - mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - mywin.mouseVisible = False + #mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) + self.mywin.mouseVisible = False # Instructions - text = visual.TextStim(win=mywin, text=self.instruction_text, color=[-1, -1, -1]) + text = visual.TextStim(win=self.mywin, text=self.instruction_text, color=[-1, -1, -1]) text.draw() - mywin.flip() + self.mywin.flip() event.waitKeys(keyList="space") - mywin.mouseVisible = True - mywin.close() \ No newline at end of file + self.mywin.mouseVisible = True + \ No newline at end of file diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index f4da0c2d..0f216dbd 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -26,21 +26,21 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): exp_name = "Visual N170" - super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) + super(VisualN170, self).__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) def load_stimulus(self): - load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) + load_image = lambda fn: visual.ImageStim(win=self.mywin, image=fn) - 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")))) + self.faces = list(map(load_image, glob(os.path.join(FACE_HOUSE, "faces", "*_3.jpg")))) + self.houses = list(map(load_image, glob(os.path.join(FACE_HOUSE, "houses", "*.3.jpg")))) - return [houses, faces] + return [self.houses, self.faces] - def present_stimulus(self): + def present_stimulus(self, ii): - label = self.trials["image_type"].iloc[ii] - image = choice(faces if label == 1 else houses) + label = self.trials["parameter"].iloc[ii] + image = choice(self.faces if label == 1 else self.houses) image.draw() # Push sample @@ -51,4 +51,5 @@ def present_stimulus(self): else: marker = self.markernames[label] self.eeg.push_sample(marker=marker, timestamp=timestamp) - \ No newline at end of file + + self.mywin.flip() \ No newline at end of file diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 54710ee3..e7883a46 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -9,6 +9,7 @@ from eegnb import generate_save_fn from eegnb.stimuli import CAT_DOG +from eegnb.devices.eeg import EEG from eegnb.experiments.Experiment import Experiment as Experiment class VisualP300(Experiment): @@ -20,18 +21,18 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) def load_stimulus(self): - load_image = lambda fn: visual.ImageStim(win=mywin, image=fn) + + load_image = lambda fn: visual.ImageStim(win=self.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")))) + self.targets = list(map(load_image, glob(os.path.join(CAT_DOG, "target-*.jpg")))) + self.nontargets = list(map(load_image, glob(os.path.join(CAT_DOG, "nontarget-*.jpg")))) - return [nontargets, targets] + return [self.nontargets, self.targets] - def present_stimulus(self): + def present_stimulus(self, ii): - label = self.trials["image_type"].iloc[ii] - image = choice(targets if label == 1 else nontargets) + label = self.trials["parameter"].iloc[ii] + image = choice(self.targets if label == 1 else self.nontargets) image.draw() # Push sample @@ -42,3 +43,5 @@ def present_stimulus(self): else: marker = self.markernames[label] self.eeg.push_sample(marker=marker, timestamp=timestamp) + + self.mywin.flip() \ No newline at end of file diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index e6a3415e..71baf9a8 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -20,7 +20,7 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, i def load_stimulus(self): - grating = visual.GratingStim(win=mywin, mask="circle", size=80, sf=0.2) + grating = visual.GratingStim(win=self.mywin, mask="circle", size=80, sf=0.2) grating_neg = visual.GratingStim( win=mywin, mask="circle", size=80, sf=0.2, phase=0.5 ) @@ -31,8 +31,8 @@ def load_stimulus(self): # Generate the possible ssvep frequencies based on monitor refresh rate def get_possible_ssvep_freqs(frame_rate, stim_type="single"): if stim_type == "single": - max_period_nb = int(frame_rate / 6) - periods = np.arange(max_period_nb) + 1 + max_period_nb = int(frame_rate / 6) + periods = np.arange(max_period_nb) + 1 if stim_type == "single": freqs = dict() @@ -77,10 +77,10 @@ def init_flicker_stim(frame_rate, cycle, soa): init_flicker_stim(frame_rate, 3, soa), ] - def present_stimulus(self): + def present_stimulus(self, ii): # Select stimulus frequency - ind = self.trials["stim_freq"].iloc[ii] + ind = self.trials["parameter"].iloc[ii] # Push sample if eeg: @@ -102,3 +102,5 @@ def present_stimulus(self): mywin.flip() grating_neg.setAutoDraw(False) pass + + self.mywin.flip() \ No newline at end of file From ee5ab09df22c4b190eab13aef0966469a5ddf7cb Mon Sep 17 00:00:00 2001 From: Parvfect Date: Wed, 6 Jul 2022 12:08:40 +0100 Subject: [PATCH 15/22] Comments and naming ticks --- eegnb/experiments/Experiment.py | 102 ++++++++++++++++-------- eegnb/experiments/visual_n170/n170.py | 15 ++-- eegnb/experiments/visual_p300/p300.py | 18 +++-- eegnb/experiments/visual_ssvep/ssvep.py | 32 +++----- eegnb/experiments/visual_vep/vep.py | 4 - 5 files changed, 97 insertions(+), 74 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index 4e986805..be31a80b 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -29,10 +29,10 @@ from eegnb import generate_save_fn from eegnb.devices.eeg import EEG -class Experiment: +class BaseExperiment: def __init__(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter): - """ Anything that must be passed as a minimum for the experiment should be initialized here """ + """ 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 @@ -47,43 +47,86 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) @abstractmethod def load_stimulus(self): - """ Needs to be overwritten by specific experiment """ + """ + 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): + 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): + def setup(self, instructions=True): + # Initializing the record duration and the marker names self.record_duration = np.float32(self.duration) self.markernames = [1, 2] - # Setup Trial list -> Common in most (csv in Unicorn) + # 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))) - # Setup Graphics - self.mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) + # Setting up Graphics + self.window = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - # Needs to be overwritten by specific experiment + # 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 - self.show_instructions() + # 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 save_fn is None: # If no save_fn passed, generate a new unnamed save file + # If no save_fn passed, generate a new unnamed save file + if 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 present(self): + 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 """ - self.setup() + # Setup the experiment, alternatively could get rid of this line, something to think about + self.setup(instructions) # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point if self.eeg: @@ -97,37 +140,26 @@ def present(self): # 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 + # Stimulus presentation overwritten by specific experiment self.present_stimulus(ii) # Offset core.wait(self.soa) - self.mywin.flip() + 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() - # Close the EEG stream + # Closing the EEG stream if self.eeg: self.eeg.stop() - # Close the window - self.mywin.close() + # Closing the window + self.window.close() - def show_instructions(self): - - self.instruction_text = self.instruction_text % self.duration - - # graphics - #mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True) - self.mywin.mouseVisible = False - - # Instructions - text = visual.TextStim(win=self.mywin, text=self.instruction_text, color=[-1, -1, -1]) - text.draw() - self.mywin.flip() - event.waitKeys(keyList="space") - - self.mywin.mouseVisible = True - \ No newline at end of file + \ No newline at end of file diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index 0f216dbd..593a64a2 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -1,3 +1,5 @@ +""" eeg-notebooks/eegnb/experiments/visual_n170/n170.py """ + from psychopy import prefs #change the pref libraty to PTB and set the latency mode to high precision prefs.hardware['audioLib'] = 'PTB' @@ -14,13 +16,12 @@ from pandas import DataFrame from psychopy import visual, core, event -from eegnb import generate_save_fn from eegnb.devices.eeg import EEG from eegnb.stimuli import FACE_HOUSE -from eegnb.experiments.Experiment import Experiment +from eegnb.experiments import Experiment -class VisualN170(Experiment): +class VisualN170(Experiment.BaseExperiment): def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): @@ -30,16 +31,16 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, def load_stimulus(self): - load_image = lambda fn: visual.ImageStim(win=self.mywin, image=fn) + load_image = lambda fn: visual.ImageStim(win=self.window, image=fn) self.faces = list(map(load_image, glob(os.path.join(FACE_HOUSE, "faces", "*_3.jpg")))) self.houses = list(map(load_image, glob(os.path.join(FACE_HOUSE, "houses", "*.3.jpg")))) return [self.houses, self.faces] - def present_stimulus(self, ii): + def present_stimulus(self, idx : int): - label = self.trials["parameter"].iloc[ii] + label = self.trials["parameter"].iloc[idx] image = choice(self.faces if label == 1 else self.houses) image.draw() @@ -52,4 +53,4 @@ def present_stimulus(self, ii): marker = self.markernames[label] self.eeg.push_sample(marker=marker, timestamp=timestamp) - self.mywin.flip() \ No newline at end of file + self.window.flip() \ No newline at end of file diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index e7883a46..1e901e15 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -1,18 +1,22 @@ + +""" eeg-notebooks/eegnb/experiments/visual_p300/p300.py """ + 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.stimuli import CAT_DOG +from eegnb.experiments import Experiment from eegnb.devices.eeg import EEG -from eegnb.experiments.Experiment import Experiment as Experiment -class VisualP300(Experiment): +class VisualP300(Experiment.BaseExperiment): def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): @@ -22,16 +26,16 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, def load_stimulus(self): - load_image = lambda fn: visual.ImageStim(win=self.mywin, image=fn) + load_image = lambda fn: visual.ImageStim(win=self.window, image=fn) # Setup graphics self.targets = list(map(load_image, glob(os.path.join(CAT_DOG, "target-*.jpg")))) self.nontargets = list(map(load_image, glob(os.path.join(CAT_DOG, "nontarget-*.jpg")))) return [self.nontargets, self.targets] - def present_stimulus(self, ii): + def present_stimulus(self, idx:int): - label = self.trials["parameter"].iloc[ii] + label = self.trials["parameter"].iloc[idx] image = choice(self.targets if label == 1 else self.nontargets) image.draw() @@ -44,4 +48,4 @@ def present_stimulus(self, ii): marker = self.markernames[label] self.eeg.push_sample(marker=marker, timestamp=timestamp) - self.mywin.flip() \ No newline at end of file + self.window.flip() \ No newline at end of file diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index 71baf9a8..c4e3835d 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -1,17 +1,8 @@ -import os -from time import time -from glob import glob -from random import choice -import numpy as np -from pandas import DataFrame -from psychopy import visual, core, event +from eegnb.experiments import Experiment -from eegnb import generate_save_fn -from eegnb.experiments.Experiment import Experiment - -class VisualSSVEP(Experiment): +class VisualSSVEP(Experiment.BaseExperiment): def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.5, soa = 3.0, jitter = 0.2): @@ -20,21 +11,18 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, i def load_stimulus(self): - grating = visual.GratingStim(win=self.mywin, mask="circle", size=80, sf=0.2) - grating_neg = visual.GratingStim( - win=mywin, mask="circle", size=80, sf=0.2, phase=0.5 - ) - fixation = visual.GratingStim( - win=mywin, size=0.2, pos=[0, 0], sf=0.2, color=[1, 0, 0], autoDraw=True - ) + grating = visual.GratingStim(win=self.window, mask="circle", size=80, sf=0.2) + + grating_neg = visual.GratingStim(win=mywin, mask="circle", size=80, sf=0.2, phase=0.5) + + fixation = visual.GratingStim(win=mywin, size=0.2, pos=[0, 0], sf=0.2, color=[1, 0, 0], autoDraw=True) # Generate the possible ssvep frequencies based on monitor refresh rate def get_possible_ssvep_freqs(frame_rate, stim_type="single"): + if stim_type == "single": max_period_nb = int(frame_rate / 6) periods = np.arange(max_period_nb) + 1 - - if stim_type == "single": freqs = dict() for p1 in periods: for p2 in periods: @@ -43,6 +31,7 @@ def get_possible_ssvep_freqs(frame_rate, stim_type="single"): freqs[f].append((p1, p2)) except: freqs[f] = [(p1, p2)] + elif stim_type == "reversal": freqs = {frame_rate / p: [(p, p)] for p in periods[::-1]} @@ -53,6 +42,7 @@ def init_flicker_stim(frame_rate, cycle, soa): if isinstance(cycle, tuple): stim_freq = frame_rate / sum(cycle) n_cycles = int(soa * stim_freq) + else: stim_freq = frame_rate / cycle cycle = (cycle, cycle) @@ -103,4 +93,4 @@ def present_stimulus(self, ii): grating_neg.setAutoDraw(False) pass - self.mywin.flip() \ No newline at end of file + self.window.flip() \ No newline at end of file diff --git a/eegnb/experiments/visual_vep/vep.py b/eegnb/experiments/visual_vep/vep.py index f1370f94..1acd1b11 100644 --- a/eegnb/experiments/visual_vep/vep.py +++ b/eegnb/experiments/visual_vep/vep.py @@ -1,8 +1,4 @@ -import numpy as np -from pandas import DataFrame -from psychopy import visual, core, event from time import time, strftime, gmtime -from optparse import OptionParser from pylsl import StreamInfo, StreamOutlet from eegnb.experiments.Experiment import Experiment From cb7895bd71b338d251cb089f7f65d6a61c652611 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Wed, 6 Jul 2022 12:14:11 +0100 Subject: [PATCH 16/22] More comments --- eegnb/experiments/visual_n170/n170.py | 18 +++++++++++++----- eegnb/experiments/visual_p300/p300.py | 2 +- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index 593a64a2..dbf9e373 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -25,26 +25,34 @@ class VisualN170(Experiment.BaseExperiment): def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2): - + + # Set experiment name exp_name = "Visual N170" + # Calling the super class constructor to initialize the experiment variables super(VisualN170, self).__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) def load_stimulus(self): + # Loading Images from the folder load_image = lambda fn: visual.ImageStim(win=self.window, image=fn) - + + # Setting up images for the stimulus self.faces = list(map(load_image, glob(os.path.join(FACE_HOUSE, "faces", "*_3.jpg")))) self.houses = list(map(load_image, glob(os.path.join(FACE_HOUSE, "houses", "*.3.jpg")))) + # Return the list of images as a stimulus object return [self.houses, self.faces] def present_stimulus(self, idx : int): - + + # Get the label of the trial label = self.trials["parameter"].iloc[idx] + # Get the image to be presented image = choice(self.faces if label == 1 else self.houses) + # Draw the image image.draw() - # Push sample + # Pushing the sample to the EEG if self.eeg: timestamp = time() if self.eeg.backend == "muselsl": @@ -52,5 +60,5 @@ def present_stimulus(self, idx : int): else: marker = self.markernames[label] self.eeg.push_sample(marker=marker, timestamp=timestamp) - + self.window.flip() \ No newline at end of file diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 1e901e15..7cf5abb0 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -27,7 +27,7 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, def load_stimulus(self): load_image = lambda fn: visual.ImageStim(win=self.window, image=fn) - # Setup graphics + self.targets = list(map(load_image, glob(os.path.join(CAT_DOG, "target-*.jpg")))) self.nontargets = list(map(load_image, glob(os.path.join(CAT_DOG, "nontarget-*.jpg")))) From 4815d28c8937e78a332edae90f45be30120ae990 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Thu, 7 Jul 2022 20:42:53 +0100 Subject: [PATCH 17/22] Live coding demonstration --- eegnb/experiments/__init__.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/eegnb/experiments/__init__.py b/eegnb/experiments/__init__.py index e69de29b..9c59082d 100644 --- a/eegnb/experiments/__init__.py +++ b/eegnb/experiments/__init__.py @@ -0,0 +1,2 @@ +from .visual_n170.n170 import VisualN170 +from .visual_p300.p300 import VisualP300 \ No newline at end of file From 3ec461a6aa1110fc3a5c78c5a1686828766025a0 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Fri, 8 Jul 2022 15:54:55 +0100 Subject: [PATCH 18/22] ssvep adapted --- eegnb/experiments/__init__.py | 3 +- eegnb/experiments/visual_ssvep/ssvep.py | 63 ++++++++++++++++--------- 2 files changed, 42 insertions(+), 24 deletions(-) diff --git a/eegnb/experiments/__init__.py b/eegnb/experiments/__init__.py index 9c59082d..19726bf3 100644 --- a/eegnb/experiments/__init__.py +++ b/eegnb/experiments/__init__.py @@ -1,2 +1,3 @@ from .visual_n170.n170 import VisualN170 -from .visual_p300.p300 import VisualP300 \ No newline at end of file +from .visual_p300.p300 import VisualP300 +from .visual_ssvep.ssvep import VisualSSVEP \ No newline at end of file diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index c4e3835d..6940eb51 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -1,5 +1,17 @@ from eegnb.experiments import Experiment +import os +from time import time +from glob import glob +from random import choice + +import numpy as np +from pandas import DataFrame +from psychopy import visual, core, event + + +from eegnb.devices.eeg import EEG +from eegnb import generate_save_fn class VisualSSVEP(Experiment.BaseExperiment): @@ -11,18 +23,19 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, i def load_stimulus(self): - grating = visual.GratingStim(win=self.window, mask="circle", size=80, sf=0.2) + self.grating = visual.GratingStim(win=self.window, mask="circle", size=80, sf=0.2) - grating_neg = visual.GratingStim(win=mywin, mask="circle", size=80, sf=0.2, phase=0.5) + self.grating_neg = visual.GratingStim(win=self.window, mask="circle", size=80, sf=0.2, phase=0.5) - fixation = visual.GratingStim(win=mywin, size=0.2, pos=[0, 0], sf=0.2, color=[1, 0, 0], autoDraw=True) + fixation = visual.GratingStim(win=self.window, size=0.2, pos=[0, 0], sf=0.2, color=[1, 0, 0], autoDraw=True) # Generate the possible ssvep frequencies based on monitor refresh rate def get_possible_ssvep_freqs(frame_rate, stim_type="single"): - + + max_period_nb = int(frame_rate / 6) + periods = np.arange(max_period_nb) + 1 + if stim_type == "single": - max_period_nb = int(frame_rate / 6) - periods = np.arange(max_period_nb) + 1 freqs = dict() for p1 in periods: for p2 in periods: @@ -51,29 +64,33 @@ def init_flicker_stim(frame_rate, cycle, soa): return {"cycle": cycle, "freq": stim_freq, "n_cycles": n_cycles} # Set up stimuli - frame_rate = np.round(mywin.getActualFrameRate()) # Frame rate, in Hz + frame_rate = np.round(self.window.getActualFrameRate()) # Frame rate, in Hz freqs = get_possible_ssvep_freqs(frame_rate, stim_type="reversal") - + self.stim_patterns = [ + init_flicker_stim(frame_rate, 2, self.soa), + init_flicker_stim(frame_rate, 3, self.soa), + ] + print( ( "Flickering frequencies (Hz): {}\n".format( - [stim_patterns[0]["freq"], stim_patterns[1]["freq"]] + [self.stim_patterns[0]["freq"], self.stim_patterns[1]["freq"]] ) ) ) return [ - init_flicker_stim(frame_rate, 2, soa), - init_flicker_stim(frame_rate, 3, soa), + init_flicker_stim(frame_rate, 2, self.soa), + init_flicker_stim(frame_rate, 3, self.soa), ] - def present_stimulus(self, ii): + def present_stimulus(self, idx): # Select stimulus frequency - ind = self.trials["parameter"].iloc[ii] + ind = self.trials["parameter"].iloc[idx] # Push sample - if eeg: + if self.eeg: timestamp = time() if self.eeg.backend == "muselsl": marker = [self.markernames[ind]] @@ -82,15 +99,15 @@ def present_stimulus(self, ii): self.eeg.push_sample(marker=marker, timestamp=timestamp) # Present flickering stim - for _ in range(int(stim_patterns[ind]["n_cycles"])): - grating.setAutoDraw(True) - for _ in range(int(stim_patterns[ind]["cycle"][0])): - mywin.flip() - grating.setAutoDraw(False) - grating_neg.setAutoDraw(True) - for _ in range(stim_patterns[ind]["cycle"][1]): - mywin.flip() - grating_neg.setAutoDraw(False) + for _ in range(int(self.stim_patterns[ind]["n_cycles"])): + self.grating.setAutoDraw(True) + for _ in range(int(self.stim_patterns[ind]["cycle"][0])): + self.window.flip() + self.grating.setAutoDraw(False) + self.grating_neg.setAutoDraw(True) + for _ in range(self.stim_patterns[ind]["cycle"][1]): + self.window.flip() + self.grating_neg.setAutoDraw(False) pass self.window.flip() \ No newline at end of file From b1000bd5ece17922cab43b642fc01706bed39a53 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Mon, 11 Jul 2022 15:00:44 +0100 Subject: [PATCH 19/22] Adapting Auditory Oddball --- eegnb/experiments/Experiment.py | 5 +- eegnb/experiments/auditory_oddball/aob.py | 221 ++++++++-------------- 2 files changed, 78 insertions(+), 148 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index be31a80b..91054ca5 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -3,10 +3,9 @@ Specific experiments are implemented as sub classes that overload a load_stimulus and present_stimulus method - Running each experiment: -obj = VisualP300({parametrs}) -obj.present() +obj = VisualP300({parameters}) +obj.run() """ from abc import ABC, abstractmethod diff --git a/eegnb/experiments/auditory_oddball/aob.py b/eegnb/experiments/auditory_oddball/aob.py index d9d28896..11a892ae 100644 --- a/eegnb/experiments/auditory_oddball/aob.py +++ b/eegnb/experiments/auditory_oddball/aob.py @@ -3,156 +3,87 @@ from psychopy import visual, core, event, sound from time import time - -__title__ = "Auditory oddball (orig)" - - -def present( - save_fn = None, - eeg=None, - duration=120, - n_trials=2010, - iti=0.3, - soa=0.2, - jitter=0.2, - secs=0.2, - volume=0.8, - random_state=42, - s1_freq="C", - s2_freq="D", - s1_octave=5, - s2_octave=6, -): - - """ - - Auditory Oddball Experiment - =========================== - - - Parameters: - ----------- - - duration - duration of the recording in seconds (default 10) - - n_trials - number of trials (default 10) - - iti - intertrial interval (default 0.3) - - soa - stimulus onset asynchrony, = interval between end of stimulus - and next trial (default 0.2) - - jitter - jitter in the intertrial intervals (default 0.2) - - secs - duration of the sound in seconds (default 0.2) - - volume - volume of the sounds in [0,1] (default 0.8) - - random_state - random seed (default 42) - - s1_freq - frequency of first tone - s2_freq - frequency of second tone - - s1_octave - octave of first tone - s2_octave - octave of second tone - - """ - - # Set up trial parameters - np.random.seed(random_state) - markernames = [1, 2] - record_duration = np.float32(duration) - - # Initialize stimuli - aud1 = sound.Sound(s1_freq, octave=s1_octave, secs=secs) - aud1.setVolume(volume) - aud2 = sound.Sound(s2_freq, octave=s2_octave, secs=secs) - aud2.setVolume(volume) - auds = [aud1, aud2] - - # Setup trial list - sound_ind = np.random.binomial(1, 0.25, n_trials) - itis = iti + np.random.rand(n_trials) * jitter - trials = DataFrame(dict(sound_ind=sound_ind, iti=itis)) - trials["soa"] = soa - trials["secs"] = secs - - # Setup graphics - mywin = visual.Window( - [1920, 1080], monitor="testMonitor", units="deg", fullscr=True - ) - fixation = visual.GratingStim(win=mywin, size=0.2, pos=[0, 0], sf=0, rgb=[1, 0, 0]) - fixation.setAutoDraw(True) - mywin.flip() - - # Show the instructions screen - show_instructions(10) - - # Start EEG Stream, wait for signal to settle, and then pull timestamp for start point - if eeg: - eeg.start(save_fn, duration=record_duration) - start = time() - - # Iterate through the events - for ii, trial in trials.iterrows(): - - # Intertrial interval - core.wait(trial["iti"]) - +from eegnb.devices.eeg import EEG +from eegnb.experiments import Experiment + + +class AuditoryOddball(Experiment.BaseExperiment): + + def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, iti = 0.3, soa = 0.2, jitter = 0.2, secs=0.2, volume=0.8, random_state=42, s1_freq="C", s2_freq="D", s1_octave=5, s2_octave=6): + """ + + Auditory Oddball Experiment + =========================== + + Unique Parameters: + ----------- + + secs - duration of the sound in seconds (default 0.2) + + volume - volume of the sounds in [0,1] (default 0.8) + + random_state - random seed (default 42) + + s1_freq - frequency of first tone + s2_freq - frequency of second tone + + s1_octave - octave of first tone + s2_octave - octave of second tone + + """ + + exp_name = "Auditory Oddball" + super().__init__(exp_name, duration, eeg, save_fn, n_trials, iti, soa, jitter) + self.secs = secs + self.volume = volume + self.random_state = random_state + self.s1_freq = s1_freq + self.s2_freq = s2_freq + self.s1_octave = s1_octave + self.s2_octave = s2_octave + + def load_stimulus(self): + + # Set up trial parameters + np.random.seed(random_state) + + # Initialize stimuli + aud1, aud2 = sound.Sound(s1_freq, octave=s1_octave, secs=secs), sound.Sound(s2_freq, octave=s2_octave, secs=secs) + aud1.setVolume(volume) + aud2.setVolume(volume) + self.auds = [aud1, aud2] + + # Setup trial list + sound_ind = np.random.binomial(1, 0.25, n_trials) + itis = iti + np.random.rand(n_trials) * jitter + self.trials = DataFrame(dict(sound_ind=sound_ind, iti=itis)) + self.trials["soa"] = soa + self.trials["secs"] = secs + + self.fixation = visual.GratingStim(win=self.window, size=0.2, pos=[0, 0], sf=0, rgb=[1, 0, 0]) + self.fixation.setAutoDraw(True) + self.window.flip() + + return + + def present_stimulus(self, idx : int): + + # The wait time between trials differs might have to be an option that's hard coded + # core.wait(trial["iti"]) + + # itterrows iterates as series pair - might have to be tested since I am not passing trial + trial = self.trials[idx] + # Select and play sound ind = int(trial["sound_ind"]) - auds[ind].stop() - auds[ind].play() + self.auds[ind].stop() + self.auds[ind].play() # Push sample - if eeg: + if self.eeg: timestamp = time() - marker = [markernames[ind]] + marker = [self.markernames[ind]] marker = list(map(int, marker)) + self.eeg.push_sample(marker=marker, timestamp=timestamp) - eeg.push_sample(marker=marker, timestamp=timestamp) - - mywin.flip() - - # Offset - core.wait(soa) - if len(event.getKeys()) > 0: - break - if (time() - start) > record_duration: - break - event.clearEvents() - - # Cleanup - if eeg: - eeg.stop() - - mywin.close() - - -def show_instructions(duration): - - instruction_text = """ - Welcome to the aMMN 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. - - """ - 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() From d9f53b233da527e5cff333aaa568bcb2b8cf1cdb Mon Sep 17 00:00:00 2001 From: Parvfect Date: Mon, 25 Jul 2022 21:06:30 +0100 Subject: [PATCH 20/22] changing save_fn to self.save_fun --- eegnb/experiments/Experiment.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index 91054ca5..79534bf0 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -88,7 +88,7 @@ def setup(self, instructions=True): # Checking for EEG to setup the EEG stream if self.eeg: # If no save_fn passed, generate a new unnamed save file - if save_fn is None: + if self.save_fn is None: # Generating a random int for the filename random_id = random.randint(1000,10000) # Generating save function From dc97fb5569b854a19a53e5e632a821d7b97dffb3 Mon Sep 17 00:00:00 2001 From: Parvfect Date: Fri, 29 Jul 2022 01:09:35 +0100 Subject: [PATCH 21/22] This maybe the last big change --- eegnb/experiments/Experiment.py | 8 +++++-- eegnb/experiments/__init__.py | 3 ++- eegnb/experiments/auditory_oddball/aob.py | 26 ++++++++++------------- eegnb/experiments/visual_n170/n170.py | 2 +- eegnb/experiments/visual_p300/p300.py | 2 +- eegnb/experiments/visual_ssvep/ssvep.py | 2 +- 6 files changed, 22 insertions(+), 21 deletions(-) diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py index 79534bf0..35da046a 100644 --- a/eegnb/experiments/Experiment.py +++ b/eegnb/experiments/Experiment.py @@ -126,11 +126,15 @@ def run(self, instructions=True): # 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 @@ -140,7 +144,7 @@ def run(self, instructions=True): core.wait(self.iti + np.random.rand() * self.jitter) # Stimulus presentation overwritten by specific experiment - self.present_stimulus(ii) + self.present_stimulus(ii, trial) # Offset core.wait(self.soa) diff --git a/eegnb/experiments/__init__.py b/eegnb/experiments/__init__.py index 19726bf3..e17326ff 100644 --- a/eegnb/experiments/__init__.py +++ b/eegnb/experiments/__init__.py @@ -1,3 +1,4 @@ from .visual_n170.n170 import VisualN170 from .visual_p300.p300 import VisualP300 -from .visual_ssvep.ssvep import VisualSSVEP \ No newline at end of file +from .visual_ssvep.ssvep import VisualSSVEP +from .auditory_oddball.aob import AuditoryOddball \ No newline at end of file diff --git a/eegnb/experiments/auditory_oddball/aob.py b/eegnb/experiments/auditory_oddball/aob.py index 11a892ae..32ccd1b2 100644 --- a/eegnb/experiments/auditory_oddball/aob.py +++ b/eegnb/experiments/auditory_oddball/aob.py @@ -43,22 +43,23 @@ def __init__(self, duration=120, eeg: EEG=None, save_fn=None, n_trials = 2010, i self.s2_octave = s2_octave def load_stimulus(self): + """ Loads the Stimulus """ # Set up trial parameters - np.random.seed(random_state) + np.random.seed(self.random_state) # Initialize stimuli - aud1, aud2 = sound.Sound(s1_freq, octave=s1_octave, secs=secs), sound.Sound(s2_freq, octave=s2_octave, secs=secs) - aud1.setVolume(volume) - aud2.setVolume(volume) + aud1, aud2 = sound.Sound(self.s1_freq, octave=self.s1_octave, secs=self.secs), sound.Sound(self.s2_freq, octave=self.s2_octave, secs=self.secs) + aud1.setVolume(self.volume) + aud2.setVolume(self.volume) self.auds = [aud1, aud2] # Setup trial list - sound_ind = np.random.binomial(1, 0.25, n_trials) - itis = iti + np.random.rand(n_trials) * jitter + sound_ind = np.random.binomial(1, 0.25, self.n_trials) + itis = self.iti + np.random.rand(self.n_trials) * self.jitter self.trials = DataFrame(dict(sound_ind=sound_ind, iti=itis)) - self.trials["soa"] = soa - self.trials["secs"] = secs + self.trials["soa"] = self.soa + self.trials["secs"] = self.secs self.fixation = visual.GratingStim(win=self.window, size=0.2, pos=[0, 0], sf=0, rgb=[1, 0, 0]) self.fixation.setAutoDraw(True) @@ -66,14 +67,9 @@ def load_stimulus(self): return - def present_stimulus(self, idx : int): - - # The wait time between trials differs might have to be an option that's hard coded - # core.wait(trial["iti"]) + def present_stimulus(self, idx : int, trial): + """ Presents the Stimulus """ - # itterrows iterates as series pair - might have to be tested since I am not passing trial - trial = self.trials[idx] - # Select and play sound ind = int(trial["sound_ind"]) self.auds[ind].stop() diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py index dbf9e373..0ff0dbed 100644 --- a/eegnb/experiments/visual_n170/n170.py +++ b/eegnb/experiments/visual_n170/n170.py @@ -43,7 +43,7 @@ def load_stimulus(self): # Return the list of images as a stimulus object return [self.houses, self.faces] - def present_stimulus(self, idx : int): + def present_stimulus(self, idx : int, trial): # Get the label of the trial label = self.trials["parameter"].iloc[idx] diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py index 7cf5abb0..33f6f3e2 100644 --- a/eegnb/experiments/visual_p300/p300.py +++ b/eegnb/experiments/visual_p300/p300.py @@ -33,7 +33,7 @@ def load_stimulus(self): return [self.nontargets, self.targets] - def present_stimulus(self, idx:int): + def present_stimulus(self, idx:int, trial): label = self.trials["parameter"].iloc[idx] image = choice(self.targets if label == 1 else self.nontargets) diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py index 6940eb51..b30baffd 100644 --- a/eegnb/experiments/visual_ssvep/ssvep.py +++ b/eegnb/experiments/visual_ssvep/ssvep.py @@ -84,7 +84,7 @@ def init_flicker_stim(frame_rate, cycle, soa): init_flicker_stim(frame_rate, 3, self.soa), ] - def present_stimulus(self, idx): + def present_stimulus(self, idx, trial): # Select stimulus frequency ind = self.trials["parameter"].iloc[idx] From ef3dba0316a8c9901298ae92e01cb22a4bc7c7de Mon Sep 17 00:00:00 2001 From: Parvfect Date: Wed, 3 Aug 2022 16:42:57 +0100 Subject: [PATCH 22/22] utils file changed, changes work through cli --- eegnb/cli/utils.py | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/eegnb/cli/utils.py b/eegnb/cli/utils.py index 00a769c5..322a16dd 100644 --- a/eegnb/cli/utils.py +++ b/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:")