[docs]class FourClassMotorImagery(StimuliAPI):
"""Classic arrow cues for motor imagery."""
# ----------------------------------------------------------------------
def __init__(self, *args, **kwargs):
""""""
super().__init__(*args, **kwargs)
self.add_stylesheet('styles.css')
self.show_cross()
self.dashboard <= w.label('4-Class motor imagery', 'headline4')
self.dashboard <= w.checkbox(
label='Cues',
options=[[cue, True] for cue in UNICODE_CUES],
on_change=None,
id='cues',
)
self.dashboard <= w.slider(
label='Trials per class:',
min=1,
max=100,
value=10,
step=1,
discrete=True,
marks=True,
id='repetitions',
)
self.dashboard <= w.slider(
label='Stimulus duration',
min=2000,
max=6000,
value=4000,
step=100,
unit='ms',
id='duration',
)
self.dashboard <= w.range_slider(
label='Stimulus onset asynchronously',
min=1000,
max=3000,
value_lower=2000,
value_upper=3000,
step=100,
unit='ms',
id='soa',
)
self.dashboard <= w.switch(
label='Record EEG',
checked=False,
id='record',
)
self.dashboard <= w.switch(
label='External marker synchronizer',
checked=False,
on_change=self.synchronizer,
id='record',
)
self.dashboard <= w.button('Start run', on_click=self.start)
self.dashboard <= w.button('Stop run', on_click=self.stop)
# ----------------------------------------------------------------------
[docs] def start(self) -> None:
"""Start the run.
A run consist in a consecutive trials execution.
"""
if w.get_value('record'):
self.start_record()
self.build_trials()
timer.set_timeout(lambda: self.run_pipeline(
self.pipeline_trial, self.trials, callback=self.stop_run), 2000)
# ----------------------------------------------------------------------
[docs] def stop(self) -> None:
"""Stop pipeline execution."""
self.stop_pipeline()
# ----------------------------------------------------------------------
[docs] def stop_run(self) -> None:
"""Stop pipeline execution."""
self.soa()
if w.get_value('record'):
timer.set_timeout(self.stop_record, 2000)
# ----------------------------------------------------------------------
[docs] def build_trials(self) -> None:
"""Define the `trials` and `pipeline trials`.
The `trials` consist (in this case) in a list of cues.
The `pipeline trials` is a set of couples `(callable, duration)` that
define a single trial, this list of functions are executed asynchronously
and repeated for each trial.
"""
self.trials = w.get_value('cues') * w.get_value('repetitions')
random.shuffle(self.trials)
self.pipeline_trial = [
(self.soa, 'soa'), # `soa` is a range reference
(self.trial, 'duration'), # `duration` is a slider reference
]
# ----------------------------------------------------------------------
[docs] def soa(self, *args) -> None:
"""Stimulus onset asynchronously.
This is a pipeline method, that explains the unused `*args` arguments.
"""
if element := getattr(self, 'cue_placeholder', None):
element.html = ''
# ----------------------------------------------------------------------
[docs] def trial(self, cue: Literal['Right', 'Left', 'Up', 'Bottom']) -> None:
"""Cue visualization.
This is a pipeline method, that means it receives the respective trial
as argument each time is called.
"""
if not hasattr(self, 'cue_placeholder'):
self.cue_placeholder = html.SPAN('', id='cue')
self.stimuli_area <= self.cue_placeholder
self.send_marker(cue)
self.cue_placeholder.html = UNICODE_CUES[cue]
self.cue_placeholder.style = {'display': 'flex'}
# ----------------------------------------------------------------------
[docs] def synchronizer(self, value: bool) -> None:
"""Show or hide synchronizer."""
if value:
self.show_synchronizer()
else:
self.hide_synchronizer()
if __name__ == '__main__':
FourClassMotorImagery()