12483232754

Committed 24 Dec 2024 02:59PM UTC coverage: 96.004% (-0.4%) from 96.367%

Build # 12483232754

Build Type

Pull #535

github

Committed by

web-flow

Commit Message

Merge 1607bb008 into 5cf3ecc7e

Pull Request Pull Request #535: Add type checking with mypy

Run Details

117 of 124 new or added lines in 15 files covered. (94.35%)

7 existing lines in 4 files now uncovered.

1033 of 1076 relevant lines covered (96.0%)

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

pyiron / executorlib / 12483232754

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