aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py
diff options
context:
space:
mode:
authorNikita Slyusarev <nslus@yandex-team.com>2022-02-10 16:46:52 +0300
committerDaniil Cherednik <dcherednik@yandex-team.ru>2022-02-10 16:46:52 +0300
commitcd77cecfc03a3eaf87816af28a33067c4f0cdb59 (patch)
tree1308e0bae862d52e0020d881fe758080437fe389 /contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py
parentcdae02d225fb5b3afbb28990e79a7ac6c9125327 (diff)
downloadydb-cd77cecfc03a3eaf87816af28a33067c4f0cdb59.tar.gz
Restoring authorship annotation for Nikita Slyusarev <nslus@yandex-team.com>. Commit 1 of 2.
Diffstat (limited to 'contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py')
-rw-r--r--contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py568
1 files changed, 284 insertions, 284 deletions
diff --git a/contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py b/contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py
index e1e0e56393..9b3a1be04d 100644
--- a/contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py
+++ b/contrib/python/prompt-toolkit/py2/prompt_toolkit/interface.py
@@ -10,30 +10,30 @@ import six
import sys
import textwrap
import threading
-import time
-import types
+import time
+import types
import weakref
-from subprocess import Popen
-
+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 .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 .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
+from .utils import Event
# Following import is required for backwards compatibility.
from .buffer import AcceptAction
@@ -57,23 +57,23 @@ class CommandLineInterface(object):
: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`.
+ 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 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
+ from .shortcuts import create_output
self.application = application
- self.eventloop = eventloop
+ self.eventloop = eventloop
self._is_running = False
# Inputs and outputs.
@@ -84,15 +84,15 @@ class CommandLineInterface(object):
assert isinstance(application.buffers, BufferMapping)
self.buffers = application.buffers
- #: EditingMode.VI or EditingMode.EMACS
- self.editing_mode = application.editing_mode
-
+ #: 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()
-
+ #: 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(
@@ -125,20 +125,20 @@ class CommandLineInterface(object):
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)
+ # 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()
+ self.reset()
+ self.on_initialize += self.application.on_initialize
+ self.on_initialize.fire()
@property
def layout(self):
@@ -174,7 +174,7 @@ class CommandLineInterface(object):
ensures that it's only called while typing if the
`complete_while_typing` filter is enabled.
"""
- def on_text_insert(_):
+ def on_text_insert(_):
# Only complete when "complete_while_typing" is enabled.
if buffer.completer and buffer.complete_while_typing():
completer_function()
@@ -187,18 +187,18 @@ class CommandLineInterface(object):
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 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):
@@ -293,14 +293,14 @@ class CommandLineInterface(object):
self.renderer.reset()
self.input_processor.reset()
self.layout.reset()
- self.vi_state.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()
+ self.on_reset.fire()
@property
def in_paste_mode(self):
@@ -333,17 +333,17 @@ class CommandLineInterface(object):
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)
-
+ # 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
@@ -357,9 +357,9 @@ class CommandLineInterface(object):
self.render_counter += 1
self.renderer.render(self, self.layout, is_done=self.is_done)
- # Fire render event.
- self.on_render.fire()
-
+ # Fire render event.
+ self.on_render.fire()
+
def _on_resize(self):
"""
When the window size changes, we erase the current output and request
@@ -368,7 +368,7 @@ class CommandLineInterface(object):
"""
# 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.erase(leave_alternate_screen=False, erase_title=False)
self.renderer.request_absolute_cursor_position()
self._redraw()
@@ -387,7 +387,7 @@ class CommandLineInterface(object):
c()
del self.pre_run_callables[:]
- def run(self, reset_current_buffer=False, pre_run=None):
+ 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.
@@ -401,7 +401,7 @@ class CommandLineInterface(object):
try:
self._is_running = True
- self.on_start.fire()
+ self.on_start.fire()
self.reset()
# Call pre_run.
@@ -416,16 +416,16 @@ class CommandLineInterface(object):
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()
-
+
+ # 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.on_stop.fire()
self._is_running = False
# Return result.
@@ -442,41 +442,41 @@ class CommandLineInterface(object):
This is only available on Python >3.3, with asyncio.
"""
- # Inline import, because it slows down startup when asyncio is not
- # needed.
- import 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)
+ @asyncio.coroutine
+ def run():
+ assert pre_run is None or callable(pre_run)
- try:
- self._is_running = True
+ try:
+ self._is_running = True
- self.on_start.fire()
+ self.on_start.fire()
self.reset()
- # Call pre_run.
+ # 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()
+ 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.
@@ -488,9 +488,9 @@ class CommandLineInterface(object):
"""
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.
+ 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`.
@@ -514,8 +514,8 @@ class CommandLineInterface(object):
raise RuntimeError('Another sub application started already.')
# Erase current application.
- if not _from_application_generator:
- self.renderer.erase()
+ if not _from_application_generator:
+ self.renderer.erase()
# Callback when the sub app is done.
def done():
@@ -523,17 +523,17 @@ class CommandLineInterface(object):
# 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()
+ 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()
+ if not _from_application_generator:
+ self.renderer.request_absolute_cursor_position()
+ self._redraw()
# Deliver result.
if done_callback:
@@ -636,7 +636,7 @@ class CommandLineInterface(object):
self.renderer.reset() # Make sure to disable mouse mode, etc...
else:
self.renderer.erase()
- self._return_value = None
+ self._return_value = None
# Run system command.
if cooked_mode:
@@ -652,63 +652,63 @@ class CommandLineInterface(object):
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_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
@@ -716,57 +716,57 @@ class CommandLineInterface(object):
: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 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 to use the same input/output file descriptors as the one,
+ # used to run this application.
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()
-
+ 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):
+ 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.)
+
+ :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.)
@@ -775,13 +775,13 @@ class CommandLineInterface(object):
# 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)
-
+ # 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):
@@ -835,16 +835,16 @@ class CommandLineInterface(object):
"""
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 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
@@ -871,13 +871,13 @@ class CommandLineInterface(object):
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[:]
-
+ 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 \
@@ -886,28 +886,28 @@ class CommandLineInterface(object):
set_completions = True
select_first_anyway = False
- # When the common part has to be inserted, and there
+ # 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.
+ # 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
+ 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
+ # which does * expansion, like '*.py', with
# exactly one match.)
if len(completions) == 1:
select_first_anyway = True
@@ -918,7 +918,7 @@ class CommandLineInterface(object):
go_to_first=select_first or select_first_anyway,
go_to_last=select_last)
self.invalidate()
- elif not buffer.complete_state:
+ elif not buffer.complete_state:
# Otherwise, restart thread.
async_completer()
@@ -953,8 +953,8 @@ class CommandLineInterface(object):
suggestion = buffer.auto_suggest.get_suggestion(self, buffer, document)
def callback():
- suggest_thread_running[0] = False
-
+ 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:
@@ -984,18 +984,18 @@ class CommandLineInterface(object):
"""
return _StdoutProxy(self, raw=raw)
- def patch_stdout_context(self, raw=False, patch_stdout=True, patch_stderr=True):
+ 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`.
+
+ :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)
+ return _PatchStdoutContext(
+ self.stdout_proxy(raw=raw),
+ patch_stdout=patch_stdout, patch_stderr=patch_stderr)
def create_eventloop_callbacks(self):
return _InterfaceEventLoopCallbacks(self)
@@ -1031,46 +1031,46 @@ class _InterfaceEventLoopCallbacks(EventLoopCallbacks):
def input_timeout(self):
cli = self._active_cli
- cli.on_input_timeout.fire()
+ 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
-
+ 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()
+ # (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):
+ 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
+ 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
+ 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
+ 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