20846335294

Committed 09 Jan 2026 08:47AM UTC coverage: 79.725%. Remained the same

Build # 20846335294

Build Type

Pull #35

github

Committed by

web-flow

Commit Message

Merge bef43bc34 into 1c98d33f5

Pull Request Pull Request #35: Group level vbm

Coverage Stats

0 of 3 new or added lines in 2 files covered. (0.0%)

11 existing lines in 1 file now uncovered.

1451 of 1820 relevant lines covered (79.73%)

0.8 hits per line

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

51.65

/brainprep/interfaces/utils.py

##########################################################################
# NSAp - Copyright (C) CEA, 2021 - 2025
# 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.
##########################################################################

"""
Tools.
"""


import glob
import gzip
import shutil

import nibabel
import numpy as np
import pandas as pd
from scipy.stats import pearsonr

from ..reporting import log_runtime
from ..typing import (
    Directory,
    File,
)
from ..utils import (
    coerceparams,
    coerce_to_path,
    outputdir,
    parse_bids_keys,
)
from ..wrappers import pywrapper


@coerceparams
@outputdir
@log_runtime(
    bunched=False)
@pywrapper
def mask_diff(
        im1_file: File,
        im2_file: File,
        output_dir: Directory,
        entities: dict,
        threshold: float = 0.6,
        dryrun: bool = False) -> tuple[File]:
    """
    Computes a defacing mask by thresholding the intensity difference between
    two input images.

    Parameters
    ----------
    im1_file : File
        Path to the first image.
    im2_file : File
        Path to the second image.
    output_dir : Directory
        Directory where the defacing mask will be saved.
    entities : dict
        A dictionary of parsed BIDS entities including modality.
    threshold : float
        Threshold for intensity difference to define the mask. Default 0.6.
    dryrun : bool
        If True, skip actual computation and file writing. Default False.

    Returns
    -------
    mask_file : File
        Path to the saved mask image.
    """
    basename = "sub-{sub}_ses-{ses}_run-{run}_mod-T1w_defacemask".format(
        **entities)
    mask_file = output_dir / f"{basename}.nii.gz"

    if not dryrun:
        im1 = nibabel.load(im1_file)
        im2 = nibabel.load(im2_file)
        mask = np.abs(im1.get_fdata() - im2.get_fdata())
        indices = np.where(mask > threshold)
        mask[...] = 0
        mask[indices] = 1
        im_mask = nibabel.Nifti1Image(mask, im1.affine)
        nibabel.save(im_mask, mask_file)

    return (mask_file, )


@coerceparams
@outputdir
@log_runtime(
    bunched=False)
@pywrapper
def copyfiles(
        source_image_files: list[File],
        destination_image_files: list[File],
        output_dir: Directory,
        dryrun: bool = False) -> None:
    """
    Copy input image files.

    Parameters
    ----------
    source_image_files : list[File]
        Path to the image to be copied.
    destination_image_files : list[File]
        Path to the locations where images will be copied.
    output_dir : Directory
        Directory where the images are copied.
    dryrun : bool
        If True, skip actual computation and file writing. Default False.
    """
    if not dryrun:
        for src_path, dest_path in zip(source_image_files,
                                       destination_image_files,
                                       strict=True):
            shutil.copy(src_path, dest_path)


@coerceparams
@outputdir
@log_runtime(
    bunched=False)
@pywrapper
def movedir(
        source_dir: Directory,
        output_dir: Directory,
        content: bool = False,
        dryrun: bool = False) -> tuple[Directory]:
    """
    Move input directory.

    Parameters
    ----------
    source_dir : Directory
        Path to the directory to be moved.
    output_dir : Directory
        Directory where the folder is moved.
    content : bool
        If True, copy only the content of the source directory. Default False.
    dryrun : bool
        If True, skip actual computation and file writing. Default False.

    Returns
    -------
    target_directory : Directory
        Path to the moved directory.

    Raises
    ------
    ValueError
        If `source_dir` is not a directory.
    """
    target_directory = output_dir
    if not content:
        target_directory /= source_dir.name
    if not dryrun:
        if not source_dir.is_dir():
            raise ValueError(
                f"Source '{source_dir}' is not a directory."
            )
        if not content:
            shutil.copytree(source_dir, output_dir)
        else:
            for item in source_dir.iterdir():
                target = output_dir / item.name
                if item.is_dir():
                    shutil.copytree(item, target, dirs_exist_ok=True)
                else:
                    shutil.copy2(item, target)
            if not any(source_dir.iterdir()):
                source_dir.rmdir()
    return (target_directory, )


