我们从Python开源项目中，提取了以下24个代码示例，用于说明如何使用signal.SIGWINCH。
def init_signals(self): # Set up signals through the event loop API. self.loop.add_signal_handler(signal.SIGQUIT, self.handle_quit, signal.SIGQUIT, None) self.loop.add_signal_handler(signal.SIGTERM, self.handle_exit, signal.SIGTERM, None) self.loop.add_signal_handler(signal.SIGINT, self.handle_quit, signal.SIGINT, None) self.loop.add_signal_handler(signal.SIGWINCH, self.handle_winch, signal.SIGWINCH, None) self.loop.add_signal_handler(signal.SIGUSR1, self.handle_usr1, signal.SIGUSR1, None) self.loop.add_signal_handler(signal.SIGABRT, self.handle_abort, signal.SIGABRT, None) # Don't let SIGTERM and SIGUSR1 disturb active requests # by interrupting system calls signal.siginterrupt(signal.SIGTERM, False) signal.siginterrupt(signal.SIGUSR1, False)
def prepare(self): # per-readline preparations: self.__svtermstate = tcgetattr(self.input_fd) raw = self.__svtermstate.copy() raw.iflag |= termios.ICRNL raw.iflag &= ~(termios.BRKINT | termios.INPCK | termios.ISTRIP | termios.IXON) raw.oflag &= ~termios.OPOST raw.cflag &= ~(termios.CSIZE | termios.PARENB) raw.cflag |= (termios.CS8) raw.lflag &= ~(termios.ICANON | termios.ECHO | termios.IEXTEN | (termios.ISIG * 1)) raw.cc[termios.VMIN] = 1 raw.cc[termios.VTIME] = 0 tcsetattr(self.input_fd, termios.TCSADRAIN, raw) self.screen = [] self.height, self.width = self.getheightwidth() self.__buffer = [] self.__posxy = 0, 0 self.__gone_tall = 0 self.__move = self.__move_short self.__offset = 0 self.__maybe_write_code(self._smkx) try: self.old_sigwinch = signal.signal( signal.SIGWINCH, self.__sigwinch) except ValueError: pass
def restore(self): self.__maybe_write_code(self._rmkx) self.flushoutput() tcsetattr(self.input_fd, termios.TCSADRAIN, self.__svtermstate) if hasattr(self, 'old_sigwinch'): signal.signal(signal.SIGWINCH, self.old_sigwinch) del self.old_sigwinch
def init_signals(self): # Set up signals through the event loop API. loop = asyncio.get_event_loop() loop.add_signal_handler(signal.SIGQUIT, self.handle_quit, signal.SIGQUIT, None) loop.add_signal_handler(signal.SIGTERM, self.handle_exit, signal.SIGTERM, None) loop.add_signal_handler(signal.SIGINT, self.handle_quit, signal.SIGINT, None) loop.add_signal_handler(signal.SIGWINCH, self.handle_winch, signal.SIGWINCH, None) loop.add_signal_handler(signal.SIGUSR1, self.handle_usr1, signal.SIGUSR1, None) loop.add_signal_handler(signal.SIGABRT, self.handle_abort, signal.SIGABRT, None) # Don't let SIGTERM and SIGUSR1 disturb active requests # by interrupting system calls signal.siginterrupt(signal.SIGTERM, False) signal.siginterrupt(signal.SIGUSR1, False)
def start(self): """Start Start trapping WINCH signals and resizing the PTY. This method saves the previous WINCH handler so it can be restored on `stop()`. """ def handle(signum, frame): if signum == signal.SIGWINCH: LOG.debug("Send command to resize the tty session") self.client.handle_resize() self.original_handler = signal.signal(signal.SIGWINCH, handle)
def stop(self): """stop Stop trapping WINCH signals and restore the previous WINCH handler. """ if self.original_handler is not None: signal.signal(signal.SIGWINCH, self.original_handler)
def resize(self, rows, cols, em_dimensions=None, ctrl_l=True): """ Resizes the child process's terminal window to *rows* and *cols* by first sending it a TIOCSWINSZ event and then sending ctrl-l. If *em_dimensions* are provided they will be updated along with the rows and cols. The sending of ctrl-l can be disabled by setting *ctrl_l* to False. """ logging.debug( "Resizing term %s to rows: %s, cols: %s, em_dimensions=%s" % (self.term_id, rows, cols, em_dimensions)) if rows < 2: rows = 24 if cols < 2: cols = 80 self.rows = rows self.cols = cols self.term.resize(rows, cols, em_dimensions) # Sometimes the resize doesn't actually apply (for whatever reason) # so to get around this we have to send a different value than the # actual value we want then send our actual value. It's a bug outside # of Gate One that I have no idea how to isolate but this has proven to # be an effective workaround. import fcntl, termios s = struct.pack("HHHH", rows, cols, 0, 0) try: fcntl.ioctl(self.fd, termios.TIOCSWINSZ, s) except IOError: # Process already ended--no big deal return try: os.kill(self.pid, signal.SIGWINCH) # Send the resize signal except OSError: return # Process is dead. Can happen when things go quickly if ctrl_l: self.write(u'\x0c') # ctrl-l
def __exit__(self, *a): # Quit UI application. if self.app.is_running: self.app.set_result(None) # Remove WINCH handler. if self._has_sigwinch: self._loop.add_signal_handler(signal.SIGWINCH, self._previous_winch_handler) self._thread.join()
def _window_resize(self, signum, frame): """Signal handler for SIGWINCH. Generates a message with the current demensions of the terminal and puts it in the input_queue. :param signum: the signal number being handled :type signum: int :param frame: current stack frame :type frame: frame """ # Determine the size of our terminal, and create the message to be sent rows, columns = os.popen('stty size', 'r').read().split() message = { 'type': 'ATTACH_CONTAINER_INPUT', 'attach_container_input': { 'type': 'PROCESS_IO', 'process_io': { 'type': 'CONTROL', 'control': { 'type': 'TTY_INFO', 'tty_info': { 'window_size': { 'rows': int(rows), 'columns': int(columns)}}}}}} self.input_queue.put(self.encoder.encode(message))
def prepare(self): # per-readline preparations: self.__svtermstate = tcgetattr(self.input_fd) raw = self.__svtermstate.copy() raw.iflag &=~ (termios.BRKINT | termios.INPCK | termios.ISTRIP | termios.IXON) raw.oflag &=~ (termios.OPOST) raw.cflag &=~ (termios.CSIZE|termios.PARENB) raw.cflag |= (termios.CS8) raw.lflag &=~ (termios.ICANON|termios.ECHO| termios.IEXTEN|(termios.ISIG*1)) raw.cc[termios.VMIN] = 1 raw.cc[termios.VTIME] = 0 tcsetattr(self.input_fd, termios.TCSADRAIN, raw) self.screen = [] self.height, self.width = self.getheightwidth() self.__buffer = [] self.__posxy = 0, 0 self.__gone_tall = 0 self.__move = self.__move_short self.__offset = 0 self.__maybe_write_code(self._smkx) try: self.old_sigwinch = signal.signal( signal.SIGWINCH, self.__sigwinch) except ValueError: pass
def window_resize(self, lines, columns): """ update window size in kernel, then send SIGWINCH to fg process """ try: fcntl.ioctl(self.fd, termios.TIOCSWINSZ, struct.pack("HHHH", lines, columns, 0, 0)) os.kill(self.pid, signal.SIGWINCH) except: pass # vim:foldmethod=marker
def run(self, stdin, callbacks): # The PosixEventLoop basically sets up a callback for sigwinch to # monitor terminal resizes, and a callback for when stdin is ready. # libuv can do the stdin with the TTY handler, so we'll give that a # whirl self.sigw = pyuv.Signal(self.realloop) self.sigw.start(self.sigwinch, signal.SIGWINCH) self.tty = pyuv.TTY(self.realloop, sys.stdin.fileno(), True) self.tty.start_read(self.ttyread) self._callbacks = callbacks self.inputstream = prompt_toolkit.terminal.vt100_input.InputStream(callbacks.feed_key) return self.realloop.run()
def run(self, stdin, callbacks): # The PosixEventLoop basically sets up a callback for sigwinch to # monitor terminal resizes, and a callback for when stdin is ready. # libuv can do the stdin with the TTY handler, so we'll give that a # whirl self.sigw = pyuv.Signal(self.realloop) self.sigw.start(self.sigwinch, signal.SIGWINCH) self.tty = pyuv.TTY(self.realloop, sys.stdin.fileno(), True) self.tty.start_read(self.ttyread) self._callbacks = callbacks self.inputstream = prompt_toolkit.terminal.vt100_input.InputStream(callbacks.feed_key) #print dir(callbacks) return self.realloop.run()
def run_as_coroutine(self, stdin, callbacks): """ The input 'event loop'. """ assert isinstance(callbacks, EventLoopCallbacks) # Create reader class. stdin_reader = PosixStdinReader(stdin.fileno()) if self.closed: raise Exception('Event loop already closed.') inputstream = InputStream(callbacks.feed_key) try: # Create a new Future every time. self._stopped_f = asyncio.Future(loop=self.loop) # Handle input timouts def timeout_handler(): """ When no input has been received for INPUT_TIMEOUT seconds, flush the input stream and fire the timeout event. """ inputstream.flush() callbacks.input_timeout() timeout = AsyncioTimeout(INPUT_TIMEOUT, timeout_handler, self.loop) # Catch sigwinch def received_winch(): self.call_from_executor(callbacks.terminal_size_changed) self.loop.add_signal_handler(signal.SIGWINCH, received_winch) # Read input data. def stdin_ready(): data = stdin_reader.read() inputstream.feed(data) timeout.reset() # Quit when the input stream was closed. if stdin_reader.closed: self.stop() self.loop.add_reader(stdin.fileno(), stdin_ready) # Block this coroutine until stop() has been called. for f in self._stopped_f: yield f finally: # Clean up. self.loop.remove_reader(stdin.fileno()) self.loop.remove_signal_handler(signal.SIGWINCH) # Don't trigger any timeout events anymore. timeout.stop()
def test_Region_sync_custom_image_not_synced_uses_progress_bar(tmpdir, monkeypatch): """Test Region.sync_custom performs create and handles when its already synced.""" region = make_Region(quiet=False) mock_progress_bar = MagicMock() mock_progress_bar.signal_set = True mock_progress_bar_class = MagicMock() mock_progress_bar_class.return_value = mock_progress_bar mock_progress_bar.term_width = 80 monkeypatch.setattr(sys.stdout, "isatty", lambda: True) monkeypatch.setattr(region_module, "ProgressBar", mock_progress_bar_class) mock_signal = Mock() monkeypatch.setattr(signal, "signal", mock_signal) image_path = tmpdir.join("image.tar.gz") image_path.write(b"data") def fake_create( # pylint: disable=too-many-arguments,unused-argument name, arch, stream, title=None, filetype=None, progress_callback=None): """Fakes the create method. Calls progress_callback twice with 0% then 100%. """ progress_callback(0) progress_callback(1) # Call fake_create instead of the mock. region.origin.BootResources.create.side_effect = fake_create region.sync_custom("image", { "path": str(image_path), "architecture": "amd64/generic", "title": "My Title", }) assert mock_progress_bar.start.called is True assert [call(0), call(1)] == mock_progress_bar.update.call_args_list assert call(signal.SIGWINCH, signal.SIG_DFL) == mock_signal.call_args assert ( call( "custom/image uploaded", level=MessageLevel.SUCCESS, replace=True, fill=80) == region.print_msg.call_args)
def __enter__(self): # Create UI Application. title_toolbar = ConditionalContainer( Window(FormattedTextControl(lambda: self.title), height=1, style='class:progressbar,title'), filter=Condition(lambda: self.title is not None)) bottom_toolbar = ConditionalContainer( Window(FormattedTextControl(lambda: self.bottom_toolbar, style='class:bottom-toolbar.text'), style='class:bottom-toolbar', height=1), filter=~is_done & renderer_height_is_known & Condition(lambda: self.bottom_toolbar is not None)) def width_for_formatter(formatter): return formatter.get_width(progress_bar=self) progress_controls = [ Window(content=_ProgressControl(self, f), width=width_for_formatter(f)) for f in self.formatters ] self.app = Application( min_redraw_interval=.05, layout=Layout(HSplit([ title_toolbar, VSplit(progress_controls, height=lambda: D( preferred=len(self.counters), max=len(self.counters))), Window(), bottom_toolbar, ])), style=self.style, key_bindings=self.key_bindings, output=self.output, input=self.input) # Run application in different thread. def run(): with _auto_refresh_context(self.app, .3): try: self.app.run() except Exception as e: traceback.print_exc() print(e) self._thread = threading.Thread(target=run) self._thread.start() # Attach WINCH signal handler in main thread. # (Interrupt that we receive during resize events.) self._has_sigwinch = hasattr(signal, 'SIGWINCH') and in_main_thread() if self._has_sigwinch: self._previous_winch_handler = self._loop.add_signal_handler( signal.SIGWINCH, self.app.invalidate) return self
def run(self): """Run the helper threads in this class which enable streaming of STDIN/STDOUT/STDERR between the CLI and the Mesos Agent API. If a tty is requested, we take over the current terminal and put it into raw mode. We make sure to reset the terminal back to its original settings before exiting. """ # Without a TTY. if not self.tty: try: self._start_threads() self.exit_event.wait() except Exception as e: self.exception = e if self.exception: raise self.exception return # With a TTY. if util.is_windows_platform(): raise DCOSException( "Running with the '--tty' flag is not supported on windows.") if not sys.stdin.isatty(): raise DCOSException( "Must be running in a tty to pass the '--tty flag'.") fd = sys.stdin.fileno() oldtermios = termios.tcgetattr(fd) try: if self.interactive: tty.setraw(fd, when=termios.TCSANOW) self._window_resize(signal.SIGWINCH, None) signal.signal(signal.SIGWINCH, self._window_resize) self._start_threads() self.exit_event.wait() except Exception as e: self.exception = e termios.tcsetattr( sys.stdin.fileno(), termios.TCSAFLUSH, oldtermios) if self.exception: raise self.exception
def run_as_coroutine(self, stdin, callbacks): """ The input 'event loop'. """ assert isinstance(callbacks, EventLoopCallbacks) # Create reader class. stdin_reader = PosixStdinReader(stdin.fileno()) if self.closed: raise Exception('Event loop already closed.') inputstream = InputStream(callbacks.feed_key) try: # Create a new Future every time. self._stopped_f = asyncio.Future() # Handle input timouts def timeout_handler(): """ When no input has been received for INPUT_TIMEOUT seconds, flush the input stream and fire the timeout event. """ inputstream.flush() callbacks.input_timeout() timeout = AsyncioTimeout(INPUT_TIMEOUT, timeout_handler, self.loop) # Catch sigwinch def received_winch(): self.call_from_executor(callbacks.terminal_size_changed) self.loop.add_signal_handler(signal.SIGWINCH, received_winch) # Read input data. def stdin_ready(): data = stdin_reader.read() inputstream.feed(data) timeout.reset() # Quit when the input stream was closed. if stdin_reader.closed: self.stop() self.loop.add_reader(stdin.fileno(), stdin_ready) # Block this coroutine until stop() has been called. for f in self._stopped_f: yield f finally: # Clean up. self.loop.remove_reader(stdin.fileno()) self.loop.remove_signal_handler(signal.SIGWINCH) # Don't trigger any timeout events anymore. timeout.stop()
