Используешь Popen(), перехватываешь ввод-вывод (
stdin, stdout и stderr, передаёшь туда константу PIPE). Дальше используешь свойства stdin, stdout, stderr - это файлоподобные объекты (пайпы).
Тут есть ряд подводных камней. Обычно чтение из файла - блокирующая операция, т.е. пока дочерний процесс не напишет что-то в пайп - твоя программа будет висеть и не реагировать на действия пользователя. В питоне неблокирующее чтение из пайпа доступно только под unix, а под виндой - только начиная с 3.12. Если нужна поддержка винды, будет практичнее запустить отдельный поток, который читает данные из процесса. Например, так:
import collections
import signal
import subprocess
import threading
import time
class ChildProcessReader(threading.Thread):
def __init__(self, args: list[str], *, encoding: str | None = None, errors: str = 'ignore', exit_timeout: float = 1.0):
super().__init__(name=f'Monitor {args!r}', daemon=True)
self._encoding = encoding
self._errors = errors
self._exit_timeout = exit_timeout
self._queue = collections.deque()
self._child = subprocess.Popen(args,
stdin=subprocess.DEVNULL, # стандартный вход отключен
stdout=subprocess.PIPE, # перехватываем стандартный вывод
stderr=subprocess.STDOUT, # стандартный вывод ошибок направляем на стандартый вывод
)
self.start()
@property
def running(self) -> bool:
'''Возвращает True, если процесс ещё жив.'''
return self._child is not None
@property
def output(self) -> collections.deque:
'''Очередь, в которую будут складываться получаемые от процесса строки.'''
return self._queue
def stop(self) -> None:
'''Останавливает процесс - сначала "по хорошему", потом принудительно.'''
if self._child is None:
return
try:
self._child.send_signal(signal.CTRL_C_EVENT)
self._child.wait(self._exit_timeout)
except subprocess.TimeoutExpired:
self._child.terminate()
except KeyboardInterrupt:
pass
finally:
self._child = None
def run(self) -> None:
'''Тело потока.'''
try:
while self._child.poll() is None: # рабочий цикл крутится, пока дочерний процесс жив.
stdout_data = self._child.stdout.read1()
if self._encoding is not None:
stdout_data = stdout_data.decode(self._encoding, errors=self._errors)
self._queue.append(stdout_data)
except KeyboardInterrupt:
pass
finally: # по выходу из цикла завершаем процесс
self.stop()
if __name__ == '__main__':
child = ChildProcessReader(
# выполняемая команда
['ping', '-t', '127.0.0.1'],
# кодировка, в которой отдаёт данные твой процесс. None - отдавать байты как есть.
encoding='cp866', errors='ignore',
# сколько ждать завершения процесса
exit_timeout=1.0
)
# процесс (и поток) стартуют немедленно по созданию экземпляра класса
try:
while child.running:
# вместо просто цикла тебе лучше периодически выполнять код ниже иным способом
# например, если у тебя GUI на tkinter, используй метод .after()
try: # пытаемся получить очередную строку от процесса
data = child.output.popleft()
except IndexError: # ничего нет, буфер пуст
# это просто реакция на отсутствие текста. Можно вообще ничего не делать.
print('.', end='', flush=True)
time.sleep(0.1) #
else: # получили строку, обрабатываем
print(f'\n> {data!r}')
except KeyboardInterrupt:
print('\nStopping')
finally: # в итоге надо будет прибить дочерний процесс
child.stop()
