Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Posix-specific implementation of process utilities.
3This file is only meant to be imported by process.py, not by end-users.
4"""
6#-----------------------------------------------------------------------------
7# Imports
8#-----------------------------------------------------------------------------
10# Stdlib
11import errno
12import os
13import subprocess as sp
14import sys
16# Our own
17from ._process_common import getoutput as getoutput, arg_split
18from IPython.utils.encoding import DEFAULT_ENCODING
20__all__ = ["getoutput", "arg_split", "system", "check_pid"]
22#-----------------------------------------------------------------------------
23# Function definitions
24#-----------------------------------------------------------------------------
26class ProcessHandler:
27 """Execute subprocesses under the control of pexpect.
28 """
29 # Timeout in seconds to wait on each reading of the subprocess' output.
30 # This should not be set too low to avoid cpu overusage from our side,
31 # since we read in a loop whose period is controlled by this timeout.
32 read_timeout: float = 0.05
34 # Timeout to give a process if we receive SIGINT, between sending the
35 # SIGINT to the process and forcefully terminating it.
36 terminate_timeout: float = 0.2
38 # File object where stdout and stderr of the subprocess will be written
39 logfile = None
41 # Shell to call for subprocesses to execute
42 _sh: str | None = None
44 @property
45 def sh(self) -> str | None:
46 if self._sh is None:
47 import pexpect
48 shell_name = os.environ.get("SHELL", "sh")
49 self._sh = pexpect.which(shell_name)
50 if self._sh is None:
51 raise OSError('"{}" shell not found'.format(shell_name))
53 return self._sh
55 def __init__(self) -> None:
56 """Arguments are used for pexpect calls."""
57 self.logfile = sys.stdout
59 def getoutput(self, cmd: str) -> str | None:
60 """Run a command and return its stdout/stderr as a string.
62 Parameters
63 ----------
64 cmd : str
65 A command to be executed in the system shell.
67 Returns
68 -------
69 output : str
70 A string containing the combination of stdout and stderr from the
71 subprocess, in whatever order the subprocess originally wrote to its
72 file descriptors (so the order of the information in this string is the
73 correct order as would be seen if running the command in a terminal).
74 """
75 import pexpect
77 assert self.sh is not None
78 try:
79 res = pexpect.run(self.sh, args=["-c", cmd])
80 assert isinstance(res, str)
81 return res.replace("\r\n", "\n")
82 except KeyboardInterrupt:
83 print('^C', file=sys.stderr, end='')
84 return None
86 def system(self, cmd: str) -> int:
87 """Execute a command in a subshell.
89 Parameters
90 ----------
91 cmd : str
92 A command to be executed in the system shell.
94 Returns
95 -------
96 int : child's exitstatus
97 """
98 import pexpect
100 # Get likely encoding for the output.
101 enc = DEFAULT_ENCODING
103 # Patterns to match on the output, for pexpect. We read input and
104 # allow either a short timeout or EOF
105 patterns = [pexpect.TIMEOUT, pexpect.EOF]
106 # the index of the EOF pattern in the list.
107 # even though we know it's 1, this call means we don't have to worry if
108 # we change the above list, and forget to change this value:
109 EOF_index = patterns.index(pexpect.EOF)
110 # The size of the output stored so far in the process output buffer.
111 # Since pexpect only appends to this buffer, each time we print we
112 # record how far we've printed, so that next time we only print *new*
113 # content from the buffer.
114 out_size = 0
115 assert self.sh is not None
116 try:
117 # Since we're not really searching the buffer for text patterns, we
118 # can set pexpect's search window to be tiny and it won't matter.
119 # We only search for the 'patterns' timeout or EOF, which aren't in
120 # the text itself.
121 #child = pexpect.spawn(pcmd, searchwindowsize=1)
122 if hasattr(pexpect, 'spawnb'):
123 child = pexpect.spawnb(self.sh, args=['-c', cmd]) # Pexpect-U
124 else:
125 child = pexpect.spawn(self.sh, args=['-c', cmd]) # Vanilla Pexpect
126 flush = sys.stdout.flush
127 while True:
128 # res is the index of the pattern that caused the match, so we
129 # know whether we've finished (if we matched EOF) or not
130 res_idx = child.expect_list(patterns, self.read_timeout)
131 print(child.before[out_size:].decode(enc, 'replace'), end='')
132 flush()
133 if res_idx==EOF_index:
134 break
135 # Update the pointer to what we've already printed
136 out_size = len(child.before)
137 except KeyboardInterrupt:
138 # We need to send ^C to the process. The ascii code for '^C' is 3
139 # (the character is known as ETX for 'End of Text', see
140 # curses.ascii.ETX).
141 child.sendline(chr(3))
142 # Read and print any more output the program might produce on its
143 # way out.
144 try:
145 out_size = len(child.before)
146 child.expect_list(patterns, self.terminate_timeout)
147 print(child.before[out_size:].decode(enc, 'replace'), end='')
148 sys.stdout.flush()
149 except KeyboardInterrupt:
150 # Impatient users tend to type it multiple times
151 pass
152 finally:
153 # Ensure the subprocess really is terminated
154 child.terminate(force=True)
155 # add isalive check, to ensure exitstatus is set:
156 child.isalive()
158 # We follow the subprocess pattern, returning either the exit status
159 # as a positive number, or the terminating signal as a negative
160 # number.
161 # on Linux, sh returns 128+n for signals terminating child processes on Linux
162 # on BSD (OS X), the signal code is set instead
163 if child.exitstatus is None:
164 # on WIFSIGNALED, pexpect sets signalstatus, leaving exitstatus=None
165 if child.signalstatus is None:
166 # this condition may never occur,
167 # but let's be certain we always return an integer.
168 return 0
169 return -child.signalstatus
170 if child.exitstatus > 128:
171 return -(child.exitstatus - 128)
172 return child.exitstatus
175# Make system() with a functional interface for outside use. Note that we use
176# getoutput() from the _common utils, which is built on top of popen(). Using
177# pexpect to get subprocess output produces difficult to parse output, since
178# programs think they are talking to a tty and produce highly formatted output
179# (ls is a good example) that makes them hard.
180system = ProcessHandler().system
183def check_pid(pid: int) -> bool:
184 try:
185 os.kill(pid, 0)
186 except OSError as err:
187 if err.errno == errno.ESRCH:
188 return False
189 elif err.errno == errno.EPERM:
190 # Don't have permission to signal the process - probably means it exists
191 return True
192 raise
193 else:
194 return True