<commit_before>import codecs
import errno
import fcntl
import io
import os
import pty
import resource
import signal
import struct
import sys
import termios
import time

# Constants
from pty import (STDIN_FILENO, CHILD)

from .util import which

_platform = sys.platform.lower()

# Solaris uses internal __fork_pty(). All others use pty.fork().
_is_solaris = not (
    _platform.startswith('solaris') or
    _platform.startswith('sunos'))

_is_irix = _platform.startswith('irix')

if _is_solaris:
    use_native_pty_fork = False
    from . import _fork_pty
else:
    use_native_pty_fork = True

PY3 = sys.version_info[0] >= 3

if PY3:
    def _byte(i):
        return bytes([i])
else:
    def _byte(i):
        return chr(i)
    
    class FileNotFoundError(OSError): pass
    class TimeoutError(OSError): pass

_EOF, _INTR = None, None

def _make_eof_intr():
    """Set constants _EOF and _INTR.
    
    This avoids doing potentially costly operations on module load.
    """
    global _EOF, _INTR
    if (_EOF is not None) and (_INTR is not None):
        pass

    # inherit EOF and INTR definitions from controlling process.
    try:
        from termios import VEOF, VINTR
        try:
            fd = sys.__stdin__.fileno()
        except ValueError:
            # ValueError: I/O operation on closed file
            fd = sys.__stdout__.fileno()
        intr = ord(termios.tcgetattr(fd)[6][VINTR])
        eof = ord(termios.tcgetattr(fd)[6][VEOF])
    except (ImportError, OSError, IOError, ValueError, termios.error):
        # unless the controlling process is also not a terminal,
        # such as cron(1), or when stdin and stdout are both closed.
        # Fall-back to using CEOF and CINTR. There
        try:
            from termios import CEOF, CINTR
            (intr, eof) = (CINTR, CEOF)
        except ImportError:
            #                         ^C, ^D
            (intr, eof) = (3, 4)
    
    _INTR = _byte(intr)
    _EOF = _byte(eof)

class PtyProcessError(Exception):
    """Generic error class for this package."""

# setecho and setwinsize are pulled out here because on some platforms, we need
# to do this from the child before we exec()
    
def _setecho(fd, state):
    errmsg = 'setecho() may not be called on this platform'

    try:
        attr = termios.tcgetattr(fd)
    except termios.error as err:
        if err.args[0] == errno.EINVAL:
            raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg))
        raise

    if state:
        attr[3] = attr[3] | termios.ECHO
    else:
        attr[3] = attr[3] & ~termios.ECHO

    try:
        # I tried TCSADRAIN and TCSAFLUSH, but these were inconsistent and
        # blocked on some platforms. TCSADRAIN would probably be ideal.
        termios.tcsetattr(fd, termios.TCSANOW, attr)
    except IOError as err:
        if err.args[0] == errno.EINVAL:
            raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg))
        raise

def _setwinsize(fd, rows, cols):
    # Some very old platforms have a bug that causes the value for
    # termios.TIOCSWINSZ to be truncated. There was a hack here to work
    # around this, but it caused problems with newer platforms so has been
    # removed. For details see https://github.com/pexpect/pexpect/issues/39
    TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)
    # Note, assume ws_xpixel and ws_ypixel are zero.
    s = struct.pack('HHHH', rows, cols, 0, 0)
    fcntl.ioctl(fd, TIOCSWINSZ, s)

