Source code for cmp.stages.common

# 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 common parent classes for stages."""

# Libraries imports
import os

from traits.api import *

[docs]class Stage(HasTraits): """Parent class that extends `HasTraits` and represents a processing pipeline stage. It is extended by the various pipeline stage subclasses. Attributes ---------- bids_subject_label : traits.Str BIDS subject (participant) label bids_session_label : traits.Str BIDS session label bids_dir : traits.Str BIDS dataset root directory output_dir : traits.Str Output directory inspect_outputs : traits.Dict Dictionary of stage outputs with corresponding commands for visual inspection (Initialization: 'Outputs not available') inspect_outputs_enum : traits.Enum Choice of output to be visually inspected (values='inspect_outputs') enabled : traits.Bool Stage enabled in the pipeline (Default: True) config : Instance(HasTraits) Instance of stage configuration See Also -------- cmp.stages.preprocessing.preprocessing.PreprocessingStage cmp.stages.diffusion.diffusion.DiffusionStage cmp.stages.registration.registration.RegistrationStage cmp.stages.connectome.connectome.ConnectomeStage cmp.stages.preprocessing.fmri_preprocessing.PreprocessingStage cmp.stages.functional.functionalMRI.FunctionalMRIStage cmp.stages.connectome.fmri_connectome.ConnectomeStage """ bids_subject_label = Str(desc="BIDS subject (participant) label") bids_session_label = Str("", desc="BIDS session label") bids_dir = Str output_dir = Str inspect_outputs = ["Outputs not available"] inspect_outputs_enum = Enum(values="inspect_outputs") inspect_outputs_dict = Dict enabled = True config = Instance(HasTraits)
[docs] def is_running(self): """Return the number of unfinished files in the stage. Returns ------- nb_of_unfinished_files : int Number of unfinished files in the stage """ unfinished_files = [ os.path.join(dirpath, f) for dirpath, dirnames, files in os.walk(self.stage_dir) for f in files if f.endswith("_unfinished.json") ] nb_of_unfinished_files = len(unfinished_files) return nb_of_unfinished_files