pseudo terminal utilities bugs no signal handling doesn t set slave termios and window size only tested on linux freebsd and macos see w richard stevens 1992 advanced programming in the unix environment chapter 19 steen lumholt with additions by guido from select import select import os import sys import tty names imported directly for test mocking purposes from os import close waitpid from tty import setraw tcgetattr tcsetattr all openpty fork spawn stdinfileno 0 stdoutfileno 1 stderrfileno 2 child 0 def openpty masteropen masterfd slavename open a pty master and return the fd and the filename of the slave end deprecated use openpty instead import warnings warnings warnuse pty openpty instead deprecationwarning stacklevel2 remove api in 3 14 try masterfd slavefd os openpty except attributeerror oserror pass else slavename os ttynameslavefd os closeslavefd return masterfd slavename return openterminal def openterminal slaveopenttyname slavefd open the pty slave and acquire the controlling terminal returning opened filedescriptor deprecated use openpty instead import warnings warnings warnuse pty openpty instead deprecationwarning stacklevel2 remove api in 3 14 result os openttyname os ordwr try from fcntl import ioctl ipush except importerror return result try ioctlresult ipush ptem ioctlresult ipush ldterm except oserror pass return result def fork os forkpty already set us session leader parent and child process default read function return os readfd 1024 def copymasterfd masterreadread stdinreadread if we write more than ttyndisc is willing to buffer we may block indefinitely so we set masterfd to nonblocking temporarily during the copy operation restore blocking mode for backwards compatibility some oses signal eof by returning an empty byte string some throw oserrors unreachable so we clean up create a spawned process if isinstanceargv str argv argv sys audit pty spawn argv pid masterfd fork if pid child os execlpargv0 argv try mode tcgetattrstdinfileno setrawstdinfileno restore true except tty error this is the same as termios error restore false try copymasterfd masterread stdinread finally if restore tcsetattrstdinfileno tty tcsaflush mode closemasterfd return waitpidpid 01 bugs no signal handling doesn t set slave termios and window size only tested on linux freebsd and macos see w richard stevens 1992 advanced programming in the unix environment chapter 19 steen lumholt with additions by guido names imported directly for test mocking purposes openpty master_fd slave_fd open a pty master slave pair using os openpty if possible master_open master_fd slave_name open a pty master and return the fd and the filename of the slave end deprecated use openpty instead remove api in 3 14 open pty master and return master_fd tty_name slave_open tty_name slave_fd open the pty slave and acquire the controlling terminal returning opened filedescriptor deprecated use openpty instead remove api in 3 14 fork pid master_fd fork and make the child a session leader with a controlling terminal os forkpty already set us session leader parent and child process default read function parent copy loop copies pty master standard output master_read standard input pty master stdin_read if we write more than tty ndisc is willing to buffer we may block indefinitely so we set master_fd to non blocking temporarily during the copy operation restore blocking mode for backwards compatibility some oses signal eof by returning an empty byte string some throw oserrors reached eof assume the child process has exited and is unreachable so we clean up create a spawned process this is the same as termios error
from select import select
import os
import sys
import tty


from os import close, waitpid
from tty import setraw, tcgetattr, tcsetattr

__all__ = ["openpty", "fork", "spawn"]

STDIN_FILENO = 0
STDOUT_FILENO = 1
STDERR_FILENO = 2

CHILD = 0

def openpty():
    

    try:
        return os.openpty()
    except (AttributeError, OSError):
        pass
    master_fd, slave_name = _open_terminal()
    slave_fd = slave_open(slave_name)
    return master_fd, slave_fd

def master_open():
    

    import warnings
    warnings.warn("Use pty.openpty() instead.", DeprecationWarning, stacklevel=2)  

    try:
        master_fd, slave_fd = os.openpty()
    except (AttributeError, OSError):
        pass
    else:
        slave_name = os.ttyname(slave_fd)
        os.close(slave_fd)
        return master_fd, slave_name

    return _open_terminal()

def _open_terminal():
    
    for x in 'pqrstuvwxyzPQRST':
        for y in '0123456789abcdef':
            pty_name = '/dev/pty' + x + y
            try:
                fd = os.open(pty_name, os.O_RDWR)
            except OSError:
                continue
            return (fd, '/dev/tty' + x + y)
    raise OSError('out of pty devices')

def slave_open(tty_name):
    

    import warnings
    warnings.warn("Use pty.openpty() instead.", DeprecationWarning, stacklevel=2)  

    result = os.open(tty_name, os.O_RDWR)
    try:
        from fcntl import ioctl, I_PUSH
    except ImportError:
        return result
    try:
        ioctl(result, I_PUSH, "ptem")
        ioctl(result, I_PUSH, "ldterm")
    except OSError:
        pass
    return result

def fork():
    

    try:
        pid, fd = os.forkpty()
    except (AttributeError, OSError):
        pass
    else:
        if pid == CHILD:
            try:
                os.setsid()
            except OSError:
                
                pass
        return pid, fd

    master_fd, slave_fd = openpty()
    pid = os.fork()
    if pid == CHILD:
        os.close(master_fd)
        os.login_tty(slave_fd)
    else:
        os.close(slave_fd)

    
    return pid, master_fd

def _read(fd):
    
    return os.read(fd, 1024)

def _copy(master_fd, master_read=_read, stdin_read=_read):
    
    if os.get_blocking(master_fd):
        
        
        
        os.set_blocking(master_fd, False)
        try:
            _copy(master_fd, master_read=master_read, stdin_read=stdin_read)
        finally:
            
            os.set_blocking(master_fd, True)
        return
    high_waterlevel = 4096
    stdin_avail = master_fd != STDIN_FILENO
    stdout_avail = master_fd != STDOUT_FILENO
    i_buf = b''
    o_buf = b''
    while 1:
        rfds = []
        wfds = []
        if stdin_avail and len(i_buf) < high_waterlevel:
            rfds.append(STDIN_FILENO)
        if stdout_avail and len(o_buf) < high_waterlevel:
            rfds.append(master_fd)
        if stdout_avail and len(o_buf) > 0:
            wfds.append(STDOUT_FILENO)
        if len(i_buf) > 0:
            wfds.append(master_fd)

        rfds, wfds, _xfds = select(rfds, wfds, [])

        if STDOUT_FILENO in wfds:
            try:
                n = os.write(STDOUT_FILENO, o_buf)
                o_buf = o_buf[n:]
            except OSError:
                stdout_avail = False

        if master_fd in rfds:
            
            
            try:
                data = master_read(master_fd)
            except OSError:
                data = b""
            if not data:  
                return    
                          
            o_buf += data

        if master_fd in wfds:
            n = os.write(master_fd, i_buf)
            i_buf = i_buf[n:]

        if stdin_avail and STDIN_FILENO in rfds:
            data = stdin_read(STDIN_FILENO)
            if not data:
                stdin_avail = False
            else:
                i_buf += data

def spawn(argv, master_read=_read, stdin_read=_read):
    
    if isinstance(argv, str):
        argv = (argv,)
    sys.audit('pty.spawn', argv)

    pid, master_fd = fork()
    if pid == CHILD:
        os.execlp(argv[0], *argv)

    try:
        mode = tcgetattr(STDIN_FILENO)
        setraw(STDIN_FILENO)
        restore = True
    except tty.error:    
        restore = False

    try:
        _copy(master_fd, master_read, stdin_read)
    finally:
        if restore:
            tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode)

    close(master_fd)
    return waitpid(pid, 0)[1]