23547419711

Committed 25 Mar 2026 02:52PM UTC coverage: 74.094% (-1.0%) from 75.06%

Build # 23547419711

Build Type

push

github

Committed by

AGrigis

Commit Message

brainprep: fix CI.

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

358 existing lines in 17 files now uncovered.

1410 of 1903 relevant lines covered (74.09%)

0.74 hits per line

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

63.41

/brainprep/interfaces/utils.py

##########################################################################
# NSAp - Copyright (C) CEA, 2021 - 2026
# 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 getpass
import gzip
import shutil
import socket
from collections import OrderedDict

import nibabel
import numpy as np
import pandas as pd

from ..decorators import (
    CoerceparamsHook,
    CommandLineWrapperHook,
    LogRuntimeHook,
    OutputdirHook,
    PythonWrapperHook,
    step,
)
from ..typing import (
    Directory,
    File,
)
from ..utils import (
    make_run_id,
)


@step(
    hooks=[
        CoerceparamsHook(),
        OutputdirHook(),
        LogRuntimeHook(
            bunched=False
        ),
        PythonWrapperHook(),
    ]
)
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, )


@step(
    hooks=[
        CoerceparamsHook(),
        OutputdirHook(),
        LogRuntimeHook(
            bunched=False
        ),
        PythonWrapperHook(),
    ]
)
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)


@step(
    hooks=[
        CoerceparamsHook(),
        OutputdirHook(),
        LogRuntimeHook(
            bunched=False
        ),
        PythonWrapperHook(),
    ]
)
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, )


@step(
    hooks=[
        CoerceparamsHook(),
        LogRuntimeHook(
            bunched=False
        ),
        PythonWrapperHook(),
    ]
)
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, )


