Canlı konsol oluşturmak için python'un pty'sini kullanın

Question

Bir sunucuda oluşturulacak soket üzerinden, stdout'u yayınlayan bir sunucuda uzaktan yürütecek bir yürütme ortamı/kabuğu oluşturmaya çalışıyorum. Şu anda PIPE ile subprocess.run'u kullanma yaklaşımını denedim. Sorun, işlem tamamlandıktan sonra stdout'u almamdır. Benim elde etmek istediğim, line-by-line, sözde terminal uygulamasıdır.

Benim şu anki uygulama

def greeter(): 
    for _ in range(10): 
     print('hello world') 

greeter() 

ve kabuğunda

test.py

>>> 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 

i, nasıl pty ile bile bu basit uygulama girişiminde çalışırsanız bir tane mi?

Answer 1

Im emin etrafında bir yerlerde bir dupe gidecekseniz ama o zaman çok uzun bir süre için yoğun bir mücadele olacak Windows üzerinde ise hızla

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

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

Answer 2

onu bulamadım, ben de acı için üzgünüm tahammül edersiniz (orada). Ancak, Linux'taysanız, pexpect modülünü kullanabilirsiniz. Pexpect, iki yönlü iletişim gerçekleştirebileceğiniz bir arka plan alt süreci oluşturmanıza olanak tanır. Bu, her tür sistem otomasyonu için kullanışlıdır, ancak çok yaygın bir kullanım durumu ssh'dir.

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!') 

Çok faydalı bir SSH bağlantısı gibi karmaşık bir şey işlemek için bir sınıf oluşturmak için bulduk, ancak kullanım durumu yeterince basitse uygun ya da gerekli olmayabilir. Pexpect.cefore tür bayttan önce ve aradığınız deseni atlamamak garip olabilir, bu yüzden bunu sizin için en azından işleyen bir işlev oluşturmak mantıklı olabilir.

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

Çocuk işlemine mesaj göndermek istiyorsanız, child.sendline (satır) 'ı kullanabilirsiniz. Daha fazla ayrıntı için, bağlandığım belgeleri gözden geçirin.

Umarım yardımcı olmuş olurum!

Answer 3

bir tarayıcıda bu işleyebilen eğer bilmiyorum ama böyle stdout'u hemen almak böylece modül gibi bir program çalıştırabilirsiniz:

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() 

Sen hiç test.py içinde uyku ekleyebilir

from time import sleep 

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

greeter() 

Answer 4

uygulamanız stdout'ta verileri okuma ve daha sonra bir WebSocket'e bunu yazmak gibi, birden fazla görevi ile asenkron çalışmaya gidiyor, ben asyncio kullanmanızı öneririz: farkı görmek.

import asyncio.subprocess 
import os 

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


async def on_websocket(request): 
    # Prepare aiohttp's 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's 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) 

Bu aiohttp kullanır ve web_ws ve subprocess streams örnekler dayanmaktadır: burada

bir yöntem çalışır ve bir WebSocket içine çıkış yönlendirir bir örnektir.