Используйте python pty для создания живой консоли

Ответ 1

Я уверен, что это обман вокруг, но я не смог найти его быстро

process = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE,bufsize=0)

for out in iter(process.stdout.readline, b''):
    print(out)

Ответ 2

Если ваше приложение будет работать асинхронно с несколькими задачами, например, считывать данные из stdout, а затем записывать их в websocket, я предлагаю использовать asyncio.

Вот пример, который запускает процесс и перенаправляет его вывод в websocket:

import asyncio.subprocess
import os

from aiohttp.web import (Application, Response, WebSocketResponse, WSMsgType,
                         run_app)


async def on_websocket(request):
    # Prepare aiohttp websocket...
    resp = WebSocketResponse()
    await resp.prepare(request)
    # ... and store in a global dictionary so it can be closed on shutdown
    request.app['sockets'].append(resp)

    process = await asyncio.create_subprocess_exec(sys.executable,
                                                   '/tmp/test.py',
                                                    stdout=asyncio.subprocess.PIPE,
                                                    stderr=asyncio.subprocess.PIPE,
                                                    bufsize=0)
    # Schedule reading from stdout and stderr as asynchronous tasks.
    stdout_f = asyncio.ensure_future(p.stdout.readline())
    stderr_f = asyncio.ensure_future(p.stderr.readline())

    # returncode will be set upon process termination.
    while p.returncode is None:
        # Wait for a line in either stdout or stderr.
        await asyncio.wait((stdout_f, stderr_f), return_when=asyncio.FIRST_COMPLETED)

        # If task is done, then line is available.
        if stdout_f.done():
            line = stdout_f.result().encode()
            stdout_f = asyncio.ensure_future(p.stdout.readline())
            await ws.send_str(f'stdout: {line}')

        if stderr_f.done():
            line = stderr_f.result().encode()
            stderr_f = asyncio.ensure_future(p.stderr.readline())
            await ws.send_str(f'stderr: {line}')

    return resp


async def on_shutdown(app):
    for ws in app['sockets']:
        await ws.close()    


async def init(loop):
    app = Application()
    app['sockets'] = []
    app.router.add_get('/', on_websocket)
    app.on_shutdown.append(on_shutdown)
    return app


loop = asyncio.get_event_loop()
app = loop.run_until_complete(init())
run_app(app)

Он использует aiohttp и основан на web_ws и примеры подпроцессов.

Ответ 3

Если вы находитесь в Windows, вы очень долго будете сражаться в гору, и я сожалею о боли, которую вы перенесете (там был). Однако, если вы работаете в Linux, вы можете использовать модуль pexpect. Pexpect позволяет вам генерировать фоновый дочерний процесс, с которым вы можете осуществлять двунаправленную связь. Это полезно для всех типов системной автоматизации, но очень распространенным случаем является ssh.

import pexpect

child   = pexpect.spawn('python3 test.py')
message = 'hello world'

while True:
    try:
        child.expect(message)
    except pexpect.exceptions.EOF:
        break
    input('child sent: "%s"\nHit enter to continue: ' %
         (message + child.before.decode()))

print('reached end of file!')

Я нашел очень полезным создать класс для обработки чего-то сложного, как ssh-соединение, но если ваш пример использования достаточно прост, который может быть неприемлемым или необходимым. Путь pexpect.before имеет тип байтов и опускает шаблон, который вы ищете, может быть неудобным, поэтому имеет смысл создать функцию, которая обрабатывает это для вас, по крайней мере.

def get_output(child, message):
    return(message + child.before.decode())

Если вы хотите отправлять сообщения дочернему процессу, вы можете использовать child.sendline(строка). Для получения дополнительной информации ознакомьтесь с документацией, которую я связал.

Надеюсь, я смог помочь!

Ответ 4

Я не знаю, можете ли вы сделать это в браузере, но вы можете запустить программу, такую ​​как модуль, чтобы сразу получить stdout следующим образом:

import importlib
from importlib.machinery import SourceFileLoader

class Program:

    def __init__(self, path, name=''):
        self.path = path
        self.name = name
        if self.path:
            if not self.name:
                self.get_name()
            self.loader = importlib.machinery.SourceFileLoader(self.name, self.path)
            self.spec = importlib.util.spec_from_loader(self.loader.name, self.loader)
            self.mod = importlib.util.module_from_spec(self.spec)
        return

    def get_name(self):
        extension = '.py' #change this if self.path is not python program with extension .py
        self.name = self.path.split('\\')[-1].strip('.py')
        return

    def load(self):
        self.check()
        self.loader.exec_module(self.mod)
        return

    def check(self):
        if not self.path:
            Error('self.file is NOT defined.'.format(path)).throw()
        return

file_path = 'C:\\Users\\RICHGang\\Documents\\projects\\stackoverflow\\ptyconsole\\test.py'
file_name = 'test'
prog = Program(file_path, file_name)   
prog.load()

Вы можете добавить sleep в test.py, чтобы увидеть разницу:

from time import sleep

def greeter():
    for i in range(10):
        sleep(0.3)
        print('hello world')

greeter()