class PtyProcess(object):
    '''This class represents a process running in a pseudoterminal.
    
    The main constructor is the :meth:`spawn` classmethod.
    '''
    string_type = bytes
    if PY3:
        linesep = os.linesep.encode('ascii')
        crlf = '\r\n'.encode('ascii')

        @staticmethod
        def write_to_stdout(b):
            try:
                return sys.stdout.buffer.write(b)
            except AttributeError:
                # If stdout has been replaced, it may not have .buffer
                return sys.stdout.write(b.decode('ascii', 'replace'))
    else:
        linesep = os.linesep
        crlf = '\r\n'
        write_to_stdout = sys.stdout.write

    encoding = None
    
    argv = None
    env = None
    launch_dir = None

    def __init__(self, pid, fd):
        _make_eof_intr()  # Ensure _EOF and _INTR are calculated
        self.pid = pid
        self.fd = fd
        self.fileobj = self.fileobj_bytes = io.open(fd, 'r+b', buffering=0)

        self.terminated = False
        self.closed = False
        self.exitstatus = None
        self.signalstatus = None
        # status returned by os.waitpid
        self.status = None
        self.flag_eof = False
        # Delay used before sending data to child. Time in seconds.
        # Most Linux machines don't like this to be below 0.03 (30 ms).
        self.delaybeforesend = 0.05
        # Used by close() to give kernel time to update process status.
        # Time in seconds.
        self.delayafterclose = 0.1
        # Used by terminate() to give kernel time to update process status.
        # Time in seconds.
        self.delayafterterminate = 0.1

    @classmethod
    def spawn(cls, argv, cwd=None, env=None, echo=True, before_exec=()):
        '''Start the given command in a child process in a pseudo terminal.
        
        This does all the fork/exec type of stuff for a pty, and returns an
        instance of PtyProcess.
        '''
        # Note that it is difficult for this method to fail.
        # You cannot detect if the child process cannot start.
        # So the only way you can tell if the child process started
        # or not is to try to read from the file descriptor. If you get
        # EOF immediately then it means that the child is already dead.
        # That may not necessarily be bad because you may have spawned a child
        # that performs some task; creates no stdout output; and then dies.<commit_msg>Type checking for argv (because it's easy to try passing a string)<commit_after>import codecs
import errno
import fcntl
import io
import os
import pty
import resource
import signal
import struct
import sys
import termios
import time

# Constants
from pty import (STDIN_FILENO, CHILD)

from .util import which

_platform = sys.platform.lower()

# Solaris uses internal __fork_pty(). All others use pty.fork().
_is_solaris = not (
    _platform.startswith('solaris') or
    _platform.startswith('sunos'))

_is_irix = _platform.startswith('irix')

if _is_solaris:
    use_native_pty_fork = False
    from . import _fork_pty
else:
    use_native_pty_fork = True

PY3 = sys.version_info[0] >= 3

if PY3:
    def _byte(i):
        return bytes([i])
else:
    def _byte(i):
        return chr(i)
    
    class FileNotFoundError(OSError): pass
    class TimeoutError(OSError): pass

_EOF, _INTR = None, None

def _make_eof_intr():
    """Set constants _EOF and _INTR.
    
    This avoids doing potentially costly operations on module load.
    """
    global _EOF, _INTR
    if (_EOF is not None) and (_INTR is not None):
        pass

    # inherit EOF and INTR definitions from controlling process.
    try:
        from termios import VEOF, VINTR
        try:
            fd = sys.__stdin__.fileno()
        except ValueError:
            # ValueError: I/O operation on closed file
            fd = sys.__stdout__.fileno()
        intr = ord(termios.tcgetattr(fd)[6][VINTR])
        eof = ord(termios.tcgetattr(fd)[6][VEOF])
    except (ImportError, OSError, IOError, ValueError, termios.error):
        # unless the controlling process is also not a terminal,
        # such as cron(1), or when stdin and stdout are both closed.
        # Fall-back to using CEOF and CINTR. There
        try:
            from termios import CEOF, CINTR
            (intr, eof) = (CINTR, CEOF)
        except ImportError:
            #                         ^C, ^D
            (intr, eof) = (3, 4)
    
    _INTR = _byte(intr)
    _EOF = _byte(eof)

class PtyProcessError(Exception):
    """Generic error class for this package."""

# setecho and setwinsize are pulled out here because on some platforms, we need
# to do this from the child before we exec()
    
