11907763573

Committed 19 Nov 2024 06:52AM UTC coverage: 95.648%. Remained the same

Build # 11907763573

Build Type

push

github

Committed by

web-flow

Commit Message

[pre-commit.ci] pre-commit autoupdate (#502)

* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/astral-sh/ruff-pre-commit: v0.7.3 → v0.7.4](https://github.com/astral-sh/ruff-pre-commit/compare/v0.7.3...v0.7.4)

* Update environment-openmpi.yml

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Jan Janssen <jan-janssen@users.noreply.github.com>

Run Details

945 of 988 relevant lines covered (95.65%)

0.96 hits per line

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

94.44

/executorlib/cache/executor.py

import os
from typing import Optional

from executorlib.base.executor import ExecutorBase
from executorlib.cache.shared import execute_tasks_h5
from executorlib.cache.subprocess_spawner import (
    execute_in_subprocess,
    terminate_subprocess,
)
from executorlib.standalone.inputcheck import (
    check_executor,
    check_flux_executor_pmi_mode,
    check_hostname_localhost,
    check_max_workers_and_cores,
    check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
    from executorlib.cache.queue_spawner import execute_with_pysqa
except ImportError:
    # If pysqa is not available fall back to executing tasks in a subprocess
    execute_with_pysqa = execute_in_subprocess


class FileExecutor(ExecutorBase):
    def __init__(
        self,
        cache_directory: str = "cache",
        resource_dict: Optional[dict] = None,
        execute_function: callable = execute_with_pysqa,
        terminate_function: Optional[callable] = None,
        pysqa_config_directory: Optional[str] = None,
        backend: Optional[str] = None,
        disable_dependencies: bool = False,
    ):
        """
        Initialize the FileExecutor.

        Args:
            cache_directory (str, optional): The directory to store cache files. Defaults to "cache".
            resource_dict (dict): A dictionary of resources required by the task. With the following keys:
                              - cores (int): number of MPI cores to be used for each function call
                              - cwd (str/None): current working directory where the parallel python task is executed
            execute_function (callable, optional): The function to execute tasks. Defaults to execute_in_subprocess.
            terminate_function (callable, optional): The function to terminate the tasks.
            pysqa_config_directory (str, optional): path to the pysqa config directory (only for pysqa based backend).
            backend (str, optional): name of the backend used to spawn tasks.
            disable_dependencies (boolean): Disable resolving future objects during the submission.
        """
        super().__init__(max_cores=None)
        default_resource_dict = {
            "cores": 1,
            "cwd": None,
        }
        if resource_dict is None:
            resource_dict = {}
        resource_dict.update(
            {k: v for k, v in default_resource_dict.items() if k not in resource_dict}
        )
        if execute_function == execute_in_subprocess and terminate_function is None:
            terminate_function = terminate_subprocess
        cache_directory_path = os.path.abspath(cache_directory)
        os.makedirs(cache_directory_path, exist_ok=True)
        self._set_process(
            RaisingThread(
                target=execute_tasks_h5,
                kwargs={
                    "future_queue": self._future_queue,
                    "execute_function": execute_function,
                    "cache_directory": cache_directory_path,
                    "resource_dict": resource_dict,
                    "terminate_function": terminate_function,
                    "pysqa_config_directory": pysqa_config_directory,
                    "backend": backend,
                    "disable_dependencies": disable_dependencies,
                },
            )
        )


def create_file_executor(
    max_workers: int = 1,
    backend: str = "flux_submission",
    max_cores: int = 1,
    cache_directory: Optional[str] = None,
    resource_dict: Optional[dict] = None,
    flux_executor=None,
    flux_executor_pmi_mode: Optional[str] = None,
    flux_executor_nesting: bool = False,
    pysqa_config_directory: Optional[str] = None,
    hostname_localhost: Optional[bool] = None,
    block_allocation: bool = False,
    init_function: Optional[callable] = None,
    disable_dependencies: bool = False,
):
    if cache_directory is None:
        cache_directory = "executorlib_cache"
    if block_allocation:
        raise ValueError(
            "The option block_allocation is not available with the pysqa based backend."
        )
    if init_function is not None:
        raise ValueError(
            "The option to specify an init_function is not available with the pysqa based backend."
        )
    check_flux_executor_pmi_mode(flux_executor_pmi_mode=flux_executor_pmi_mode)
    check_max_workers_and_cores(max_cores=max_cores, max_workers=max_workers)
    check_hostname_localhost(hostname_localhost=hostname_localhost)
    check_executor(executor=flux_executor)
    check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)
    return FileExecutor(
        cache_directory=cache_directory,
        resource_dict=resource_dict,
        pysqa_config_directory=pysqa_config_directory,
        backend=backend.split("_submission")[0],
        disable_dependencies=disable_dependencies,
    )

1	import os	1✔
2	from typing import Optional	1✔
3
4	from executorlib.base.executor import ExecutorBase	1✔
5	from executorlib.cache.shared import execute_tasks_h5	1✔
6	from executorlib.cache.subprocess_spawner import (	1✔
7	execute_in_subprocess,
8	terminate_subprocess,
9	)
10	from executorlib.standalone.inputcheck import (	1✔
11	check_executor,
12	check_flux_executor_pmi_mode,
13	check_hostname_localhost,
14	check_max_workers_and_cores,
15	check_nested_flux_executor,
16	)
17	from executorlib.standalone.thread import RaisingThread	1✔
18
19	try:	1✔
20	from executorlib.cache.queue_spawner import execute_with_pysqa	1✔
21	except ImportError:	×
22	# If pysqa is not available fall back to executing tasks in a subprocess
23	execute_with_pysqa = execute_in_subprocess	×
24
25
26	class FileExecutor(ExecutorBase):	1✔
27	def __init__(	1✔
28	self,
29	cache_directory: str = "cache",
30	resource_dict: Optional[dict] = None,
31	execute_function: callable = execute_with_pysqa,
32	terminate_function: Optional[callable] = None,
33	pysqa_config_directory: Optional[str] = None,
34	backend: Optional[str] = None,
35	disable_dependencies: bool = False,
36	):
37	"""
38	Initialize the FileExecutor.
39
40	Args:
41	cache_directory (str, optional): The directory to store cache files. Defaults to "cache".
42	resource_dict (dict): A dictionary of resources required by the task. With the following keys:
43	- cores (int): number of MPI cores to be used for each function call
44	- cwd (str/None): current working directory where the parallel python task is executed
45	execute_function (callable, optional): The function to execute tasks. Defaults to execute_in_subprocess.
46	terminate_function (callable, optional): The function to terminate the tasks.
47	pysqa_config_directory (str, optional): path to the pysqa config directory (only for pysqa based backend).
48	backend (str, optional): name of the backend used to spawn tasks.
49	disable_dependencies (boolean): Disable resolving future objects during the submission.
50	"""
51	super().__init__(max_cores=None)	1✔
52	default_resource_dict = {	1✔
53	"cores": 1,
54	"cwd": None,
55	}
56	if resource_dict is None:	1✔
57	resource_dict = {}	1✔
58	resource_dict.update(	1✔
59	{k: v for k, v in default_resource_dict.items() if k not in resource_dict}
60	)
61	if execute_function == execute_in_subprocess and terminate_function is None:	1✔
62	terminate_function = terminate_subprocess	1✔
63	cache_directory_path = os.path.abspath(cache_directory)	1✔
64	os.makedirs(cache_directory_path, exist_ok=True)	1✔
65	self._set_process(	1✔
66	RaisingThread(
67	target=execute_tasks_h5,
68	kwargs={
69	"future_queue": self._future_queue,
70	"execute_function": execute_function,
71	"cache_directory": cache_directory_path,
72	"resource_dict": resource_dict,
73	"terminate_function": terminate_function,
74	"pysqa_config_directory": pysqa_config_directory,
75	"backend": backend,
76	"disable_dependencies": disable_dependencies,
77	},
78	)
79	)
80
81
82	def create_file_executor(	1✔
83	max_workers: int = 1,
84	backend: str = "flux_submission",
85	max_cores: int = 1,
86	cache_directory: Optional[str] = None,
87	resource_dict: Optional[dict] = None,
88	flux_executor=None,
89	flux_executor_pmi_mode: Optional[str] = None,
90	flux_executor_nesting: bool = False,
91	pysqa_config_directory: Optional[str] = None,
92	hostname_localhost: Optional[bool] = None,
93	block_allocation: bool = False,
94	init_function: Optional[callable] = None,
95	disable_dependencies: bool = False,
96	):
97	if cache_directory is None:	1✔
98	cache_directory = "executorlib_cache"	1✔
99	if block_allocation:	1✔
100	raise ValueError(	1✔
101	"The option block_allocation is not available with the pysqa based backend."
102	)
103	if init_function is not None:	1✔
104	raise ValueError(	1✔
105	"The option to specify an init_function is not available with the pysqa based backend."
106	)
107	check_flux_executor_pmi_mode(flux_executor_pmi_mode=flux_executor_pmi_mode)	1✔
108	check_max_workers_and_cores(max_cores=max_cores, max_workers=max_workers)	1✔
109	check_hostname_localhost(hostname_localhost=hostname_localhost)	1✔
110	check_executor(executor=flux_executor)	1✔
111	check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)	1✔
112	return FileExecutor(	1✔
113	cache_directory=cache_directory,
114	resource_dict=resource_dict,
115	pysqa_config_directory=pysqa_config_directory,
116	backend=backend.split("_submission")[0],
117	disable_dependencies=disable_dependencies,
118	)

pyiron / executorlib / 11907763573

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