pythonのptyを使ってライブコンソールを作成する

Question

ブラウザでレンダリングするソケット上でstdout、errをストリームするサーバー上でリモート実行する実行環境/シェルを作成しようとしています。私は現在、subprocess.runにPIPEを使ってアプローチを試みました。問題は、プロセスが完了した後に標準出力が得られることです。私が達成したいのは、行単位で擬似端末を実装することです。

私の現在の実装

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

greeter() 

とシェルで

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 

私は、どのようにptyでも、この単純な実装をしようしようとした場合それは何ですか？

Answer 1

イム必ずどこかの周りにだまされやすい人をtheresのが、私は、Windows上にある場合、あなたは非常に長い時間のために苦しい戦いを戦うことができる、と私は痛みのために申し訳ありません迅速

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

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

Answer 2

それを見つけるcouldntのあなたは耐えられるでしょう（そこにいました）。ただし、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がバイト型で、検索しているパターンを省略する方法は扱いにくいかもしれないので、少なくともあなたのためにこれを処理する関数を作成することは意味があります。

メッセージを子プロセスに送信する場合は、child.sendline（行）を使用できます。詳細については、リンク先のドキュメントをご覧ください。

私は助けてくれることを望みます！

Answer 3

ブラウザでこれをレンダリングすることができるかどうかは分かりませんが、あなたはすぐにこのように標準出力を得るように、あなたはモジュールのようなプログラムを実行することができます。

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

あなたがにtest.pyで睡眠を追加することができます

from time import sleep 

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

greeter() 

Answer 4

アプリケーションが標準出力からデータを読み、その後の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'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) 

それはaiohttpを使用し、web_wsとsubprocess streams例に基づいています。