我假设你pty是出于Q中概述的原因而使用的:为什么不仅仅使用管道(popen())?(到目前为止,所有其他答案都将忽略你的“注意:我不想一次打印所有内容”)。
pty仅在docs中说过是Linux :
由于伪终端处理高度依赖平台,因此有代码仅适用于
Linux。(Linux代码应该可以在其他平台上运行,但尚未经过测试。)
目前尚不清楚它在其他操作系统上的运行情况。
你可以尝试pexpect:
import sysimport pexpectpexpect.run("ruby ruby_sleep.rb", logfile=sys.stdout)或stdbuf在非交互模式下启用行缓冲:
from subprocess import Popen, PIPE, STDOUTproc = Popen(['stdbuf', '-oL', 'ruby', 'ruby_sleep.rb'], bufsize=1, stdout=PIPE, stderr=STDOUT, close_fds=True)for line in iter(proc.stdout.readline, b''): print line,proc.stdout.close()proc.wait()
或者pty基于@Antti Haapala的答案从stdlib 使用:
#!/usr/bin/env pythonimport errnoimport osimport ptyfrom subprocess import Popen, STDOUTmaster_fd, slave_fd = pty.openpty() # provide tty to enable # line-buffering on ruby's sideproc = Popen(['ruby', 'ruby_sleep.rb'], stdin=slave_fd, stdout=slave_fd, stderr=STDOUT, close_fds=True)os.close(slave_fd)try: while 1: try: data = os.read(master_fd, 512) except OSError as e: if e.errno != errno.EIO: raise break # EIO means EOF on some systems else: if not data: # EOF break print('got ' + repr(data))finally: os.close(master_fd) if proc.poll() is None: proc.kill() proc.wait()print("This is reached!")这三个代码示例都立即打印“ hello”(在看到第一个EOL时)。
此处保留旧的更复杂的代码示例,因为可能会在SO的其他帖子中进行引用和讨论
或者使用pty基于@Antti Haapala的答案:
import osimport ptyimport selectfrom subprocess import Popen, STDOUTmaster_fd, slave_fd = pty.openpty() # provide tty to enable # line-buffering on ruby's sideproc = Popen(['ruby', 'ruby_sleep.rb'], stdout=slave_fd, stderr=STDOUT, close_fds=True)timeout = .04 # secondswhile 1: ready, _, _ = select.select([master_fd], [], [], timeout) if ready: data = os.read(master_fd, 512) if not data: break print("got " + repr(data)) elif proc.poll() is not None: # select timeout assert not select.select([master_fd], [], [], 0)[0] # detect race condition break # proc exitedos.close(slave_fd) # can't do it sooner: it leads to errno.EIO erroros.close(master_fd)proc.wait()print("This is reached!")


