Source code for CPAC.utils.extract_parameters

# Copyright (C) 2013-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/>.
[docs] def merge(output_dir, scan_name, threshold, motion_f, power_f, flag): """Merge power parameters and motion parameters file.""" import os import re if threshold is None: filename = scan_name + "_all_params.csv" filename = filename.lstrip("_") outfile = os.path.join(output_dir, filename) else: filename = scan_name + threshold + "_all_params.csv" filename = filename.lstrip("_") outfile = os.path.join(output_dir, filename) float(re.sub(r"[a-zA-Z_]", "", threshold)) # Read in the motion and power parameters files try: motion = open(motion_f, "r").readlines() except Exception as e: err_string = ( "\n\n[!] CPAC says: Could not read the motion " "parameters file.\n\nFilepath: %s\n\nError details: %s" "\n\n" % (motion_f, e) ) raise Exception(err_string) try: power = open(power_f, "r").readlines() except Exception as e: err_string = ( "\n\n[!] CPAC says: Could not read the power " "parameters file.\n\nFilepath: %s\n\nError details: %s" "\n\n" % (power_f, e) ) raise Exception(err_string) # Write the combined motion and power parameters CSV file try: if flag: f = open(outfile, "w") m = motion[0].strip("\n") p = ",".join(power[0].split(",")[1:]) f.write(m + p) else: f = open(outfile, "a") m = motion[1] p = ",".join(power[1].split(",")[2:]) f.write(m + p + "\n") f.close() except Exception as e: err_string = ( "\n\n[!] CPAC says: Could not create or open the motion " "and power parameters CSV file. Ensure you have write " "permissions for the directory it is writing to.\n\n" "Attempted write path: %s\n\nError details: %s\n\n" % (outfile, e) ) raise Exception(err_string)
[docs] def grab(output_dir, scrubbing): """ Method to grab all the motion parameters and power parameters file from each subject for each pipeline and merge them. Parameters ---------- output_dir : string Path to the datasink output directory of CPAC """ import glob import os import re pipelines = glob.glob(os.path.join(output_dir, "pipeline*")) for p in pipelines: scan_list = [] threshold_list = [] pattern1 = re.compile(r"(\w)*scan(\w)*(\d)*(\w)*[/]") pattern2 = re.compile(r"(\w)*threshold_[-+]?([0-9]*\.[0-9]+|[0-9]+)") scans = glob.glob(os.path.join(p, "*/power_params/*/*")) # get the unique scans and threshold value for s in scans: val = re.search(pattern1, s) if val: scan_list.append(val.group(0).rstrip("/")) val = re.search(pattern2, s) if val: threshold_list.append(val.group(0)) scan_list = set(scan_list) threshold_list = set(threshold_list) for scan in scan_list: for threshold in threshold_list: Flag = 1 # merge files for each subject for sub in os.listdir(p): sub = os.path.join(p, sub) motion_file = os.path.join( sub, "motion_params", scan, "motion_parameters.txt" ) power_file = os.path.join( sub, "power_params", scan, threshold, "pow_params.txt" ) if os.path.exists(motion_file) and os.path.exists(power_file): merge(p, scan, threshold, motion_file, power_file, Flag) Flag = 0 if 0 in scrubbing: for sub in os.listdir(p): sub = os.path.join(p, sub) motion_file = os.path.join( sub, "motion_params", scan, "motion_parameters.txt" ) power_file = os.path.join( sub, "power_params", scan, "pow_params.txt" ) if os.path.exists(motion_file) and os.path.exists(power_file): threshold = None merge(p, scan, threshold, motion_file, power_file, Flag) Flag = 0 return threshold
[docs] def run(output_path, scrubbing): return grab(output_path, scrubbing)
if __name__ == "__main__": import sys if len(sys.argv) == 2: grab(sys.argv[1], [0]) else: print("Usage: python extract_parameters.py /path/to/output/dir") # noqa: T201