13116578816

Committed 03 Feb 2025 03:11PM UTC coverage: 96.536%. Remained the same

Build # 13116578816

Build Type

Pull #555

github

Committed by

web-flow

Commit Message

Merge 70bacefb9 into 5ec2a2015

Pull Request Pull Request #555: Add linting

Run Details

14 of 15 new or added lines in 7 files covered. (93.33%)

17 existing lines in 7 files now uncovered.

1059 of 1097 relevant lines covered (96.54%)

0.97 hits per line

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

93.18

/executorlib/standalone/interactive/spawner.py

import subprocess
from abc import ABC
from typing import Optional

MPI_COMMAND = "mpiexec"


class BaseSpawner(ABC):
    def __init__(
        self,
        cwd: Optional[str] = None,
        cores: int = 1,
        openmpi_oversubscribe: bool = False,
    ):
        """
        Base class for interface implementations.

        Args:
            cwd (str): The current working directory.
            cores (int, optional): The number of cores to use. Defaults to 1.
            openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
        """
        self._cwd = cwd
        self._cores = cores
        self._openmpi_oversubscribe = openmpi_oversubscribe

    def bootup(
        self,
        command_lst: list[str],
    ):
        """
        Method to start the interface.

        Args:
            command_lst (list[str]): The command list to execute.
        """
        raise NotImplementedError

    def shutdown(self, wait: bool = True):
        """
        Method to shutdown the interface.

        Args:
            wait (bool, optional): Whether to wait for the interface to shutdown. Defaults to True.
        """
        raise NotImplementedError

    def poll(self):
        """
        Method to check if the interface is running.

        Returns:
            bool: True if the interface is running, False otherwise.
        """
        raise NotImplementedError


class SubprocessSpawner(BaseSpawner):
    def __init__(
        self,
        cwd: Optional[str] = None,
        cores: int = 1,
        openmpi_oversubscribe: bool = False,
        threads_per_core: int = 1,
    ):
        """
        Subprocess interface implementation.

        Args:
            cwd (str, optional): The current working directory. Defaults to None.
            cores (int, optional): The number of cores to use. Defaults to 1.
            threads_per_core (int, optional): The number of threads per core. Defaults to 1.
            oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
        """
        super().__init__(
            cwd=cwd,
            cores=cores,
            openmpi_oversubscribe=openmpi_oversubscribe,
        )
        self._process: Optional[subprocess.Popen] = None
        self._threads_per_core = threads_per_core

    def bootup(
        self,
        command_lst: list[str],
    ):
        """
        Method to start the subprocess interface.

        Args:
            command_lst (list[str]): The command list to execute.
        """
        self._process = subprocess.Popen(
            args=self.generate_command(command_lst=command_lst),
            cwd=self._cwd,
            stdin=subprocess.DEVNULL,
        )

    def generate_command(self, command_lst: list[str]) -> list[str]:
        """
        Method to generate the command list.

        Args:
            command_lst (list[str]): The command list.

        Returns:
            list[str]: The generated command list.
        """
        return command_lst

    def shutdown(self, wait: bool = True):
        """
        Method to shutdown the subprocess interface.

        Args:
            wait (bool, optional): Whether to wait for the interface to shutdown. Defaults to True.
        """
        if self._process is not None:
            self._process.communicate()
            self._process.terminate()
            if wait:
                self._process.wait()
        self._process = None

    def poll(self) -> bool:
        """
        Method to check if the subprocess interface is running.

        Returns:
            bool: True if the interface is running, False otherwise.
        """
        return self._process is not None and self._process.poll() is None


class MpiExecSpawner(SubprocessSpawner):
    def generate_command(self, command_lst: list[str]) -> list[str]:
        """
        Generate the command list for the MPIExec interface.

        Args:
            command_lst (list[str]): The command list.

        Returns:
            list[str]: The generated command list.
        """
        command_prepend_lst = generate_mpiexec_command(
            cores=self._cores,
            openmpi_oversubscribe=self._openmpi_oversubscribe,
        )
        return super().generate_command(
            command_lst=command_prepend_lst + command_lst,
        )


def generate_mpiexec_command(
    cores: int, openmpi_oversubscribe: bool = False
) -> list[str]:
    """
    Generate the command list for the MPIExec interface.

    Args:
        cores (int): The number of cores.
        openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.

    Returns:
        list[str]: The generated command list.
    """
    if cores == 1:
        return []
    else:
        command_prepend_lst = [MPI_COMMAND, "-n", str(cores)]
        if openmpi_oversubscribe:
            command_prepend_lst += ["--oversubscribe"]
        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
