22624503433

Committed 03 Mar 2026 01:08PM UTC coverage: 74.755% (+0.2%) from 74.577%

Build # 22624503433

Build Type

push

github

Committed by

AGrigis

Commit Message

brainprep/__init__: organize traceback with rich.

Coverage Stats

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

29 existing lines in 2 files now uncovered.

1525 of 2040 relevant lines covered (74.75%)

0.75 hits per line

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

81.99

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

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

import inspect
import json
import re
import uuid
from collections.abc import Callable, Iterable
from pathlib import Path
from typing import (
    Any,
    Union,
    get_args,
    get_origin,
)

from decorator import decorator

from .._version import __version__
from ..typing import (
    Directory,
    File,
)
from .color import (
    print_info,
    print_warn,
)


@decorator
def bids(
        func: Callable,
        process: str | None = None,
        bids_file: File | Iterable[File] | None = None,
        container: str | None = None,
        add_subjects: bool = False,
        longitudinal: bool = False,
        *args: Any,
        **kw: Any) -> Callable:
    """
    BIDS specification.

    Decorator that computes a BIDS-compliant output directory path
    based on the input BIDS file and injects it into the function.

    Decorator that ensures BIDS-compliant metadata is written to the output
    directory.

    Parameters
    ----------
    func : Callable
        The function to be decorated.
    process : str | None
        Name of the processing pipeline (e.g., 'fmriprep', 'custom'). Default
        None.
    bids_file : File |Iterable[File] | None
        Name of the argument in the function that contains the BIDS file path.
        Default None.
    container : str | None
        The name of the container (e.g., Docker image) used to run the
        pipeline. Default None.
    add_subjects : bool
        If True, add a 'subjects' upper level directory in the output
        directory, for instance to regroup subject level data. Default False.
    longitudinal : bool
        If True, add a 'longitudinal' upper level directory in the output
        directory. Default False.
    *args : Any
        Positional arguments passed to `func`.
    **kw : Any
        Keyword arguments passed to `func`.

    Returns
    -------
    wrapper : Callable
        A wrapped function with computed 'output_dir' injected.

    Raises
    ------
    ValueError
        If the decorated function has no `bids_file` or 'output_dir'
        arguments.
    """
    inputs = inspect.getcallargs(func, *args, **kw)

    if process is None:
        return func(**inputs)

    subject_level = bids_file is not None
    output_dir = (
        Path(inputs["output_dir"]) /
        "derivatives" /
        process
    )
    if longitudinal:
        output_dir /= "longitudinal"
    if add_subjects:
        output_dir /= "subjects"
    if subject_level:
        for key in (bids_file, "output_dir"):
            if key not in inputs:
                raise ValueError(
                    f"The 'bids' decorator needs a '{key}' function argument."
                )
        if isinstance(inputs[bids_file], (list, tuple)):
            entities = parse_bids_keys(inputs[bids_file][0])
        else:
            entities = parse_bids_keys(inputs[bids_file])
        output_dir = (
            output_dir /
            f"sub-{entities['sub']}" /
            f"ses-{entities['ses']}"
        )
    metadata_file = (
        Path(inputs["output_dir"]) /
        "derivatives" /
        process /
        "dataset_description.json"
    )

    metadata_file.parent.mkdir(parents=True, exist_ok=True)
    inputs["output_dir"] = output_dir

    if not metadata_file.is_file():
        metadata = {
            "Name": f"{func.__module__}.{func.__name__}",
            "BIDSVersion": "1.8.0",
            "DatasetType": "derivative",
            "GeneratedBy": [
                {
                    "Name": "brainprep",
                    "Version": __version__,
                    "CodeURL": ("https://github.com/neurospin-deepinsight/"
                                "brainprep"),
                }
            ],
        }
        if container is not None:
            metadata["GeneratedBy"][0].update(
                {
                    "Container": {
                        "Type": "docker",
                        "Tag": f"{container}:{__version__}"
                      }
                }
            )
        with metadata_file.open("w", encoding="utf-8") as of:
            json.dump(metadata, of, indent=4)

    return func(**inputs)