@outputdir
@log_runtime(
    bunched=False)
@pywrapper
def ungzfile(
        input_file: File,
        output_file: File,
        output_dir: Directory,
        dryrun: bool = False) -> tuple[File]:
    """
    Ungzip input file.

    Parameters
    ----------
    input_file : File
        Path to the file to ungzip.
    output_file : File
        Path to the ungzip file.
    output_dir : Directory
        Directory where the unzip file is created.
    dryrun : bool
        If True, skip actual computation and file writing. Default False.

    Returns
    -------
    output_file : File
        Path to the ungzip file.

    Raises
    ------
    ValueError
        If the input file is not compressed.
    """
    if input_file.suffix != ".gz":
        raise ValueError(
            f"The input file is not compressed: {input_file}"
        )

    if not dryrun:
        with gzip.open(input_file, "rb") as gzfobj:
            output_file.write_bytes(gzfobj.read())

    return (output_file, )


@outputdir
@log_runtime(
    bunched=False)
@pywrapper
def mean_correlation(
        image_files_regex: str,
        atlas_file: File,
        output_dir: Directory,
        correlation_threshold: float = 0.5,
        dryrun: bool = False) -> tuple[File]:
    """
    Compute the mean Pearson correlation between a reference image and a list
    of other images.

    Parameters
    ----------
    image_files_regex : str
        A REGEX to image files, each representing an image of the same shape
        and geometry as `atlas_file`.
    atlas_file : File
        An file representing the reference image.
    output_dir : Directory
        Directory where a TSV file containing the mean correlation values is
        created.
    correlation_threshold : float
        Quality control threshold on the correlation score. Default 0.5.
    dryrun : bool
        If True, skip actual computation and file writing. Default False.

    Returns
    -------
    correlations_file : File
        A TSV file containing the Pearson correlation coefficient between the
        atlas image and each image pointed by the input REGEX.

    Raises
    ------
    ValueError
        If the atlas and an image have incompatible shape or geometry.
    """
    correlations_file = output_dir / "mean_correlations.tsv"
    correlations_file.parent.mkdir(parents=True, exist_ok=True)

    if not dryrun:

        image_files = coerce_to_path(glob.glob(
            str(image_files_regex)),
            expected_type=list[File]
        )
        atlas_im = nibabel.load(atlas_file)
        atlas_arr = atlas_im.get_fdata()

        scores = pd.DataFrame(
            columns=(
                "participant_id",
                "session",
                "run",
                "mean_correlation",
            )
        )
        for path in image_files:
            entities = parse_bids_keys(path)
            im = nibabel.load(path)
            arr = atlas_im.get_fdata()
            if atlas_arr.shape != arr.shape:
                raise ValueError(
                    f"Atlas and image have incompatible shape: {path}"
                )
            if not np.allclose(atlas_im.affine, im.affine):
                raise ValueError(
                    f"Atlas and image have incompatible orientation: {path}"
                )
            corr, _ = pearsonr(
                atlas_arr.flatten(),
                arr.flatten(),
            )
            scores.loc[len(scores)] = [
                entities["sub"],
                entities["ses"],
                entities["run"],
                corr,
            ]

        scores["qc"] = (
            scores["mean_correlation"] > correlation_threshold
        ).astype(int)
        scores.to_csv(
            correlations_file,
            index=False,
            sep="\t",
        )

    return (correlations_file, )