7
8	class BaseSpawner(ABC):	1✔
9	def __init__(	1✔
10	self,
11	cwd: Optional[str] = None,
12	cores: int = 1,
13	openmpi_oversubscribe: bool = False,
14	):
15	"""
16	Base class for interface implementations.
17
18	Args:
19	cwd (str): The current working directory.
20	cores (int, optional): The number of cores to use. Defaults to 1.
21	openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
22	"""
23	self._cwd = cwd	1✔
24	self._cores = cores	1✔
25	self._openmpi_oversubscribe = openmpi_oversubscribe	1✔
26
27	def bootup(	1✔
28	self,
29	command_lst: list[str],
30	):
31	"""
32	Method to start the interface.
33
34	Args:
35	command_lst (list[str]): The command list to execute.
36	"""
UNCOV 37	raise NotImplementedError	×
38
39	def shutdown(self, wait: bool = True):	1✔
40	"""
41	Method to shutdown the interface.
42
43	Args:
44	wait (bool, optional): Whether to wait for the interface to shutdown. Defaults to True.
45	"""
UNCOV 46	raise NotImplementedError	×
47
48	def poll(self):	1✔
49	"""
50	Method to check if the interface is running.
51
52	Returns:
53	bool: True if the interface is running, False otherwise.
54	"""
UNCOV 55	raise NotImplementedError	×
56
57
58	class SubprocessSpawner(BaseSpawner):	1✔
59	def __init__(	1✔
60	self,
61	cwd: Optional[str] = None,
62	cores: int = 1,
63	openmpi_oversubscribe: bool = False,
64	threads_per_core: int = 1,
65	):
66	"""
67	Subprocess interface implementation.
68
69	Args:
70	cwd (str, optional): The current working directory. Defaults to None.
71	cores (int, optional): The number of cores to use. Defaults to 1.
72	threads_per_core (int, optional): The number of threads per core. Defaults to 1.
73	oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
74	"""
75	super().__init__(	1✔
76	cwd=cwd,
77	cores=cores,
78	openmpi_oversubscribe=openmpi_oversubscribe,
79	)
80	self._process: Optional[subprocess.Popen] = None	1✔
81	self._threads_per_core = threads_per_core	1✔
82
83	def bootup(	1✔
84	self,
85	command_lst: list[str],
86	):
87	"""
88	Method to start the subprocess interface.
89
90	Args:
91	command_lst (list[str]): The command list to execute.
92	"""
93	self._process = subprocess.Popen(	1✔
94	args=self.generate_command(command_lst=command_lst),
95	cwd=self._cwd,
96	stdin=subprocess.DEVNULL,
97	)
98
99	def generate_command(self, command_lst: list[str]) -> list[str]:	1✔
100	"""
101	Method to generate the command list.
102
103	Args:
104	command_lst (list[str]): The command list.
105
106	Returns:
107	list[str]: The generated command list.
108	"""
109	return command_lst	1✔
110
111	def shutdown(self, wait: bool = True):	1✔
112	"""
113	Method to shutdown the subprocess interface.
114
115	Args:
116	wait (bool, optional): Whether to wait for the interface to shutdown. Defaults to True.
117	"""
118	if self._process is not None:	1✔
119	self._process.communicate()	1✔
120	self._process.terminate()	1✔
121	if wait:	1✔
122	self._process.wait()	1✔
123	self._process = None	1✔
124
125	def poll(self) -> bool:	1✔
126	"""
127	Method to check if the subprocess interface is running.
128
129	Returns:
130	bool: True if the interface is running, False otherwise.
131	"""
132	return self._process is not None and self._process.poll() is None	1✔
133
134
135	class MpiExecSpawner(SubprocessSpawner):	1✔
136	def generate_command(self, command_lst: list[str]) -> list[str]:	1✔
137	"""
138	Generate the command list for the MPIExec interface.
139
140	Args:
141	command_lst (list[str]): The command list.
142
143	Returns:
144	list[str]: The generated command list.
145	"""
146	command_prepend_lst = generate_mpiexec_command(	1✔
147	cores=self._cores,
148	openmpi_oversubscribe=self._openmpi_oversubscribe,
149	)
150	return super().generate_command(	1✔
151	command_lst=command_prepend_lst + command_lst,
152	)
153
154
155	def generate_mpiexec_command(	1✔
156	cores: int, openmpi_oversubscribe: bool = False
157	) -> list[str]:
158	"""
159	Generate the command list for the MPIExec interface.
160
161	Args:
162	cores (int): The number of cores.
163	openmpi_oversubscribe (bool, optional): Whether to oversubscribe the cores. Defaults to False.
164
165	Returns:
166	list[str]: The generated command list.
167	"""
168	if cores == 1:	1✔
169	return []	1✔
170	else:
171	command_prepend_lst = [MPI_COMMAND, "-n", str(cores)]	1✔
172	if openmpi_oversubscribe:	1✔
173	command_prepend_lst += ["--oversubscribe"]	1✔
174	return command_prepend_lst	1✔

pyiron / executorlib / 13116578816

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