В предположении, что процесс, который вы вызываете, длительный и не производит весь его вывод за один раз, это означает, что вы не можете использовать subprocess.Popen.communicate(), так как он предназначен для чтения всего вывода до конца файл.
Вам нужно будет использовать другие стандартные методы для считывания с трубы.
Как вы хотите интегрировать его с графическим интерфейсом, и процесс длительный, вам нужно будет координировать чтение его вывода с помощью основного цикла GUI. Это несколько усложняет ситуацию.
TkInter
Давайте сначала предположим, что вы хотите использовать TkInter, как в одном из ваших примеров. Это ставит перед нами несколько проблем:
- Там нет интеграции TkInter с выбрать модуль.
- В настоящее время даже нет канонической интеграции TkInter с asyncio (см. Также https://bugs.python.org/issue27546).
- Обычно рекомендуется использовать обычную основную петлю с использованием
root.update(), что позволяет нам решать проблемы с потоками, что должно было быть основано на событиях.
- У TkInter's
event_generate() отсутствует способность Tk отправлять пользовательские данные вместе с событием, поэтому мы не можем использовать события TkInter для передачи полученного результата из одного потока в другой.
Таким образом, мы будем решать с резьб (даже если я предпочел бы не), где основной поток управляет Tk GUI и вспомогательный поток считывает выход из процесса, и не хватает родной в TkInter для передачи данных, мы используем поточно-безопасную Очередь.
#!/usr/bin/env python3
from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
from threading import Thread, Event
from queue import Queue, Empty
from tkinter import Tk, Text, END
class ProcessOutputReader(Thread):
def __init__(self, queue, cmd, params=(),
group=None, name=None, daemon=True):
super().__init__(group=group, name=name, daemon=daemon)
self._stop_request = Event()
self.queue = queue
self.process = Popen((cmd,) + tuple(params),
stdout=PIPE,
stderr=STDOUT,
universal_newlines=True)
def run(self):
for line in self.process.stdout:
if self._stop_request.is_set():
# if stopping was requested, terminate the process and bail out
self.process.terminate()
break
self.queue.put(line) # enqueue the line for further processing
try:
# give process a chance to exit gracefully
self.process.wait(timeout=3)
except TimeoutExpired:
# otherwise try to terminate it forcefully
self.process.kill()
def stop(self):
# request the thread to exit gracefully during its next loop iteration
self._stop_request.set()
# empty the queue, so the thread will be woken up
# if it is blocking on a full queue
while True:
try:
self.queue.get(block=False)
except Empty:
break
self.queue.task_done() # acknowledge line has been processed
class MyConsole(Text):
def __init__(self, parent, queue, update_interval=50, process_lines=500):
super().__init__(parent)
self.queue = queue
self.update_interval = update_interval
self.process_lines = process_lines
self.after(self.update_interval, self.fetch_lines)
def fetch_lines(self):
something_inserted = False
for _ in range(self.process_lines):
try:
line = self.queue.get(block=False)
except Empty:
break
self.insert(END, line)
self.queue.task_done() # acknowledge line has been processed
# ensure scrolling the view is at most done once per interval
something_inserted = True
if something_inserted:
self.see(END)
self.after(self.update_interval, self.fetch_lines)
# create the root widget
root = Tk()
# create a queue for sending the lines from the process output reader thread
# to the TkInter main thread
line_queue = Queue(maxsize=1000)
# create a process output reader
reader = ProcessOutputReader(line_queue, 'python3', params=['-u', 'test.py'])
# create a console
console = MyConsole(root, line_queue)
reader.start() # start the process
console.pack() # make the console visible
root.mainloop() # run the TkInter main loop
reader.stop()
reader.join(timeout=5) # give thread a chance to exit gracefully
if reader.is_alive():
raise RuntimeError("process output reader failed to stop")
Благодаря вышеупомянутым предостережений, код Tkinter заканчивается немного на большей стороне.
PyQt
Использование PyQt вместо этого, мы можем значительно улучшить ситуацию, так как эта структура уже поставляется с родным способом интеграции с подпроцесс, в форме его QProcess класса.
Это означает, что мы можем сделать прочь с потоками и использовать механизм в Qt родной сигнала и Slot вместо этого.
#!/usr/bin/env python3
import sys
from PyQt5.QtCore import pyqtSignal, pyqtSlot, QProcess, QTextCodec
from PyQt5.QtGui import QTextCursor
from PyQt5.QtWidgets import QApplication, QPlainTextEdit
class ProcessOutputReader(QProcess):
produce_output = pyqtSignal(str)
def __init__(self, parent=None):
super().__init__(parent=parent)
# merge stderr channel into stdout channel
self.setProcessChannelMode(QProcess.MergedChannels)
# prepare decoding process' output to Unicode
codec = QTextCodec.codecForLocale()
self._decoder_stdout = codec.makeDecoder()
# only necessary when stderr channel isn't merged into stdout:
# self._decoder_stderr = codec.makeDecoder()
self.readyReadStandardOutput.connect(self._ready_read_standard_output)
# only necessary when stderr channel isn't merged into stdout:
# self.readyReadStandardError.connect(self._ready_read_standard_error)
@pyqtSlot()
def _ready_read_standard_output(self):
raw_bytes = self.readAllStandardOutput()
text = self._decoder_stdout.toUnicode(raw_bytes)
self.produce_output.emit(text)
# only necessary when stderr channel isn't merged into stdout:
# @pyqtSlot()
# def _ready_read_standard_error(self):
# raw_bytes = self.readAllStandardError()
# text = self._decoder_stderr.toUnicode(raw_bytes)
# self.produce_output.emit(text)
class MyConsole(QPlainTextEdit):
def __init__(self, parent=None):
super().__init__(parent=parent)
self.setReadOnly(True)
self.setMaximumBlockCount(10000) # limit console to 10000 lines
self._cursor_output = self.textCursor()
@pyqtSlot(str)
def append_output(self, text):
self._cursor_output.insertText(text)
self.scroll_to_last_line()
def scroll_to_last_line(self):
cursor = self.textCursor()
cursor.movePosition(QTextCursor.End)
cursor.movePosition(QTextCursor.Up if cursor.atBlockStart() else
QTextCursor.StartOfLine)
self.setTextCursor(cursor)
# create the application instance
app = QApplication(sys.argv)
# create a process output reader
reader = ProcessOutputReader()
# create a console and connect the process output reader to it
console = MyConsole()
reader.produce_output.connect(console.append_output)
reader.start('python3', ['-u', 'test.py']) # start the process
console.show() # make the console visible
app.exec_() # run the PyQt main loop
Мы в конечном итоге с небольшим шаблонным вытекающим из классов Qt, но с общим подходом пылесоса.
Общие соображения
Также убедитесь, что процесс вы вызываете не буферные несколько выходных линий, так как в противном случае он все равно будет выглядеть, как будто консоль застряла.
В частности, если вызываемая программа представляет собой программу python, вы можете либо убедиться, что она использует print(..., flush=True), либо вызвать ее с помощью python -u callee.py для обеспечения небуферизованного вывода.
Как захватить вывод долговременной программы и представить ее в графическом интерфейсе в Python?
Вы должны изучить модуль 'tkinter' – spicypumpkin
PyCrust (wxWidgets) - отличная интерактивная оболочка, которую вы можете включить в свои приложения. –
Я сделал и не являюсь реальным решением для добавления этого окна в графический интерфейс, все «решения» связаны с использованием «подпроцесса», а затем считывают вывод подпроцесса, и это не то, что мне нужно. –