Ответ 1
Я уверен, что это обман вокруг, но я не смог найти его быстро
process = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE,bufsize=0)
for out in iter(process.stdout.readline, b''):
print(out)
Я пытаюсь создать среду выполнения/оболочку, которая будет удаленно выполняться на сервере, который передает поток stdout, err, поверх сокета, который будет отображаться в браузере. В настоящее время я попытался использовать subprocess.run
с помощью PIPE
. Проблема в том, что я получаю stdout после завершения процесса. То, что я хочу достичь, состоит в том, чтобы получить линейную, псевдотерминальную реализацию.
Моя текущая реализация
test.py
def greeter():
for _ in range(10):
print('hello world')
greeter()
и в оболочке
>>> import subprocess
>>> result = subprocess.run(['python3', 'test.py'], stdout=subprocess.PIPE)
>>> print(result.stdout.decode('utf-8'))
hello world
hello world
hello world
hello world
hello world
hello world
hello world
hello world
hello world
hello world
Если я попытаюсь выполнить эту простую реализацию с помощью pty
, как это сделать?
Я уверен, что это обман вокруг, но я не смог найти его быстро
process = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE,bufsize=0)
for out in iter(process.stdout.readline, b''):
print(out)
Если ваше приложение будет работать асинхронно с несколькими задачами, например, считывать данные из 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 и примеры подпроцессов.
Если вы находитесь в 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(строка). Для получения дополнительной информации ознакомьтесь с документацией, которую я связал.
Надеюсь, я смог помочь!
Я не знаю, можете ли вы сделать это в браузере, но вы можете запустить программу, такую как модуль, чтобы сразу получить 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()