Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Experiment class refactor #183

Closed
wants to merge 8 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 100 additions & 0 deletions eegnb/experiments/Experiment.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
"""
Initial run of the Experiment Class Refactor base class

Derived classes have to set a few things in major:
1. load_stimulus function : returns an array of stimuli
2. present_stimulus function : presents the stimuli and pushes eeg data back and forth as needed
Additional parameters can be set from the derived class as per the initializer

"""

class Experiment:

def __init_(self, exp_name):
""" Anything that must be passed as a minimum for the experiment should be initialized here """

""" Dk if this overwrites the class variable or is worth doing
if we just assume they will overwrite """

self.exp_name= exp_name
self.instruction_text = """\nWelcome to the {} experiment!\nStay still, focus on the centre of the screen, and try not to blink. \nThis block will run for %s seconds.\n
Press spacebar to continue. \n""".format(exp_name)
self.duration=120
self.eeg:EEG=None
self.save_fn=None
self.n_trials=2010
self.iti = 0.4
self.soa = 0.3
Comment on lines +26 to +27
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@JohnGriffiths Always wondered about these confusing variable names. Are iti and soa abbreviations or something?

In any case, they should probably be given more descriptive names (and note that the values are different in different experiments, so they need to be passed too!)

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah so they are being initialized to default values and then if the experiment needs different ones it can change it after the init (otherwise I could force it to initialize too) - see visual ssvep

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They are standard technical acronyms in psych / cog neuro:

ITI = Inter-trial interval

SOA = Stimulus onset asynchrony

( A related term that we don't have here is ISI (inter-stimulus interval) )

These absolutely should be more clearly defined and consistently used.

self.jitter = 0.2

def setup(self):

self.record_duration = np.float32(self.duration)
self.markernames = [1, 2]

# Setup Trial list -> Common in most (csv in Unicorn)
self.parameter = np.random.binomial(1, 0.5, n_trials)
self.trials = DataFrame(dict(parameter=parameter, timestamp=np.zeros(n_trials)))

# Setup Graphics
mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True)

# Needs to be overwritten by specific experiment
self.stim = self.load_stimulus()

# Show Instruction Screen
self.show_instructions(duration=duration)

# Establish save function
if self.save_fn is None: # If no save_fn passed, generate a new unnamed save file
random_id = random.randint(1000,10000)
self.save_fn = generate_save_fn(eeg.device_name, experiement_id, random_id, random_id, "unnamed")
print(
f"No path for a save file was passed to the experiment. Saving data to {save_fn}"
)

def present(self):
""" Do the present operation for a bunch of experiments """

# Start EEG Stream, wait for signal to settle, and then pull timestamp for start point
if eeg:
eeg.start(self.save_fn, duration=self.record_duration + 5)

start = time()

# Iterate through the events
for ii, trial in self.trials.iterrows():

# Intertrial interval
core.wait(self.iti + np.random.rand() * self.jitter)

# Some form of presenting the stimulus - sometimes order changed in lower files like ssvep
self.present_stimulus(self.trials, ii, self.eeg, self.markernames)

# Offset
mywin.flip()
if len(event.getKeys()) > 0 or (time() - start) > self.record_duration:
break
event.clearEvents()

# Close the EEG stream
if eeg:
eeg.stop()


def show_instructions(self):

self.instruction_text = self.instruction_text % duration

# graphics
mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True)
mywin.mouseVisible = False

# Instructions
text = visual.TextStim(win=mywin, text=self.instruction_text, color=[-1, -1, -1])
text.draw()
mywin.flip()
event.waitKeys(keyList="space")

mywin.mouseVisible = True
mywin.close()
13 changes: 13 additions & 0 deletions eegnb/experiments/Experiment_readme.txt
Original file line number Diff line number Diff line change
@@ -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
123 changes: 28 additions & 95 deletions eegnb/experiments/visual_n170/n170.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,107 +17,40 @@
from eegnb import generate_save_fn
from eegnb.devices.eeg import EEG
from eegnb.stimuli import FACE_HOUSE
from Experiment import Experiment

__title__ = "Visual N170"


def present(duration=120, eeg: EEG=None, save_fn=None,
n_trials = 2010, iti = 0.4, soa = 0.3, jitter = 0.2):
def load_stimulus():
load_image = lambda fn: visual.ImageStim(win=mywin, image=fn)