@decorator
def outputdir(
        func: Callable,
        plotting: bool = False,
        quality_check: bool = False,
        morphometry: bool = False,
        *args: Any,
        **kw: Any) -> Callable:
    """
    Create the output directory for the decorated function.

    This decorator ensures that the output directory exists before the
    wrapped function is executed. Optional subdirectories can also be
    created, such as a ``figures`` directory for plots or a ``quality_check``
    directory for quality check outputs or ``morphometry`` directory
    for morphometry outputs.

    Parameters
    ----------
    func : Callable
        The function to be decorated.
    plotting : bool
        If True, add a ``figures`` upper level directory in the output
        directory. Default False.
    quality_check : bool
        If True, add a ``quality_check`` upper level directory in the output
        directory. Default False.
    morphometry : bool
        If True, add a ``morphometry`` upper level directory in the output
        directory. Default False.
    *args : Any
        Positional arguments passed to ``func``.
    **kw : Any
        Keyword arguments passed to ``func``.

    Returns
    -------
    wrapper : Callable
        A wrapped function with the ``output_dir`` created on disk.

    Raises
    ------
    ValueError
        If the decorated function has no ``output_dir`` argument.
    """
    inputs = inspect.getcallargs(func, *args, **kw)

    if "output_dir" not in inputs:
        raise ValueError(
            "The 'outputdir' decorator needs a 'output_dir' function argument."
        )

    if plotting:
        inputs["output_dir"] = (
            Path(inputs["output_dir"]) /
            "figures"
        )
    if quality_check:
        inputs["output_dir"] = (
            Path(inputs["output_dir"]) /
            "quality_check"
        )
    if morphometry:
        inputs["output_dir"] = (
            Path(inputs["output_dir"]) /
            "morphometry"
        )

    Path(inputs["output_dir"]).mkdir(parents=True, exist_ok=True)

    return func(**inputs)


@decorator
def coerceparams(
        func: Callable,
        *args: Any,
        **kw: Any) -> Callable:
    """
    Convert annotated arguments before calling the decorated function.

    This decorator inspects the type annotations of the wrapped function
    and performs two automatic conversions:

    - Arguments annotated as ``File`` or ``Directory`` are converted into
      ``pathlib.Path`` instances.
    - Arguments annotated as list types are parsed from comma-separated
      strings into Python lists.

    Parameters
    ----------
    func : Callable
        The function to be decorated.
    *args : Any
        Positional arguments passed to ``func``.
    **kw : Any
        Keyword arguments passed to ``func``.

    Returns
    -------
    Callable
        A wrapped function in which arguments annotated as ``File`` or
        ``Directory`` are converted to ``pathlib.Path`` objects, and list-typed
        arguments are coerced from comma-separated strings into lists.

    Raises
    ------
    ValueError
        If the decorated function contains arguments without type annotations.
    """
    inputs = inspect.getcallargs(func, *args, **kw)
    sig = inspect.signature(func)

    for name, param in sig.parameters.items():
        if param.annotation is inspect.Parameter.empty:
            raise ValueError(
                "The decorated function must only have typed arguments."
            )
        inputs[name] = coerce_to_path(
            coerce_to_list(
                inputs[name],
                param.annotation,
            ),
            param.annotation,
        )

    return func(**inputs)


def coerce_to_list(
        value: Any,
        expected_type: type) -> Any:
    """
    Coerce a value into a list when the expected type annotation indicates
    a list or tuple.

    Parameters
    ----------
    value : Any
        The input value to be coerced.
    expected_type : type
        The expected type annotation (e.g., `File`, `List[File]`,
        `Dict[str, Directory]`, `Union[str, Directory]`).

    Returns
    -------
    typed_value : Any
        The coerced value, with `list` converted to `list`.

    Notes
    -----
    - Comma-separated strings (e.g., ``"a,b,c"``) are split into lists.
    - Single non-list values are wrapped into a list.
    - Existing lists or tuples are returned as lists.
    """
    if value is None:
        return value

    origin = get_origin(expected_type)

    if origin in {list, tuple}:
        if isinstance(value, str) and "," in value:
            return value.split(",")
        if not isinstance(value, (list, tuple)):
            return [value]
        return list(value)

    return value


def coerce_to_path(
        value: Any,
        expected_type: type) -> Any:
    """
    Recursively convert values to `pathlib.Path` based on expected type
    annotations.

    Parameters
    ----------
    value : Any
        The input value to be coerced.
    expected_type : type
        The expected type annotation (e.g., `File`, `List[File]`,
        `Dict[str, Directory]`, `Union[str, Directory]`).

    Returns
    -------
    typed_value : Any
        The coerced value, with `File` and `Directory` converted to
        `pathlib.Path`.
    """
    origin = get_origin(expected_type)
    args = get_args(expected_type)

    if value is None:
        return value

    if expected_type in {File, Directory}:
        return Path(value).resolve()

    if origin is Union and (File in args or Directory in args):
        return Path(value).resolve()

    if origin in {list, tuple, set} and args:
        container_type = origin
        inner_type = args[0]
        return container_type(coerce_to_path(inner_value, inner_type)
                              for inner_value in value)

    if origin is dict and len(args) == 2:
        _, val_type = args
        return {key: coerce_to_path(val, val_type)
                for key, val in value.items()}

    return value


def parse_bids_keys(
        bids_path: File,
        full_path: bool = False) -> dict[str]:
    """
    Parse BIDS entities and modality from a filename or path with validation.

    This function extracts BIDS entities (e.g., subject, session, task,
    run) from a BIDS-compliant filename or full path. It also identifies the
    modality and applies default values when certain entities are missing.

    When the `ses` entity is absent, it defaults to "01". This provides ensures
    consistent downstream file handling.

    When the `run` entity is absent, a deterministic 5-digit identifier is
    generated from the filename using UUID. This produces a short, stable
    hash so that the same filename always yields the same default run value.

    Parameters
    ----------
    bids_path : File
        The BIDS file to parse.
    full_path: bool
        If True, extract entities from the full input path rather than
        only the filename. Default is False.

    Returns
    -------
    entities : dict[str]
        A dictionary containing the parsed BIDS entities and the detected
        modality. Missing entities such as `ses` and `run` are filled with
        default values.

    Notes
    -----
    This procedure ensures that each BIDS file has a unique run identifier
    within its folder. It checks whether the current run value appears more
    than once, assigns a UUID-style fallback if needed, and warns if even
    that fallback is not unique.
    """
    # Extract the filename from the path id necessary
    filename = str(bids_path) if full_path else bids_path.name

    # Regex pattern for BIDS entities
    entity_pattern = (
        r"(?P<entity>(sub|ses|task|acq|run|echo|rec|dir|mod|ce|part|space|res|"
        r"recording))"
        r"-(?P<value>[^_/]+)"
    )
    entities = {}
    for match in re.finditer(entity_pattern, filename):
        entity = match.group("entity")
        value = match.group("value")
        entities[entity] = value

    # Extract modality (suffix before extension)
    suffix_pattern = (
        r"_(?P<modality>[a-zA-Z0-9]+)(?=\.(nii|nii\.gz|json|tsv|edf|vhdr"
        r"|eeg|bvec|bval|csv))"
    )
    modality_match = re.search(suffix_pattern, filename)
    if modality_match:
        entities["modality"] = modality_match.group("modality")

    # Update modality
    if "mod" not in entities and "modality" in entities:
        entities["mod"] = entities["modality"]

    # Define default values for missing entities
    defaults = {
        "ses": "01",
        "run": make_run_id(filename)[1],
    }

    # Fill in missing entities with defaults
    run_in_entities = "run" in entities
    for key, default in defaults.items():
        entities.setdefault(key, default)

    # Check integrity
    status = check_run(bids_path, entities, full_path)
    if run_in_entities and not status:
        print_info(
            "Multiple files with same run ID detected, using UUID instead."
        )
        entities["run"] = defaults["run"]
        status = check_run(bids_path, entities, full_path)
    if not status:
        print_warn(
            f"The generated UUID is not unique: {bids_path}"
        )

    return entities


def check_run(
        bids_path: File,
        entities: dict[str],
        full_path: bool = False) -> bool:
    """
    Scan the folder containing a BIDS file and verify that the run entity
    associated with the file appears exactly once among all matching files.

    Parameters
    ----------
    bids_path : File
        A BIDS file.
    entities : dict[str]
        Dictionary of parsed BIDS entities for the file, including the
        modality.
    full_path : bool
        If True, extract entities from the full path instead of only the
        filename. Default False.

    Returns
    -------
    bool
        True if the run identifier occurs exactly once among all matching
        files in the folder, False otherwise.
    """
    filename = str(bids_path) if full_path else bids_path.name
    ext = "".join(bids_path.suffixes)
    entity_pattern = (
        r"(?P<entity>(run))"
        r"-(?P<value>[^_/]+)"
    )
    pattern = f"sub-*{entities['modality']}*{ext}"

    all_entities = []
    for bids_path_ in bids_path.parent.glob(pattern):
        filename_ = str(bids_path_) if full_path else bids_path_.name
        entities_ = {"filename": filename_}

        # Extract run entity if present
        for match in re.finditer(entity_pattern, str(bids_path_)):
            entity = match.group("entity")
            value = match.group("value")
            entities_[entity] = value

        # If run is missing, generate one
        if "run" not in entities_:
            entities_["run"] = make_run_id(filename)[1]

        all_entities.append(entities_)

    # Count how many times the current file's run appears
    all_run_ids = [item["run"] for item in all_entities]
    count = all_run_ids.count(entities["run"])

    return count == 1


def make_run_id(
        filename: str) -> tuple[str, str]:
    """
    Generate a deterministic identifier and a 5-digit short code from a
    filename.

    This function computes a UUIDv5 using the URL namespace and the provided
    filename, converts the UUID to its integer representation, and returns both
    the full integer-based code and its first five digits. The result is stable
    and reproducible: the same filename always produces the same values.

    Parameters
    ----------
    filename : str
        The filename used as the seed for generating the identifiers.

    Returns
    -------
    code : str
        The full integer representation of the UUIDv5 derived from the
        filename.
    short_code : str
        The first five digits of the UUID-derived code, used as a compact ID.
    """
    code = str(uuid.uuid5(uuid.NAMESPACE_URL, filename).int)
    return code, code[:5]


def sidecar_from_file(
        image_file: File) -> File:
    """
    Infers the corresponding JSON sidecar file for a given NIfTI image file.

    This function checks that the input file has a ``.nii.gz`` extension and
    attempts to locate a sidecar ``.json`` file with the same base name. If
    either condition fails, it raises a ValueError.

    Parameters
    ----------
    image_file : File
        The NIfTI image file for which to infer the JSON sidecar.

    Returns
    -------
    sidecar_file : File
        Path to the inferred JSON sidecar file.

    Raises
    ------
    ValueError
        If the input file does not have a `.nii.gz` extension or if the
        corresponding JSON sidecar file does not exist.

    Examples
    --------
    >>> from pathlib import Path
    >>> from brainprep.utils import sidecar_from_file
    >>>
    >>> image_file = Path("/tmp/sub-01_T1w.nii.gz")
    >>> sidecar_file = Path("/tmp/sub-01_T1w.json")
    >>> sidecar_file.touch()
    >>>
    >>> sidecar_from_file(image_file)
    PosixPath('/tmp/sub-01_T1w.json')
    """
    if not str(image_file).endswith(".nii.gz"):
        raise ValueError(
            f"Input image file must be in NIIGZ format: {image_file}"
        )
    sidecar_file = Path(str(image_file).replace(".nii.gz", ".json"))
    if not sidecar_file.is_file():
        raise ValueError(
            f"Sidecar inferred from input image file not found: {sidecar_file}"
        )
    return sidecar_file


def find_stack_level() -> int:
    """
    Return the index of the first stack frame outside the ``brainprep``
    package.

    This function walks backward through the current call stack and finds the
    first frame whose file path does not belong to the ``brainprep`` package
    directory. Test files (i.e., files whose names start with ``test_``) are
    always treated as external. This is useful for producing cleaner warnings
    and error messages by pointing to user code rather than internal library
    frames.

    Returns
    -------
    int
        The number of internal frames to skip before reaching user code.

    Notes
    -----
    Adapted from the pandas codebase.

    Examples
    --------
    >>> import warnings
    >>> from brainprep.utils import find_stack_level
    >>>
    >>> def load_data(path):
    ...     if not path.exists():
    ...         warnings.warn(
    ...             "The provided path does not exist.",
    ...             stacklevel=find_stack_level()
    ...         )
    """
    import brainprep

    pkg_dir = Path(brainprep.__file__).parent

    # https://stackoverflow.com/questions/17407119/python-inspect-stack-is-slow
    frame = inspect.currentframe()
    try:
        n = 0
        while frame:
            filename = inspect.getfile(frame)
            is_test_file = Path(filename).name.startswith("test_")
            in_nilearn_code = filename.startswith(str(pkg_dir))
            if not in_nilearn_code or is_test_file:
                break
            frame = frame.f_back
            n += 1
    finally:
        # See note in
        # https://docs.python.org/3/library/inspect.html#inspect.Traceback
        del frame
    return n

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	Module that contains some utility functions.
11	"""
12
13	import inspect	1✔
14	import json	1✔
15	import re	1✔
16	import uuid	1✔
17	from collections.abc import Callable, Iterable	1✔
18	from pathlib import Path	1✔
19	from typing import (	1✔
20	Any,
21	Union,
22	get_args,
23	get_origin,
24	)
25
26	from decorator import decorator	1✔
27
28	from .._version import __version__	1✔
29	from ..typing import (	1✔
30	Directory,
31	File,
32	)
33	from .color import (	1✔
34	print_info,
35	print_warn,
36	)
37
38
39	@decorator	1✔
40	def bids(	1✔
41	func: Callable,
42	process: str \| None = None,
43	bids_file: File \| Iterable[File] \| None = None,
44	container: str \| None = None,
45	add_subjects: bool = False,
46	longitudinal: bool = False,
47	*args: Any,
48	**kw: Any) -> Callable:
49	"""
50	BIDS specification.
51
52	Decorator that computes a BIDS-compliant output directory path
53	based on the input BIDS file and injects it into the function.
54
55	Decorator that ensures BIDS-compliant metadata is written to the output
56	directory.
57
58	Parameters
59	----------
60	func : Callable
61	The function to be decorated.
62	process : str \| None
63	Name of the processing pipeline (e.g., 'fmriprep', 'custom'). Default
64	None.
65	bids_file : File \|Iterable[File] \| None
66	Name of the argument in the function that contains the BIDS file path.
67	Default None.
68	container : str \| None
69	The name of the container (e.g., Docker image) used to run the
70	pipeline. Default None.
71	add_subjects : bool
72	If True, add a 'subjects' upper level directory in the output
73	directory, for instance to regroup subject level data. Default False.
74	longitudinal : bool
75	If True, add a 'longitudinal' upper level directory in the output
76	directory. Default False.
77	*args : Any
78	Positional arguments passed to `func`.
79	**kw : Any
80	Keyword arguments passed to `func`.
81
82	Returns
83	-------
84	wrapper : Callable
85	A wrapped function with computed 'output_dir' injected.
86
87	Raises
88	------
89	ValueError
90	If the decorated function has no `bids_file` or 'output_dir'
91	arguments.
92	"""
93	inputs = inspect.getcallargs(func, args, *kw)	1✔
94
95	if process is None:	1✔
UNCOV 96	return func(**inputs)	×
97
98	subject_level = bids_file is not None	1✔
99	output_dir = (	1✔
100	Path(inputs["output_dir"]) /
101	"derivatives" /
102	process
103	)
104	if longitudinal:	1✔
105	output_dir /= "longitudinal"	1✔
106	if add_subjects:	1✔
107	output_dir /= "subjects"	1✔
108	if subject_level:	1✔
109	for key in (bids_file, "output_dir"):	1✔
110	if key not in inputs:	1✔
UNCOV 111	raise ValueError(	×
112	f"The 'bids' decorator needs a '{key}' function argument."
113	)
114	if isinstance(inputs[bids_file], (list, tuple)):	1✔
115	entities = parse_bids_keys(inputs[bids_file][0])	1✔
116	else:
117	entities = parse_bids_keys(inputs[bids_file])	1✔
118	output_dir = (	1✔
119	output_dir /
120	f"sub-{entities['sub']}" /
121	f"ses-{entities['ses']}"
122	)
123	metadata_file = (	1✔
124	Path(inputs["output_dir"]) /
125	"derivatives" /
126	process /
127	"dataset_description.json"
128	)
129
130	metadata_file.parent.mkdir(parents=True, exist_ok=True)	1✔
131	inputs["output_dir"] = output_dir	1✔
132
133	if not metadata_file.is_file():	1✔
134	metadata = {	1✔
135	"Name": f"{func.__module__}.{func.__name__}",
136	"BIDSVersion": "1.8.0",
137	"DatasetType": "derivative",
138	"GeneratedBy": [
139	{
140	"Name": "brainprep",
141	"Version": __version__,
142	"CodeURL": ("https://github.com/neurospin-deepinsight/"
143	"brainprep"),
144	}
145	],
146	}
147	if container is not None:	1✔
148	metadata["GeneratedBy"][0].update(	1✔
149	{
150	"Container": {
151	"Type": "docker",
152	"Tag": f"{container}:{__version__}"
153	}
154	}
155	)
156	with metadata_file.open("w", encoding="utf-8") as of:	1✔
157	json.dump(metadata, of, indent=4)	1✔
158
159	return func(**inputs)	1✔
160
161
162	@decorator	1✔
163	def outputdir(	1✔
164	func: Callable,
165	plotting: bool = False,
166	quality_check: bool = False,
167	morphometry: bool = False,
168	*args: Any,
169	**kw: Any) -> Callable:
170	"""
171	Create the output directory for the decorated function.
172
173	This decorator ensures that the output directory exists before the
174	wrapped function is executed. Optional subdirectories can also be
175	created, such as a ``figures`` directory for plots or a ``quality_check``
176	directory for quality check outputs or ``morphometry`` directory
177	for morphometry outputs.
178
179	Parameters
180	----------
181	func : Callable
182	The function to be decorated.
183	plotting : bool
184	If True, add a ``figures`` upper level directory in the output
185	directory. Default False.
186	quality_check : bool
187	If True, add a ``quality_check`` upper level directory in the output
188	directory. Default False.
189	morphometry : bool
190	If True, add a ``morphometry`` upper level directory in the output
191	directory. Default False.
192	*args : Any
193	Positional arguments passed to ``func``.
194	**kw : Any
195	Keyword arguments passed to ``func``.
196
197	Returns
198	-------
199	wrapper : Callable
200	A wrapped function with the ``output_dir`` created on disk.
201
202	Raises
203	------
204	ValueError
205	If the decorated function has no ``output_dir`` argument.
206	"""
207	inputs = inspect.getcallargs(func, args, *kw)	1✔
208
209	if "output_dir" not in inputs:	1✔
UNCOV 210	raise ValueError(	×
211	"The 'outputdir' decorator needs a 'output_dir' function argument."
212	)
213
214	if plotting:	1✔
215	inputs["output_dir"] = (	1✔
216	Path(inputs["output_dir"]) /
217	"figures"
218	)
219	if quality_check:	1✔
220	inputs["output_dir"] = (	1✔
221	Path(inputs["output_dir"]) /
222	"quality_check"
223	)
224	if morphometry:	1✔
225	inputs["output_dir"] = (	1✔
226	Path(inputs["output_dir"]) /
227	"morphometry"
228	)
229
230	Path(inputs["output_dir"]).mkdir(parents=True, exist_ok=True)	1✔
231
232	return func(**inputs)	1✔
233
234
235	@decorator	1✔
236	def coerceparams(	1✔
237	func: Callable,
238	*args: Any,
239	**kw: Any) -> Callable:
240	"""
241	Convert annotated arguments before calling the decorated function.
242
243	This decorator inspects the type annotations of the wrapped function
244	and performs two automatic conversions:
245
246	- Arguments annotated as ``File`` or ``Directory`` are converted into
247	``pathlib.Path`` instances.
248	- Arguments annotated as list types are parsed from comma-separated
249	strings into Python lists.
250
251	Parameters
252	----------
253	func : Callable
254	The function to be decorated.
255	*args : Any
256	Positional arguments passed to ``func``.
257	**kw : Any
258	Keyword arguments passed to ``func``.
259
260	Returns
261	-------
262	Callable
263	A wrapped function in which arguments annotated as ``File`` or
264	``Directory`` are converted to ``pathlib.Path`` objects, and list-typed
265	arguments are coerced from comma-separated strings into lists.
266
267	Raises
268	------
269	ValueError
270	If the decorated function contains arguments without type annotations.
271	"""
272	inputs = inspect.getcallargs(func, args, *kw)	1✔
273	sig = inspect.signature(func)	1✔
274
275	for name, param in sig.parameters.items():	1✔
276	if param.annotation is inspect.Parameter.empty:	1✔
UNCOV 277	raise ValueError(	×
278	"The decorated function must only have typed arguments."
279	)
280	inputs[name] = coerce_to_path(	1✔
281	coerce_to_list(
282	inputs[name],
283	param.annotation,
284	),
285	param.annotation,
286	)
287
288	return func(**inputs)	1✔
289
290
291	def coerce_to_list(	1✔
292	value: Any,
293	expected_type: type) -> Any:
294	"""
295	Coerce a value into a list when the expected type annotation indicates
296	a list or tuple.
297
298	Parameters
299	----------
300	value : Any
301	The input value to be coerced.
302	expected_type : type
303	The expected type annotation (e.g., `File`, `List[File]`,
304	`Dict[str, Directory]`, `Union[str, Directory]`).
305
306	Returns
307	-------
308	typed_value : Any
309	The coerced value, with `list` converted to `list`.
310
311	Notes
312	-----
313	- Comma-separated strings (e.g., ``"a,b,c"``) are split into lists.
314	- Single non-list values are wrapped into a list.
315	- Existing lists or tuples are returned as lists.
316	"""
317	if value is None:	1✔
318	return value	1✔
319
320	origin = get_origin(expected_type)	1✔
321
322	if origin in {list, tuple}:	1✔
323	if isinstance(value, str) and "," in value:	1✔
UNCOV 324	return value.split(",")	×
325	if not isinstance(value, (list, tuple)):	1✔
UNCOV 326	return [value]	×
327	return list(value)	1✔
328
329	return value	1✔
330
331
332	def coerce_to_path(	1✔
333	value: Any,
334	expected_type: type) -> Any:
335	"""
336	Recursively convert values to `pathlib.Path` based on expected type
337	annotations.
338
339	Parameters
340	----------
341	value : Any
342	The input value to be coerced.
343	expected_type : type
344	The expected type annotation (e.g., `File`, `List[File]`,
345	`Dict[str, Directory]`, `Union[str, Directory]`).
346
347	Returns
348	-------
349	typed_value : Any
350	The coerced value, with `File` and `Directory` converted to
351	`pathlib.Path`.
352	"""
353	origin = get_origin(expected_type)	1✔
354	args = get_args(expected_type)	1✔
355
356	if value is None:	1✔
357	return value	1✔
358
359	if expected_type in {File, Directory}:	1✔
360	return Path(value).resolve()	1✔
361
362	if origin is Union and (File in args or Directory in args):	1✔
UNCOV 363	return Path(value).resolve()	×
364
365	if origin in {list, tuple, set} and args:	1✔
366	container_type = origin	1✔
367	inner_type = args[0]	1✔
368	return container_type(coerce_to_path(inner_value, inner_type)	1✔
369	for inner_value in value)
370
371	if origin is dict and len(args) == 2:	1✔
UNCOV 372	_, val_type = args	×
UNCOV 373	return {key: coerce_to_path(val, val_type)	×
374	for key, val in value.items()}
375
376	return value	1✔
377
378
379	def parse_bids_keys(	1✔
380	bids_path: File,
381	full_path: bool = False) -> dict[str]:
382	"""
383	Parse BIDS entities and modality from a filename or path with validation.
384
385	This function extracts BIDS entities (e.g., subject, session, task,
386	run) from a BIDS-compliant filename or full path. It also identifies the
387	modality and applies default values when certain entities are missing.
388
389	When the `ses` entity is absent, it defaults to "01". This provides ensures
390	consistent downstream file handling.
391
392	When the `run` entity is absent, a deterministic 5-digit identifier is
393	generated from the filename using UUID. This produces a short, stable
394	hash so that the same filename always yields the same default run value.
395
396	Parameters
397	----------
398	bids_path : File
399	The BIDS file to parse.
400	full_path: bool
401	If True, extract entities from the full input path rather than
402	only the filename. Default is False.
403
404	Returns
405	-------
406	entities : dict[str]
407	A dictionary containing the parsed BIDS entities and the detected
408	modality. Missing entities such as `ses` and `run` are filled with
409	default values.
410
411	Notes
412	-----
413	This procedure ensures that each BIDS file has a unique run identifier
414	within its folder. It checks whether the current run value appears more
415	than once, assigns a UUID-style fallback if needed, and warns if even
416	that fallback is not unique.
417	"""
418	# Extract the filename from the path id necessary
419	filename = str(bids_path) if full_path else bids_path.name	1✔
420
421	# Regex pattern for BIDS entities
422	entity_pattern = (	1✔
423	r"(?P<entity>(sub\|ses\|task\|acq\|run\|echo\|rec\|dir\|mod\|ce\|part\|space\|res\|"
424	r"recording))"
425	r"-(?P<value>[^_/]+)"
426	)
427	entities = {}	1✔
428	for match in re.finditer(entity_pattern, filename):	1✔
429	entity = match.group("entity")	1✔
430	value = match.group("value")	1✔
431	entities[entity] = value	1✔
432
433	# Extract modality (suffix before extension)
434	suffix_pattern = (	1✔
435	r"_(?P<modality>[a-zA-Z0-9]+)(?=\.(nii\|nii\.gz\|json\|tsv\|edf\|vhdr"
436	r"\|eeg\|bvec\|bval\|csv))"
437	)
438	modality_match = re.search(suffix_pattern, filename)	1✔
439	if modality_match:	1✔
440	entities["modality"] = modality_match.group("modality")	1✔
441
442	# Update modality
443	if "mod" not in entities and "modality" in entities:	1✔
444	entities["mod"] = entities["modality"]	1✔
445
446	# Define default values for missing entities
447	defaults = {	1✔
448	"ses": "01",
449	"run": make_run_id(filename)[1],
450	}
451
452	# Fill in missing entities with defaults
453	run_in_entities = "run" in entities	1✔
454	for key, default in defaults.items():	1✔
455	entities.setdefault(key, default)	1✔
456
457	# Check integrity
458	status = check_run(bids_path, entities, full_path)	1✔
459	if run_in_entities and not status:	1✔
460	print_info(	1✔
461	"Multiple files with same run ID detected, using UUID instead."
462	)
463	entities["run"] = defaults["run"]	1✔
464	status = check_run(bids_path, entities, full_path)	1✔
465	if not status:	1✔
466	print_warn(	1✔
467	f"The generated UUID is not unique: {bids_path}"
468	)
469
470	return entities	1✔
471
472
473	def check_run(	1✔
474	bids_path: File,
475	entities: dict[str],
476	full_path: bool = False) -> bool:
477	"""
478	Scan the folder containing a BIDS file and verify that the run entity
479	associated with the file appears exactly once among all matching files.
480
481	Parameters
482	----------
483	bids_path : File
484	A BIDS file.
485	entities : dict[str]
486	Dictionary of parsed BIDS entities for the file, including the
487	modality.
488	full_path : bool
489	If True, extract entities from the full path instead of only the
490	filename. Default False.
491
492	Returns
493	-------
494	bool
495	True if the run identifier occurs exactly once among all matching
496	files in the folder, False otherwise.
497	"""
498	filename = str(bids_path) if full_path else bids_path.name	1✔
499	ext = "".join(bids_path.suffixes)	1✔
500	entity_pattern = (	1✔
501	r"(?P<entity>(run))"
502	r"-(?P<value>[^_/]+)"
503	)
504	pattern = f"sub-{entities['modality']}{ext}"	1✔
505
506	all_entities = []	1✔
507	for bids_path_ in bids_path.parent.glob(pattern):	1✔
508	filename_ = str(bids_path_) if full_path else bids_path_.name	1✔
509	entities_ = {"filename": filename_}	1✔
510
511	# Extract run entity if present
512	for match in re.finditer(entity_pattern, str(bids_path_)):	1✔
UNCOV 513	entity = match.group("entity")	×
514	value = match.group("value")	×
UNCOV 515	entities_[entity] = value	×
516
517	# If run is missing, generate one
518	if "run" not in entities_:	1✔
519	entities_["run"] = make_run_id(filename)[1]	1✔
520
521	all_entities.append(entities_)	1✔
522
523	# Count how many times the current file's run appears
524	all_run_ids = [item["run"] for item in all_entities]	1✔
525	count = all_run_ids.count(entities["run"])	1✔
526
527	return count == 1	1✔
528
529
530	def make_run_id(	1✔
531	filename: str) -> tuple[str, str]:
532	"""
533	Generate a deterministic identifier and a 5-digit short code from a
534	filename.
535
536	This function computes a UUIDv5 using the URL namespace and the provided
537	filename, converts the UUID to its integer representation, and returns both
538	the full integer-based code and its first five digits. The result is stable
539	and reproducible: the same filename always produces the same values.
540
541	Parameters
542	----------
543	filename : str
544	The filename used as the seed for generating the identifiers.
545
546	Returns
547	-------
548	code : str
549	The full integer representation of the UUIDv5 derived from the
550	filename.
551	short_code : str
552	The first five digits of the UUID-derived code, used as a compact ID.
553	"""
554	code = str(uuid.uuid5(uuid.NAMESPACE_URL, filename).int)	1✔
555	return code, code[:5]	1✔
556
557
558	def sidecar_from_file(	1✔
559	image_file: File) -> File:
560	"""
561	Infers the corresponding JSON sidecar file for a given NIfTI image file.
562
563	This function checks that the input file has a ``.nii.gz`` extension and
564	attempts to locate a sidecar ``.json`` file with the same base name. If
565	either condition fails, it raises a ValueError.
566
567	Parameters
568	----------
569	image_file : File
570	The NIfTI image file for which to infer the JSON sidecar.
571
572	Returns
573	-------
574	sidecar_file : File
575	Path to the inferred JSON sidecar file.
576
577	Raises
578	------
579	ValueError
580	If the input file does not have a `.nii.gz` extension or if the
581	corresponding JSON sidecar file does not exist.
582
583	Examples
584	--------
585	>>> from pathlib import Path
586	>>> from brainprep.utils import sidecar_from_file
587	>>>
588	>>> image_file = Path("/tmp/sub-01_T1w.nii.gz")
589	>>> sidecar_file = Path("/tmp/sub-01_T1w.json")
590	>>> sidecar_file.touch()
591	>>>
592	>>> sidecar_from_file(image_file)
593	PosixPath('/tmp/sub-01_T1w.json')
594	"""
595	if not str(image_file).endswith(".nii.gz"):	1✔
UNCOV 596	raise ValueError(	×
597	f"Input image file must be in NIIGZ format: {image_file}"
598	)
599	sidecar_file = Path(str(image_file).replace(".nii.gz", ".json"))	1✔
600	if not sidecar_file.is_file():	1✔
UNCOV 601	raise ValueError(	×
602	f"Sidecar inferred from input image file not found: {sidecar_file}"
603	)
604	return sidecar_file	1✔
605
606
607	def find_stack_level() -> int:	1✔
608	"""
609	Return the index of the first stack frame outside the ``brainprep``
610	package.
611
612	This function walks backward through the current call stack and finds the
613	first frame whose file path does not belong to the ``brainprep`` package
614	directory. Test files (i.e., files whose names start with ``test_``) are
615	always treated as external. This is useful for producing cleaner warnings
616	and error messages by pointing to user code rather than internal library
617	frames.
618
619	Returns
620	-------
621	int
622	The number of internal frames to skip before reaching user code.
623
624	Notes
625	-----
626	Adapted from the pandas codebase.
627
628	Examples
629	--------
630	>>> import warnings
631	>>> from brainprep.utils import find_stack_level
632	>>>
633	>>> def load_data(path):
634	... if not path.exists():
635	... warnings.warn(
636	... "The provided path does not exist.",
637	... stacklevel=find_stack_level()
638	... )
639	"""
UNCOV 640	import brainprep	×
641
UNCOV 642	pkg_dir = Path(brainprep.__file__).parent	×
643
644	# https://stackoverflow.com/questions/17407119/python-inspect-stack-is-slow
UNCOV 645	frame = inspect.currentframe()	×
UNCOV 646	try:	×
UNCOV 647	n = 0	×
UNCOV 648	while frame:	×
UNCOV 649	filename = inspect.getfile(frame)	×
UNCOV 650	is_test_file = Path(filename).name.startswith("test_")	×
UNCOV 651	in_nilearn_code = filename.startswith(str(pkg_dir))	×
UNCOV 652	if not in_nilearn_code or is_test_file:	×
UNCOV 653	break	×
UNCOV 654	frame = frame.f_back	×
UNCOV 655	n += 1	×
656	finally:
657	# See note in
658	# https://docs.python.org/3/library/inspect.html#inspect.Traceback
UNCOV 659	del frame	×
UNCOV 660	return n	×

neurospin-deepinsight / brainprep / 22624503433

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