9890133465

Committed 11 Jul 2024 10:52AM UTC coverage: 93.776%. Remained the same

Build # 9890133465

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #370 from pyiron/dependabot/pip/matplotlib-3.9.1

Bump matplotlib from 3.9.0 to 3.9.1

Run Details

889 of 948 relevant lines covered (93.78%)

0.94 hits per line

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

94.12

/pympipool/shared/interface.py

import subprocess
from abc import ABC
from typing import Optional

MPI_COMMAND = "mpiexec"
SLURM_COMMAND = "srun"


class BaseInterface(ABC):
    def __init__(self, cwd: str, cores: int = 1, oversubscribe: bool = False):
        self._cwd = cwd
        self._cores = cores
        self._oversubscribe = oversubscribe

    def bootup(
        self,
        command_lst: list[str],
        prefix_name: Optional[str] = None,
        prefix_path: Optional[str] = None,
    ):
        raise NotImplementedError

    def shutdown(self, wait: bool = True):
        raise NotImplementedError

    def poll(self):
        raise NotImplementedError


class SubprocessInterface(BaseInterface):
    def __init__(
        self,
        cwd: Optional[str] = None,
        cores: int = 1,
        oversubscribe: bool = False,
    ):
        super().__init__(
            cwd=cwd,
            cores=cores,
            oversubscribe=oversubscribe,
        )
        self._process = None

    def bootup(
        self,
        command_lst: list[str],
        prefix_name: Optional[str] = None,
        prefix_path: Optional[str] = None,
    ):
        if prefix_name is None and prefix_path is None:
            self._process = subprocess.Popen(
                args=self.generate_command(command_lst=command_lst),
                cwd=self._cwd,
                stdin=subprocess.DEVNULL,
            )
        else:
            import conda_subprocess

            self._process = conda_subprocess.Popen(
                args=self.generate_command(command_lst=command_lst),
                cwd=self._cwd,
                stdin=subprocess.DEVNULL,
                prefix_path=prefix_path,
                prefix_name=prefix_name,
            )

    def generate_command(self, command_lst: list[str]) -> list[str]:
        return command_lst

    def shutdown(self, wait: bool = True):
        self._process.communicate()
        self._process.terminate()
        if wait:
            self._process.wait()
        self._process = None

    def poll(self):
        return self._process is not None and self._process.poll() is None


class MpiExecInterface(SubprocessInterface):
    def generate_command(self, command_lst: list[str]):
        command_prepend_lst = generate_mpiexec_command(
            cores=self._cores,
            oversubscribe=self._oversubscribe,
        )
        return super().generate_command(
            command_lst=command_prepend_lst + command_lst,
        )


class SrunInterface(SubprocessInterface):
    def __init__(
        self,
        cwd: Optional[str] = None,
        cores: int = 1,
        threads_per_core: int = 1,
        gpus_per_core: int = 0,
        oversubscribe: bool = False,
        command_line_argument_lst: list[str] = [],
    ):
        super().__init__(
            cwd=cwd,
            cores=cores,
            oversubscribe=oversubscribe,
        )
        self._threads_per_core = threads_per_core
        self._gpus_per_core = gpus_per_core
        self._command_line_argument_lst = command_line_argument_lst

    def generate_command(self, command_lst: list[str]) -> list[str]:
        command_prepend_lst = generate_slurm_command(
            cores=self._cores,
            cwd=self._cwd,
            threads_per_core=self._threads_per_core,
            gpus_per_core=self._gpus_per_core,
            oversubscribe=self._oversubscribe,
            command_line_argument_lst=self._command_line_argument_lst,
        )
        return super().generate_command(
            command_lst=command_prepend_lst + command_lst,
        )


def generate_mpiexec_command(cores: int, oversubscribe: bool = False) -> list[str]:
    if cores == 1:
        return []
    else:
        command_prepend_lst = [MPI_COMMAND, "-n", str(cores)]
        if oversubscribe:
            command_prepend_lst += ["--oversubscribe"]
        return command_prepend_lst


def generate_slurm_command(
    cores: int,
    cwd: str,
    threads_per_core: int = 1,
    gpus_per_core: int = 0,
    oversubscribe: bool = False,
    command_line_argument_lst: list[str] = [],
) -> list[str]:
    command_prepend_lst = [SLURM_COMMAND, "-n", str(cores)]
    if cwd is not None:
        command_prepend_lst += ["-D", cwd]
    if threads_per_core > 1:
        command_prepend_lst += ["--cpus-per-task" + str(threads_per_core)]
    if gpus_per_core > 0:
        command_prepend_lst += ["--gpus-per-task=" + str(gpus_per_core)]
    if oversubscribe:
        command_prepend_lst += ["--oversubscribe"]
    if len(command_line_argument_lst) > 0:
        command_prepend_lst += command_line_argument_lst
    return command_prepend_lst

