path: root/contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py

""" 
The main `CommandLineInterface` class and logic. 
""" 
from __future__ import unicode_literals 
 
import functools 
import os 
import signal 
import six 
import sys 
import textwrap 
import threading 
import time
import types
import weakref 
 
from subprocess import Popen

from .application import Application, AbortAction 
from .buffer import Buffer 
from .buffer_mapping import BufferMapping 
from .completion import CompleteEvent, get_common_complete_suffix
from .enums import SEARCH_BUFFER 
from .eventloop.base import EventLoop 
from .eventloop.callbacks import EventLoopCallbacks 
from .filters import Condition 
from .input import StdinInput, Input 
from .key_binding.input_processor import InputProcessor 
from .key_binding.input_processor import KeyPress
from .key_binding.registry import Registry
from .key_binding.vi_state import ViState
from .keys import Keys
from .output import Output 
from .renderer import Renderer, print_tokens 
from .search_state import SearchState 
from .utils import Event
 
# Following import is required for backwards compatibility. 
from .buffer import AcceptAction 
 
__all__ = ( 
    'AbortAction', 
    'CommandLineInterface', 
) 
 
 
class CommandLineInterface(object): 
    """ 
    Wrapper around all the other classes, tying everything together. 
 
    Typical usage:: 
 
        application = Application(...) 
        cli = CommandLineInterface(application, eventloop) 
        result = cli.run() 
        print(result) 
 
    :param application: :class:`~prompt_toolkit.application.Application` instance. 
    :param eventloop: The :class:`~prompt_toolkit.eventloop.base.EventLoop` to 
                      be used when `run` is called. The easiest way to create
                      an eventloop is by calling
                      :meth:`~prompt_toolkit.shortcuts.create_eventloop`.
    :param input: :class:`~prompt_toolkit.input.Input` instance. 
    :param output: :class:`~prompt_toolkit.output.Output` instance. (Probably 
                   Vt100_Output or Win32Output.) 
    """ 
    def __init__(self, application, eventloop=None, input=None, output=None): 
        assert isinstance(application, Application) 
        assert isinstance(eventloop, EventLoop), 'Passing an eventloop is required.'
        assert output is None or isinstance(output, Output) 
        assert input is None or isinstance(input, Input) 
 
        from .shortcuts import create_output
 
        self.application = application 
        self.eventloop = eventloop
        self._is_running = False 
 
        # Inputs and outputs. 
        self.output = output or create_output() 
        self.input = input or StdinInput(sys.stdin) 
 
        #: The input buffers. 
        assert isinstance(application.buffers, BufferMapping) 
        self.buffers = application.buffers 
 
        #: EditingMode.VI or EditingMode.EMACS
        self.editing_mode = application.editing_mode

        #: Quoted insert. This flag is set if we go into quoted insert mode.
        self.quoted_insert = False

        #: Vi state. (For Vi key bindings.)
        self.vi_state = ViState()

        #: The `Renderer` instance. 
        # Make sure that the same stdout is used, when a custom renderer has been passed. 
        self.renderer = Renderer( 
            self.application.style, 
            self.output, 
            use_alternate_screen=application.use_alternate_screen, 
            mouse_support=application.mouse_support) 
 
        #: Render counter. This one is increased every time the UI is rendered. 
        #: It can be used as a key for caching certain information during one 
        #: rendering. 
        self.render_counter = 0 
 
        #: When there is high CPU, postpone the renderering max x seconds. 
        #: '0' means: don't postpone. '.5' means: try to draw at least twice a second. 
        self.max_render_postpone_time = 0  # E.g. .5 
 
        # Invalidate flag. When 'True', a repaint has been scheduled. 
        self._invalidated = False 
 
        #: The `InputProcessor` instance. 
        self.input_processor = InputProcessor(application.key_bindings_registry, weakref.ref(self)) 
 
        self._async_completers = {}  # Map buffer name to completer function. 
 
        # Pointer to sub CLI. (In chain of CLI instances.) 
        self._sub_cli = None  # None or other CommandLineInterface instance. 
 
        # Call `add_buffer` for each buffer. 
        for name, b in self.buffers.items(): 
            self.add_buffer(name, b) 
 
        # Events.
        self.on_buffer_changed = Event(self, application.on_buffer_changed)
        self.on_initialize = Event(self, application.on_initialize)
        self.on_input_timeout = Event(self, application.on_input_timeout)
        self.on_invalidate = Event(self, application.on_invalidate)
        self.on_render = Event(self, application.on_render)
        self.on_reset = Event(self, application.on_reset)
        self.on_start = Event(self, application.on_start)
        self.on_stop = Event(self, application.on_stop)
 
        # Trigger initialize callback. 
        self.reset()
        self.on_initialize += self.application.on_initialize
        self.on_initialize.fire()
 
    @property 
    def layout(self): 
        return self.application.layout 
 
    @property 
    def clipboard(self): 
        return self.application.clipboard 
 
    @property
    def pre_run_callables(self):
        return self.application.pre_run_callables

    def add_buffer(self, name, buffer, focus=False): 
        """ 
        Insert a new buffer. 
        """ 
        assert isinstance(buffer, Buffer) 
        self.buffers[name] = buffer 
 
        if focus: 
            self.buffers.focus(name) 
 
        # Create asynchronous completer / auto suggestion. 
        auto_suggest_function = self._create_auto_suggest_function(buffer) 
        completer_function = self._create_async_completer(buffer) 
        self._async_completers[name] = completer_function 
 
        # Complete/suggest on text insert. 
        def create_on_insert_handler(): 
            """ 
            Wrapper around the asynchronous completer and auto suggestion, that 
            ensures that it's only called while typing if the 
            `complete_while_typing` filter is enabled. 
            """ 
            def on_text_insert(_):
                # Only complete when "complete_while_typing" is enabled. 
                if buffer.completer and buffer.complete_while_typing(): 
                    completer_function() 
 
                # Call auto_suggest. 
                if buffer.auto_suggest: 
                    auto_suggest_function() 
 
            return on_text_insert 
 
        buffer.on_text_insert += create_on_insert_handler() 
 
        def buffer_changed(_):
            """
            When the text in a buffer changes.
            (A paste event is also a change, but not an insert. So we don't
            want to do autocompletions in this case, but we want to propagate
            the on_buffer_changed event.)
            """
            # Trigger on_buffer_changed.
            self.on_buffer_changed.fire()

        buffer.on_text_changed += buffer_changed

    def start_completion(self, buffer_name=None, select_first=False, 
                         select_last=False, insert_common_part=False, 
                         complete_event=None): 
        """ 
        Start asynchronous autocompletion of this buffer. 
        (This will do nothing if a previous completion was still in progress.) 
        """ 
        buffer_name = buffer_name or self.current_buffer_name 
        completer = self._async_completers.get(buffer_name) 
 
        if completer: 
            completer(select_first=select_first, 
                      select_last=select_last, 
                      insert_common_part=insert_common_part, 
                      complete_event=CompleteEvent(completion_requested=True)) 
 
    @property 
    def current_buffer_name(self): 
        """ 
        The name of the current  :class:`.Buffer`. (Or `None`.) 
        """ 
        return self.buffers.current_name(self) 
 
    @property 
    def current_buffer(self): 
        """ 
        The currently focussed :class:`~.Buffer`. 
 
        (This returns a dummy :class:`.Buffer` when none of the actual buffers 
        has the focus. In this case, it's really not practical to check for 
        `None` values or catch exceptions every time.) 
        """ 
        return self.buffers.current(self) 
 
    def focus(self, buffer_name): 
        """ 
        Focus the buffer with the given name on the focus stack. 
        """ 
        self.buffers.focus(self, buffer_name) 
 
    def push_focus(self, buffer_name): 
        """ 
        Push to the focus stack. 
        """ 
        self.buffers.push_focus(self, buffer_name) 
 
    def pop_focus(self): 
        """ 
        Pop from the focus stack. 
        """ 
        self.buffers.pop_focus(self) 
 
    @property 
    def terminal_title(self): 
        """ 
        Return the current title to be displayed in the terminal. 
        When this in `None`, the terminal title remains the original. 
        """ 
        result = self.application.get_title() 
 
        # Make sure that this function returns a unicode object, 
        # and not a byte string. 
        assert result is None or isinstance(result, six.text_type) 
        return result 
 
    @property 
    def is_searching(self): 
        """ 
        True when we are searching. 
        """ 
        return self.current_buffer_name == SEARCH_BUFFER 
 
    def reset(self, reset_current_buffer=False): 
        """ 
        Reset everything, for reading the next input. 
 
        :param reset_current_buffer: XXX: not used anymore. The reason for
            having this option in the past was when this CommandLineInterface
            is run multiple times, that we could reset the buffer content from
            the previous run. This is now handled in the AcceptAction.
        """ 
        # Notice that we don't reset the buffers. (This happens just before 
        # returning, and when we have multiple buffers, we clearly want the 
        # content in the other buffers to remain unchanged between several 
        # calls of `run`. (And the same is true for the focus stack.) 
 
        self._exit_flag = False 
        self._abort_flag = False 
 
        self._return_value = None 
 
        self.renderer.reset() 
        self.input_processor.reset() 
        self.layout.reset() 
        self.vi_state.reset()
 
        # Search new search state. (Does also remember what has to be 
        # highlighted.) 
        self.search_state = SearchState(ignore_case=Condition(lambda: self.is_ignoring_case)) 
 
        # Trigger reset event. 
        self.on_reset.fire()
 
    @property 
    def in_paste_mode(self): 
        """ True when we are in paste mode. """ 
        return self.application.paste_mode(self) 
 
    @property 
    def is_ignoring_case(self): 
        """ True when we currently ignore casing. """ 
        return self.application.ignore_case(self) 
 
    def invalidate(self): 
        """ 
        Thread safe way of sending a repaint trigger to the input event loop. 
        """ 
        # Never schedule a second redraw, when a previous one has not yet been 
        # executed. (This should protect against other threads calling 
        # 'invalidate' many times, resulting in 100% CPU.) 
        if self._invalidated: 
            return 
        else: 
            self._invalidated = True 
 
        # Trigger event. 
        self.on_invalidate.fire() 
 
        if self.eventloop is not None: 
            def redraw(): 
                self._invalidated = False 
                self._redraw() 
 
            # Call redraw in the eventloop (thread safe). 
            # Usually with the high priority, in order to make the application
            # feel responsive, but this can be tuned by changing the value of
            # `max_render_postpone_time`.
            if self.max_render_postpone_time:
                _max_postpone_until = time.time() + self.max_render_postpone_time
            else:
                _max_postpone_until = None
 
            self.eventloop.call_from_executor(
                redraw, _max_postpone_until=_max_postpone_until)

    # Depracated alias for 'invalidate'. 
    request_redraw = invalidate 
 
    def _redraw(self): 
        """ 
        Render the command line again. (Not thread safe!) (From other threads, 
        or if unsure, use :meth:`.CommandLineInterface.invalidate`.) 
        """ 
        # Only draw when no sub application was started. 
        if self._is_running and self._sub_cli is None: 
            self.render_counter += 1 
            self.renderer.render(self, self.layout, is_done=self.is_done) 
 
            # Fire render event.
            self.on_render.fire()

    def _on_resize(self): 
        """ 
        When the window size changes, we erase the current output and request 
        again the cursor position. When the CPR answer arrives, the output is 
        drawn again. 
        """ 
        # Erase, request position (when cursor is at the start position) 
        # and redraw again. -- The order is important. 
        self.renderer.erase(leave_alternate_screen=False, erase_title=False)
        self.renderer.request_absolute_cursor_position() 
        self._redraw() 
 
    def _load_next_buffer_indexes(self):
        for buff, index in self._next_buffer_indexes.items():
            if buff in self.buffers:
                self.buffers[buff].working_index = index

    def _pre_run(self, pre_run=None):
        " Called during `run`. "
        if pre_run:
            pre_run()

        # Process registered "pre_run_callables" and clear list.
        for c in self.pre_run_callables:
            c()
        del self.pre_run_callables[:]

    def run(self, reset_current_buffer=False, pre_run=None):
        """ 
        Read input from the command line. 
        This runs the eventloop until a return value has been set. 
 
        :param reset_current_buffer: XXX: Not used anymore.
        :param pre_run: Callable that is called right after the reset has taken 
            place. This allows custom initialisation. 
        """ 
        assert pre_run is None or callable(pre_run) 
 
        try: 
            self._is_running = True 
 
            self.on_start.fire()
            self.reset()
 
            # Call pre_run. 
            self._pre_run(pre_run)
 
            # Run eventloop in raw mode. 
            with self.input.raw_mode(): 
                self.renderer.request_absolute_cursor_position() 
                self._redraw() 
 
                self.eventloop.run(self.input, self.create_eventloop_callbacks()) 
        finally: 
            # Clean up renderer. (This will leave the alternate screen, if we use 
            # that.) 

            # If exit/abort haven't been called set, but another exception was
            # thrown instead for some reason, make sure that we redraw in exit
            # mode.
            if not self.is_done:
                self._exit_flag = True
                self._redraw()

            self.renderer.reset() 
            self.on_stop.fire()
            self._is_running = False 
 
        # Return result. 
        return self.return_value() 
 
    try: 
        # The following `run_async` function is compiled at runtime 
        # because it contains syntax which is not supported on older Python 
        # versions. (A 'return' inside a generator.) 
        six.exec_(textwrap.dedent(''' 
        def run_async(self, reset_current_buffer=True, pre_run=None): 
            """ 
            Same as `run`, but this returns a coroutine. 
 
            This is only available on Python >3.3, with asyncio. 
            """ 
            # Inline import, because it slows down startup when asyncio is not
            # needed.
            import asyncio
 
            @asyncio.coroutine
            def run():
                assert pre_run is None or callable(pre_run)
 
                try:
                    self._is_running = True
 
                    self.on_start.fire()
                    self.reset()
 
                    # Call pre_run.
                    self._pre_run(pre_run)
 
                    with self.input.raw_mode():
                        self.renderer.request_absolute_cursor_position()
                        self._redraw()
 
                        yield from self.eventloop.run_as_coroutine(
                                self.input, self.create_eventloop_callbacks())
 
                    return self.return_value()
                finally:
                    if not self.is_done:
                        self._exit_flag = True
                        self._redraw()

                    self.renderer.reset()
                    self.on_stop.fire()
                    self._is_running = False

            return run()
        ''')) 
    except SyntaxError: 
        # Python2, or early versions of Python 3. 
        def run_async(self, reset_current_buffer=True, pre_run=None): 
            """ 
            Same as `run`, but this returns a coroutine. 
 
            This is only available on Python >3.3, with asyncio. 
            """ 
            raise NotImplementedError 
 
    def run_sub_application(self, application, done_callback=None, erase_when_done=False,
                            _from_application_generator=False):
        # `erase_when_done` is deprecated, set Application.erase_when_done instead.
        """ 
        Run a sub :class:`~prompt_toolkit.application.Application`. 
 
        This will suspend the main application and display the sub application 
        until that one returns a value. The value is returned by calling 
        `done_callback` with the result. 
 
        The sub application will share the same I/O of the main application. 
        That means, it uses the same input and output channels and it shares 
        the same event loop. 
 
        .. note:: Technically, it gets another Eventloop instance, but that is 
            only a proxy to our main event loop. The reason is that calling 
            'stop' --which returns the result of an application when it's 
            done-- is handled differently. 
        """ 
        assert isinstance(application, Application) 
        assert done_callback is None or callable(done_callback) 
 
        if self._sub_cli is not None: 
            raise RuntimeError('Another sub application started already.') 
 
        # Erase current application. 
        if not _from_application_generator:
            self.renderer.erase()
 
        # Callback when the sub app is done. 
        def done(): 
            # Redraw sub app in done state. 
            # and reset the renderer. (This reset will also quit the alternate 
            # screen, if the sub application used that.) 
            sub_cli._redraw() 
            if erase_when_done or application.erase_when_done:
                sub_cli.renderer.erase()
            sub_cli.renderer.reset() 
            sub_cli._is_running = False  # Don't render anymore. 
 
            self._sub_cli = None 
 
            # Restore main application. 
            if not _from_application_generator:
                self.renderer.request_absolute_cursor_position()
                self._redraw()
 
            # Deliver result. 
            if done_callback: 
                done_callback(sub_cli.return_value()) 
 
        # Create sub CommandLineInterface. 
        sub_cli = CommandLineInterface( 
            application=application, 
            eventloop=_SubApplicationEventLoop(self, done), 
            input=self.input, 
            output=self.output) 
        sub_cli._is_running = True  # Allow rendering of sub app. 
 
        sub_cli._redraw() 
        self._sub_cli = sub_cli 
 
    def exit(self): 
        """ 
        Set exit. When Control-D has been pressed. 
        """ 
        on_exit = self.application.on_exit 
        self._exit_flag = True 
        self._redraw() 
 
        if on_exit == AbortAction.RAISE_EXCEPTION: 
            def eof_error(): 
                raise EOFError() 
            self._set_return_callable(eof_error) 
 
        elif on_exit == AbortAction.RETRY: 
            self.reset() 
            self.renderer.request_absolute_cursor_position() 
            self.current_buffer.reset() 
 
        elif on_exit == AbortAction.RETURN_NONE: 
            self.set_return_value(None) 
 
    def abort(self): 
        """ 
        Set abort. When Control-C has been pressed. 
        """ 
        on_abort = self.application.on_abort 
        self._abort_flag = True 
        self._redraw() 
 
        if on_abort == AbortAction.RAISE_EXCEPTION: 
            def keyboard_interrupt(): 
                raise KeyboardInterrupt() 
            self._set_return_callable(keyboard_interrupt) 
 
        elif on_abort == AbortAction.RETRY: 
            self.reset() 
            self.renderer.request_absolute_cursor_position() 
            self.current_buffer.reset() 
 
        elif on_abort == AbortAction.RETURN_NONE: 
            self.set_return_value(None) 
 
    # Deprecated aliase for exit/abort. 
    set_exit = exit 
    set_abort = abort 
 
    def set_return_value(self, document): 
        """ 
        Set a return value. The eventloop can retrieve the result it by calling 
        `return_value`. 
        """ 
        self._set_return_callable(lambda: document) 
        self._redraw()  # Redraw in "done" state, after the return value has been set. 
 
    def _set_return_callable(self, value): 
        assert callable(value) 
        self._return_value = value 
 
        if self.eventloop: 
            self.eventloop.stop() 
 
    def run_in_terminal(self, func, render_cli_done=False, cooked_mode=True):
        """ 
        Run function on the terminal above the prompt. 
 
        What this does is first hiding the prompt, then running this callable 
        (which can safely output to the terminal), and then again rendering the 
        prompt which causes the output of this function to scroll above the 
        prompt. 
 
        :param func: The callable to execute. 
        :param render_cli_done: When True, render the interface in the 
                'Done' state first, then execute the function. If False, 
                erase the interface first. 
        :param cooked_mode: When True (the default), switch the input to
                cooked mode while executing the function.
 
        :returns: the result of `func`. 
        """ 
        # Draw interface in 'done' state, or erase. 
        if render_cli_done: 
            self._return_value = True 
            self._redraw() 
            self.renderer.reset()  # Make sure to disable mouse mode, etc... 
        else: 
            self.renderer.erase() 
        self._return_value = None
 
        # Run system command. 
        if cooked_mode:
            with self.input.cooked_mode():
                result = func()
        else:
            result = func() 
 
        # Redraw interface again. 
        self.renderer.reset() 
        self.renderer.request_absolute_cursor_position() 
        self._redraw() 
 
        return result 
 
    def run_application_generator(self, coroutine, render_cli_done=False):
        """
        EXPERIMENTAL
        Like `run_in_terminal`, but takes a generator that can yield Application instances.

        Example:

            def f():
                yield Application1(...)
                print('...')
                yield Application2(...)
            cli.run_in_terminal_async(f)

        The values which are yielded by the given coroutine are supposed to be
        `Application` instances that run in the current CLI, all other code is
        supposed to be CPU bound, so except for yielding the applications,
        there should not be any user interaction or I/O in the given function.
        """
        # Draw interface in 'done' state, or erase.
        if render_cli_done:
            self._return_value = True
            self._redraw()
            self.renderer.reset()  # Make sure to disable mouse mode, etc...
        else:
            self.renderer.erase()
        self._return_value = None

        # Loop through the generator.
        g = coroutine()
        assert isinstance(g, types.GeneratorType)

        def step_next(send_value=None):
            " Execute next step of the coroutine."
            try:
                # Run until next yield, in cooked mode.
                with self.input.cooked_mode():
                    result = g.send(send_value)
            except StopIteration:
                done()
            except:
                done()
                raise
            else:
                # Process yielded value from coroutine.
                assert isinstance(result, Application)
                self.run_sub_application(result, done_callback=step_next,
                                         _from_application_generator=True)

        def done():
            # Redraw interface again.
            self.renderer.reset()
            self.renderer.request_absolute_cursor_position()
            self._redraw()

        # Start processing coroutine.
        step_next()

    def run_system_command(self, command): 
        """ 
        Run system command (While hiding the prompt. When finished, all the 
        output will scroll above the prompt.) 
 
        :param command: Shell command to be executed. 
        """ 
        def wait_for_enter():
            """
            Create a sub application to wait for the enter key press.
            This has two advantages over using 'input'/'raw_input':
            - This will share the same input/output I/O.
            - This doesn't block the event loop.
            """
            from .shortcuts import create_prompt_application

            registry = Registry()

            @registry.add_binding(Keys.ControlJ)
            @registry.add_binding(Keys.ControlM)
            def _(event):
                event.cli.set_return_value(None)

            application = create_prompt_application(
                message='Press ENTER to continue...',
                key_bindings_registry=registry)
            self.run_sub_application(application)

        def run(): 
            # Try to use the same input/output file descriptors as the one,
            # used to run this application.
            try: 
                input_fd = self.input.fileno()
            except AttributeError:
                input_fd = sys.stdin.fileno()
            try:
                output_fd = self.output.fileno()
            except AttributeError:
                output_fd = sys.stdout.fileno()
 
            # Run sub process.
            # XXX: This will still block the event loop.
            p = Popen(command, shell=True,
                      stdin=input_fd, stdout=output_fd)
            p.wait()

            # Wait for the user to press enter.
            wait_for_enter()

        self.run_in_terminal(run) 
 
    def suspend_to_background(self, suspend_group=True):
        """ 
        (Not thread safe -- to be called from inside the key bindings.) 
        Suspend process. 

        :param suspend_group: When true, suspend the whole process group.
            (This is the default, and probably what you want.)
        """ 
        # Only suspend when the opperating system supports it. 
        # (Not on Windows.) 
        if hasattr(signal, 'SIGTSTP'): 
            def run(): 
                # Send `SIGSTP` to own process. 
                # This will cause it to suspend. 
 
                # Usually we want the whole process group to be suspended. This
                # handles the case when input is piped from another process.
                if suspend_group:
                    os.kill(0, signal.SIGTSTP)
                else:
                    os.kill(os.getpid(), signal.SIGTSTP)

            self.run_in_terminal(run) 
 
    def print_tokens(self, tokens, style=None): 
        """ 
        Print a list of (Token, text) tuples to the output. 
        (When the UI is running, this method has to be called through 
        `run_in_terminal`, otherwise it will destroy the UI.) 
 
        :param style: Style class to use. Defaults to the active style in the CLI. 
        """ 
        print_tokens(self.output, tokens, style or self.application.style) 
 
    @property 
    def is_exiting(self): 
        """ 
        ``True`` when the exit flag as been set. 
        """ 
        return self._exit_flag 
 
    @property 
    def is_aborting(self): 
        """ 
        ``True`` when the abort flag as been set. 
        """ 
        return self._abort_flag 
 
    @property 
    def is_returning(self): 
        """ 
        ``True`` when a return value has been set. 
        """ 
        return self._return_value is not None 
 
    def return_value(self): 
        """ 
        Get the return value. Not that this method can throw an exception. 
        """ 
        # Note that it's a method, not a property, because it can throw 
        # exceptions. 
        if self._return_value: 
            return self._return_value() 
 
    @property 
    def is_done(self): 
        return self.is_exiting or self.is_aborting or self.is_returning 
 
    def _create_async_completer(self, buffer): 
        """ 
        Create function for asynchronous autocompletion. 
        (Autocomplete in other thread.) 
        """ 
        complete_thread_running = [False]  # By ref. 
 
        def completion_does_nothing(document, completion):
            """
            Return `True` if applying this completion doesn't have any effect.
            (When it doesn't insert any new text.
            """
            text_before_cursor = document.text_before_cursor
            replaced_text = text_before_cursor[
                len(text_before_cursor) + completion.start_position:]
            return replaced_text == completion.text

        def async_completer(select_first=False, select_last=False, 
                            insert_common_part=False, complete_event=None): 
            document = buffer.document 
            complete_event = complete_event or CompleteEvent(text_inserted=True) 
 
            # Don't start two threads at the same time. 
            if complete_thread_running[0]: 
                return 
 
            # Don't complete when we already have completions. 
            if buffer.complete_state or not buffer.completer: 
                return 
 
            # Otherwise, get completions in other thread. 
            complete_thread_running[0] = True 
 
            def run(): 
                completions = list(buffer.completer.get_completions(document, complete_event)) 
 
                def callback(): 
                    """ 
                    Set the new complete_state in a safe way. Don't replace an 
                    existing complete_state if we had one. (The user could have 
                    pressed 'Tab' in the meantime. Also don't set it if the text 
                    was changed in the meantime. 
                    """ 
                    complete_thread_running[0] = False

                    # When there is only one completion, which has nothing to add, ignore it.
                    if (len(completions) == 1 and
                            completion_does_nothing(document, completions[0])):
                        del completions[:]

                    # Set completions if the text was not yet changed. 
                    if buffer.text == document.text and \ 
                            buffer.cursor_position == document.cursor_position and \ 
                            not buffer.complete_state: 
 
                        set_completions = True 
                        select_first_anyway = False 
 
                        # When the common part has to be inserted, and there
                        # is a common part. 
                        if insert_common_part: 
                            common_part = get_common_complete_suffix(document, completions) 
                            if common_part: 
                                # Insert the common part, update completions.
                                buffer.insert_text(common_part) 
                                if len(completions) > 1:
                                    # (Don't call `async_completer` again, but
                                    # recalculate completions. See:
                                    # https://github.com/ipython/ipython/issues/9658)
                                    completions[:] = [
                                        c.new_completion_from_position(len(common_part))
                                        for c in completions]
                                else:
                                    set_completions = False
                            else: 
                                # When we were asked to insert the "common" 
                                # prefix, but there was no common suffix but 
                                # still exactly one match, then select the 
                                # first. (It could be that we have a completion 
                                # which does * expansion, like '*.py', with
                                # exactly one match.) 
                                if len(completions) == 1: 
                                    select_first_anyway = True 
 
                        if set_completions: 
                            buffer.set_completions( 
                                completions=completions, 
                                go_to_first=select_first or select_first_anyway, 
                                go_to_last=select_last) 
                        self.invalidate() 
                    elif not buffer.complete_state:
                        # Otherwise, restart thread. 
                        async_completer() 
 
                if self.eventloop: 
                    self.eventloop.call_from_executor(callback) 
 
            self.eventloop.run_in_executor(run) 
        return async_completer 
 
    def _create_auto_suggest_function(self, buffer): 
        """ 
        Create function for asynchronous auto suggestion. 
        (AutoSuggest in other thread.) 
        """ 
        suggest_thread_running = [False]  # By ref. 
 
        def async_suggestor(): 
            document = buffer.document 
 
            # Don't start two threads at the same time. 
            if suggest_thread_running[0]: 
                return 
 
            # Don't suggest when we already have a suggestion. 
            if buffer.suggestion or not buffer.auto_suggest: 
                return 
 
            # Otherwise, get completions in other thread. 
            suggest_thread_running[0] = True 
 
            def run(): 
                suggestion = buffer.auto_suggest.get_suggestion(self, buffer, document) 
 
                def callback(): 
                    suggest_thread_running[0] = False

                    # Set suggestion only if the text was not yet changed. 
                    if buffer.text == document.text and \ 
                            buffer.cursor_position == document.cursor_position: 
 
                        # Set suggestion and redraw interface. 
                        buffer.suggestion = suggestion 
                        self.invalidate() 
                    else: 
                        # Otherwise, restart thread. 
                        async_suggestor() 
 
                if self.eventloop: 
                    self.eventloop.call_from_executor(callback) 
 
            self.eventloop.run_in_executor(run) 
        return async_suggestor 
 
    def stdout_proxy(self, raw=False): 
        """ 
        Create an :class:`_StdoutProxy` class which can be used as a patch for 
        `sys.stdout`. Writing to this proxy will make sure that the text 
        appears above the prompt, and that it doesn't destroy the output from 
        the renderer. 
 
        :param raw: (`bool`) When True, vt100 terminal escape sequences are not 
                    removed/escaped. 
        """ 
        return _StdoutProxy(self, raw=raw) 
 
    def patch_stdout_context(self, raw=False, patch_stdout=True, patch_stderr=True):
        """ 
        Return a context manager that will replace ``sys.stdout`` with a proxy 
        that makes sure that all printed text will appear above the prompt, and 
        that it doesn't destroy the output from the renderer. 

        :param patch_stdout: Replace `sys.stdout`.
        :param patch_stderr: Replace `sys.stderr`.
        """ 
        return _PatchStdoutContext(
            self.stdout_proxy(raw=raw),
            patch_stdout=patch_stdout, patch_stderr=patch_stderr)
 
    def create_eventloop_callbacks(self): 
        return _InterfaceEventLoopCallbacks(self) 
 
 
