我们从Python开源项目中，提取了以下16个代码示例，用于说明如何使用tty.TCSANOW。
def closed(self): global old log.msg('closed %s' % self) log.msg(repr(self.conn.channels)) if not options['nocache']: # fork into the background if os.fork(): if old: fd = sys.stdin.fileno() tty.tcsetattr(fd, tty.TCSANOW, old) if (options['command'] and options['tty']) or \ not options['notty']: signal.signal(signal.SIGWINCH, signal.SIG_DFL) os._exit(0) os.setsid() for i in range(3): try: os.close(i) except OSError, e: import errno if e.errno != errno.EBADF: raise
def execCommand(self, proto, cmd): from twisted.internet import reactor uid, gid = self.avatar.getUserGroupId() homeDir = self.avatar.getHomeDir() shell = self.avatar.getShell() or '/bin/sh' command = (shell, '-c', cmd) peer = self.avatar.conn.transport.transport.getPeer() host = self.avatar.conn.transport.transport.getHost() self.environ['SSH_CLIENT'] = '%s %s %s' % (peer.host, peer.port, host.port) if self.ptyTuple: self.getPtyOwnership() self.pty = reactor.spawnProcess(proto, \ shell, command, self.environ, homeDir, uid, gid, usePTY = self.ptyTuple or 0) if self.ptyTuple: self.addUTMPEntry() if self.modes: self.setModes() # else: # tty.setraw(self.pty.pipes[0].fileno(), tty.TCSANOW) self.avatar.conn.transport.transport.setTcpNoDelay(1)
def _leaveRawMode(): global _inRawMode if not _inRawMode: return fd = sys.stdin.fileno() tty.tcsetattr(fd, tty.TCSANOW, _savedMode) _inRawMode = 0
def _enterRawMode(): global _inRawMode, _savedMode if _inRawMode: return fd = sys.stdin.fileno() try: old = tty.tcgetattr(fd) new = old[:] except: log.msg('not a typewriter!') else: # iflage new[0] = new[0] | tty.IGNPAR new[0] = new[0] & ~(tty.ISTRIP | tty.INLCR | tty.IGNCR | tty.ICRNL | tty.IXON | tty.IXANY | tty.IXOFF) if hasattr(tty, 'IUCLC'): new[0] = new[0] & ~tty.IUCLC # lflag new[3] = new[3] & ~(tty.ISIG | tty.ICANON | tty.ECHO | tty.ECHO | tty.ECHOE | tty.ECHOK | tty.ECHONL) if hasattr(tty, 'IEXTEN'): new[3] = new[3] & ~tty.IEXTEN #oflag new[1] = new[1] & ~tty.OPOST new[6][tty.VMIN] = 1 new[6][tty.VTIME] = 0 _savedMode = old tty.tcsetattr(fd, tty.TCSANOW, new) #tty.setraw(fd) _inRawMode = 1
def __init__(self, infile): if not infile.isatty(): raise NotTTYException() self.file = infile # prepare for getch self.save_attr = tty.tcgetattr(self.file) newattr = self.save_attr[:] newattr[3] &= ~tty.ECHO & ~tty.ICANON tty.tcsetattr(self.file, tty.TCSANOW, newattr)
def _leaveRawMode(): global _inRawMode if not _inRawMode: return fd = sys.stdin.fileno() tty.tcsetattr(fd, tty.TCSANOW, _savedRawMode) _inRawMode = 0
def _enterRawMode(): global _inRawMode, _savedRawMode if _inRawMode: return fd = sys.stdin.fileno() try: old = tty.tcgetattr(fd) new = old[:] except: log.msg('not a typewriter!') else: # iflage new[0] = new[0] | tty.IGNPAR new[0] = new[0] & ~(tty.ISTRIP | tty.INLCR | tty.IGNCR | tty.ICRNL | tty.IXON | tty.IXANY | tty.IXOFF) if hasattr(tty, 'IUCLC'): new[0] = new[0] & ~tty.IUCLC # lflag new[3] = new[3] & ~(tty.ISIG | tty.ICANON | tty.ECHO | tty.ECHO | tty.ECHOE | tty.ECHOK | tty.ECHONL) if hasattr(tty, 'IEXTEN'): new[3] = new[3] & ~tty.IEXTEN #oflag new[1] = new[1] & ~tty.OPOST new[6][tty.VMIN] = 1 new[6][tty.VTIME] = 0 _savedRawMode = old tty.tcsetattr(fd, tty.TCSANOW, new) #tty.setraw(fd) _inRawMode = 1
def setModes(self): pty = self.pty attr = tty.tcgetattr(pty.fileno()) for mode, modeValue in self.modes: if mode not in ttymodes.TTYMODES: continue ttyMode = ttymodes.TTYMODES[mode] if len(ttyMode) == 2: # Flag. flag, ttyAttr = ttyMode if not hasattr(tty, ttyAttr): continue ttyval = getattr(tty, ttyAttr) if modeValue: attr[flag] = attr[flag] | ttyval else: attr[flag] = attr[flag] & ~ttyval elif ttyMode == 'OSPEED': attr[tty.OSPEED] = getattr(tty, 'B%s' % (modeValue,)) elif ttyMode == 'ISPEED': attr[tty.ISPEED] = getattr(tty, 'B%s' % (modeValue,)) else: if not hasattr(tty, ttyMode): continue ttyval = getattr(tty, ttyMode) attr[tty.CC][ttyval] = chr(modeValue) tty.tcsetattr(pty.fileno(), tty.TCSANOW, attr)
def run(): global options, old args = sys.argv[1:] if '-l' in args: # cvs is an idiot i = args.index('-l') args = args[i:i+2]+args del args[i+2:i+4] for arg in args[:]: try: i = args.index(arg) if arg[:2] == '-o' and args[i+1][0]!='-': args[i:i+2] = [] # suck on it scp except ValueError: pass options = ClientOptions() try: options.parseOptions(args) except usage.UsageError, u: print 'ERROR: %s' % u options.opt_help() sys.exit(1) if options['log']: if options['logfile']: if options['logfile'] == '-': f = sys.stdout else: f = file(options['logfile'], 'a+') else: f = sys.stderr realout = sys.stdout log.startLogging(f) sys.stdout = realout else: log.discardLogs() doConnect() fd = sys.stdin.fileno() try: old = tty.tcgetattr(fd) except: old = None try: oldUSR1 = signal.signal(signal.SIGUSR1, lambda *a: reactor.callLater(0, reConnect)) except: oldUSR1 = None try: reactor.run() finally: if old: tty.tcsetattr(fd, tty.TCSANOW, old) if oldUSR1: signal.signal(signal.SIGUSR1, oldUSR1) if (options['command'] and options['tty']) or not options['notty']: signal.signal(signal.SIGWINCH, signal.SIG_DFL) if sys.stdout.isatty() and not options['command']: print 'Connection to %s closed.' % options['host'] sys.exit(exitStatus)
def open(self, command, env={}): """ Create subprocess using forkpty() """ # parse command command_arr = shlex.split(command) executable = command_arr[0] args = command_arr # try to fork a new pty try: self.pid, self.fd = pty.fork() except: return False # child proc, replace with command after altering terminal attributes if self.pid == 0: # set requested environment variables for k in env.keys(): os.environ[k] = env[k] # set tty attributes try: attrs = tty.tcgetattr(1) attrs[0] = attrs[0] ^ tty.IGNBRK attrs[0] = attrs[0] | tty.BRKINT | tty.IXANY | tty.IMAXBEL attrs[2] = attrs[2] | tty.HUPCL attrs[3] = attrs[3] | tty.ICANON | tty.ECHO | tty.ISIG | tty.ECHOKE attrs[6][tty.VMIN] = 1 attrs[6][tty.VTIME] = 0 tty.tcsetattr(1, tty.TCSANOW, attrs) except: pass # replace this process with the subprocess os.execvp(executable, args) # else master, do nothing else: pass
def run(): global options, old args = sys.argv[1:] if '-l' in args: # cvs is an idiot i = args.index('-l') args = args[i:i+2]+args del args[i+2:i+4] for arg in args[:]: try: i = args.index(arg) if arg[:2] == '-o' and args[i+1][0]!='-': args[i:i+2] = [] # suck on it scp except ValueError: pass options = ClientOptions() try: options.parseOptions(args) except usage.UsageError as u: print('ERROR: %s' % u) options.opt_help() sys.exit(1) if options['log']: if options['logfile']: if options['logfile'] == '-': f = sys.stdout else: f = open(options['logfile'], 'a+') else: f = sys.stderr realout = sys.stdout log.startLogging(f) sys.stdout = realout else: log.discardLogs() doConnect() fd = sys.stdin.fileno() try: old = tty.tcgetattr(fd) except: old = None try: oldUSR1 = signal.signal(signal.SIGUSR1, lambda *a: reactor.callLater(0, reConnect)) except: oldUSR1 = None try: reactor.run() finally: if old: tty.tcsetattr(fd, tty.TCSANOW, old) if oldUSR1: signal.signal(signal.SIGUSR1, oldUSR1) if (options['command'] and options['tty']) or not options['notty']: signal.signal(signal.SIGWINCH, signal.SIG_DFL) if sys.stdout.isatty() and not options['command']: print('Connection to %s closed.' % options['host']) sys.exit(exitStatus)
