Source code for cmp.stages.eeg.preprocessing

# Copyright (C) 2009-2022, Ecole Polytechnique Federale de Lausanne (EPFL) and
# Hospital Center and University of Lausanne (UNIL-CHUV), Switzerland, and CMP3 contributors
# All rights reserved.
#
#  This software is distributed under the open-source license Modified BSD.

"""Definition of config and stage classes for computing brain parcellation."""

# General imports
import os
from traits.api import (
    HasTraits, Enum, Instance, Float, Str
)

# Nipype imports
import nipype.pipeline.engine as pe

# Own imports
from cmp.stages.common import Stage
from cmtklib.bids.io import (
    CustomEEGPreprocBIDSFile, CustomEEGEventsBIDSFile,
    CustomEEGElectrodesBIDSFile, CustomEEGCartoolElectrodesBIDSFile,
    __cmp_directory__
)
from cmtklib.interfaces.mne import EEGLAB2fif


[docs]class EEGPreprocessingConfig(HasTraits): """Class used to store configuration parameters of a :class:`~cmp.stages.eeg.preparer.EEGPreprocessingStage` instance. Attributes ---------- task_label : Str Task label (e.g. `_task-<label>_`) eeg_ts_file : CustomEEGPreprocBIDSFile Instance of :obj:`~cmtklib.bids.io.CustomEEGPreprocBIDSFile` that describes the input BIDS-formatted preprocessed EEG file events_file : CustomEEGEventsBIDSFile Instance of :obj:`~cmtklib.bids.io.CustomEEGEventsBIDSFile` that describes the input BIDS-formatted EEG events file electrodes_file_fmt : Enum(["BIDS", "Cartool"]) Select the type of tabular file describing electrode positions bids_electrodes_file : CustomEEGElectrodesBIDSFile Instance of :obj:`~cmtklib.bids.io.CustomEEGElectrodesBIDSFile` that describes the input BIDS-compliant EEG electrode file cartool_electrodes_file : CustomEEGCartoolElectrodesBIDSFile Instance of :obj:`~cmtklib.bids.io.CustomEEGCartoolElectrodesBIDSFile` that describes the input BIDS-formatted EEG electrode file created by Cartool t_min : Float Start time of the epochs in seconds, relative to the time-locked event (Default: -0.2) t_max : Float End time of the epochs in seconds, relative to the time-locked event (Default: 0.5) See Also -------- cmp.stages.eeg.preparer.EEGPreprocessingStage """ task_label = Str("Undefined", desc="Task label (e.g. _task-<label>_)") eeg_ts_file = Instance( CustomEEGPreprocBIDSFile, (), desc="Instance of :obj:`~cmtklib.bids.io.CustomEEGPreprocBIDSFile`" "that describes the input BIDS-formatted preprocessed EEG file" ) events_file = Instance( CustomEEGEventsBIDSFile, (), desc="Instance of :obj:`~cmtklib.bids.io.CustomEEGEventsBIDSFile`" "that describes the input BIDS-formatted EEG events file" ) electrodes_file_fmt = Enum( "BIDS", "Cartool", desc="Select the type of tabular file describing electrode positions" ) bids_electrodes_file = Instance( CustomEEGElectrodesBIDSFile, (), desc="Instance of :obj:`~cmtklib.bids.io.CustomEEGElectrodesBIDSFile`" "that describes the input BIDS-compliant EEG electrode file" ) cartool_electrodes_file = Instance( CustomEEGCartoolElectrodesBIDSFile, (), desc="Instance of :obj:`~cmtklib.bids.io.CustomEEGCartoolElectrodesBIDSFile`" "that describes the input BIDS-formatted EEG electrode file created by Cartool" ) t_min = Float(-0.2, desc="Start time of the epochs in seconds, relative to the time-locked event.") t_max = Float(0.5, desc="End time of the epochs in seconds, relative to the time-locked event.") def _task_label_changed(self, new): self.eeg_ts_file.task = new self.events_file.task = new self.bids_electrodes_file.task = new def __str__(self): str_repr = '\tEEGPreprocessingConfig:\n' str_repr += f'\t\t* task_label: {self.task_label}\n' str_repr += f'\t\t* eeg_ts_file: {self.eeg_ts_file}\n' str_repr += f'\t\t* events_file: {self.events_file}\n' str_repr += f'\t\t* electrodes_file_fmt: {self.electrodes_file_fmt}\n' str_repr += f'\t\t* cartool_electrodes_file: {self.cartool_electrodes_file}\n' str_repr += f'\t\t* bids_electrodes_file: {self.bids_electrodes_file}\n' str_repr += f'\t\t* t_min: {self.t_min}\n' str_repr += f'\t\t* t_max: {self.t_max}\n' return str_repr
[docs]class EEGPreprocessingStage(Stage): """Class that represents the preprocessing stage of a :class:`~cmp.pipelines.functional.eeg.EEGPipeline`. This stage consists of converting EEGLab `.set` EEG files to MNE Epochs in `.fif` format, the format used in the rest of the pipeline by calling, if necessary the following interface: - :class:`~cmtklib.interfaces.mne.EEGLAB2fif`: Reads eeglab data and converts them to MNE format (`.fif` file extension). See Also -------- cmp.pipelines.functional.eeg.EEGPipeline cmp.stages.eeg.preparer.EEGPreprocessingConfig """ def __init__(self, subject, session, bids_dir, output_dir): """Constructor of a :class:`~cmp.stages.eeg.prerocessing.EEGPreprocessingStage` instance.""" self.name = 'eeg_preprocessing_stage' self.bids_subject_label = subject self.bids_session_label = session self.bids_dir = bids_dir self.output_dir = output_dir self.config = EEGPreprocessingConfig() self.inputs = [ "eeg_ts_file", "events_file", "electrodes_file", ] self.outputs = [ "epochs_file", ]
[docs] def create_workflow(self, flow, inputnode, outputnode): """Create the stage workflow. Parameters ---------- flow : nipype.pipeline.engine.Workflow The nipype.pipeline.engine.Workflow instance of the EEG pipeline inputnode : nipype.interfaces.utility.IdentityInterface Identity interface describing the inputs of the stage outputnode : nipype.interfaces.utility.IdentityInterface Identity interface describing the outputs of the stage """ if self.config.eeg_ts_file.extension == "set": events_derivatives_dir = self.config.events_file.get_toolbox_derivatives_dir() eeglab2fif_node = pe.Node( interface=EEGLAB2fif( out_epochs_fif_fname='epo.fif', event_ids=self.config.events_file.extract_event_ids_from_json_sidecar( base_dir=(self.bids_dir if events_derivatives_dir == "" else os.path.join(self.bids_dir, 'derivatives', events_derivatives_dir)), subject=self.bids_subject_label, session=self.bids_session_label ), t_min=self.config.t_min, t_max=self.config.t_max ), name="eeglab2fif" ) # fmt: off flow.connect( [ (inputnode, eeglab2fif_node, [('eeg_ts_file', 'eeg_ts_file'), ('events_file', 'events_file'), ('electrodes_file', 'electrodes_file')]), (eeglab2fif_node, outputnode, [('epochs_file', 'epochs_file')]) ] ) # fmt: on else: # Already in fif format # fmt: off flow.connect( [ (inputnode, outputnode, [('eeg_ts_file', 'epochs_file')]) ] )
# fmt: on
[docs] def define_inspect_outputs(self): """Update the `inspect_outputs` class attribute. It contains a dictionary of stage outputs with corresponding commands for visual inspection. """ self.inspect_outputs_dict = {} subject_derivatives_dir = os.path.join( self.output_dir, __cmp_directory__, self.bids_subject_label ) if self.bids_session_label and self.bids_session_label != "": subject_derivatives_dir = os.path.join( subject_derivatives_dir, self.bids_session_label ) subject_part = (f'{self.bids_subject_label}_{self.bids_session_label}' if self.bids_session_label and self.bids_session_label != "" else f'{self.bids_subject_label}') epo_fname = f'{subject_part}_task-{self.config.task_label}_epo.fif' epo_file = os.path.join(subject_derivatives_dir, "eeg", epo_fname) if os.path.exists(epo_file): self.inspect_outputs_dict[f"Epochs * Electrodes time series (Task:{self.config.task_label})"] = [ "visualize_eeg_pipeline_outputs", "--epo_file", epo_file ] if not self.inspect_outputs_dict: self.inspect_outputs = ["Outputs not available"] else: self.inspect_outputs = sorted( [key for key in list(self.inspect_outputs_dict.keys())], key=str.lower )