|
""" |
|
`Fish-style <http://fishshell.com/>`_ like auto-suggestion. |
|
|
|
While a user types input in a certain buffer, suggestions are generated |
|
(asynchronously.) Usually, they are displayed after the input. When the cursor |
|
presses the right arrow and the cursor is at the end of the input, the |
|
suggestion will be inserted. |
|
|
|
If you want the auto suggestions to be asynchronous (in a background thread), |
|
because they take too much time, and could potentially block the event loop, |
|
then wrap the :class:`.AutoSuggest` instance into a |
|
:class:`.ThreadedAutoSuggest`. |
|
""" |
|
from __future__ import annotations |
|
|
|
from abc import ABCMeta, abstractmethod |
|
from typing import TYPE_CHECKING, Callable |
|
|
|
from prompt_toolkit.eventloop import run_in_executor_with_context |
|
|
|
from .document import Document |
|
from .filters import Filter, to_filter |
|
|
|
if TYPE_CHECKING: |
|
from .buffer import Buffer |
|
|
|
__all__ = [ |
|
"Suggestion", |
|
"AutoSuggest", |
|
"ThreadedAutoSuggest", |
|
"DummyAutoSuggest", |
|
"AutoSuggestFromHistory", |
|
"ConditionalAutoSuggest", |
|
"DynamicAutoSuggest", |
|
] |
|
|
|
|
|
class Suggestion: |
|
""" |
|
Suggestion returned by an auto-suggest algorithm. |
|
|
|
:param text: The suggestion text. |
|
""" |
|
|
|
def __init__(self, text: str) -> None: |
|
self.text = text |
|
|
|
def __repr__(self) -> str: |
|
return "Suggestion(%s)" % self.text |
|
|
|
|
|
class AutoSuggest(metaclass=ABCMeta): |
|
""" |
|
Base class for auto suggestion implementations. |
|
""" |
|
|
|
@abstractmethod |
|
def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None: |
|
""" |
|
Return `None` or a :class:`.Suggestion` instance. |
|
|
|
We receive both :class:`~prompt_toolkit.buffer.Buffer` and |
|
:class:`~prompt_toolkit.document.Document`. The reason is that auto |
|
suggestions are retrieved asynchronously. (Like completions.) The |
|
buffer text could be changed in the meantime, but ``document`` contains |
|
the buffer document like it was at the start of the auto suggestion |
|
call. So, from here, don't access ``buffer.text``, but use |
|
``document.text`` instead. |
|
|
|
:param buffer: The :class:`~prompt_toolkit.buffer.Buffer` instance. |
|
:param document: The :class:`~prompt_toolkit.document.Document` instance. |
|
""" |
|
|
|
async def get_suggestion_async( |
|
self, buff: Buffer, document: Document |
|
) -> Suggestion | None: |
|
""" |
|
Return a :class:`.Future` which is set when the suggestions are ready. |
|
This function can be overloaded in order to provide an asynchronous |
|
implementation. |
|
""" |
|
return self.get_suggestion(buff, document) |
|
|
|
|
|
class ThreadedAutoSuggest(AutoSuggest): |
|
""" |
|
Wrapper that runs auto suggestions in a thread. |
|
(Use this to prevent the user interface from becoming unresponsive if the |
|
generation of suggestions takes too much time.) |
|
""" |
|
|
|
def __init__(self, auto_suggest: AutoSuggest) -> None: |
|
self.auto_suggest = auto_suggest |
|
|
|
def get_suggestion(self, buff: Buffer, document: Document) -> Suggestion | None: |
|
return self.auto_suggest.get_suggestion(buff, document) |
|
|
|
async def get_suggestion_async( |
|
self, buff: Buffer, document: Document |
|
) -> Suggestion | None: |
|
""" |
|
Run the `get_suggestion` function in a thread. |
|
""" |
|
|
|
def run_get_suggestion_thread() -> Suggestion | None: |
|
return self.get_suggestion(buff, document) |
|
|
|
return await run_in_executor_with_context(run_get_suggestion_thread) |
|
|
|
|
|
class DummyAutoSuggest(AutoSuggest): |
|
""" |
|
AutoSuggest class that doesn't return any suggestion. |
|
""" |
|
|
|
def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None: |
|
return None |
|
|
|
|
|
class AutoSuggestFromHistory(AutoSuggest): |
|
""" |
|
Give suggestions based on the lines in the history. |
|
""" |
|
|
|
def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None: |
|
history = buffer.history |
|
|
|
|
|
text = document.text.rsplit("\n", 1)[-1] |
|
|
|
|
|
if text.strip(): |
|
|
|
for string in reversed(list(history.get_strings())): |
|
for line in reversed(string.splitlines()): |
|
if line.startswith(text): |
|
return Suggestion(line[len(text) :]) |
|
|
|
return None |
|
|
|
|
|
class ConditionalAutoSuggest(AutoSuggest): |
|
""" |
|
Auto suggest that can be turned on and of according to a certain condition. |
|
""" |
|
|
|
def __init__(self, auto_suggest: AutoSuggest, filter: bool | Filter) -> None: |
|
self.auto_suggest = auto_suggest |
|
self.filter = to_filter(filter) |
|
|
|
def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None: |
|
if self.filter(): |
|
return self.auto_suggest.get_suggestion(buffer, document) |
|
|
|
return None |
|
|
|
|
|
class DynamicAutoSuggest(AutoSuggest): |
|
""" |
|
Validator class that can dynamically returns any Validator. |
|
|
|
:param get_validator: Callable that returns a :class:`.Validator` instance. |
|
""" |
|
|
|
def __init__(self, get_auto_suggest: Callable[[], AutoSuggest | None]) -> None: |
|
self.get_auto_suggest = get_auto_suggest |
|
|
|
def get_suggestion(self, buff: Buffer, document: Document) -> Suggestion | None: |
|
auto_suggest = self.get_auto_suggest() or DummyAutoSuggest() |
|
return auto_suggest.get_suggestion(buff, document) |
|
|
|
async def get_suggestion_async( |
|
self, buff: Buffer, document: Document |
|
) -> Suggestion | None: |
|
auto_suggest = self.get_auto_suggest() or DummyAutoSuggest() |
|
return await auto_suggest.get_suggestion_async(buff, document) |
|
|