# Copyright (C) 2019-2024 C-PAC Developers# This file is part of C-PAC.# C-PAC is free software: you can redistribute it and/or modify it under# the terms of the GNU Lesser General Public License as published by the# Free Software Foundation, either version 3 of the License, or (at your# option) any later version.# C-PAC is distributed in the hope that it will be useful, but WITHOUT# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public# License for more details.# You should have received a copy of the GNU Lesser General Public# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.fromCPAC.utils.monitoringimportWFLOGGER
[docs]defsetup_test_wf(s3_prefix,paths_list,test_name,workdirs_to_keep=None):"""Set up a basic template Nipype workflow for testing single nodes or small sub-workflows. """importosimportshutilfromCPAC.pipelineimportnipype_pipeline_engineaspefromCPAC.utils.datasourceimportcheck_for_s3fromCPAC.utils.interfaces.datasinkimportDataSinktest_dir=os.path.join(os.getcwd(),test_name)work_dir=os.path.join(test_dir,"workdir")out_dir=os.path.join(test_dir,"output")ifos.path.exists(out_dir):try:shutil.rmtree(out_dir)except:passifos.path.exists(work_dir):fordirnameinos.listdir(work_dir):ifworkdirs_to_keep:forkeepdirinworkdirs_to_keep:WFLOGGER.info("%s --- %s\n",dirname,keepdir)ifkeepdirindirname:continuetry:shutil.rmtree(os.path.join(work_dir,dirname))except:passlocal_paths={}forsubpathinpaths_list:s3_path=os.path.join(s3_prefix,subpath)local_path=check_for_s3(s3_path,dl_dir=test_dir)local_paths[subpath]=local_pathwf=pe.Workflow(name=test_name)wf.base_dir=os.path.join(work_dir)wf.config["execution"]={"hash_method":"timestamp","crashdump_dir":os.path.abspath(test_dir),}ds=pe.Node(DataSink(),name=f"sinker_{test_name}")ds.inputs.base_directory=out_dirds.inputs.parameterization=Truereturn(wf,ds,local_paths)