15466399122

Committed 05 Jun 2025 11:57AM UTC coverage: 35.744% (-0.2%) from 35.977%

Build # 15466399122

Build Type

Pull #19

github

Committed by

web-flow

Commit Message

Merge 2f8a6cef3 into 2f230a9ce

Pull Request Pull Request #19: Automatic brain mask computation in quasi raw + bug fixed for software version

Run Details

13 of 54 new or added lines in 3 files covered. (24.07%)

3 existing lines in 1 file now uncovered.

692 of 1936 relevant lines covered (35.74%)

0.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

47.62

/brainprep/utils.py

# -*- coding: utf-8 -*-
##########################################################################
# NSAp - Copyright (C) CEA, 2021 - 2022
# Distributed under the terms of the CeCILL-B license, as published by
# the CEA-CNRS-INRIA. Refer to the LICENSE file or to
# http://www.cecill.info/licences/Licence_CeCILL-B_V1-en.html
# for details.
##########################################################################

"""
Module that contains some utility functions.
"""

# Imports
import os
import re
import sys
import gzip
import subprocess
import numpy as np
import pandas as pd
import nibabel
from .color_utils import print_command, print_error


def execute_command(command):
    """ Execute a command.

    Parameters
    ----------
    command: list of str
        the command to be executed.
    """
    print_command(" ".join(command))
    proc = subprocess.Popen(
        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    output, error = proc.communicate()
    if proc.returncode != 0:
        raise ValueError(
            "\nCommand {0} failed:\n\n- output:\n{1}\n\n- error: "
            "{2}\n\n".format(" ".join(command),
                             output.decode("utf8"),
                             error.decode("utf8")))


def check_command(command):
    """ Check if a command is installed.

    .. note:: This function is based on which linux command.

    Parameters
    ----------
    command: str
        the name of the command to locate.
    """
    if sys.platform != "linux":
        raise ValueError("This code works only on a linux machine.")
    process = subprocess.Popen(
        ["which", command], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = process.communicate()
    stdout = stdout.decode("utf8")
    stderr = stderr.decode("utf8")
    exitcode = process.returncode
    if exitcode != 0:
        print_error("Command {0}: {1}".format(command, stderr))
        raise ValueError("Impossible to locate command '{0}'.".format(command))


def check_version(package_name, check_pkg_version):
    """ Check installed version of a package.

    .. note:: This function is based on dpkg linux command.

    Parameters
    ----------
    package_name: str
        the name of the package we want to check the version.
    """
    process = subprocess.Popen(
        ["dpkg", "-s", package_name],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdout, stderr = process.communicate()
    stdout = stdout.decode("utf8")
    stderr = stderr.decode("utf8")
    exitcode = process.returncode
    if check_pkg_version:
        # local computer installation
        if exitcode != 0:
            version = None
            print_error("Version {0}: {1}".format(package_name, stderr))
            raise ValueError(
                "Impossible to check package '{0}' version."
                .format(package_name))
        else:
            versions = re.findall("Version: .*$", stdout, re.MULTILINE)
            version = "|".join(versions)
    else:
        # specific installation
        version = "custom install (no check)."
    print("{0} - {1}".format(package_name, version))


def write_matlabbatch(template, nii_files, tpm_file, darteltpm_file,
                      session, batch_file, outdir, model_long=1):
    """ Complete matlab batch from template and unzip T1w file in the outdir.
        Create the outdir.

    Parameters
    ----------
    template: str
        path to template batch to be completed.
    nii_files: list
        the Nifti images to be processed.
    tpm_file: str
        path to the SPM TPM file.
    darteltpm_file: str
        path to the CAT12 tempalte file.
    batch_file: str
        Filepath to the matlabbatch
    outdir: str
        the destination folder for cat12vbm outputs.
    session: str
        the session names, usefull for longitudinal preprocessings.
        Warning session and nii files must be in the same order.
    model_long: int
        longitudinal model choice, default 1.
        1 short time (weeks), 2 long time (years) between images sessions.
    """
    nii_files_str = ""
    if session:
        outdir = [os.path.join(outdir, ses) for ses in session]
    if not isinstance(outdir, list):
        outdir = [outdir]
    for idx, path in enumerate(nii_files):
        nii_files_str += "'{0}' \n".format(
            ungzip_file(path, outdir=outdir[idx]))
    with open(template, "r") as of:
        stream = of.read()
        stream = stream.format(model_long=model_long, anat_file=nii_files_str,
                               tpm_file=tpm_file,
                               darteltpm_file=darteltpm_file)
    with open(batch_file, "w") as of:
        of.write(stream)


def ungzip_file(zfile, prefix="u", outdir=None):
    """ Copy and ungzip the input file.

    Parameters
    ----------
    zfile: str
        input file to ungzip.
    prefix: str, default 'u'
        the prefix of the result file.
    outdir: str, default None)
        the output directory where ungzip file is saved. If not set use the
        input image directory.

    Returns
    -------
    unzfile: str
        the ungzip file.
    """
    # Checks
    if not os.path.isfile(zfile):
        raise ValueError("'{0}' is not a valid filename.".format(zfile))
    if outdir is not None:
        if not os.path.isdir(outdir):
            raise ValueError("'{0}' is not a valid directory.".format(outdir))
    else:
        outdir = os.path.dirname(zfile)

    # Get the file descriptors
    base, extension = os.path.splitext(zfile)
    basename = os.path.basename(base)

    # Ungzip only known extension
    if extension in [".gz"]:
        basename = prefix + basename
        unzfile = os.path.join(outdir, basename)
        with gzip.open(zfile, "rb") as gzfobj:
            data = gzfobj.read()
        with open(unzfile, "wb") as openfile:
            openfile.write(data)

    # Default, unknown compression extension: the input file is returned
    else:
        unzfile = zfile

    return unzfile


def get_bids_keys(filename):
    """ Extract BIDS 'participant_id', 'session' and 'run' keys from a
    filename.

    Parameters
    ----------
    filename: str
        a bids path.

    Returns
    -------
    keys: dict
        the retrieved BIDS keys/values, no key if no match. Add the file name
        in 'ni_path'.
    """
    keys = {}
    participant_re = re.compile("sub-([^_/]+)")
    session_re = re.compile("ses-([^_/]+)")
    run_re = re.compile("run-([a-zA-Z0-9]+)")
    for name, regex in (("participant_id", participant_re),
                        ("session", session_re),
                        ("run", run_re)):
        match = regex.findall(filename)
        if len(set(match)) != 1:
            if name == "participant_id":
                raise ValueError(
                    "Found several or no '{}' in path '{}'.".format(
                        name, filename))
        else:
            keys[name] = match[0]
    if "run" not in keys:
        keys["run"] = "1"
    if "session" not in keys:
        keys["session"] = "V1"
    keys["ni_path"] = filename
    return keys


def load_images(img_files, check_same_referential=True):
    """ Load a list of images in a BIDS organisation: check that all images
    are in the same referential.

    Parameters
    ----------
    img_files: list of str (n_subjects, )
        path to images.

    Returns
    -------
    imgs_arr: array (n_subjects, 1, image_axis0, image_axis1, ...)
        the generated array.
    df: pandas DataFrame
        description of the array with columns 'participant_id',
        'session', 'run', 'ni_path'.
    """
    ref_affine = None
    ref_shape = None
    data = []
    info = {}
    for path in img_files:
        keys = get_bids_keys(path)
        participant_id = keys["participant_id"]
        session = keys.get("session", "V1")
        run = keys.get("run", "1")
        img = nibabel.load(path)
        if ref_affine is None:
            ref_affine = img.affine
            ref_shape = img.shape
        else:
            assert np.allclose(ref_affine, img.affine), "Different affine."
            assert ref_shape == img.shape, "Different shape."
        data.append(np.expand_dims(img.get_fdata(), axis=0))
        info.setdefault("participant_id", []).append(participant_id)
        info.setdefault("session", []).append(session)
        info.setdefault("run", []).append(run)
        info.setdefault("ni_path", []).append(path)
    df = pd.DataFrame.from_dict(info)
    imgs_arr = np.asarray(data)
    return imgs_arr, df


def create_clickable(path_or_url):
    """ Foramt a path or a URL as a HTML href.

    Parameters
    ----------
    path_or_url: str
        a path or a URL.

    Returns
    -------
    url: str
        a href formated URL.
    """
    url = "<a href='{}' target='_blank'>&plus;</a>".format(path_or_url)
    return url


def listify(obj):
    """ Function to transform a coma separated string to a list of string.

    Parameters
    ----------
    obj: list or string
        the input data.

    Returns
    -------
    list: list
        the list of input data or input data.
    """
    if not isinstance(obj, list):
        return obj.split(",")
    else:
        return obj


def cp_file(src, dst):
    """ Copy a file from src to dst.

    Parameters
    ----------
    src: str
        the source file.
    dst: str
        the destination file.
    """
    if not os.path.isfile(src):
        raise ValueError("Source file '{}' does not exist.".format(src))
    if not os.path.isdir(os.path.dirname(dst)):
        raise ValueError("Destination directory '{}' does not exist."
                         .format(os.path.dirname(dst)))
    execute_command(["cp", src, dst])

1	# -- coding: utf-8 --
2	##########################################################################
3	# NSAp - Copyright (C) CEA, 2021 - 2022
4	# Distributed under the terms of the CeCILL-B license, as published by
5	# the CEA-CNRS-INRIA. Refer to the LICENSE file or to
6	# http://www.cecill.info/licences/Licence_CeCILL-B_V1-en.html
7	# for details.
8	##########################################################################
9
10	"""
11	Module that contains some utility functions.
12	"""
13
14	# Imports
15	import os	1✔
16	import re	1✔
17	import sys	1✔
18	import gzip	1✔
19	import subprocess	1✔
20	import numpy as np	1✔
21	import pandas as pd	1✔
22	import nibabel	1✔
23	from .color_utils import print_command, print_error	1✔
24
25
26	def execute_command(command):	1✔
27	""" Execute a command.
28
29	Parameters
30	----------
31	command: list of str
32	the command to be executed.
33	"""
34	print_command(" ".join(command))	1✔
35	proc = subprocess.Popen(	1✔
36	command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
37	output, error = proc.communicate()	1✔
38	if proc.returncode != 0:	1✔
39	raise ValueError(	×
40	"\nCommand {0} failed:\n\n- output:\n{1}\n\n- error: "
41	"{2}\n\n".format(" ".join(command),
42	output.decode("utf8"),
43	error.decode("utf8")))
44
45
46	def check_command(command):	1✔
47	""" Check if a command is installed.
48
49	.. note:: This function is based on which linux command.
50
51	Parameters
52	----------
53	command: str
54	the name of the command to locate.
55	"""
56	if sys.platform != "linux":	1✔
57	raise ValueError("This code works only on a linux machine.")	×
58	process = subprocess.Popen(	1✔
59	["which", command], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
60	stdout, stderr = process.communicate()	1✔
61	stdout = stdout.decode("utf8")	1✔
62	stderr = stderr.decode("utf8")	1✔
63	exitcode = process.returncode	1✔
64	if exitcode != 0:	1✔
65	print_error("Command {0}: {1}".format(command, stderr))	×
66	raise ValueError("Impossible to locate command '{0}'.".format(command))	×
67
68
69	def check_version(package_name, check_pkg_version):	1✔
70	""" Check installed version of a package.
71
72	.. note:: This function is based on dpkg linux command.
73
74	Parameters
75	----------
76	package_name: str
77	the name of the package we want to check the version.
78	"""
79	process = subprocess.Popen(	1✔
80	["dpkg", "-s", package_name],
81	stdout=subprocess.PIPE, stderr=subprocess.PIPE)
82	stdout, stderr = process.communicate()	1✔
83	stdout = stdout.decode("utf8")	1✔
84	stderr = stderr.decode("utf8")	1✔
85	exitcode = process.returncode	1✔
86	if check_pkg_version:	1✔
87	# local computer installation
88	if exitcode != 0:	1✔
89	version = None	×
90	print_error("Version {0}: {1}".format(package_name, stderr))	×
91	raise ValueError(	×
92	"Impossible to check package '{0}' version."
93	.format(package_name))
94	else:
95	versions = re.findall("Version: .*$", stdout, re.MULTILINE)	1✔
96	version = "\|".join(versions)	1✔
97	else:
98	# specific installation
99	version = "custom install (no check)."	1✔
100	print("{0} - {1}".format(package_name, version))	1✔
101
102
103	def write_matlabbatch(template, nii_files, tpm_file, darteltpm_file,	1✔
104	session, batch_file, outdir, model_long=1):
105	""" Complete matlab batch from template and unzip T1w file in the outdir.
106	Create the outdir.
107
108	Parameters
109	----------
110	template: str
111	path to template batch to be completed.
112	nii_files: list
113	the Nifti images to be processed.
114	tpm_file: str
115	path to the SPM TPM file.
116	darteltpm_file: str
117	path to the CAT12 tempalte file.
118	batch_file: str
119	Filepath to the matlabbatch
120	outdir: str
121	the destination folder for cat12vbm outputs.
122	session: str
123	the session names, usefull for longitudinal preprocessings.
124	Warning session and nii files must be in the same order.
125	model_long: int
126	longitudinal model choice, default 1.
127	1 short time (weeks), 2 long time (years) between images sessions.
128	"""
129	nii_files_str = ""	1✔
130	if session:	1✔
131	outdir = [os.path.join(outdir, ses) for ses in session]	1✔
132	if not isinstance(outdir, list):	1✔
133	outdir = [outdir]	×
134	for idx, path in enumerate(nii_files):	1✔
135	nii_files_str += "'{0}' \n".format(	1✔
136	ungzip_file(path, outdir=outdir[idx]))
137	with open(template, "r") as of:	1✔
138	stream = of.read()	1✔
139	stream = stream.format(model_long=model_long, anat_file=nii_files_str,	1✔
140	tpm_file=tpm_file,
141	darteltpm_file=darteltpm_file)
142	with open(batch_file, "w") as of:	1✔
143	of.write(stream)	1✔
144
145
146	def ungzip_file(zfile, prefix="u", outdir=None):	1✔
147	""" Copy and ungzip the input file.
148
149	Parameters
150	----------
151	zfile: str
152	input file to ungzip.
153	prefix: str, default 'u'
154	the prefix of the result file.
155	outdir: str, default None)
156	the output directory where ungzip file is saved. If not set use the
157	input image directory.
158
159	Returns
160	-------
161	unzfile: str
162	the ungzip file.
163	"""
164	# Checks
165	if not os.path.isfile(zfile):	1✔
166	raise ValueError("'{0}' is not a valid filename.".format(zfile))	×
167	if outdir is not None:	1✔
168	if not os.path.isdir(outdir):	1✔
169	raise ValueError("'{0}' is not a valid directory.".format(outdir))	×
170	else:
171	outdir = os.path.dirname(zfile)	×
172
173	# Get the file descriptors
174	base, extension = os.path.splitext(zfile)	1✔
175	basename = os.path.basename(base)	1✔
176
177	# Ungzip only known extension
178	if extension in [".gz"]:	1✔
179	basename = prefix + basename	×
180	unzfile = os.path.join(outdir, basename)	×
181	with gzip.open(zfile, "rb") as gzfobj:	×
182	data = gzfobj.read()	×
183	with open(unzfile, "wb") as openfile:	×
184	openfile.write(data)	×
185
186	# Default, unknown compression extension: the input file is returned
187	else:
188	unzfile = zfile	1✔
189
190	return unzfile	1✔
191
192
193	def get_bids_keys(filename):	1✔
194	""" Extract BIDS 'participant_id', 'session' and 'run' keys from a
195	filename.
196
197	Parameters
198	----------
199	filename: str
200	a bids path.
201
202	Returns
203	-------
204	keys: dict
205	the retrieved BIDS keys/values, no key if no match. Add the file name
206	in 'ni_path'.
207	"""
208	keys = {}	×
209	participant_re = re.compile("sub-([^_/]+)")	×
210	session_re = re.compile("ses-([^_/]+)")	×
211	run_re = re.compile("run-([a-zA-Z0-9]+)")	×
212	for name, regex in (("participant_id", participant_re),	×
213	("session", session_re),
214	("run", run_re)):
215	match = regex.findall(filename)	×
216	if len(set(match)) != 1:	×
217	if name == "participant_id":	×
218	raise ValueError(	×
219	"Found several or no '{}' in path '{}'.".format(
220	name, filename))
221	else:
222	keys[name] = match[0]	×
223	if "run" not in keys:	×
224	keys["run"] = "1"	×
225	if "session" not in keys:	×
226	keys["session"] = "V1"	×
227	keys["ni_path"] = filename	×
228	return keys	×
229
230
231	def load_images(img_files, check_same_referential=True):	1✔
232	""" Load a list of images in a BIDS organisation: check that all images
233	are in the same referential.
234
235	Parameters
236	----------
237	img_files: list of str (n_subjects, )
238	path to images.
239
240	Returns
241	-------
242	imgs_arr: array (n_subjects, 1, image_axis0, image_axis1, ...)
243	the generated array.
244	df: pandas DataFrame
245	description of the array with columns 'participant_id',
246	'session', 'run', 'ni_path'.
247	"""
248	ref_affine = None	×
249	ref_shape = None	×
250	data = []	×
251	info = {}	×
252	for path in img_files:	×
253	keys = get_bids_keys(path)	×
254	participant_id = keys["participant_id"]	×
255	session = keys.get("session", "V1")	×
256	run = keys.get("run", "1")	×
257	img = nibabel.load(path)	×
258	if ref_affine is None:	×
259	ref_affine = img.affine	×
260	ref_shape = img.shape	×
261	else:
262	assert np.allclose(ref_affine, img.affine), "Different affine."	×
263	assert ref_shape == img.shape, "Different shape."	×
264	data.append(np.expand_dims(img.get_fdata(), axis=0))	×
265	info.setdefault("participant_id", []).append(participant_id)	×
266	info.setdefault("session", []).append(session)	×
267	info.setdefault("run", []).append(run)	×
268	info.setdefault("ni_path", []).append(path)	×
269	df = pd.DataFrame.from_dict(info)	×
270	imgs_arr = np.asarray(data)	×
271	return imgs_arr, df	×
272
273
274	def create_clickable(path_or_url):	1✔
275	""" Foramt a path or a URL as a HTML href.
276
277	Parameters
278	----------
279	path_or_url: str
280	a path or a URL.
281
282	Returns
283	-------
284	url: str
285	a href formated URL.
286	"""
287	url = "<a href='{}' target='_blank'>+</a>".format(path_or_url)	×
288	return url	×
289
290
291	def listify(obj):	1✔
292	""" Function to transform a coma separated string to a list of string.
293
294	Parameters
295	----------
296	obj: list or string
297	the input data.
298
299	Returns
300	-------
301	list: list
302	the list of input data or input data.
303	"""
304	if not isinstance(obj, list):	×
305	return obj.split(",")	×
306	else:
307	return obj	×
308
309
310	def cp_file(src, dst):	1✔
311	""" Copy a file from src to dst.
312
313	Parameters
314	----------
315	src: str
316	the source file.
317	dst: str
318	the destination file.
319	"""
NEW 320	if not os.path.isfile(src):	×
NEW 321	raise ValueError("Source file '{}' does not exist.".format(src))	×
NEW 322	if not os.path.isdir(os.path.dirname(dst)):	×
NEW 323	raise ValueError("Destination directory '{}' does not exist."	×
324	.format(os.path.dirname(dst)))
NEW 325	execute_command(["cp", src, dst])	×

neurospin-deepinsight / brainprep / 15466399122

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous