18969220961

Committed 31 Oct 2025 10:04AM UTC coverage: 96.881% (+4.3%) from 92.572%

Build # 18969220961

Build Type

Pull #281

github

Committed by

web-flow

Commit Message

Merge a474cda75 into 1406de2c3

Pull Request Pull Request #281: [DRAFT] Job rework

Run Details

566 of 572 new or added lines in 6 files covered. (98.95%)

10 existing lines in 1 file now uncovered.

3852 of 3976 relevant lines covered (96.88%)

0.97 hits per line

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

64.49

/src/osekit/utils/core_utils.py

from __future__ import annotations

import json
import os
import shutil
import time
from bisect import bisect
from importlib.resources import as_file
from importlib.util import find_spec
from pathlib import Path
from typing import NamedTuple

try:
    import tomllib
except ModuleNotFoundError:
    import tomli as tomllib


from osekit.config import global_logging_context as glc
from osekit.config import print_logger

_is_grp_supported = bool(find_spec("grp"))


@glc.set_logger(print_logger)
def display_folder_storage_info(dir_path: str) -> None:
    """Print a detail of the current disk usage."""
    usage = shutil.disk_usage(dir_path)

    def str_usage(key: str, value: int) -> str:
        return f"{f'{key} storage space:':<30}{f'{round(value / (1024**4), 1)} TB':>10}"

    total = str_usage("Total", usage.total)
    used = str_usage("Used", usage.used)
    free = str_usage("Available", usage.free)
    glc.logger.info("%s\n%s\n%s\n%s", total, used, f"{'-' * 30:^40}", free)


def read_config(raw_config: str | dict | Path) -> NamedTuple:
    """Read the given configuration file or dict and converts it to a namedtuple. Only TOML and JSON formats are accepted for now.

    Parameter
    ---------
    raw_config : `str` or `Path` or `dict`
        The path of the configuration file, or the dict object containing the configuration.

    Returns
    -------
    config : `namedtuple`
        The configuration as a `namedtuple` object.

    Raises
    ------
    FileNotFoundError
        Raised if the raw_config is a string that does not correspond to a valid path, or the raw_config file is not in TOML, JSON or YAML formats.
    TypeError
        Raised if the raw_config is anything else than a string, a PurePath or a dict.
    NotImplementedError
        Raised if the raw_config file is in YAML format

    """
    match raw_config:
        case Path():
            with as_file(raw_config) as input_config:
                raw_config = input_config

        case str():
            if not Path(raw_config).is_file:
                raise FileNotFoundError(
                    f"The configuration file {raw_config} does not exist.",
                )

        case dict():
            pass
        case _:
            raise TypeError(
                "The raw_config must be either of type str, dict or Traversable.",
            )

    if not isinstance(raw_config, dict):
        with open(raw_config, "rb") as input_config:
            match Path(raw_config).suffix:
                case ".toml":
                    raw_config = tomllib.load(input_config)
                case ".json":
                    raw_config = json.load(input_config)
                case ".yaml":
                    raise NotImplementedError(
                        "YAML support will eventually get there (unfortunately)",
                    )
                case _:
                    raise FileNotFoundError(
                        f"The provided configuration file extension ({Path(raw_config).suffix} is not a valid extension. Please use .toml or .json files.",
                    )

    return raw_config


def chmod_if_needed(path: Path, mode: int) -> None:
    """Change the permission of a path if user doesn't have read and write access.

    Parameters
    ----------
    path: Path
        Path of the file or folder in which permission should be changed.
    mode: int
        Permissions as used by os.chmod()

    """
    if not _is_grp_supported:
        return
    if all(os.access(path, p) for p in (os.R_OK, os.W_OK)):
        return

    try:
        path.chmod(mode)
    except PermissionError as e:
        message = (
            f"You do not have the permission to write to {path}, "
            "nor to change its permissions."
        )
        glc.logger.error(message)
        raise PermissionError(message) from e


def change_owner_group(path: Path, owner_group: str) -> None:
    """Change the owner group of the given path.

    Parameters
    ----------
    path:
        Path of which the owner group should be changed.
    owner_group:
        The new owner group.
        A warning is logged if the grp module is supported (Unix os) but
        no owner_group is passed.

    """
    if not _is_grp_supported:
        return
    if owner_group is None:
        glc.logger.warning("You did not set the group owner of the dataset.")
        return
    if path.group() == owner_group:
        return
    glc.logger.debug("Setting osekit permission to the dataset..")

    try:
        import grp

        gid = grp.getgrnam(owner_group).gr_gid
    except KeyError as e:
        message = f"Group {owner_group} does not exist."
        glc.logger.error(message)
        raise KeyError(message) from e

    try:
        os.chown(path, -1, gid)
    except PermissionError as e:
        message = (
            f"You do not have the permission to change the owner of {path}."
            f"The group owner has not been changed "
            f"from {path.group()} to {owner_group}."
        )
        glc.logger.error(message)
        raise PermissionError(message) from e


def get_umask() -> int:
    """Return the current umask."""
    umask = os.umask(0)
    os.umask(umask)
    return umask


def file_indexes_per_batch(
    total_nb_files: int,
    nb_batches: int,
) -> list[tuple[int, int]]:
    """Compute the start and stop file indexes for each batch.

    The number of files is equitably distributed among batches.
    Example: 10 files distributed among 4 batches will lead to
    batches indexes [(0,3), (3,6), (6,8), (8,10)].

    Parameters
    ----------
    total_nb_files: int
        Number of files processed by ball batches
    nb_batches: int
        Number of batches in the analysis

    Returns
    -------
    list[tuple[int,int]]:
    A list of tuples representing the start and stop index of files processed by each
    batch in the analysis.

    Examples
    --------
    >>> file_indexes_per_batch(10,4)
    [(0, 3), (3, 6), (6, 8), (8, 10)]
    >>> file_indexes_per_batch(1448,10)
    [(0, 145), (145, 290), (290, 435), (435, 580), (580, 725), (725, 870), (870, 1015), (1015, 1160), (1160, 1304), (1304, 1448)]

    """  # noqa: E501
    batch_lengths = [
        length
        for length in nb_files_per_batch(total_nb_files, nb_batches)
        if length > 0
    ]
    return [
        (sum(batch_lengths[:b]), sum(batch_lengths[:b]) + batch_lengths[b])
        for b in range(len(batch_lengths))
    ]


def nb_files_per_batch(total_nb_files: int, nb_batches: int) -> list[int]:
    """Compute the number of files processed by each batch in the analysis.

    The number of files is equitably distributed among batches.
    Example: 10 files distributed among 4 batches will lead to
    batches containing [3,3,2,2] files.

    Parameters
    ----------
    total_nb_files: int
        Number of files processed by ball batches
    nb_batches: int
        Number of batches in the analysis

    Returns
    -------
    list(int):
    A list representing the number of files processed by each batch in the analysis.

    Examples
    --------
    >>> nb_files_per_batch(10,4)
    [3, 3, 2, 2]
    >>> nb_files_per_batch(1448,10)
    [145, 145, 145, 145, 145, 145, 145, 145, 144, 144]

    """
    return [
        total_nb_files // nb_batches + (1 if i < total_nb_files % nb_batches else 0)
        for i in range(nb_batches)
    ]


def locked(lock_file: Path) -> callable:
    """Use a lock file for managing priorities between processes.

    If the specified lock file already exists, the decorated function execution will be
    suspended until the lock file is removed.

    The lock_file will then be created before the execution of the decorated function,
    and removed once the function has been executed.

    Parameters
    ----------
    lock_file: Path
        Path to the lock file.

    """

    def inner(func: callable) -> callable:
        def wrapper(*args: any, **kwargs: any) -> any:
            # Wait for the lock to be released
            while lock_file.exists():
                time.sleep(1)

            # Create lock file
            lock_file.touch()

            r = func(*args, **kwargs)

            # Release lock file
            lock_file.unlink()

            return r

        return wrapper

    return inner


def get_closest_value_index(target: float, values: list[float]) -> int:
    """Get the index of the closest value in a list from a target value.

    Parameters
    ----------
    target: float
        Target value from which the closest value is to be found.
    values: list[float]
        List of values in which the closest value from target is
        to be found.

    Returns
    -------
    int
        Index of the closest value from target in values.

    """
    closest_upper_index = min(bisect(values, target), len(values) - 1)
    closest_lower_index = max(0, closest_upper_index - 1)
    return min(
        (closest_lower_index, closest_upper_index),
        key=lambda i: abs(values[i] - target),
    )

1	from __future__ import annotations	1✔
2
3	import json	1✔
4	import os	1✔
5	import shutil	1✔
6	import time	1✔
7	from bisect import bisect	1✔
8	from importlib.resources import as_file	1✔
9	from importlib.util import find_spec	1✔
10	from pathlib import Path	1✔
11	from typing import NamedTuple	1✔
12
13	try:	1✔
14	import tomllib	1✔
15	except ModuleNotFoundError:	×
16	import tomli as tomllib	×
17
18
19	from osekit.config import global_logging_context as glc	1✔
20	from osekit.config import print_logger	1✔
21
22	_is_grp_supported = bool(find_spec("grp"))	1✔
23
24
25	@glc.set_logger(print_logger)	1✔
26	def display_folder_storage_info(dir_path: str) -> None:	1✔
27	"""Print a detail of the current disk usage."""
28	usage = shutil.disk_usage(dir_path)	×
29
30	def str_usage(key: str, value: int) -> str:	×
31	return f"{f'{key} storage space:':<30}{f'{round(value / (1024**4), 1)} TB':>10}"	×
32
33	total = str_usage("Total", usage.total)	×
34	used = str_usage("Used", usage.used)	×
35	free = str_usage("Available", usage.free)	×
36	glc.logger.info("%s\n%s\n%s\n%s", total, used, f"{'-' * 30:^40}", free)	×
37
38
39	def read_config(raw_config: str \| dict \| Path) -> NamedTuple:	1✔
40	"""Read the given configuration file or dict and converts it to a namedtuple. Only TOML and JSON formats are accepted for now.
41
42	Parameter
43	---------
44	raw_config : `str` or `Path` or `dict`
45	The path of the configuration file, or the dict object containing the configuration.
46
47	Returns
48	-------
49	config : `namedtuple`
50	The configuration as a `namedtuple` object.
51
52	Raises
53	------
54	FileNotFoundError
55	Raised if the raw_config is a string that does not correspond to a valid path, or the raw_config file is not in TOML, JSON or YAML formats.
56	TypeError
57	Raised if the raw_config is anything else than a string, a PurePath or a dict.
58	NotImplementedError
59	Raised if the raw_config file is in YAML format
60
61	"""
UNCOV 62	match raw_config:	×
UNCOV 63	case Path():	×
UNCOV 64	with as_file(raw_config) as input_config:	×
UNCOV 65	raw_config = input_config	×
66
67	case str():	×
68	if not Path(raw_config).is_file:	×
69	raise FileNotFoundError(	×
70	f"The configuration file {raw_config} does not exist.",
71	)
72
73	case dict():	×
74	pass	×
75	case _:	×
76	raise TypeError(	×
77	"The raw_config must be either of type str, dict or Traversable.",
78	)
79
UNCOV 80	if not isinstance(raw_config, dict):	×
UNCOV 81	with open(raw_config, "rb") as input_config:	×
UNCOV 82	match Path(raw_config).suffix:	×
UNCOV 83	case ".toml":	×
UNCOV 84	raw_config = tomllib.load(input_config)	×
85	case ".json":	×
86	raw_config = json.load(input_config)	×
87	case ".yaml":	×
88	raise NotImplementedError(	×
89	"YAML support will eventually get there (unfortunately)",
90	)
91	case _:	×
92	raise FileNotFoundError(	×
93	f"The provided configuration file extension ({Path(raw_config).suffix} is not a valid extension. Please use .toml or .json files.",
94	)
95
UNCOV 96	return raw_config	×
97
98
99	def chmod_if_needed(path: Path, mode: int) -> None:	1✔
100	"""Change the permission of a path if user doesn't have read and write access.
101
102	Parameters
103	----------
104	path: Path
105	Path of the file or folder in which permission should be changed.
106	mode: int
107	Permissions as used by os.chmod()
108
109	"""
110	if not _is_grp_supported:	1✔
111	return	1✔
112	if all(os.access(path, p) for p in (os.R_OK, os.W_OK)):	1✔
113	return	1✔
114
115	try:	1✔
116	path.chmod(mode)	1✔
117	except PermissionError as e:	1✔
118	message = (	1✔
119	f"You do not have the permission to write to {path}, "
120	"nor to change its permissions."
121	)
122	glc.logger.error(message)	1✔
123	raise PermissionError(message) from e	1✔
124
125
126	def change_owner_group(path: Path, owner_group: str) -> None:	1✔
127	"""Change the owner group of the given path.
128
129	Parameters
130	----------
131	path:
132	Path of which the owner group should be changed.
133	owner_group:
134	The new owner group.
135	A warning is logged if the grp module is supported (Unix os) but
136	no owner_group is passed.
137
138	"""
139	if not _is_grp_supported:	1✔
140	return	1✔
141	if owner_group is None:	1✔
142	glc.logger.warning("You did not set the group owner of the dataset.")	×
143	return	×
144	if path.group() == owner_group:	1✔
145	return	×
146	glc.logger.debug("Setting osekit permission to the dataset..")	1✔
147
148	try:	1✔
149	import grp	1✔
150
151	gid = grp.getgrnam(owner_group).gr_gid	1✔
152	except KeyError as e:	1✔
153	message = f"Group {owner_group} does not exist."	1✔
154	glc.logger.error(message)	1✔
155	raise KeyError(message) from e	1✔
156
157	try:	1✔
158	os.chown(path, -1, gid)	1✔
159	except PermissionError as e:	1✔
160	message = (	1✔
161	f"You do not have the permission to change the owner of {path}."
162	f"The group owner has not been changed "
163	f"from {path.group()} to {owner_group}."
164	)
165	glc.logger.error(message)	1✔
166	raise PermissionError(message) from e	1✔
167
168
169	def get_umask() -> int:	1✔
170	"""Return the current umask."""
171	umask = os.umask(0)	×
172	os.umask(umask)	×
173	return umask	×
174
175
176	def file_indexes_per_batch(	1✔
177	total_nb_files: int,
178	nb_batches: int,
179	) -> list[tuple[int, int]]:
180	"""Compute the start and stop file indexes for each batch.
181
182	The number of files is equitably distributed among batches.
183	Example: 10 files distributed among 4 batches will lead to
184	batches indexes [(0,3), (3,6), (6,8), (8,10)].
185
186	Parameters
187	----------
188	total_nb_files: int
189	Number of files processed by ball batches
190	nb_batches: int
191	Number of batches in the analysis
192
193	Returns
194	-------
195	list[tuple[int,int]]:
196	A list of tuples representing the start and stop index of files processed by each
197	batch in the analysis.
198
199	Examples
200	--------
201	>>> file_indexes_per_batch(10,4)
202	[(0, 3), (3, 6), (6, 8), (8, 10)]
203	>>> file_indexes_per_batch(1448,10)
204	[(0, 145), (145, 290), (290, 435), (435, 580), (580, 725), (725, 870), (870, 1015), (1015, 1160), (1160, 1304), (1304, 1448)]
205
206	""" # noqa: E501
207	batch_lengths = [	1✔
208	length
209	for length in nb_files_per_batch(total_nb_files, nb_batches)
210	if length > 0
211	]
212	return [	1✔
213	(sum(batch_lengths[:b]), sum(batch_lengths[:b]) + batch_lengths[b])
214	for b in range(len(batch_lengths))
215	]
216
217
218	def nb_files_per_batch(total_nb_files: int, nb_batches: int) -> list[int]:	1✔
219	"""Compute the number of files processed by each batch in the analysis.
220
221	The number of files is equitably distributed among batches.
222	Example: 10 files distributed among 4 batches will lead to
223	batches containing [3,3,2,2] files.
224
225	Parameters
226	----------
227	total_nb_files: int
228	Number of files processed by ball batches
229	nb_batches: int
230	Number of batches in the analysis
231
232	Returns
233	-------
234	list(int):
235	A list representing the number of files processed by each batch in the analysis.
236
237	Examples
238	--------
239	>>> nb_files_per_batch(10,4)
240	[3, 3, 2, 2]
241	>>> nb_files_per_batch(1448,10)
242	[145, 145, 145, 145, 145, 145, 145, 145, 144, 144]
243
244	"""
245	return [	1✔
246	total_nb_files // nb_batches + (1 if i < total_nb_files % nb_batches else 0)
247	for i in range(nb_batches)
248	]
249
250
251	def locked(lock_file: Path) -> callable:	1✔
252	"""Use a lock file for managing priorities between processes.
253
254	If the specified lock file already exists, the decorated function execution will be
255	suspended until the lock file is removed.
256
257	The lock_file will then be created before the execution of the decorated function,
258	and removed once the function has been executed.
259
260	Parameters
261	----------
262	lock_file: Path
263	Path to the lock file.
264
265	"""
266
267	def inner(func: callable) -> callable:	1✔
268	def wrapper(args: any, *kwargs: any) -> any:	1✔
269	# Wait for the lock to be released
270	while lock_file.exists():	1✔
271	time.sleep(1)	1✔
272
273	# Create lock file
274	lock_file.touch()	1✔
275
276	r = func(args, *kwargs)	1✔
277
278	# Release lock file
279	lock_file.unlink()	1✔
280
281	return r	1✔
282
283	return wrapper	1✔
284
285	return inner	1✔
286
287
288	def get_closest_value_index(target: float, values: list[float]) -> int:	1✔
289	"""Get the index of the closest value in a list from a target value.
290
291	Parameters
292	----------
293	target: float
294	Target value from which the closest value is to be found.
295	values: list[float]
296	List of values in which the closest value from target is
297	to be found.
298
299	Returns
300	-------
301	int
302	Index of the closest value from target in values.
303
304	"""
305	closest_upper_index = min(bisect(values, target), len(values) - 1)	1✔
306	closest_lower_index = max(0, closest_upper_index - 1)	1✔
307	return min(	1✔
308	(closest_lower_index, closest_upper_index),
309	key=lambda i: abs(values[i] - target),
310	)

Project-OSmOSE / OSEkit / 18969220961

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