def _setecho(fd, state):
    errmsg = 'setecho() may not be called on this platform'

    try:
        attr = termios.tcgetattr(fd)
    except termios.error as err:
        if err.args[0] == errno.EINVAL:
            raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg))
        raise

    if state:
        attr[3] = attr[3] | termios.ECHO
    else:
        attr[3] = attr[3] & ~termios.ECHO

    try:
        # I tried TCSADRAIN and TCSAFLUSH, but these were inconsistent and
        # blocked on some platforms. TCSADRAIN would probably be ideal.
        termios.tcsetattr(fd, termios.TCSANOW, attr)
    except IOError as err:
        if err.args[0] == errno.EINVAL:
            raise IOError(err.args[0], '%s: %s.' % (err.args[1], errmsg))
        raise

def _setwinsize(fd, rows, cols):
    # Some very old platforms have a bug that causes the value for
    # termios.TIOCSWINSZ to be truncated. There was a hack here to work
    # around this, but it caused problems with newer platforms so has been
    # removed. For details see https://github.com/pexpect/pexpect/issues/39
    TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)
    # Note, assume ws_xpixel and ws_ypixel are zero.
    s = struct.pack('HHHH', rows, cols, 0, 0)
    fcntl.ioctl(fd, TIOCSWINSZ, s)

class PtyProcess(object):
    '''This class represents a process running in a pseudoterminal.
    
    The main constructor is the :meth:`spawn` classmethod.
    '''
    string_type = bytes
    if PY3:
        linesep = os.linesep.encode('ascii')
        crlf = '\r\n'.encode('ascii')

        @staticmethod
        def write_to_stdout(b):
            try:
                return sys.stdout.buffer.write(b)
            except AttributeError:
                # If stdout has been replaced, it may not have .buffer
                return sys.stdout.write(b.decode('ascii', 'replace'))
    else:
        linesep = os.linesep
        crlf = '\r\n'
        write_to_stdout = sys.stdout.write

    encoding = None
    
    argv = None
    env = None
    launch_dir = None

    def __init__(self, pid, fd):
        _make_eof_intr()  # Ensure _EOF and _INTR are calculated
        self.pid = pid
        self.fd = fd
        self.fileobj = self.fileobj_bytes = io.open(fd, 'r+b', buffering=0)

        self.terminated = False
        self.closed = False
        self.exitstatus = None
        self.signalstatus = None
        # status returned by os.waitpid
        self.status = None
        self.flag_eof = False
        # Delay used before sending data to child. Time in seconds.
        # Most Linux machines don't like this to be below 0.03 (30 ms).
        self.delaybeforesend = 0.05
        # Used by close() to give kernel time to update process status.
        # Time in seconds.
        self.delayafterclose = 0.1
        # Used by terminate() to give kernel time to update process status.
        # Time in seconds.
        self.delayafterterminate = 0.1

    @classmethod
    def spawn(cls, argv, cwd=None, env=None, echo=True, before_exec=()):
        '''Start the given command in a child process in a pseudo terminal.
        
        This does all the fork/exec type of stuff for a pty, and returns an
        instance of PtyProcess.
        '''
        # Note that it is difficult for this method to fail.
        # You cannot detect if the child process cannot start.
        # So the only way you can tell if the child process started
        # or not is to try to read from the file descriptor. If you get
        # EOF immediately then it means that the child is already dead.
        # That may not necessarily be bad because you may have spawned a child
        # that performs some task; creates no stdout output; and then dies.

        if not isinstance(argv, (list, tuple)):
            raise TypeError("Expected a list or tuple for argv, got %r" % argv)
{'fraction_non_alphanumeric': 0.08753117206982544, 'fraction_numerical': 0.007813798836242726, 'mean_word_length': 2.319812362030905, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3677575', 'n_tokens_mistral': 3780, 'n_tokens_neox': 3626, 'n_words': 1312}