1	import subprocess	1✔
2	from abc import ABC	1✔
3	from typing import Optional	1✔
4
5	MPI_COMMAND = "mpiexec"	1✔
6	SLURM_COMMAND = "srun"	1✔
7
8
9	class BaseInterface(ABC):	1✔
10	def __init__(self, cwd: str, cores: int = 1, oversubscribe: bool = False):	1✔
11	self._cwd = cwd	1✔
12	self._cores = cores	1✔
13	self._oversubscribe = oversubscribe	1✔
14
15	def bootup(	1✔
16	self,
17	command_lst: list[str],
18	prefix_name: Optional[str] = None,
19	prefix_path: Optional[str] = None,
20	):
21	raise NotImplementedError	×
22
23	def shutdown(self, wait: bool = True):	1✔
24	raise NotImplementedError	×
25
26	def poll(self):	1✔
27	raise NotImplementedError	×
28
29
30	class SubprocessInterface(BaseInterface):	1✔
31	def __init__(	1✔
32	self,
33	cwd: Optional[str] = None,
34	cores: int = 1,
35	oversubscribe: bool = False,
36	):
37	super().__init__(	1✔
38	cwd=cwd,
39	cores=cores,
40	oversubscribe=oversubscribe,
41	)
42	self._process = None	1✔
43
44	def bootup(	1✔
45	self,
46	command_lst: list[str],
47	prefix_name: Optional[str] = None,
48	prefix_path: Optional[str] = None,
49	):
50	if prefix_name is None and prefix_path is None:	1✔
51	self._process = subprocess.Popen(	1✔
52	args=self.generate_command(command_lst=command_lst),
53	cwd=self._cwd,
54	stdin=subprocess.DEVNULL,
55	)
56	else:
57	import conda_subprocess	1✔
58
59	self._process = conda_subprocess.Popen(	1✔
60	args=self.generate_command(command_lst=command_lst),
61	cwd=self._cwd,
62	stdin=subprocess.DEVNULL,
63	prefix_path=prefix_path,
64	prefix_name=prefix_name,
65	)
66
67	def generate_command(self, command_lst: list[str]) -> list[str]:	1✔
68	return command_lst	1✔
69
70	def shutdown(self, wait: bool = True):	1✔
71	self._process.communicate()	1✔
72	self._process.terminate()	1✔
73	if wait:	1✔
74	self._process.wait()	1✔
75	self._process = None	1✔
76
77	def poll(self):	1✔
78	return self._process is not None and self._process.poll() is None	1✔
79
80
81	class MpiExecInterface(SubprocessInterface):	1✔
82	def generate_command(self, command_lst: list[str]):	1✔
83	command_prepend_lst = generate_mpiexec_command(	1✔
84	cores=self._cores,
85	oversubscribe=self._oversubscribe,
86	)
87	return super().generate_command(	1✔
88	command_lst=command_prepend_lst + command_lst,
89	)
90
91
92	class SrunInterface(SubprocessInterface):	1✔
93	def __init__(	1✔
94	self,
95	cwd: Optional[str] = None,
96	cores: int = 1,
97	threads_per_core: int = 1,
98	gpus_per_core: int = 0,
99	oversubscribe: bool = False,
100	command_line_argument_lst: list[str] = [],
101	):
102	super().__init__(	1✔
103	cwd=cwd,
104	cores=cores,
105	oversubscribe=oversubscribe,
106	)
107	self._threads_per_core = threads_per_core	1✔
108	self._gpus_per_core = gpus_per_core	1✔
109	self._command_line_argument_lst = command_line_argument_lst	1✔
110
111	def generate_command(self, command_lst: list[str]) -> list[str]:	1✔
112	command_prepend_lst = generate_slurm_command(	1✔
113	cores=self._cores,
114	cwd=self._cwd,
115	threads_per_core=self._threads_per_core,
116	gpus_per_core=self._gpus_per_core,
117	oversubscribe=self._oversubscribe,
118	command_line_argument_lst=self._command_line_argument_lst,
119	)
120	return super().generate_command(	1✔
121	command_lst=command_prepend_lst + command_lst,
122	)
123
124
125	def generate_mpiexec_command(cores: int, oversubscribe: bool = False) -> list[str]:	1✔
126	if cores == 1:	1✔
127	return []	1✔
128	else:
129	command_prepend_lst = [MPI_COMMAND, "-n", str(cores)]	1✔
130	if oversubscribe:	1✔
131	command_prepend_lst += ["--oversubscribe"]	1✔
132	return command_prepend_lst	1✔
133
134
135	def generate_slurm_command(	1✔
136	cores: int,
137	cwd: str,
138	threads_per_core: int = 1,
139	gpus_per_core: int = 0,
140	oversubscribe: bool = False,
141	command_line_argument_lst: list[str] = [],
142	) -> list[str]:
143	command_prepend_lst = [SLURM_COMMAND, "-n", str(cores)]	1✔
144	if cwd is not None:	1✔
145	command_prepend_lst += ["-D", cwd]	1✔
146	if threads_per_core > 1:	1✔
147	command_prepend_lst += ["--cpus-per-task" + str(threads_per_core)]	×
148	if gpus_per_core > 0:	1✔
149	command_prepend_lst += ["--gpus-per-task=" + str(gpus_per_core)]	1✔
150	if oversubscribe:	1✔
151	command_prepend_lst += ["--oversubscribe"]	1✔
152	if len(command_line_argument_lst) > 0:	1✔
153	command_prepend_lst += command_line_argument_lst	1✔
154	return command_prepend_lst	1✔

pyiron / pympipool / 9890133465

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