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

89.19

/pympipool/shell/interactive.py

import queue
import subprocess
import threading
from concurrent.futures import Future
from time import sleep
from typing import Optional

from pympipool.shared.executor import ExecutorBase, cancel_items_in_queue
from pympipool.shared.thread import RaisingThread


def wait_for_process_to_stop(process: threading.Thread, sleep_interval: float = 10e-10):
    """
    Wait for the subprocess.Popen() process to stop executing

    Args:
        process (subprocess.Popen): process object
        sleep_interval (float): interval to sleep during poll() calls
    """
    while process.poll() is None:
        sleep(sleep_interval)


def execute_single_task(future_queue: queue.Queue):
    """
    Process items received via the queue.

    Args:
        future_queue (queue.Queue):
    """
    process = None
    while True:
        task_dict = future_queue.get()
        if "shutdown" in task_dict.keys() and task_dict["shutdown"]:
            if process is not None and process.poll() is None:
                process.stdin.flush()
                process.stdin.close()
                process.stdout.close()
                process.stderr.close()
                process.terminate()
                wait_for_process_to_stop(process=process)
            future_queue.task_done()
            # future_queue.join()
            break
        elif "init" in task_dict.keys() and task_dict["init"]:
            process = subprocess.Popen(
                *task_dict["args"],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                **task_dict["kwargs"],
            )
        elif "future" in task_dict.keys():
            if process is None:
                raise ValueError("process not initialized")
            elif process.poll() is None:
                f = task_dict.pop("future")
                if f.set_running_or_notify_cancel():
                    try:
                        process.stdin.write(task_dict["input"])
                        process.stdin.flush()
                        lines_count = 0
                        output = ""
                        while True:
                            output_current = process.stdout.readline()
                            output += output_current
                            lines_count += 1
                            if (
                                task_dict["stop_read_pattern"] is not None
                                and task_dict["stop_read_pattern"] in output_current
                            ):
                                break
                            elif (
                                task_dict["lines_to_read"] is not None
                                and task_dict["lines_to_read"] == lines_count
                            ):
                                break
                        f.set_result(output)
                    except Exception as thread_exception:
                        future_queue.task_done()
                        f.set_exception(exception=thread_exception)
                        raise thread_exception
                    else:
                        future_queue.task_done()
            else:
                raise ValueError("process exited")


class ShellExecutor(ExecutorBase):
    """
    In contrast to the other pympipool.shell.SubprocessExecutor and the pympipool.Executor the pympipool.shell.ShellExecutor
    can only execute a single process at a given time. Still it adds the capability to interact with this process during
    its execution. The initialization of the pympipool.shell.ShellExecutor takes the same input arguments as the
    subprocess.Popen() call for the standard library to start a subprocess.

    Examples

        >>> from pympipool import ShellExecutor
        >>> with ShellExecutor(["python", "count.py"], universal_newlines=True) as exe:
        >>>     future_lines = exe.submit(string_input="4", lines_to_read=5)
        >>>     print(future_lines.done(), future_lines.result(), future_lines.done())
        (False, "0\n1\n2\n3\ndone\n", True)

        >>> from pympipool import ShellExecutor
        >>> with ShellExecutor(["python", "count.py"], universal_newlines=True) as exe:
        >>>     future_pattern = exe.submit(string_input="4", stop_read_pattern="done")
        >>>     print(future_pattern.done(), future_pattern.result(), future_pattern.done())
        (False, "0\n1\n2\n3\ndone\n", True)
    """

    def __init__(self, *args, **kwargs):
        super().__init__()
        self._set_process(
            process=RaisingThread(
                target=execute_single_task,
                kwargs={
                    "future_queue": self._future_queue,
                },
            ),
        )
        self._future_queue.put({"init": True, "args": args, "kwargs": kwargs})

    def submit(
        self,
        string_input: str,
        lines_to_read: Optional[int] = None,
        stop_read_pattern: Optional[str] = None,
    ):
        """
        Submit the input as a string to the executable. In addition to the input the ShellExecutor also needs a measure
        to identify the completion of the execution. This can either be provided based on the number of lines to read
        using the `lines_to_read` parameter or by providing a string pattern using the `stop_read_pattern` to stop
        reading new lines. One of these two stopping criteria has to be defined.

        Args:
            string_input (str): Input to be communicated to the underlying executable
            lines_to_read (None/int): integer number of lines to read from the command line (optional)
            stop_read_pattern (None/str): string pattern to indicate the command line output is completed (optional)

        Returns:
            A Future representing the given call.
        """
        if lines_to_read is None and stop_read_pattern is None:
            raise ValueError(
                "Either the number of lines_to_read (int) or the stop_read_pattern (str) has to be defined."
            )
        if string_input[-1:] != "\n":
            string_input += "\n"
        f = Future()
        self._future_queue.put(
            {
                "future": f,
                "input": string_input,
                "lines_to_read": lines_to_read,
                "stop_read_pattern": stop_read_pattern,
            }
        )
        return f

    def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
        """Clean-up the resources associated with the Executor.

        It is safe to call this method several times. Otherwise, no other
        methods can be called after this one.

        Args:
            wait: If True then shutdown will not return until all running
                futures have finished executing and the resources used by the
                parallel_executors have been reclaimed.
            cancel_futures: If True then shutdown will cancel all pending
                futures. Futures that are completed or running will not be
                cancelled.
        """
        if cancel_futures:
            cancel_items_in_queue(que=self._future_queue)
        self._future_queue.put({"shutdown": True, "wait": wait})
        if wait:
            self._process.join()
            # self._future_queue.join()
        self._process = None
        self._future_queue = None

