Ответ 1
#!/usr/bin/env python
"""Start process; wait 2 seconds; kill the process; print all process output."""
import subprocess
import tempfile
import time
def main():
# open temporary file (it automatically deleted when it is closed)
# `Popen` requires `f.fileno()` so `SpooledTemporaryFile` adds nothing here
f = tempfile.TemporaryFile()
# start process, redirect stdout
p = subprocess.Popen(["top"], stdout=f)
# wait 2 seconds
time.sleep(2)
# kill process
#NOTE: if it doesn't kill the process then `p.wait()` blocks forever
p.terminate()
p.wait() # wait for the process to terminate otherwise the output is garbled
# print saved output
f.seek(0) # rewind to the beginning of the file
print f.read(),
f.close()
if __name__=="__main__":
main()
Хвостовые решения, которые печатают только часть вывода
Вы можете прочитать выход процесса в другом потоке и сохранить необходимое количество последних строк в очереди:
import collections
import subprocess
import time
import threading
def read_output(process, append):
for line in iter(process.stdout.readline, ""):
append(line)
def main():
# start process, redirect stdout
process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
try:
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines) # atomic .append()
t = threading.Thread(target=read_output, args=(process, q.append))
t.daemon = True
t.start()
#
time.sleep(2)
finally:
process.terminate() #NOTE: it doesn't ensure the process termination
# print saved lines
print ''.join(q)
if __name__=="__main__":
main()
Для этого варианта требуется q.append() быть атомарной. В противном случае выход может быть поврежден.
signal.alarm() решение
Вы можете использовать signal.alarm() для вызова process.terminate() после указанного таймаута вместо чтения в другом потоке. Хотя он может не очень хорошо взаимодействовать с модулем subprocess. На основе @Alex Martelli ответ:
import collections
import signal
import subprocess
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
def main():
# start process, redirect stdout
process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
# set signal handler
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(2) # produce SIGALRM in 2 seconds
try:
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines)
for line in iter(process.stdout.readline, ""):
q.append(line)
signal.alarm(0) # cancel alarm
except Alarm:
process.terminate()
finally:
# print saved lines
print ''.join(q)
if __name__=="__main__":
main()
Этот подход работает только в системах * nix. Он может блокироваться, если process.stdout.readline() не возвращается.
threading.Timer решение
import collections
import subprocess
import threading
def main():
# start process, redirect stdout
process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
# terminate process in timeout seconds
timeout = 2 # seconds
timer = threading.Timer(timeout, process.terminate)
timer.start()
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(process.stdout, maxlen=number_of_lines)
timer.cancel()
# print saved lines
print ''.join(q),
if __name__=="__main__":
main()
Этот подход также должен работать на Windows. Здесь я использовал process.stdout как итеративный; он может ввести дополнительную буферизацию вывода, вы можете переключиться на подход iter(process.stdout.readline, ""), если это нежелательно. если процесс не заканчивается на process.terminate(), тогда скрипты зависают.
Без потоков, без решения сигналов
import collections
import subprocess
import sys
import time
def main():
args = sys.argv[1:]
if not args:
args = ['top']
# start process, redirect stdout
process = subprocess.Popen(args, stdout=subprocess.PIPE, close_fds=True)
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines)
timeout = 2 # seconds
now = start = time.time()
while (now - start) < timeout:
line = process.stdout.readline()
if not line:
break
q.append(line)
now = time.time()
else: # on timeout
process.terminate()
# print saved lines
print ''.join(q),
if __name__=="__main__":
main()
В этом варианте не используются ни потоки, ни сигналы, но он выводит искаженный вывод в терминал. Он будет блокироваться, если process.stdout.readline() блокирует.