Use python's pty to create a live console

Question

I'm trying to create an execution environment/shell that will remotely execute on a server, which streams the stdout,err,in over the socket to be rendered in a browser. I currently have tried the approach of using subprocess.run with a PIPE. The Problem is that I get the stdout after the process has completed. What i want to achieve is to get a line-by-line, pseudo-terminal sort of implementation.

My current implementation

test.py

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

greeter()

and in the shell

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

If i try to attempt even this simple implementation with pty, how does one do it?

Answer 1

If your application is going to work asynchronously with multiple tasks, like reading data from stdout and then writing it to a websocket, I suggest using asyncio.

Here is an example that runs a process and redirects its output into a 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)

It uses aiohttp and is based on the web_ws and subprocess streams examples.

Answer 2

Im sure theres a dupe around somewhere but i couldnt find it quickly

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

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

Answer 3

If you are on Windows then you will be fighting an uphill battle for a very long time, and I am sorry for the pain you will endure (been there). If you are on Linux, however, you can use the pexpect module. Pexpect allows you to spawn a background child process which you can perform bidirectional communication with. This is useful for all types of system automation, but a very common use case is 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!')

I have found it very useful to create a class to handle something complicated like an ssh connection, but if your use case is simple enough that might not be appropriate or necessary. The way pexpect.before is of type bytes and omits the pattern you are searching for can be awkward, so it may make sense to create a function that handles this for you at the very least.

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

If you want to send messages to the child process, you can use child.sendline(line). For more details, check out the documentation I linked.

I hope I was able to help!

Answer 4

I don't know if you can render this in a browser, but you can run a program like module so you get stdout immediately like this:

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

You can add sleep in test.py to see the difference:

from time import sleep

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

greeter()

Answer 5

Take a look at terminado. Works on Windows and Linux.

Jupiter Lab uses it.