13

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?

2
  • Check this out: stackoverflow.com/questions/1606795/… Commented Aug 21, 2017 at 4:15
  • Try using bufsize=1 parameter to subprocess to set line buffer, and use iter(result.stdout.readline, b'') to read the stdout wrapped in while True loop Commented Aug 23, 2017 at 10:51

5 Answers 5

6
+25

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.

Sign up to request clarification or add additional context in comments.

Comments

5

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) 

3 Comments

this would still wait for the cmd to finish and then the for loop will start. I want more of an async implementation, that's why I want to know more about pty
this should stream it in realtime ... you might want to set your buffersize to zero if you are not finding this to be the case
@IshanKhare> this will stream in realtime. The Popen function starts the program in the background and returns immediately. Anything the program outputs will be read immediately. Note that the reads are buffered though, so the reads will return once a large enough chunk has been read (that's why if you test with too simple examples you could think it waits). You can disable buffering with bufsize=0 if you really want fully realtime reads at the expense of performance.
2

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!

Comments

1

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

Comments

0

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

Jupiter Lab uses it.

Comments

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.