class _InterfaceEventLoopCallbacks(EventLoopCallbacks): 
    """ 
    Callbacks on the :class:`.CommandLineInterface` object, to which an 
    eventloop can talk. 
    """ 
    def __init__(self, cli): 
        assert isinstance(cli, CommandLineInterface) 
        self.cli = cli 
 
    @property 
    def _active_cli(self): 
        """ 
        Return the active `CommandLineInterface`. 
        """ 
        cli = self.cli 
 
        # If there is a sub CLI. That one is always active. 
        while cli._sub_cli: 
            cli = cli._sub_cli 
 
        return cli 
 
    def terminal_size_changed(self): 
        """ 
        Report terminal size change. This will trigger a redraw. 
        """ 
        self._active_cli._on_resize() 
 
    def input_timeout(self): 
        cli = self._active_cli 
        cli.on_input_timeout.fire()
 
    def feed_key(self, key_press): 
        """ 
        Feed a key press to the CommandLineInterface. 
        """ 
        assert isinstance(key_press, KeyPress)
        cli = self._active_cli

        # Feed the key and redraw. 
        # (When the CLI is in 'done' state, it should return to the event loop
        # as soon as possible. Ignore all key presses beyond this point.)
        if not cli.is_done:
            cli.input_processor.feed(key_press)
            cli.input_processor.process_keys()
 
 
