我们从Python开源项目中,提取了以下37个代码示例,用于说明如何使用tty.TCSAFLUSH。
def spawn(argv, master_read=_read, stdin_read=_read): """Create a spawned process.""" if type(argv) == type(''): argv = (argv,) pid, master_fd = fork() if pid == CHILD: os.execlp(argv[0], *argv) try: mode = tty.tcgetattr(STDIN_FILENO) tty.setraw(STDIN_FILENO) restore = 1 except tty.error: # This is the same as termios.error restore = 0 try: _copy(master_fd, master_read, stdin_read) except (IOError, OSError): if restore: tty.tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode) os.close(master_fd)
def spawn(argv, master_read=_read, stdin_read=_read): """Create a spawned process.""" if type(argv) == type(''): argv = (argv,) pid, master_fd = fork() if pid == CHILD: os.execlp(argv[0], *argv) try: mode = tty.tcgetattr(STDIN_FILENO) tty.setraw(STDIN_FILENO) restore = 1 except tty.error: # This is the same as termios.error restore = 0 try: _copy(master_fd, master_read, stdin_read) except OSError: if restore: tty.tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode) os.close(master_fd)
def spawn(argv, master_read=_read, stdin_read=_read): """Create a spawned process.""" if type(argv) == type(''): argv = (argv,) pid, master_fd = fork() if pid == CHILD: os.execlp(argv[0], *argv) try: mode = tty.tcgetattr(STDIN_FILENO) tty.setraw(STDIN_FILENO) restore = 1 except tty.error: # This is the same as termios.error restore = 0 try: _copy(master_fd, master_read, stdin_read) except OSError: if restore: tty.tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode) os.close(master_fd) return os.waitpid(pid, 0)[1]
def ttypager(text): """Page through text on a text terminal.""" lines = split(plain(text), '\n') try: import tty fd = sys.stdin.fileno() old = tty.tcgetattr(fd) tty.setcbreak(fd) getchar = lambda: sys.stdin.read(1) except (ImportError, AttributeError): tty = None getchar = lambda: sys.stdin.readline()[:-1][:1] try: r = inc = os.environ.get('LINES', 25) - 1 sys.stdout.write(join(lines[:inc], '\n') + '\n') while lines[r:]: sys.stdout.write('-- more --') sys.stdout.flush() c = getchar() if c in ('q', 'Q'): sys.stdout.write('\r \r') break elif c in ('\r', '\n'): sys.stdout.write('\r \r' + lines[r] + '\n') r = r + 1 continue if c in ('b', 'B', '\x1b'): r = r - inc - inc if r < 0: r = 0 sys.stdout.write('\n' + join(lines[r:r+inc], '\n') + '\n') r = r + inc finally: if tty: tty.tcsetattr(fd, tty.TCSAFLUSH, old)
def ttypager(text): """Page through text on a text terminal.""" lines = plain(_encode(plain(text), getattr(sys.stdout, 'encoding', _encoding))).split('\n') try: import tty fd = sys.stdin.fileno() old = tty.tcgetattr(fd) tty.setcbreak(fd) getchar = lambda: sys.stdin.read(1) except (ImportError, AttributeError): tty = None getchar = lambda: sys.stdin.readline()[:-1][:1] try: r = inc = os.environ.get('LINES', 25) - 1 sys.stdout.write(join(lines[:inc], '\n') + '\n') while lines[r:]: sys.stdout.write('-- more --') sys.stdout.flush() c = getchar() if c in ('q', 'Q'): sys.stdout.write('\r \r') break elif c in ('\r', '\n'): sys.stdout.write('\r \r' + lines[r] + '\n') r = r + 1 continue if c in ('b', 'B', '\x1b'): r = r - inc - inc if r < 0: r = 0 sys.stdout.write('\n' + join(lines[r:r+inc], '\n') + '\n') r = r + inc finally: if tty: tty.tcsetattr(fd, tty.TCSAFLUSH, old)
def ttypager(text): """Page through text on a text terminal.""" lines = plain(text).split('\n') try: import tty fd = sys.stdin.fileno() old = tty.tcgetattr(fd) tty.setcbreak(fd) getchar = lambda: sys.stdin.read(1) except (ImportError, AttributeError): tty = None getchar = lambda: sys.stdin.readline()[:-1][:1] try: r = inc = os.environ.get('LINES', 25) - 1 sys.stdout.write('\n'.join(lines[:inc]) + '\n') while lines[r:]: sys.stdout.write('-- more --') sys.stdout.flush() c = getchar() if c in ('q', 'Q'): sys.stdout.write('\r \r') break elif c in ('\r', '\n'): sys.stdout.write('\r \r' + lines[r] + '\n') r = r + 1 continue if c in ('b', 'B', '\x1b'): r = r - inc - inc if r < 0: r = 0 sys.stdout.write('\n' + '\n'.join(lines[r:r+inc]) + '\n') r = r + inc finally: if tty: tty.tcsetattr(fd, tty.TCSAFLUSH, old)
def _del_tty(self): if self.mode: # restore the tty mode for stdout tty.tcsetattr(pty.STDIN_FILENO, tty.TCSAFLUSH, self.mode)
def interact(self, escape_character=chr(29), input_filter=None, output_filter=None): '''This gives control of the child process to the interactive user (the human at the keyboard). Keystrokes are sent to the child process, and the stdout and stderr output of the child process is printed. This simply echos the child stdout and child stderr to the real stdout and it echos the real stdin to the child stdin. When the user types the escape_character this method will return None. The escape_character will not be transmitted. The default for escape_character is entered as ``Ctrl - ]``, the very same as BSD telnet. To prevent escaping, escape_character may be set to None. If a logfile is specified, then the data sent and received from the child process in interact mode is duplicated to the given log. You may pass in optional input and output filter functions. These functions should take a string and return a string. The output_filter will be passed all the output from the child process. The input_filter will be passed all the keyboard input from the user. The input_filter is run BEFORE the check for the escape_character. Note that if you change the window size of the parent the SIGWINCH signal will not be passed through to the child. If you want the child window size to change when the parent's window size changes then do something like the following example:: import pexpect, struct, fcntl, termios, signal, sys def sigwinch_passthrough (sig, data): s = struct.pack("HHHH", 0, 0, 0, 0) a = struct.unpack('hhhh', fcntl.ioctl(sys.stdout.fileno(), termios.TIOCGWINSZ , s)) global p p.setwinsize(a[0],a[1]) # Note this 'p' global and used in sigwinch_passthrough. p = pexpect.spawn('/bin/bash') signal.signal(signal.SIGWINCH, sigwinch_passthrough) p.interact() ''' # Flush the buffer. self.write_to_stdout(self.buffer) self.stdout.flush() self.buffer = self.string_type() mode = tty.tcgetattr(self.STDIN_FILENO) tty.setraw(self.STDIN_FILENO) if escape_character is not None and PY3: escape_character = escape_character.encode('latin-1') try: self.__interact_copy(escape_character, input_filter, output_filter) finally: tty.tcsetattr(self.STDIN_FILENO, tty.TCSAFLUSH, mode)
def setecho(self, state): """This sets the terminal echo mode on or off. Note that anything the child sent before the echo will be lost, so you should be sure that your input buffer is empty before you call setecho(). For example, the following will work as expected:: p = pexpect.spawn('cat') p.sendline ('1234') # We will see this twice (once from tty echo and again from cat). p.expect (['1234']) p.expect (['1234']) p.setecho(False) # Turn off tty echo p.sendline ('abcd') # We will set this only once (echoed by cat). p.sendline ('wxyz') # We will set this only once (echoed by cat) p.expect (['abcd']) p.expect (['wxyz']) The following WILL NOT WORK because the lines sent before the setecho will be lost:: p = pexpect.spawn('cat') p.sendline ('1234') # We will see this twice (once from tty echo and again from cat). p.setecho(False) # Turn off tty echo p.sendline ('abcd') # We will set this only once (echoed by cat). p.sendline ('wxyz') # We will set this only once (echoed by cat) p.expect (['1234']) p.expect (['1234']) p.expect (['abcd']) p.expect (['wxyz']) """ self.child_fd attr = termios.tcgetattr(self.child_fd) if state: attr[3] = attr[3] | termios.ECHO else: attr[3] = attr[3] & ~termios.ECHO # I tried TCSADRAIN and TCSAFLUSH, but these were inconsistent # and blocked on some platforms. TCSADRAIN is probably ideal if it # worked. termios.tcsetattr(self.child_fd, termios.TCSANOW, attr)
def ttypager(text): """Page through text on a text terminal.""" lines = plain(_encode(plain(text), getattr(sys.stdout, 'encoding', _encoding))).split('\n') try: import tty fd = sys.stdin.fileno() old = tty.tcgetattr(fd) tty.setcbreak(fd) getchar = lambda: sys.stdin.read(1) except (ImportError, AttributeError): tty = None getchar = lambda: sys.stdin.readline()[:-1][:1] try: try: h = int(os.environ.get('LINES', 0)) except ValueError: h = 0 if h <= 1: h = 25 r = inc = h - 1 sys.stdout.write(join(lines[:inc], '\n') + '\n') while lines[r:]: sys.stdout.write('-- more --') sys.stdout.flush() c = getchar() if c in ('q', 'Q'): sys.stdout.write('\r \r') break elif c in ('\r', '\n'): sys.stdout.write('\r \r' + lines[r] + '\n') r = r + 1 continue if c in ('b', 'B', '\x1b'): r = r - inc - inc if r < 0: r = 0 sys.stdout.write('\n' + join(lines[r:r+inc], '\n') + '\n') r = r + inc finally: if tty: tty.tcsetattr(fd, tty.TCSAFLUSH, old)
def setecho(self, state): '''This sets the terminal echo mode on or off. Note that anything the child sent before the echo will be lost, so you should be sure that your input buffer is empty before you call setecho(). For example, the following will work as expected:: p = pexpect.spawn('cat') # Echo is on by default. p.sendline('1234') # We expect see this twice from the child... p.expect(['1234']) # ... once from the tty echo... p.expect(['1234']) # ... and again from cat itself. p.setecho(False) # Turn off tty echo p.sendline('abcd') # We will set this only once (echoed by cat). p.sendline('wxyz') # We will set this only once (echoed by cat) p.expect(['abcd']) p.expect(['wxyz']) The following WILL NOT WORK because the lines sent before the setecho will be lost:: p = pexpect.spawn('cat') p.sendline('1234') p.setecho(False) # Turn off tty echo p.sendline('abcd') # We will set this only once (echoed by cat). p.sendline('wxyz') # We will set this only once (echoed by cat) p.expect(['1234']) p.expect(['1234']) p.expect(['abcd']) p.expect(['wxyz']) Not supported on platforms where ``isatty()`` returns False. ''' errmsg = 'setecho() may not be called on this platform' try: attr = termios.tcgetattr(self.child_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(self.child_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 self.echo = state
def interact(self, escape_character=chr(29), input_filter=None, output_filter=None): '''This gives control of the child process to the interactive user (the human at the keyboard). Keystrokes are sent to the child process, and the stdout and stderr output of the child process is printed. This simply echos the child stdout and child stderr to the real stdout and it echos the real stdin to the child stdin. When the user types the escape_character this method will stop. The default for escape_character is ^]. This should not be confused with ASCII 27 -- the ESC character. ASCII 29 was chosen for historical merit because this is the character used by 'telnet' as the escape character. The escape_character will not be sent to the child process. You may pass in optional input and output filter functions. These functions should take a string and return a string. The output_filter will be passed all the output from the child process. The input_filter will be passed all the keyboard input from the user. The input_filter is run BEFORE the check for the escape_character. Note that if you change the window size of the parent the SIGWINCH signal will not be passed through to the child. If you want the child window size to change when the parent's window size changes then do something like the following example:: import pexpect, struct, fcntl, termios, signal, sys def sigwinch_passthrough (sig, data): s = struct.pack("HHHH", 0, 0, 0, 0) a = struct.unpack('hhhh', fcntl.ioctl(sys.stdout.fileno(), termios.TIOCGWINSZ , s)) global p p.setwinsize(a[0],a[1]) # Note this 'p' global and used in sigwinch_passthrough. p = pexpect.spawn('/bin/bash') signal.signal(signal.SIGWINCH, sigwinch_passthrough) p.interact() ''' # Flush the buffer. self.write_to_stdout(self.buffer) self.stdout.flush() self.buffer = self.string_type() mode = tty.tcgetattr(self.STDIN_FILENO) tty.setraw(self.STDIN_FILENO) if PY3: escape_character = escape_character.encode('latin-1') try: self.__interact_copy(escape_character, input_filter, output_filter) finally: tty.tcsetattr(self.STDIN_FILENO, tty.TCSAFLUSH, mode)
def ttypager(text): """Page through text on a text terminal.""" lines = plain(_escape_stdout(text)).split('\n') try: import tty fd = sys.stdin.fileno() old = tty.tcgetattr(fd) tty.setcbreak(fd) getchar = lambda: sys.stdin.read(1) except (ImportError, AttributeError, io.UnsupportedOperation): tty = None getchar = lambda: sys.stdin.readline()[:-1][:1] try: try: h = int(os.environ.get('LINES', 0)) except ValueError: h = 0 if h <= 1: h = 25 r = inc = h - 1 sys.stdout.write('\n'.join(lines[:inc]) + '\n') while lines[r:]: sys.stdout.write('-- more --') sys.stdout.flush() c = getchar() if c in ('q', 'Q'): sys.stdout.write('\r \r') break elif c in ('\r', '\n'): sys.stdout.write('\r \r' + lines[r] + '\n') r = r + 1 continue if c in ('b', 'B', '\x1b'): r = r - inc - inc if r < 0: r = 0 sys.stdout.write('\n' + '\n'.join(lines[r:r+inc]) + '\n') r = r + inc finally: if tty: tty.tcsetattr(fd, tty.TCSAFLUSH, old)
def interact(self, escape_character=chr(29), input_filter=None, output_filter=None): """This gives control of the child process to the interactive user (the human at the keyboard). Keystrokes are sent to the child process, and the stdout and stderr output of the child process is printed. This simply echos the child stdout and child stderr to the real stdout and it echos the real stdin to the child stdin. When the user types the escape_character this method will stop. The default for escape_character is ^]. This should not be confused with ASCII 27 -- the ESC character. ASCII 29 was chosen for historical merit because this is the character used by 'telnet' as the escape character. The escape_character will not be sent to the child process. You may pass in optional input and output filter functions. These functions should take a string and return a string. The output_filter will be passed all the output from the child process. The input_filter will be passed all the keyboard input from the user. The input_filter is run BEFORE the check for the escape_character. Note that if you change the window size of the parent the SIGWINCH signal will not be passed through to the child. If you want the child window size to change when the parent's window size changes then do something like the following example:: import pexpect, struct, fcntl, termios, signal, sys def sigwinch_passthrough (sig, data): s = struct.pack("HHHH", 0, 0, 0, 0) a = struct.unpack('hhhh', fcntl.ioctl(sys.stdout.fileno(), termios.TIOCGWINSZ , s)) global p p.setwinsize(a[0],a[1]) p = pexpect.spawn('/bin/bash') # Note this is global and used in sigwinch_passthrough. signal.signal(signal.SIGWINCH, sigwinch_passthrough) p.interact() """ # Flush the buffer. self.stdout.write(self.buffer) self.stdout.flush() self.buffer = '' mode = tty.tcgetattr(self.STDIN_FILENO) tty.setraw(self.STDIN_FILENO) try: self.__interact_copy(escape_character, input_filter, output_filter) finally: tty.tcsetattr(self.STDIN_FILENO, tty.TCSAFLUSH, mode)