1	##########################################################################
2	# NSAp - Copyright (C) CEA, 2021 - 2025
3	# Distributed under the terms of the CeCILL-B license, as published by
4	# the CEA-CNRS-INRIA. Refer to the LICENSE file or to
5	# http://www.cecill.info/licences/Licence_CeCILL-B_V1-en.html
6	# for details.
7	##########################################################################
8
9	"""
10	Tools.
11	"""
12
13
14	import glob	1✔
15	import gzip	1✔
16	import shutil	1✔
17
18	import nibabel	1✔
19	import numpy as np	1✔
20	import pandas as pd	1✔
21	from scipy.stats import pearsonr	1✔
22
23	from ..reporting import log_runtime	1✔
24	from ..typing import (	1✔
25	Directory,
26	File,
27	)
28	from ..utils import (	1✔
29	coerceparams,
30	coerce_to_path,
31	outputdir,
32	parse_bids_keys,
33	)
34	from ..wrappers import pywrapper	1✔
35
36
37	@coerceparams	1✔
38	@outputdir	1✔
39	@log_runtime(	1✔
40	bunched=False)
41	@pywrapper	1✔
42	def mask_diff(	1✔
43	im1_file: File,
44	im2_file: File,
45	output_dir: Directory,
46	entities: dict,
47	threshold: float = 0.6,
48	dryrun: bool = False) -> tuple[File]:
49	"""
50	Computes a defacing mask by thresholding the intensity difference between
51	two input images.
52
53	Parameters
54	----------
55	im1_file : File
56	Path to the first image.
57	im2_file : File
58	Path to the second image.
59	output_dir : Directory
60	Directory where the defacing mask will be saved.
61	entities : dict
62	A dictionary of parsed BIDS entities including modality.
63	threshold : float
64	Threshold for intensity difference to define the mask. Default 0.6.
65	dryrun : bool
66	If True, skip actual computation and file writing. Default False.
67
68	Returns
69	-------
70	mask_file : File
71	Path to the saved mask image.
72	"""
73	basename = "sub-{sub}_ses-{ses}_run-{run}_mod-T1w_defacemask".format(	×
74	**entities)
75	mask_file = output_dir / f"{basename}.nii.gz"	×
76
77	if not dryrun:	×
78	im1 = nibabel.load(im1_file)	×
79	im2 = nibabel.load(im2_file)	×
80	mask = np.abs(im1.get_fdata() - im2.get_fdata())	×
81	indices = np.where(mask > threshold)	×
82	mask[...] = 0	×
83	mask[indices] = 1	×
84	im_mask = nibabel.Nifti1Image(mask, im1.affine)	×
85	nibabel.save(im_mask, mask_file)	×
86
87	return (mask_file, )	×
88
89
90	@coerceparams	1✔
91	@outputdir	1✔
92	@log_runtime(	1✔
93	bunched=False)
94	@pywrapper	1✔
95	def copyfiles(	1✔
96	source_image_files: list[File],
97	destination_image_files: list[File],
98	output_dir: Directory,
99	dryrun: bool = False) -> None:
100	"""
101	Copy input image files.
102
103	Parameters
104	----------
105	source_image_files : list[File]
106	Path to the image to be copied.
107	destination_image_files : list[File]
108	Path to the locations where images will be copied.
109	output_dir : Directory
110	Directory where the images are copied.
111	dryrun : bool
112	If True, skip actual computation and file writing. Default False.
113	"""
114	if not dryrun:	1✔
115	for src_path, dest_path in zip(source_image_files,	×
116	destination_image_files,
117	strict=True):
118	shutil.copy(src_path, dest_path)	×
119
120
121	@coerceparams	1✔
122	@outputdir	1✔
123	@log_runtime(	1✔
124	bunched=False)
125	@pywrapper	1✔
126	def movedir(	1✔
127	source_dir: Directory,
128	output_dir: Directory,
129	content: bool = False,
130	dryrun: bool = False) -> tuple[Directory]:
131	"""
132	Move input directory.
133
134	Parameters
135	----------
136	source_dir : Directory
137	Path to the directory to be moved.
138	output_dir : Directory
139	Directory where the folder is moved.
140	content : bool
141	If True, copy only the content of the source directory. Default False.
142	dryrun : bool
143	If True, skip actual computation and file writing. Default False.
144
145	Returns
146	-------
147	target_directory : Directory
148	Path to the moved directory.
149
150	Raises
151	------
152	ValueError
153	If `source_dir` is not a directory.
154	"""
155	target_directory = output_dir	1✔
156	if not content:	1✔
157	target_directory /= source_dir.name	1✔
158	if not dryrun:	1✔
159	if not source_dir.is_dir():	×
160	raise ValueError(	×
161	f"Source '{source_dir}' is not a directory."
162	)
163	if not content:	×
164	shutil.copytree(source_dir, output_dir)	×
165	else:
166	for item in source_dir.iterdir():	×
167	target = output_dir / item.name	×
168	if item.is_dir():	×
169	shutil.copytree(item, target, dirs_exist_ok=True)	×
170	else:
171	shutil.copy2(item, target)	×
172	if not any(source_dir.iterdir()):	×
173	source_dir.rmdir()	×
174	return (target_directory, )	1✔
175
176
177	@outputdir	1✔
178	@log_runtime(	1✔
179	bunched=False)
180	@pywrapper	1✔
181	def ungzfile(	1✔
182	input_file: File,
183	output_file: File,
184	output_dir: Directory,
185	dryrun: bool = False) -> tuple[File]:
186	"""
187	Ungzip input file.
188
189	Parameters
190	----------
191	input_file : File
192	Path to the file to ungzip.
193	output_file : File
194	Path to the ungzip file.
195	output_dir : Directory
196	Directory where the unzip file is created.
197	dryrun : bool
198	If True, skip actual computation and file writing. Default False.
199
200	Returns
201	-------
202	output_file : File
203	Path to the ungzip file.
204
205	Raises
206	------
207	ValueError
208	If the input file is not compressed.
209	"""
210	if input_file.suffix != ".gz":	1✔
211	raise ValueError(	×
212	f"The input file is not compressed: {input_file}"
213	)
214
215	if not dryrun:	1✔
216	with gzip.open(input_file, "rb") as gzfobj:	×
217	output_file.write_bytes(gzfobj.read())	×
218
219	return (output_file, )	1✔
220
221
222	@outputdir	1✔
223	@log_runtime(	1✔
224	bunched=False)
225	@pywrapper	1✔
226	def mean_correlation(	1✔
227	image_files_regex: str,
228	atlas_file: File,
229	output_dir: Directory,
230	correlation_threshold: float = 0.5,
231	dryrun: bool = False) -> tuple[File]:
232	"""
233	Compute the mean Pearson correlation between a reference image and a list
234	of other images.
235
236	Parameters
237	----------
238	image_files_regex : str
239	A REGEX to image files, each representing an image of the same shape
240	and geometry as `atlas_file`.
241	atlas_file : File
242	An file representing the reference image.
243	output_dir : Directory
244	Directory where a TSV file containing the mean correlation values is
245	created.
246	correlation_threshold : float
247	Quality control threshold on the correlation score. Default 0.5.
248	dryrun : bool
249	If True, skip actual computation and file writing. Default False.
250
251	Returns
252	-------
253	correlations_file : File
254	A TSV file containing the Pearson correlation coefficient between the
255	atlas image and each image pointed by the input REGEX.
256
257	Raises
258	------
259	ValueError
260	If the atlas and an image have incompatible shape or geometry.
261	"""
262	correlations_file = output_dir / "mean_correlations.tsv"	1✔
263	correlations_file.parent.mkdir(parents=True, exist_ok=True)	1✔
264
265	if not dryrun:	1✔
266
NEW 267	image_files = coerce_to_path(glob.glob(	×
268	str(image_files_regex)),
269	expected_type=list[File]
270	)
271	atlas_im = nibabel.load(atlas_file)	×
272	atlas_arr = atlas_im.get_fdata()	×
273
274	scores = pd.DataFrame(	×
275	columns=(
276	"participant_id",
277	"session",
278	"run",
279	"mean_correlation",
280	)
281	)
282	for path in image_files:	×
283	entities = parse_bids_keys(path)	×
284	im = nibabel.load(path)	×
285	arr = atlas_im.get_fdata()	×
286	if atlas_arr.shape != arr.shape:	×
287	raise ValueError(	×
288	f"Atlas and image have incompatible shape: {path}"
289	)
290	if not np.allclose(atlas_im.affine, im.affine):	×
291	raise ValueError(	×
292	f"Atlas and image have incompatible orientation: {path}"
293	)
294	corr, _ = pearsonr(	×
295	atlas_arr.flatten(),
296	arr.flatten(),
297	)
298	scores.loc[len(scores)] = [	×
299	entities["sub"],
300	entities["ses"],
301	entities["run"],
302	corr,
303	]
304
305	scores["qc"] = (	×
306	scores["mean_correlation"] > correlation_threshold
307	).astype(int)
308	scores.to_csv(	×
309	correlations_file,
310	index=False,
311	sep="\t",
312	)
313
314	return (correlations_file, )	1✔

neurospin-deepinsight / brainprep / 20846335294

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