class _PatchStdoutContext(object): 
    def __init__(self, new_stdout, patch_stdout=True, patch_stderr=True):
        self.new_stdout = new_stdout 
        self.patch_stdout = patch_stdout
        self.patch_stderr = patch_stderr
 
    def __enter__(self): 
        self.original_stdout = sys.stdout 
        self.original_stderr = sys.stderr
 
        if self.patch_stdout:
            sys.stdout = self.new_stdout
        if self.patch_stderr:
            sys.stderr = self.new_stdout

    def __exit__(self, *a, **kw): 
        if self.patch_stdout:
            sys.stdout = self.original_stdout
 
        if self.patch_stderr:
            sys.stderr = self.original_stderr
 

class _StdoutProxy(object): 
    """ 
    Proxy for stdout, as returned by 
    :class:`CommandLineInterface.stdout_proxy`. 
    """ 
    def __init__(self, cli, raw=False): 
        assert isinstance(cli, CommandLineInterface) 
        assert isinstance(raw, bool) 
 
        self._lock = threading.RLock() 
        self._cli = cli 
        self._raw = raw 
        self._buffer = [] 
 
        self.errors = sys.__stdout__.errors 
        self.encoding = sys.__stdout__.encoding 
 
    def _do(self, func): 
        if self._cli._is_running: 
            run_in_terminal = functools.partial(self._cli.run_in_terminal, func) 
            self._cli.eventloop.call_from_executor(run_in_terminal) 
        else: 
            func() 
 
    def _write(self, data): 
        """ 
        Note: print()-statements cause to multiple write calls. 
              (write('line') and write('\n')). Of course we don't want to call 
              `run_in_terminal` for every individual call, because that's too 
              expensive, and as long as the newline hasn't been written, the 
              text itself is again overwritter by the rendering of the input 
              command line. Therefor, we have a little buffer which holds the 
              text until a newline is written to stdout. 
        """ 
        if '\n' in data: 
            # When there is a newline in the data, write everything before the 
            # newline, including the newline itself. 
            before, after = data.rsplit('\n', 1) 
            to_write = self._buffer + [before, '\n'] 
            self._buffer = [after] 
 
            def run(): 
                for s in to_write: 
                    if self._raw: 
                        self._cli.output.write_raw(s) 
                    else: 
                        self._cli.output.write(s) 
            self._do(run) 
        else: 
            # Otherwise, cache in buffer. 
            self._buffer.append(data) 
 
    def write(self, data): 
        with self._lock: 
            self._write(data) 
 
    def _flush(self): 
        def run(): 
            for s in self._buffer: 
                if self._raw: 
                    self._cli.output.write_raw(s) 
                else: 
                    self._cli.output.write(s) 
            self._buffer = [] 
            self._cli.output.flush() 
        self._do(run) 
 
    def flush(self): 
        """ 
        Flush buffered output. 
        """ 
        with self._lock: 
            self._flush() 
 
 