record_duration = np.float32(duration)
markernames = [1, 2]

# Setup trial list
image_type = np.random.binomial(1, 0.5, n_trials)
trials = DataFrame(dict(image_type=image_type, timestamp=np.zeros(n_trials)))

def load_image(fn):
return visual.ImageStim(win=mywin, image=fn)

# start the EEG stream, will delay 5 seconds to let signal settle

# Setup graphics
mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True)

faces = list(map(load_image, glob(os.path.join(FACE_HOUSE, "faces", "*_3.jpg"))))
houses = list(map(load_image, glob(os.path.join(FACE_HOUSE, "houses", "*.3.jpg"))))
stim = [houses, faces]

# Show the instructions screen
show_instructions(duration)

if eeg:
if save_fn is None: # If no save_fn passed, generate a new unnamed save file
random_id = random.randint(1000,10000)
save_fn = generate_save_fn(eeg.device_name, "visual_n170", random_id, random_id, "unnamed")
print(
f"No path for a save file was passed to the experiment. Saving data to {save_fn}"
)
eeg.start(save_fn, duration=record_duration + 5)

# Start EEG Stream, wait for signal to settle, and then pull timestamp for start point
start = time()

# Iterate through the events
for ii, trial in trials.iterrows():
# Inter trial interval
core.wait(iti + np.random.rand() * jitter)

# Select and display image
label = trials["image_type"].iloc[ii]
image = choice(faces if label == 1 else houses)
image.draw()

# Push sample
if eeg:
timestamp = time()
if eeg.backend == "muselsl":
marker = [markernames[label]]
else:
marker = markernames[label]
eeg.push_sample(marker=marker, timestamp=timestamp)

mywin.flip()

# offset
core.wait(soa)
mywin.flip()
if len(event.getKeys()) > 0 or (time() - start) > record_duration:
break

event.clearEvents()

# Cleanup
if eeg:
eeg.stop()

mywin.close()


def show_instructions(duration):

instruction_text = """
Welcome to the N170 experiment!

Stay still, focus on the centre of the screen, and try not to blink.

This block will run for %s seconds.

Press spacebar to continue.
return [houses, faces]

"""
instruction_text = instruction_text % duration
def present_stimulus(trials, ii, eeg, markernames):

label = trials["image_type"].iloc[ii]
image = choice(faces if label == 1 else houses)
image.draw()

# graphics
mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True)

mywin.mouseVisible = False

# Instructions
text = visual.TextStim(win=mywin, text=instruction_text, color=[-1, -1, -1])
text.draw()
mywin.flip()
event.waitKeys(keyList="space")
# Push sample
if eeg:
timestamp = time()
if eeg.backend == "muselsl":
marker = [markernames[label]]
else:
marker = markernames[label]
eeg.push_sample(marker=marker, timestamp=timestamp)
Comment on lines +41 to +45
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@JohnGriffiths This muselsl-edgecase should prob be handled directly in push_sample, and not in calling code. (not related to this PR)

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

agree.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also the whole push sample part should be inside the experiment base class; in the majority of cases just comes immediately after stim presentation and doesn't need to be repeated



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()
Comment on lines +50 to +55
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do it like this? I assumed you'd create a subclass of Experiment?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this works better because I don't need to care about redefining a bunch of stuff and thinking about defining the functions in the base class etc. This way I can use the same class everywhere and just initialize it differently for the different experiments. And since I am seperating the present_stimulus and load_stimulus functions, it does not make much of a difference

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with Erik I think the natural way of doing this is with an experiment base class that is subclassed for every experiment. The functions that you are passing here would then be (potentially decorated) methods in the subclasses.

You're right that you're saving some overhead by using the same class everywhere and not needing placeholder function definitions in the base class like

def present_stimulus(self):
  pass

...but that's pretty standard idiomatic python and isn't a huge deal.

I'm still going through this thoroughly but I'd like a bit more spelling out as to why this function passing approach is a better than subclassing from a general code base structure perspective.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I've had a look around and I think it makes more sense to have subclasses cause it can also provide readymade access to the variables from outside. Gonna convert it to that type now.