@step(
    hooks=[
        CoerceparamsHook(),
        OutputdirHook(),
        LogRuntimeHook(
            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, )


@step(
    hooks=[
        CoerceparamsHook(),
        LogRuntimeHook(
            bunched=False
        ),
        CommandLineWrapperHook(),
    ]
)
def anonfile(
        input_file: File,
        mapping: dict[str, str]) -> tuple[list[str], File]:
    """
    Anonymize a text file using sed.

    The function constructs a list of sed substitution expressions based on
    the user-provided mapping and additional system-derived identifiers
    (hostname, IP address, username). The resulting command performs
    in-place anonymization of the input file.

    Parameters
    ----------
    input_file : File
        Path to the file to anonymize.
    mapping : dict[str, str]
        Patterns to replace (keys) and their replacements (values).

    Returns
    -------
    command : list[str]
        The sed command-line used for anonization.
    output_file : File
        Path to the anonymized file.
    """
    mapping = OrderedDict(mapping)
    hostname = socket.gethostname()
    mapping.update(
        OrderedDict({
            hostname: "HOSTNAME",
            socket.gethostbyname(hostname): "X.X.X.X",
            getpass.getuser(): "USER",
        })
    )

    patterns = []
    for old, new in mapping.items():
        old_esc = old.replace("/", r"\/")
        new_esc = new.replace("/", r"\/")
        patterns.extend(["-e", f"'s/{old_esc}/{new_esc}/g'"])
    command = [
        "sed",
        *patterns,
        "-i", str(input_file)
    ]

    return command, (input_file, )

1	##########################################################################
2	# NSAp - Copyright (C) CEA, 2021 - 2026
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 getpass	1✔
14	import gzip	1✔
15	import shutil	1✔
16	import socket	1✔
17	from collections import OrderedDict	1✔
18
19	import nibabel	1✔
20	import numpy as np	1✔
21	import pandas as pd	1✔
22
23	from ..decorators import (	1✔
24	CoerceparamsHook,
25	CommandLineWrapperHook,
26	LogRuntimeHook,
27	OutputdirHook,
28	PythonWrapperHook,
29	step,
30	)
31	from ..typing import (	1✔
32	Directory,
33	File,
34	)
35	from ..utils import (	1✔
36	make_run_id,
37	)
38
39
40	@step(	1✔
41	hooks=[
42	CoerceparamsHook(),
43	OutputdirHook(),
44	LogRuntimeHook(
45	bunched=False
46	),
47	PythonWrapperHook(),
48	]
49	)
50	def maskdiff(	1✔
51	mask1_file: File,
52	mask2_file: File,
53	output_dir: Directory,
54	entities: dict,
55	inv_mask1: bool = False,
56	inv_mask2: bool = False,
57	dryrun: bool = False) -> tuple[File]:
58	"""
59	Compute summary statistics comparing two binary masks.
60
61	This function loads two binary mask images, verifies that they share
62	the same spatial dimensions and affine transformation, computes their
63	voxel-wise intersection, and writes a summary table containing voxel
64	counts and physical volumes (in mm³) for each mask and their intersection.
65
66	Parameters
67	----------
68	mask1_file : File
69	Path to the first binary mask image.
70	mask2_file : File
71	Path to the second binary mask image.
72	output_dir : Directory
73	Directory where the defacing mask will be saved.
74	entities : dict
75	A dictionary of parsed BIDS entities including modality.
76	inv_mask1 : bool
77	If True, the first mask is inverted before comparison. This is
78	useful when the mask represents an exclusion region rather than an
79	inclusion region. Default False.
80	inv_mask2 : bool
81	If True, the second mask is inverted before comparison. This is
82	useful when the mask represents an exclusion region rather than an
83	inclusion region. Default False.
84	dryrun : bool
85	If True, skip actual computation and file writing. Default False.
86
87	Returns
88	-------
89	summary_file : File
90	Path to the generated summary TSV file.
91
92	Raises
93	------
94	ValueError
95	If both masks have not identical shapes and affines.
96	"""
97	basename = "sub-{sub}_ses-{ses}_run-{run}_mod-T1w_defacemask".format(	1✔
98	**entities)
99	summary_file = output_dir / f"{basename}.tsv"	1✔
100
101	if not dryrun:	1✔
102
103	mask1_im = nibabel.load(mask1_file)	×
104	mask2_im = nibabel.load(mask2_file)	×
105	mask1 = mask1_im.get_fdata().astype(bool)	×
UNCOV 106	mask2 = mask2_im.get_fdata().astype(bool)	×
107
108	if inv_mask1:	×
UNCOV 109	mask1 = ~mask1	×
UNCOV 110	if inv_mask2:	×
UNCOV 111	mask1 = ~mask2	×
112
113	if mask1.shape != mask2.shape:	×
UNCOV 114	raise ValueError(	×
115	f"Mask shapes differ: {mask1.shape} vs {mask2.shape}. "
116	"Resampling is required."
117	)
118	if not np.allclose(mask1_im.affine, mask2_im.affine):	×
119	raise ValueError(	×
120	"Mask affines differ. Resampling is required before "
121	"intersection."
122	)
123
UNCOV 124	intersection = np.logical_and(mask1, mask2)	×
UNCOV 125	voxel_volume = np.abs(np.linalg.det(mask1_im.affine[:3, :3]))	×
126
UNCOV 127	summary_df = pd.DataFrame({	×
128	"mask": ["mask1", "mask2", "intersection"],
129	"voxels": [
130	mask1.sum(),
131	mask2.sum(),
132	intersection.sum(),
133	],
134	"volume_mm3": [
135	mask1.sum() * voxel_volume,
136	mask2.sum() * voxel_volume,
137	intersection.sum() * voxel_volume,
138	]
139	})
UNCOV 140	summary_df.to_csv(summary_file, sep="\t", index=False)	×
141
142	return (summary_file, )	1✔
143
144
145	@step(	1✔
146	hooks=[
147	CoerceparamsHook(),
148	OutputdirHook(),
149	LogRuntimeHook(
150	bunched=False
151	),
152	PythonWrapperHook(),
153	]
154	)
155	def copyfiles(	1✔
156	source_image_files: list[File],
157	destination_image_files: list[File],
158	output_dir: Directory,
159	dryrun: bool = False) -> None:
160	"""
161	Copy input image files.
162
163	Parameters
164	----------
165	source_image_files : list[File]
166	Path to the image to be copied.
167	destination_image_files : list[File]
168	Path to the locations where images will be copied.
169	output_dir : Directory
170	Directory where the images are copied.
171	dryrun : bool
172	If True, skip actual computation and file writing. Default False.
173	"""
174	if not dryrun:	1✔
UNCOV 175	for src_path, dest_path in zip(source_image_files,	×
176	destination_image_files,
177	strict=True):
UNCOV 178	shutil.copy(src_path, dest_path)	×
179
180
181	@step(	1✔
182	hooks=[
183	CoerceparamsHook(),
184	OutputdirHook(),
185	LogRuntimeHook(
186	bunched=False
187	),
188	PythonWrapperHook(),
189	]
190	)
191	def movedir(	1✔
192	source_dir: Directory,
193	output_dir: Directory,
194	content: bool = False,
195	dryrun: bool = False) -> tuple[Directory]:
196	"""
197	Move input directory.
198
199	Parameters
200	----------
201	source_dir : Directory
202	Path to the directory to be moved.
203	output_dir : Directory
204	Directory where the folder is moved.
205	content : bool
206	If True, move the content of the source directory. Default False.
207	dryrun : bool
208	If True, skip actual computation and file writing. Default False.
209
210	Returns
211	-------
212	target_directory : Directory
213	Path to the moved directory.
214
215	Raises
216	------
217	ValueError
218	If `source_dir` is not a directory.
219	"""
220	if not dryrun:	1✔
UNCOV 221	if not source_dir.is_dir():	×
UNCOV 222	raise ValueError(	×
223	f"Source '{source_dir}' is not a directory."
224	)
UNCOV 225	if not content:	×
UNCOV 226	shutil.move(source_dir, output_dir / source_dir.name)	×
227	else:
UNCOV 228	for item in source_dir.iterdir():	×
UNCOV 229	target = output_dir / item.name	×
UNCOV 230	shutil.move(item, output_dir / item.name)	×
UNCOV 231	if not any(source_dir.iterdir()):	×
UNCOV 232	source_dir.rmdir()	×
233	return (output_dir if content else output_dir / source_dir.name, )	1✔
234
235
236	@step(	1✔
237	hooks=[
238	CoerceparamsHook(),
239	LogRuntimeHook(
240	bunched=False
241	),
242	PythonWrapperHook(),
243	]
244	)
245	def ungzfile(	1✔
246	input_file: File,
247	output_file: File,
248	output_dir: Directory,
249	dryrun: bool = False) -> tuple[File]:
250	"""
251	Ungzip input file.
252
253	Parameters
254	----------
255	input_file : File
256	Path to the file to ungzip.
257	output_file : File
258	Path to the ungzip file.
259	output_dir : Directory
260	Directory where the unzip file is created.
261	dryrun : bool
262	If True, skip actual computation and file writing. Default False.
263
264	Returns
265	-------
266	output_file : File
267	Path to the ungzip file.
268
269	Raises
270	------
271	ValueError
272	If the input file is not compressed.
273	"""
274	if input_file.suffix != ".gz":	1✔
UNCOV 275	raise ValueError(	×
276	f"The input file is not compressed: {input_file}"
277	)
278
279	if not dryrun:	1✔
UNCOV 280	with gzip.open(input_file, "rb") as gzfobj:	×
UNCOV 281	output_file.write_bytes(gzfobj.read())	×
282
283	return (output_file, )	1✔
284
285
286	@step(	1✔
287	hooks=[
288	CoerceparamsHook(),
289	OutputdirHook(),
290	LogRuntimeHook(
291	bunched=False
292	),
293	]
294	)
295	def write_uuid_mapping(	1✔
296	input_file: File,
297	output_dir: Directory,
298	entities: dict,
299	name: str = "uuid_mapping.tsv",
300	full_path: bool = False) -> File:
301	"""
302	Create a TSV file that records a deterministic UUID-based mapping.
303
304	Each row contains:
305	- filename: relative path within the BIDS dataset.
306	- run_default: 5-digit deterministic run ID derived from the filename.
307	- uuid: full UUIDv5 for traceability.
308
309	Parameters
310	----------
311	input_file : File
312	Path to the file to map.
313	output_dir : Directory
314	Directory where the TSV file is created.
315	entities : dict
316	A dictionary of parsed BIDS entities including modality.
317	name : str
318	Name of the TSV file to write. Default is "uuid_mapping.tsv".
319	full_path: bool
320	If True, extract entities from the full input path rather than
321	only the filename. Default is False.
322
323	Returns
324	-------
325	output_file : File
326	Path to the written TSV file.
327	"""
328	outut_file = output_dir / f"run-{entities['run']}" / name	1✔
329	filename = str(input_file) if full_path else input_file.name	1✔
330	code, short_code = make_run_id(filename)	1✔
331
332	if short_code == entities["run"]:	1✔
333	outut_file.parent.mkdir(parents=True, exist_ok=True)	1✔
334	df = pd.DataFrame.from_dict({	1✔
335	"filename": [filename],
336	"run": [short_code],
337	"uuid": [code],
338	})
339	df.to_csv(outut_file, sep="\t", index=False)	1✔
340	else:
341	outut_file = None	1✔
342
343	return (outut_file, )	1✔
344
345
346	@step(	1✔
347	hooks=[
348	CoerceparamsHook(),
349	LogRuntimeHook(
350	bunched=False
351	),
352	CommandLineWrapperHook(),
353	]
354	)
355	def anonfile(	1✔
356	input_file: File,
357	mapping: dict[str, str]) -> tuple[list[str], File]:
358	"""
359	Anonymize a text file using sed.
360
361	The function constructs a list of sed substitution expressions based on
362	the user-provided mapping and additional system-derived identifiers
363	(hostname, IP address, username). The resulting command performs
364	in-place anonymization of the input file.
365
366	Parameters
367	----------
368	input_file : File
369	Path to the file to anonymize.
370	mapping : dict[str, str]
371	Patterns to replace (keys) and their replacements (values).
372
373	Returns
374	-------
375	command : list[str]
376	The sed command-line used for anonization.
377	output_file : File
378	Path to the anonymized file.
379	"""
380	mapping = OrderedDict(mapping)	1✔
381	hostname = socket.gethostname()	1✔
382	mapping.update(	1✔
383	OrderedDict({
384	hostname: "HOSTNAME",
385	socket.gethostbyname(hostname): "X.X.X.X",
386	getpass.getuser(): "USER",
387	})
388	)
389
390	patterns = []	1✔
391	for old, new in mapping.items():	1✔
392	old_esc = old.replace("/", r"\/")	1✔
393	new_esc = new.replace("/", r"\/")	1✔
394	patterns.extend(["-e", f"'s/{old_esc}/{new_esc}/g'"])	1✔
395	command = [	1✔
396	"sed",
397	*patterns,
398	"-i", str(input_file)
399	]
400
401	return command, (input_file, )	1✔

neurospin-deepinsight / brainprep / 23547419711

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