1	import queue	1✔
2	import subprocess	1✔
3	import threading	1✔
4	from concurrent.futures import Future	1✔
5	from time import sleep	1✔
6	from typing import Optional	1✔
7
8	from pympipool.shared.executor import ExecutorBase, cancel_items_in_queue	1✔
9	from pympipool.shared.thread import RaisingThread	1✔
10
11
12	def wait_for_process_to_stop(process: threading.Thread, sleep_interval: float = 10e-10):	1✔
13	"""
14	Wait for the subprocess.Popen() process to stop executing
15
16	Args:
17	process (subprocess.Popen): process object
18	sleep_interval (float): interval to sleep during poll() calls
19	"""
20	while process.poll() is None:	1✔
21	sleep(sleep_interval)	1✔
22
23
24	def execute_single_task(future_queue: queue.Queue):	1✔
25	"""
26	Process items received via the queue.
27
28	Args:
29	future_queue (queue.Queue):
30	"""
31	process = None	1✔
32	while True:	1✔
33	task_dict = future_queue.get()	1✔
34	if "shutdown" in task_dict.keys() and task_dict["shutdown"]:	1✔
35	if process is not None and process.poll() is None:	1✔
36	process.stdin.flush()	1✔
37	process.stdin.close()	1✔
38	process.stdout.close()	1✔
39	process.stderr.close()	1✔
40	process.terminate()	1✔
41	wait_for_process_to_stop(process=process)	1✔
42	future_queue.task_done()	1✔
43	# future_queue.join()
44	break	1✔
45	elif "init" in task_dict.keys() and task_dict["init"]:	1✔
46	process = subprocess.Popen(	1✔
47	*task_dict["args"],
48	stdin=subprocess.PIPE,
49	stdout=subprocess.PIPE,
50	stderr=subprocess.PIPE,
51	**task_dict["kwargs"],
52	)
53	elif "future" in task_dict.keys():	1✔
54	if process is None:	1✔
55	raise ValueError("process not initialized")	×
56	elif process.poll() is None:	1✔
57	f = task_dict.pop("future")	1✔
58	if f.set_running_or_notify_cancel():	1✔
59	try:	1✔
60	process.stdin.write(task_dict["input"])	1✔
61	process.stdin.flush()	1✔
62	lines_count = 0	1✔
63	output = ""	1✔
64	while True:	1✔
65	output_current = process.stdout.readline()	1✔
66	output += output_current	1✔
67	lines_count += 1	1✔
68	if (	1✔
69	task_dict["stop_read_pattern"] is not None
70	and task_dict["stop_read_pattern"] in output_current
71	):
72	break	1✔
73	elif (	1✔
74	task_dict["lines_to_read"] is not None
75	and task_dict["lines_to_read"] == lines_count
76	):
77	break	1✔
78	f.set_result(output)	1✔
79	except Exception as thread_exception:	×
80	future_queue.task_done()	×
81	f.set_exception(exception=thread_exception)	×
82	raise thread_exception	×
83	else:
84	future_queue.task_done()	1✔
85	else:
86	raise ValueError("process exited")	×
87
88
89	class ShellExecutor(ExecutorBase):	1✔
90	"""
91	In contrast to the other pympipool.shell.SubprocessExecutor and the pympipool.Executor the pympipool.shell.ShellExecutor
92	can only execute a single process at a given time. Still it adds the capability to interact with this process during
93	its execution. The initialization of the pympipool.shell.ShellExecutor takes the same input arguments as the
94	subprocess.Popen() call for the standard library to start a subprocess.
95
96	Examples
97
98	>>> from pympipool import ShellExecutor
99	>>> with ShellExecutor(["python", "count.py"], universal_newlines=True) as exe:
100	>>> future_lines = exe.submit(string_input="4", lines_to_read=5)
101	>>> print(future_lines.done(), future_lines.result(), future_lines.done())
102	(False, "0\n1\n2\n3\ndone\n", True)
103
104	>>> from pympipool import ShellExecutor
105	>>> with ShellExecutor(["python", "count.py"], universal_newlines=True) as exe:
106	>>> future_pattern = exe.submit(string_input="4", stop_read_pattern="done")
107	>>> print(future_pattern.done(), future_pattern.result(), future_pattern.done())
108	(False, "0\n1\n2\n3\ndone\n", True)
109	"""
110
111	def __init__(self, args, *kwargs):	1✔
112	super().__init__()	1✔
113	self._set_process(	1✔
114	process=RaisingThread(
115	target=execute_single_task,
116	kwargs={
117	"future_queue": self._future_queue,
118	},
119	),
120	)
121	self._future_queue.put({"init": True, "args": args, "kwargs": kwargs})	1✔
122
123	def submit(	1✔
124	self,
125	string_input: str,
126	lines_to_read: Optional[int] = None,
127	stop_read_pattern: Optional[str] = None,
128	):
129	"""
130	Submit the input as a string to the executable. In addition to the input the ShellExecutor also needs a measure
131	to identify the completion of the execution. This can either be provided based on the number of lines to read
132	using the `lines_to_read` parameter or by providing a string pattern using the `stop_read_pattern` to stop
133	reading new lines. One of these two stopping criteria has to be defined.
134
135	Args:
136	string_input (str): Input to be communicated to the underlying executable
137	lines_to_read (None/int): integer number of lines to read from the command line (optional)
138	stop_read_pattern (None/str): string pattern to indicate the command line output is completed (optional)
139
140	Returns:
141	A Future representing the given call.
142	"""
143	if lines_to_read is None and stop_read_pattern is None:	1✔
144	raise ValueError(	×
145	"Either the number of lines_to_read (int) or the stop_read_pattern (str) has to be defined."
146	)
147	if string_input[-1:] != "\n":	1✔
148	string_input += "\n"	1✔
149	f = Future()	1✔
150	self._future_queue.put(	1✔
151	{
152	"future": f,
153	"input": string_input,
154	"lines_to_read": lines_to_read,
155	"stop_read_pattern": stop_read_pattern,
156	}
157	)
158	return f	1✔
159
160	def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):	1✔
161	"""Clean-up the resources associated with the Executor.
162
163	It is safe to call this method several times. Otherwise, no other
164	methods can be called after this one.
165
166	Args:
167	wait: If True then shutdown will not return until all running
168	futures have finished executing and the resources used by the
169	parallel_executors have been reclaimed.
170	cancel_futures: If True then shutdown will cancel all pending
171	futures. Futures that are completed or running will not be
172	cancelled.
173	"""
174	if cancel_futures:	1✔
175	cancel_items_in_queue(que=self._future_queue)	×
176	self._future_queue.put({"shutdown": True, "wait": wait})	1✔
177	if wait:	1✔
178	self._process.join()	1✔
179	# self._future_queue.join()
180	self._process = None	1✔
181	self._future_queue = None	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