Copy link
Collaborator

@ErikBjare ErikBjare Jun 14, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a note/tidbit/lesson on your existing approach, that if you do:

class A:
    def b(self):
        print(self)

def c(self):
    print(self)

a = A()
print(a.b)  # <bound method A.b of <__main__.A object at 0x7f645b76c4c0>>
a.b()       # <__main__.A object at 0x7f6459ecd9c0>
a.c = c
print(a.c)  # <function c at 0x7f645c04c790>
a.c()       # TypeError: c() missing 1 required positional argument: 'self'

In other words, a.b will be bound to the instance (have access to self), but a.c wont, and would need to be bound to get access to its own class members: .https://stackoverflow.com/a/1015405/965332


mywin.mouseVisible = True
mywin.close()
112 changes: 24 additions & 88 deletions eegnb/experiments/visual_p300/p300.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,103 +10,39 @@
from eegnb import generate_save_fn
from eegnb.stimuli import CAT_DOG

__title__ = "Visual P300"


def present(duration=120, eeg=None, save_fn=None):
n_trials = 2010
iti = 0.4
soa = 0.3
jitter = 0.2
record_duration = np.float32(duration)
markernames = [1, 2]

# Setup trial list
image_type = np.random.binomial(1, 0.5, n_trials)
trials = DataFrame(dict(image_type=image_type, timestamp=np.zeros(n_trials)))

def load_image(fn):
return visual.ImageStim(win=mywin, image=fn)

def load_stimulus():
load_image = lambda fn: visual.ImageStim(win=mywin, image=fn)
# Setup graphics
mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True)

targets = list(map(load_image, glob(os.path.join(CAT_DOG, "target-*.jpg"))))
nontargets = list(map(load_image, glob(os.path.join(CAT_DOG, "nontarget-*.jpg"))))
stim = [nontargets, targets]

# Show instructions
show_instructions(duration=duration)

# start the EEG stream, will delay 5 seconds to let signal settle
if eeg:
if save_fn is None: # If no save_fn passed, generate a new unnamed save file
save_fn = generate_save_fn(eeg.device_name, "visual_p300", "unnamed")
print(
f"No path for a save file was passed to the experiment. Saving data to {save_fn}"
)
eeg.start(save_fn, duration=record_duration)

# Iterate through the events
start = time()
for ii, trial in trials.iterrows():
# Inter trial interval
core.wait(iti + np.random.rand() * jitter)

# Select and display image
label = trials["image_type"].iloc[ii]
image = choice(targets if label == 1 else nontargets)
image.draw()

# Push sample
if eeg:
timestamp = time()
if eeg.backend == "muselsl":
marker = [markernames[label]]
else:
marker = markernames[label]
eeg.push_sample(marker=marker, timestamp=timestamp)

mywin.flip()

return [nontargets, targets]

# offset
core.wait(soa)
mywin.flip()
if len(event.getKeys()) > 0 or (time() - start) > record_duration:
break
def present_stimulus(trials, ii, eeg, markernames):

event.clearEvents()
label = trials["image_type"].iloc[ii]
image = choice(targets if label == 1 else nontargets)
image.draw()

# Cleanup
# Push sample
if eeg:
eeg.stop()
mywin.close()


def show_instructions(duration):

instruction_text = """
Welcome to the P300 experiment!

Stay still, focus on the centre of the screen, and try not to blink.

This block will run for %s seconds.

Press spacebar to continue.
timestamp = time()
if eeg.backend == "muselsl":
marker = [markernames[label]]
else:
marker = markernames[label]
eeg.push_sample(marker=marker, timestamp=timestamp)

if __name__ == "__main__":

test = Experiment("Visual P300")
test.instruction_text = instruction_text
test.load_stimulus = load_stimulus
test.present_stimulus = present_stimulus
test.setup()
test.present()

"""
instruction_text = instruction_text % duration

# graphics
mywin = visual.Window([1600, 900], monitor="testMonitor", units="deg", fullscr=True)

mywin.mouseVisible = False

# Instructions
text = visual.TextStim(win=mywin, text=instruction_text, color=[-1, -1, -1])
text.draw()
mywin.flip()
event.waitKeys(keyList="space")

mywin.mouseVisible = True
mywin.close()
Loading