22577664359

Committed 02 Mar 2026 01:15PM UTC coverage: 74.874% (+0.5%) from 74.419%

Build # 22577664359

Build Type

push

github

Committed by

AGrigis

Commit Message

brainprep: linter fix.

Coverage Stats

0 of 3 new or added lines in 1 file covered. (0.0%)

67 existing lines in 4 files now uncovered.

1490 of 1990 relevant lines covered (74.87%)

0.75 hits per line

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

62.96

/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.
##########################################################################

"""
Utils functions.
"""

import gzip
import shutil

import nibabel
import numpy as np
import pandas as pd

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


@coerceparams
@outputdir
@log_runtime(
    bunched=False)
@pywrapper
def maskdiff(
        mask1_file: File,
        mask2_file: File,
        output_dir: Directory,
        entities: dict,
        inv_mask1: bool = False,
        inv_mask2: bool = False,
        dryrun: bool = False) -> tuple[File]:
    """
    Compute summary statistics comparing two binary masks.

    This function loads two binary mask images, verifies that they share
    the same spatial dimensions and affine transformation, computes their
    voxel-wise intersection, and writes a summary table containing voxel
    counts and physical volumes (in mm³) for each mask and their intersection.

    Parameters
    ----------
    mask1_file : File
        Path to the first binary mask image.
    mask2_file : File
        Path to the second binary mask image.
    output_dir : Directory
        Directory where the defacing mask will be saved.
    entities : dict
        A dictionary of parsed BIDS entities including modality.
    inv_mask1 : bool
        If True, the first mask is inverted before comparison. This is
        useful when the mask represents an exclusion region rather than an
        inclusion region. Default False.
    inv_mask2 : bool
        If True, the second mask is inverted before comparison. This is
        useful when the mask represents an exclusion region rather than an
        inclusion region. Default False.
    dryrun : bool
        If True, skip actual computation and file writing. Default False.

    Returns
    -------
    summary_file : File
        Path to the generated summary TSV file.

    Raises
    ------
    ValueError
        If both masks have not identical shapes and affines.
    """
    basename = "sub-{sub}_ses-{ses}_run-{run}_mod-T1w_defacemask".format(
        **entities)
    summary_file = output_dir / f"{basename}.tsv"

    if not dryrun:

        mask1_im = nibabel.load(mask1_file)
        mask2_im = nibabel.load(mask2_file)
        mask1 = mask1_im.get_fdata().astype(bool)
        mask2 = mask2_im.get_fdata().astype(bool)

        if inv_mask1:
            mask1 = ~mask1
        if inv_mask2:
            mask1 = ~mask2

        if mask1.shape != mask2.shape:
            raise ValueError(
                f"Mask shapes differ: {mask1.shape} vs {mask2.shape}. "
                "Resampling is required."
            )
        if not np.allclose(mask1_im.affine, mask2_im.affine):
            raise ValueError(
                "Mask affines differ. Resampling is required before "
                "intersection."
            )

        intersection = np.logical_and(mask1, mask2)
        voxel_volume = np.abs(np.linalg.det(mask1_im.affine[:3, :3]))

        summary_df = pd.DataFrame({
            "mask": ["mask1", "mask2", "intersection"],
            "voxels": [
                mask1.sum(),
                mask2.sum(),
                intersection.sum(),
            ],
            "volume_mm3": [
                mask1.sum() * voxel_volume,
                mask2.sum() * voxel_volume,
                intersection.sum() * voxel_volume,
            ]
        })
        summary_df.to_csv(summary_file, sep="\t", index=False)

    return (summary_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, move 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.
    """
    if not dryrun:
        if not source_dir.is_dir():
            raise ValueError(
                f"Source '{source_dir}' is not a directory."
            )
        if not content:
            shutil.move(source_dir, output_dir / source_dir.name)
        else:
            for item in source_dir.iterdir():
                target = output_dir / item.name
                shutil.move(item, output_dir / item.name)
            if not any(source_dir.iterdir()):
                source_dir.rmdir()
    return (output_dir if content else output_dir / source_dir.name, )


@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, )


@coerceparams
@outputdir
@log_runtime(
    bunched=False)
def write_uuid_mapping(
        input_file: File,
        output_dir: Directory,
        entities: dict,
        name: str = "uuid_mapping.tsv",
        full_path: bool = False) -> File:
    """
    Create a TSV file that records a deterministic  UUID-based mapping.

    Each row contains:
    - filename: relative path within the BIDS dataset.
    - run_default: 5-digit deterministic run ID derived from the filename.
    - uuid: full UUIDv5 for traceability.

    Parameters
    ----------
    input_file : File
        Path to the file to map.
    output_dir : Directory
        Directory where the TSV file is created.
    entities : dict
        A dictionary of parsed BIDS entities including modality.
    name : str
        Name of the TSV file to write. Default is "uuid_mapping.tsv".
    full_path: bool
        If True, extract entities from the full input path rather than
        only the filename. Default is False.

    Returns
    -------
    output_file : File
        Path to the written TSV file.
    """
    outut_file = output_dir / f"run-{entities['run']}" / name
    filename = str(input_file) if full_path else input_file.name
    code, short_code = make_run_id(filename)

    if short_code == entities["run"]:
        outut_file.parent.mkdir(parents=True, exist_ok=True)
        df = pd.DataFrame.from_dict({
            "filename": [filename],
            "run": [short_code],
            "uuid": [code],
        })
        df.to_csv(outut_file, sep="\t", index=False)
    else:
        outut_file = None

    return (outut_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	Utils functions.
11	"""
12
13	import gzip	1✔
14	import shutil	1✔
15
16	import nibabel	1✔
17	import numpy as np	1✔
18	import pandas as pd	1✔
19
20	from ..reporting import log_runtime	1✔
21	from ..typing import (	1✔
22	Directory,
23	File,
24	)
25	from ..utils import (	1✔
26	coerceparams,
27	make_run_id,
28	outputdir,
29	)
30	from ..wrappers import pywrapper	1✔
31
32
33	@coerceparams	1✔
34	@outputdir	1✔
35	@log_runtime(	1✔
36	bunched=False)
37	@pywrapper	1✔
38	def maskdiff(	1✔
39	mask1_file: File,
40	mask2_file: File,
41	output_dir: Directory,
42	entities: dict,
43	inv_mask1: bool = False,
44	inv_mask2: bool = False,
45	dryrun: bool = False) -> tuple[File]:
46	"""
47	Compute summary statistics comparing two binary masks.
48
49	This function loads two binary mask images, verifies that they share
50	the same spatial dimensions and affine transformation, computes their
51	voxel-wise intersection, and writes a summary table containing voxel
52	counts and physical volumes (in mm³) for each mask and their intersection.
53
54	Parameters
55	----------
56	mask1_file : File
57	Path to the first binary mask image.
58	mask2_file : File
59	Path to the second binary mask image.
60	output_dir : Directory
61	Directory where the defacing mask will be saved.
62	entities : dict
63	A dictionary of parsed BIDS entities including modality.
64	inv_mask1 : bool
65	If True, the first mask is inverted before comparison. This is
66	useful when the mask represents an exclusion region rather than an
67	inclusion region. Default False.
68	inv_mask2 : bool
69	If True, the second mask is inverted before comparison. This is
70	useful when the mask represents an exclusion region rather than an
71	inclusion region. Default False.
72	dryrun : bool
73	If True, skip actual computation and file writing. Default False.
74
75	Returns
76	-------
77	summary_file : File
78	Path to the generated summary TSV file.
79
80	Raises
81	------
82	ValueError
83	If both masks have not identical shapes and affines.
84	"""
85	basename = "sub-{sub}_ses-{ses}_run-{run}_mod-T1w_defacemask".format(	1✔
86	**entities)
87	summary_file = output_dir / f"{basename}.tsv"	1✔
88
89	if not dryrun:	1✔
90
91	mask1_im = nibabel.load(mask1_file)	×
92	mask2_im = nibabel.load(mask2_file)	×
93	mask1 = mask1_im.get_fdata().astype(bool)	×
UNCOV 94	mask2 = mask2_im.get_fdata().astype(bool)	×
95
96	if inv_mask1:	×
97	mask1 = ~mask1	×
98	if inv_mask2:	×
UNCOV 99	mask1 = ~mask2	×
100
101	if mask1.shape != mask2.shape:	×
UNCOV 102	raise ValueError(	×
103	f"Mask shapes differ: {mask1.shape} vs {mask2.shape}. "
104	"Resampling is required."
105	)
106	if not np.allclose(mask1_im.affine, mask2_im.affine):	×
UNCOV 107	raise ValueError(	×
108	"Mask affines differ. Resampling is required before "
109	"intersection."
110	)
111
112	intersection = np.logical_and(mask1, mask2)	×
UNCOV 113	voxel_volume = np.abs(np.linalg.det(mask1_im.affine[:3, :3]))	×
114
UNCOV 115	summary_df = pd.DataFrame({	×
116	"mask": ["mask1", "mask2", "intersection"],
117	"voxels": [
118	mask1.sum(),
119	mask2.sum(),
120	intersection.sum(),
121	],
122	"volume_mm3": [
123	mask1.sum() * voxel_volume,
124	mask2.sum() * voxel_volume,
125	intersection.sum() * voxel_volume,
126	]
127	})
UNCOV 128	summary_df.to_csv(summary_file, sep="\t", index=False)	×
129
130	return (summary_file, )	1✔
131
132
133	@coerceparams	1✔
134	@outputdir	1✔
135	@log_runtime(	1✔
136	bunched=False)
137	@pywrapper	1✔
138	def copyfiles(	1✔
139	source_image_files: list[File],
140	destination_image_files: list[File],
141	output_dir: Directory,
142	dryrun: bool = False) -> None:
143	"""
144	Copy input image files.
145
146	Parameters
147	----------
148	source_image_files : list[File]
149	Path to the image to be copied.
150	destination_image_files : list[File]
151	Path to the locations where images will be copied.
152	output_dir : Directory
153	Directory where the images are copied.
154	dryrun : bool
155	If True, skip actual computation and file writing. Default False.
156	"""
157	if not dryrun:	1✔
UNCOV 158	for src_path, dest_path in zip(source_image_files,	×
159	destination_image_files,
160	strict=True):
UNCOV 161	shutil.copy(src_path, dest_path)	×
162
163
164	@coerceparams	1✔
165	@outputdir	1✔
166	@log_runtime(	1✔
167	bunched=False)
168	@pywrapper	1✔
169	def movedir(	1✔
170	source_dir: Directory,
171	output_dir: Directory,
172	content: bool = False,
173	dryrun: bool = False) -> tuple[Directory]:
174	"""
175	Move input directory.
176
177	Parameters
178	----------
179	source_dir : Directory
180	Path to the directory to be moved.
181	output_dir : Directory
182	Directory where the folder is moved.
183	content : bool
184	If True, move the content of the source directory. Default False.
185	dryrun : bool
186	If True, skip actual computation and file writing. Default False.
187
188	Returns
189	-------
190	target_directory : Directory
191	Path to the moved directory.
192
193	Raises
194	------
195	ValueError
196	If `source_dir` is not a directory.
197	"""
198	if not dryrun:	1✔
199	if not source_dir.is_dir():	×
UNCOV 200	raise ValueError(	×
201	f"Source '{source_dir}' is not a directory."
202	)
203	if not content:	×
UNCOV 204	shutil.move(source_dir, output_dir / source_dir.name)	×
205	else:
206	for item in source_dir.iterdir():	×
207	target = output_dir / item.name	×
208	shutil.move(item, output_dir / item.name)	×
209	if not any(source_dir.iterdir()):	×
UNCOV 210	source_dir.rmdir()	×
211	return (output_dir if content else output_dir / source_dir.name, )	1✔
212
213
214	@outputdir	1✔
215	@log_runtime(	1✔
216	bunched=False)
217	@pywrapper	1✔
218	def ungzfile(	1✔
219	input_file: File,
220	output_file: File,
221	output_dir: Directory,
222	dryrun: bool = False) -> tuple[File]:
223	"""
224	Ungzip input file.
225
226	Parameters
227	----------
228	input_file : File
229	Path to the file to ungzip.
230	output_file : File
231	Path to the ungzip file.
232	output_dir : Directory
233	Directory where the unzip file is created.
234	dryrun : bool
235	If True, skip actual computation and file writing. Default False.
236
237	Returns
238	-------
239	output_file : File
240	Path to the ungzip file.
241
242	Raises
243	------
244	ValueError
245	If the input file is not compressed.
246	"""
247	if input_file.suffix != ".gz":	1✔
UNCOV 248	raise ValueError(	×
249	f"The input file is not compressed: {input_file}"
250	)
251
252	if not dryrun:	1✔
253	with gzip.open(input_file, "rb") as gzfobj:	×
UNCOV 254	output_file.write_bytes(gzfobj.read())	×
255
256	return (output_file, )	1✔
257
258
259	@coerceparams	1✔
260	@outputdir	1✔
261	@log_runtime(	1✔
262	bunched=False)
263	def write_uuid_mapping(	1✔
264	input_file: File,
265	output_dir: Directory,
266	entities: dict,
267	name: str = "uuid_mapping.tsv",
268	full_path: bool = False) -> File:
269	"""
270	Create a TSV file that records a deterministic UUID-based mapping.
271
272	Each row contains:
273	- filename: relative path within the BIDS dataset.
274	- run_default: 5-digit deterministic run ID derived from the filename.
275	- uuid: full UUIDv5 for traceability.
276
277	Parameters
278	----------
279	input_file : File
280	Path to the file to map.
281	output_dir : Directory
282	Directory where the TSV file is created.
283	entities : dict
284	A dictionary of parsed BIDS entities including modality.
285	name : str
286	Name of the TSV file to write. Default is "uuid_mapping.tsv".
287	full_path: bool
288	If True, extract entities from the full input path rather than
289	only the filename. Default is False.
290
291	Returns
292	-------
293	output_file : File
294	Path to the written TSV file.
295	"""
296	outut_file = output_dir / f"run-{entities['run']}" / name	1✔
297	filename = str(input_file) if full_path else input_file.name	1✔
298	code, short_code = make_run_id(filename)	1✔
299
300	if short_code == entities["run"]:	1✔
301	outut_file.parent.mkdir(parents=True, exist_ok=True)	1✔
302	df = pd.DataFrame.from_dict({	1✔
303	"filename": [filename],
304	"run": [short_code],
305	"uuid": [code],
306	})
307	df.to_csv(outut_file, sep="\t", index=False)	1✔
308	else:
309	outut_file = None	1✔
310
311	return (outut_file, )	1✔

neurospin-deepinsight / brainprep / 22577664359

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