diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-04 17:35:20 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-04 17:35:20 +0000 |
commit | e106bf94eff07d9a59771d9ccc4406421e18ab64 (patch) | |
tree | edb6545500e39df9c67aa918a6125bffc8ec1aee /src/prompt_toolkit/utils.py | |
parent | Initial commit. (diff) | |
download | prompt-toolkit-upstream.tar.xz prompt-toolkit-upstream.zip |
Adding upstream version 3.0.36.upstream/3.0.36upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | src/prompt_toolkit/utils.py | 328 |
1 files changed, 328 insertions, 0 deletions
diff --git a/src/prompt_toolkit/utils.py b/src/prompt_toolkit/utils.py new file mode 100644 index 0000000..4ceded3 --- /dev/null +++ b/src/prompt_toolkit/utils.py @@ -0,0 +1,328 @@ +import os +import signal +import sys +import threading +from collections import deque +from typing import ( + Callable, + ContextManager, + Deque, + Dict, + Generator, + Generic, + List, + Optional, + TypeVar, + Union, +) + +from wcwidth import wcwidth + +__all__ = [ + "Event", + "DummyContext", + "get_cwidth", + "suspend_to_background_supported", + "is_conemu_ansi", + "is_windows", + "in_main_thread", + "get_bell_environment_variable", + "get_term_environment_variable", + "take_using_weights", + "to_str", + "to_int", + "AnyFloat", + "to_float", + "is_dumb_terminal", +] + +# Used to ensure sphinx autodoc does not try to import platform-specific +# stuff when documenting win32.py modules. +SPHINX_AUTODOC_RUNNING = "sphinx.ext.autodoc" in sys.modules + +_Sender = TypeVar("_Sender", covariant=True) + + +class Event(Generic[_Sender]): + """ + Simple event to which event handlers can be attached. For instance:: + + class Cls: + def __init__(self): + # Define event. The first parameter is the sender. + self.event = Event(self) + + obj = Cls() + + def handler(sender): + pass + + # Add event handler by using the += operator. + obj.event += handler + + # Fire event. + obj.event() + """ + + def __init__( + self, sender: _Sender, handler: Optional[Callable[[_Sender], None]] = None + ) -> None: + self.sender = sender + self._handlers: List[Callable[[_Sender], None]] = [] + + if handler is not None: + self += handler + + def __call__(self) -> None: + "Fire event." + for handler in self._handlers: + handler(self.sender) + + def fire(self) -> None: + "Alias for just calling the event." + self() + + def add_handler(self, handler: Callable[[_Sender], None]) -> None: + """ + Add another handler to this callback. + (Handler should be a callable that takes exactly one parameter: the + sender object.) + """ + # Add to list of event handlers. + self._handlers.append(handler) + + def remove_handler(self, handler: Callable[[_Sender], None]) -> None: + """ + Remove a handler from this callback. + """ + if handler in self._handlers: + self._handlers.remove(handler) + + def __iadd__(self, handler: Callable[[_Sender], None]) -> "Event[_Sender]": + """ + `event += handler` notation for adding a handler. + """ + self.add_handler(handler) + return self + + def __isub__(self, handler: Callable[[_Sender], None]) -> "Event[_Sender]": + """ + `event -= handler` notation for removing a handler. + """ + self.remove_handler(handler) + return self + + +class DummyContext(ContextManager[None]): + """ + (contextlib.nested is not available on Py3) + """ + + def __enter__(self) -> None: + pass + + def __exit__(self, *a: object) -> None: + pass + + +class _CharSizesCache(Dict[str, int]): + """ + Cache for wcwidth sizes. + """ + + LONG_STRING_MIN_LEN = 64 # Minimum string length for considering it long. + MAX_LONG_STRINGS = 16 # Maximum number of long strings to remember. + + def __init__(self) -> None: + super().__init__() + # Keep track of the "long" strings in this cache. + self._long_strings: Deque[str] = deque() + + def __missing__(self, string: str) -> int: + # Note: We use the `max(0, ...` because some non printable control + # characters, like e.g. Ctrl-underscore get a -1 wcwidth value. + # It can be possible that these characters end up in the input + # text. + result: int + if len(string) == 1: + result = max(0, wcwidth(string)) + else: + result = sum(self[c] for c in string) + + # Store in cache. + self[string] = result + + # Rotate long strings. + # (It's hard to tell what we can consider short...) + if len(string) > self.LONG_STRING_MIN_LEN: + long_strings = self._long_strings + long_strings.append(string) + + if len(long_strings) > self.MAX_LONG_STRINGS: + key_to_remove = long_strings.popleft() + if key_to_remove in self: + del self[key_to_remove] + + return result + + +_CHAR_SIZES_CACHE = _CharSizesCache() + + +def get_cwidth(string: str) -> int: + """ + Return width of a string. Wrapper around ``wcwidth``. + """ + return _CHAR_SIZES_CACHE[string] + + +def suspend_to_background_supported() -> bool: + """ + Returns `True` when the Python implementation supports + suspend-to-background. This is typically `False' on Windows systems. + """ + return hasattr(signal, "SIGTSTP") + + +def is_windows() -> bool: + """ + True when we are using Windows. + """ + return sys.platform == "win32" # Not 'darwin' or 'linux2' + + +def is_windows_vt100_supported() -> bool: + """ + True when we are using Windows, but VT100 escape sequences are supported. + """ + if sys.platform == "win32": + # Import needs to be inline. Windows libraries are not always available. + from prompt_toolkit.output.windows10 import is_win_vt100_enabled + + return is_win_vt100_enabled() + + return False + + +def is_conemu_ansi() -> bool: + """ + True when the ConEmu Windows console is used. + """ + return sys.platform == "win32" and os.environ.get("ConEmuANSI", "OFF") == "ON" + + +def in_main_thread() -> bool: + """ + True when the current thread is the main thread. + """ + return threading.current_thread().__class__.__name__ == "_MainThread" + + +def get_bell_environment_variable() -> bool: + """ + True if env variable is set to true (true, TRUE, TrUe, 1). + """ + value = os.environ.get("PROMPT_TOOLKIT_BELL", "true") + return value.lower() in ("1", "true") + + +def get_term_environment_variable() -> str: + "Return the $TERM environment variable." + return os.environ.get("TERM", "") + + +_T = TypeVar("_T") + + +def take_using_weights( + items: List[_T], weights: List[int] +) -> Generator[_T, None, None]: + """ + Generator that keeps yielding items from the items list, in proportion to + their weight. For instance:: + + # Getting the first 70 items from this generator should have yielded 10 + # times A, 20 times B and 40 times C, all distributed equally.. + take_using_weights(['A', 'B', 'C'], [5, 10, 20]) + + :param items: List of items to take from. + :param weights: Integers representing the weight. (Numbers have to be + integers, not floats.) + """ + assert len(items) == len(weights) + assert len(items) > 0 + + # Remove items with zero-weight. + items2 = [] + weights2 = [] + for item, w in zip(items, weights): + if w > 0: + items2.append(item) + weights2.append(w) + + items = items2 + weights = weights2 + + # Make sure that we have some items left. + if not items: + raise ValueError("Did't got any items with a positive weight.") + + # + already_taken = [0 for i in items] + item_count = len(items) + max_weight = max(weights) + + i = 0 + while True: + # Each iteration of this loop, we fill up until by (total_weight/max_weight). + adding = True + while adding: + adding = False + + for item_i, item, weight in zip(range(item_count), items, weights): + if already_taken[item_i] < i * weight / float(max_weight): + yield item + already_taken[item_i] += 1 + adding = True + + i += 1 + + +def to_str(value: Union[Callable[[], str], str]) -> str: + "Turn callable or string into string." + if callable(value): + return to_str(value()) + else: + return str(value) + + +def to_int(value: Union[Callable[[], int], int]) -> int: + "Turn callable or int into int." + if callable(value): + return to_int(value()) + else: + return int(value) + + +AnyFloat = Union[Callable[[], float], float] + + +def to_float(value: AnyFloat) -> float: + "Turn callable or float into float." + if callable(value): + return to_float(value()) + else: + return float(value) + + +def is_dumb_terminal(term: Optional[str] = None) -> bool: + """ + True if this terminal type is considered "dumb". + + If so, we should fall back to the simplest possible form of line editing, + without cursor positioning and color support. + """ + if term is None: + return is_dumb_terminal(os.environ.get("TERM", "")) + + return term.lower() in ["dumb", "unknown"] |