class _SubApplicationEventLoop(EventLoop): 
    """ 
    Eventloop used by sub applications. 
 
    A sub application is an `Application` that is "spawned" by a parent 
    application. The parent application is suspended temporarily and the sub 
    application is displayed instead. 
 
    It doesn't need it's own event loop. The `EventLoopCallbacks` from the 
    parent application are redirected to the sub application. So if the event 
    loop that is run by the parent application detects input, the callbacks 
    will make sure that it's forwarded to the sub application. 
 
    When the sub application has a return value set, it will terminate 
    by calling the `stop` method of this event loop. This is used to 
    transfer control back to the parent application. 
    """ 
    def __init__(self, cli, stop_callback): 
        assert isinstance(cli, CommandLineInterface) 
        assert callable(stop_callback) 
 
        self.cli = cli 
        self.stop_callback = stop_callback 
 
    def stop(self): 
        self.stop_callback() 
 
    def close(self): 
        pass 
 
    def run_in_executor(self, callback): 
        self.cli.eventloop.run_in_executor(callback) 
 
    def call_from_executor(self, callback, _max_postpone_until=None): 
        self.cli.eventloop.call_from_executor( 
            callback, _max_postpone_until=_max_postpone_until) 
 
    def add_reader(self, fd, callback): 
        self.cli.eventloop.add_reader(fd, callback) 
 
    def remove_reader(self, fd): 
        self.cli.eventloop.remove_reader(fd)