11750386682

Committed 08 Nov 2024 10:19PM UTC coverage: 95.122% (-0.08%) from 95.203%

Build # 11750386682

Build Type

push

github

Committed by

web-flow

Commit Message

Fix working directory (#481)

* Fix working directory

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* shorter fix

* even shorted

* another fix

* windows fix

---------

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

Run Details

7 of 11 new or added lines in 2 files covered. (63.64%)

897 of 943 relevant lines covered (95.12%)

0.95 hits per line

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

92.59

/executorlib/standalone/cache/queue.py

import os
import subprocess
from typing import List, Optional, Union

from pysqa import QueueAdapter


def execute_with_pysqa(
    command: str,
    resource_dict: dict,
    task_dependent_lst: List[int] = [],
    config_directory: Optional[str] = None,
    backend: Optional[str] = None,
    cache_directory: Optional[str] = None,
) -> int:
    """
    Execute a command by submitting it to the queuing system

    Args:
        command (list): The command to be executed.
        task_dependent_lst (list): A list of subprocesses that the current subprocess depends on. Defaults to [].
        resource_dict (dict): resource dictionary, which defines the resources used for the execution of the function.
                              Example resource dictionary: {
                                  cwd: None,
                              }
        config_directory (str, optional): path to the config directory.
        backend (str, optional): name of the backend used to spawn tasks.
        cache_directory (str): The directory to store the HDF5 files.

    Returns:
        int: queuing system ID
    """
    if resource_dict is None:
        resource_dict = {}
    if "cwd" in resource_dict and resource_dict["cwd"] is not None:
        cwd = resource_dict["cwd"]
    else:
        cwd = cache_directory
    qa = QueueAdapter(
        directory=config_directory,
        queue_type=backend,
        execute_command=_pysqa_execute_command,
    )
    submit_kwargs = {
        "command": " ".join(command),
        "dependency_list": [str(qid) for qid in task_dependent_lst],
        "working_directory": os.path.abspath(cwd),
    }
    if "cwd" in resource_dict:
        del resource_dict["cwd"]
    unsupported_keys = [
        "threads_per_core",
        "gpus_per_core",
        "openmpi_oversubscribe",
        "slurm_cmd_args",
    ]
    for k in unsupported_keys:
        if k in resource_dict:
            del resource_dict[k]
    submit_kwargs.update(resource_dict)
    return qa.submit_job(**submit_kwargs)


def _pysqa_execute_command(
    commands: str,
    working_directory: Optional[str] = None,
    split_output: bool = True,
    shell: bool = False,
    error_filename: str = "pysqa.err",
) -> Union[str, List[str]]:
    """
    A wrapper around the subprocess.check_output function. Modified from pysqa to raise an exception if the subprocess
    fails to submit the job to the queue.

    Args:
        commands (str): The command(s) to be executed on the command line
        working_directory (str, optional): The directory where the command is executed. Defaults to None.
        split_output (bool, optional): Boolean flag to split newlines in the output. Defaults to True.
        shell (bool, optional): Additional switch to convert commands to a single string. Defaults to False.
        error_filename (str, optional): In case the execution fails, the output is written to this file. Defaults to "pysqa.err".

    Returns:
        Union[str, List[str]]: Output of the shell command either as a string or as a list of strings
    """
    if shell and isinstance(commands, list):
        commands = " ".join(commands)
    out = subprocess.check_output(
        commands,
        cwd=working_directory,
        stderr=subprocess.STDOUT,
        universal_newlines=True,
        shell=not isinstance(commands, list),
    )
    if out is not None and split_output:
        return out.split("\n")
    else:
        return out

1	import os	1✔
2	import subprocess	1✔
3	from typing import List, Optional, Union	1✔
4
5	from pysqa import QueueAdapter	1✔
6
7
8	def execute_with_pysqa(	1✔
9	command: str,
10	resource_dict: dict,
11	task_dependent_lst: List[int] = [],
12	config_directory: Optional[str] = None,
13	backend: Optional[str] = None,
14	cache_directory: Optional[str] = None,
15	) -> int:
16	"""
17	Execute a command by submitting it to the queuing system
18
19	Args:
20	command (list): The command to be executed.
21	task_dependent_lst (list): A list of subprocesses that the current subprocess depends on. Defaults to [].
22	resource_dict (dict): resource dictionary, which defines the resources used for the execution of the function.
23	Example resource dictionary: {
24	cwd: None,
25	}
26	config_directory (str, optional): path to the config directory.
27	backend (str, optional): name of the backend used to spawn tasks.
28	cache_directory (str): The directory to store the HDF5 files.
29
30	Returns:
31	int: queuing system ID
32	"""
33	if resource_dict is None:	1✔
NEW 34	resource_dict = {}	×
35	if "cwd" in resource_dict and resource_dict["cwd"] is not None:	1✔
36	cwd = resource_dict["cwd"]	1✔
37	else:
NEW 38	cwd = cache_directory	×
39	qa = QueueAdapter(	1✔
40	directory=config_directory,
41	queue_type=backend,
42	execute_command=_pysqa_execute_command,
43	)
44	submit_kwargs = {	1✔
45	"command": " ".join(command),
46	"dependency_list": [str(qid) for qid in task_dependent_lst],
47	"working_directory": os.path.abspath(cwd),
48	}
49	if "cwd" in resource_dict:	1✔
50	del resource_dict["cwd"]	1✔
51	unsupported_keys = [	1✔
52	"threads_per_core",
53	"gpus_per_core",
54	"openmpi_oversubscribe",
55	"slurm_cmd_args",
56	]
57	for k in unsupported_keys:	1✔
58	if k in resource_dict:	1✔
59	del resource_dict[k]	1✔
60	submit_kwargs.update(resource_dict)	1✔
61	return qa.submit_job(**submit_kwargs)	1✔
62
63
64	def _pysqa_execute_command(	1✔
65	commands: str,
66	working_directory: Optional[str] = None,
67	split_output: bool = True,
68	shell: bool = False,
69	error_filename: str = "pysqa.err",
70	) -> Union[str, List[str]]:
71	"""
72	A wrapper around the subprocess.check_output function. Modified from pysqa to raise an exception if the subprocess
73	fails to submit the job to the queue.
74
75	Args:
76	commands (str): The command(s) to be executed on the command line
77	working_directory (str, optional): The directory where the command is executed. Defaults to None.
78	split_output (bool, optional): Boolean flag to split newlines in the output. Defaults to True.
79	shell (bool, optional): Additional switch to convert commands to a single string. Defaults to False.
80	error_filename (str, optional): In case the execution fails, the output is written to this file. Defaults to "pysqa.err".
81
82	Returns:
83	Union[str, List[str]]: Output of the shell command either as a string or as a list of strings
84	"""
85	if shell and isinstance(commands, list):	1✔
86	commands = " ".join(commands)	1✔
87	out = subprocess.check_output(	1✔
88	commands,
89	cwd=working_directory,
90	stderr=subprocess.STDOUT,
91	universal_newlines=True,
92	shell=not isinstance(commands, list),
93	)
94	if out is not None and split_output:	1✔
95	return out.split("\n")	1✔
96	else:
97	return out	1✔

pyiron / executorlib / 11750386682

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