Source code for bci_framework.default_extensions.Resting_state_and_noise_acquisition.main

"""
===================================
Resting state and noise acquisition
===================================

"""

from bci_framework.extensions.stimuli_delivery import StimuliAPI
from bci_framework.extensions.stimuli_delivery.utils import Widgets as w
from bci_framework.extensions.stimuli_delivery.utils import Tone as t
from bci_framework.extensions.stimuli_delivery.utils import keypress

from browser import document, timer, html

TASKS = {

    'resting_open': ['Resting with open eyes',
                     '<p>Resting with <b>open</b> eyes</p>',
                     '<p>Press the button and center your gace on the fixation \
                     cross, the acquisition begins with the single beep, \
                     <b>keep this position until listen two beeps</b>.</p>'],

    'resting_close': ['Resting with close eyes',
                      '<p>Resting with <b>close</b> eyes</p>',
                      '<p>Press the button and close the eyes, the acquisition \
                      begins with the single beep, <b>keep eyes closed until \
                      listen two beeps</b>.</p>'],

    'eye_blinking': ['Eye blinking',
                     '<p>Eyes <b>blinking</b></p>',
                     '<p>Press the button and starts to blink periodically, \
                     the acquisition begins with the single beep, <b>keep this \
                     action until listen two beeps</b>.</p>'],

    'eyeball_up_down': ['Eyeball movement up/down',
                        '<p>Eyeball movement <b>Up</b> and <b>Down</b></p>',
                        '<p>Press the button and move your eyes up and down \
                        periodically, the acquisition begins with the single \
                        beep, <b>keep this action until listen two beeps</b>.</p>'],

    'eyeball_left_right': ['Eyeball movement left/right',
                           '<p>Eyeball movement <b>Left</b> and <b>Right</b></p>',
                           '<p>Press the button and move your eyes right and \
                           left periodically, the acquisition begins with the \
                           single beep, <b>keep this action until listen two \
                           beeps</b>.</p>'],

    'head_left_right': ['Head movement left/right',
                        '<p>Head movement <b>Left</b> and <b>Right</b></p>',
                        '<p>Press the button and move your head from left to \
                        right and right to left periodically, the acquisition \
                        begins with the single beep, <b>keep this action until \
                        listen two beeps</b>.</p>'],

    'jaw_clenching': ['Jaw clenching',
                      '<p>Jaw <b>clenching</b></p>',
                      '<p>Press the button and clench the jaw periodically, the \
                      acquisition begins with the single beep, <b>keep this \
                      action until listen two beeps</b>.</p>'],

}


########################################################################
[docs]class RestingNoiseAcquisition(StimuliAPI): """Resting state and noise acquisition""" # ---------------------------------------------------------------------- def __init__(self, *args, **kwargs): """""" super().__init__(*args, **kwargs) self.add_stylesheet('styles.css') self.show_cross() self.dashboard <= w.label( 'Resting state and noise acquisition<br><br>', 'headline4') for task_id in TASKS: title = TASKS[task_id][0] self.dashboard <= w.slider( title, min=0, max=5, step=0.1, value=1, unit='minutes', id=task_id) 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) self.button_start = w.button( 'Start (space)', outlined=True, on_click=self.syncrhonous_trial, id='syncrhonous-button') # ----------------------------------------------------------------------
[docs] def start(self) -> None: """Start the run. A run consist in a consecutive pipeline trials execution. """ if w.get_value('record'): self.start_record() self.build_trials()
# ----------------------------------------------------------------------
[docs] def stop(self) -> None: """Stop pipeline execution.""" self.stop_pipeline() 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 = [] self.pipeline_trial = [] for trial in TASKS: if w.get_value(trial): self.trials.append([{'title': TASKS[trial][0], 'label': TASKS[trial][1], 'instruction': TASKS[trial][2], }]) self.pipeline_trial.append([ (self.prepare, 1500), (self.trial, w.get_value(trial) * 60 * 1000), (self.end_trial, 1000), ]) self.trial_instruction(**self.trials[0][0])
# ----------------------------------------------------------------------
[docs] def syncrhonous_trial(self) -> None: """Start a trial after subject decision.""" pipeline_trial = self.pipeline_trial.pop(0) trial = self.trials.pop(0) self.run_pipeline(pipeline_trial, trial, callback=lambda: self.trial_instruction( **self.trials[0][0]))
# ----------------------------------------------------------------------
[docs] def trial_instruction(self, title: str, label: str, instruction: str) -> None: """Show the trial instructions to the subject.""" self.stimuli_area.clear() self.show_cross() self.stimuli_area <= w.label(label, 'headline2', id='hint') self.stimuli_area <= w.label(instruction, 'headline2', id='hint2') self.stimuli_area <= self.button_start keypress(self.handle_response, timeout=None)
# ----------------------------------------------------------------------
[docs] def handle_response(self, response: str) -> None: """Key pressed.""" if response == ' ': self.syncrhonous_trial()
# ----------------------------------------------------------------------
[docs] def prepare(self, title: str, label: str, instruction: str) -> None: """A small pause before acquisition.""" self.stimuli_area.clear() self.show_cross()
# ----------------------------------------------------------------------
[docs] def trial(self, title: str, label: str, instruction: str) -> None: """Full trial.""" self.beep(1)
# ----------------------------------------------------------------------
[docs] def end_trial(self, title, label, instruction) -> None: """End of trial.""" self.beep(2)
# ----------------------------------------------------------------------
[docs] def synchronizer(self, value: bool) -> None: """Show or hide synchronizer.""" if value: self.show_synchronizer() else: self.hide_synchronizer()
# ----------------------------------------------------------------------
[docs] def beep(self, n=1) -> None: """Play `n` beeps.""" note = 'C#6' gain = 1 duration = 100 t(note, duration, gain) if n > 1: for i in range(1, n): timer.set_timeout( lambda: t(note, duration, gain), (duration + 50) * i)
if __name__ == '__main__': RestingNoiseAcquisition()