1""" 
    2Abstraction of CLI Input. 
    3""" 
    4 
    5from __future__ import annotations 
    6 
    7from abc import ABCMeta, abstractmethod 
    8from contextlib import contextmanager 
    9from typing import Callable, ContextManager, Generator 
    10 
    11from prompt_toolkit.key_binding import KeyPress 
    12 
    13__all__ = [ 
    14    "Input", 
    15    "PipeInput", 
    16    "DummyInput", 
    17] 
    18 
    19 
    20class Input(metaclass=ABCMeta): 
    21    """ 
    22    Abstraction for any input. 
    23 
    24    An instance of this class can be given to the constructor of a 
    25    :class:`~prompt_toolkit.application.Application` and will also be 
    26    passed to the :class:`~prompt_toolkit.eventloop.base.EventLoop`. 
    27    """ 
    28 
    29    @abstractmethod 
    30    def fileno(self) -> int: 
    31        """ 
    32        Fileno for putting this in an event loop. 
    33        """ 
    34 
    35    @abstractmethod 
    36    def typeahead_hash(self) -> str: 
    37        """ 
    38        Identifier for storing type ahead key presses. 
    39        """ 
    40 
    41    @abstractmethod 
    42    def read_keys(self) -> list[KeyPress]: 
    43        """ 
    44        Return a list of Key objects which are read/parsed from the input. 
    45        """ 
    46 
    47    def flush_keys(self) -> list[KeyPress]: 
    48        """ 
    49        Flush the underlying parser. and return the pending keys. 
    50        (Used for vt100 input.) 
    51        """ 
    52        return [] 
    53 
    54    def flush(self) -> None: 
    55        "The event loop can call this when the input has to be flushed." 
    56        pass 
    57 
    58    @property 
    59    @abstractmethod 
    60    def closed(self) -> bool: 
    61        "Should be true when the input stream is closed." 
    62        return False 
    63 
    64    @abstractmethod 
    65    def raw_mode(self) -> ContextManager[None]: 
    66        """ 
    67        Context manager that turns the input into raw mode. 
    68        """ 
    69 
    70    @abstractmethod 
    71    def cooked_mode(self) -> ContextManager[None]: 
    72        """ 
    73        Context manager that turns the input into cooked mode. 
    74        """ 
    75 
    76    @abstractmethod 
    77    def attach(self, input_ready_callback: Callable[[], None]) -> ContextManager[None]: 
    78        """ 
    79        Return a context manager that makes this input active in the current 
    80        event loop. 
    81        """ 
    82 
    83    @abstractmethod 
    84    def detach(self) -> ContextManager[None]: 
    85        """ 
    86        Return a context manager that makes sure that this input is not active 
    87        in the current event loop. 
    88        """ 
    89 
    90    def close(self) -> None: 
    91        "Close input." 
    92        pass 
    93 
    94 
    95class PipeInput(Input): 
    96    """ 
    97    Abstraction for pipe input. 
    98    """ 
    99 
    100    @abstractmethod 
    101    def send_bytes(self, data: bytes) -> None: 
    102        """Feed byte string into the pipe""" 
    103 
    104    @abstractmethod 
    105    def send_text(self, data: str) -> None: 
    106        """Feed a text string into the pipe""" 
    107 
    108 
    109class DummyInput(Input): 
    110    """ 
    111    Input for use in a `DummyApplication` 
    112 
    113    If used in an actual application, it will make the application render 
    114    itself once and exit immediately, due to an `EOFError`. 
    115    """ 
    116 
    117    def fileno(self) -> int: 
    118        raise NotImplementedError 
    119 
    120    def typeahead_hash(self) -> str: 
    121        return f"dummy-{id(self)}" 
    122 
    123    def read_keys(self) -> list[KeyPress]: 
    124        return [] 
    125 
    126    @property 
    127    def closed(self) -> bool: 
    128        # This needs to be true, so that the dummy input will trigger an 
    129        # `EOFError` immediately in the application. 
    130        return True 
    131 
    132    def raw_mode(self) -> ContextManager[None]: 
    133        return _dummy_context_manager() 
    134 
    135    def cooked_mode(self) -> ContextManager[None]: 
    136        return _dummy_context_manager() 
    137 
    138    def attach(self, input_ready_callback: Callable[[], None]) -> ContextManager[None]: 
    139        # Call the callback immediately once after attaching. 
    140        # This tells the callback to call `read_keys` and check the 
    141        # `input.closed` flag, after which it won't receive any keys, but knows 
    142        # that `EOFError` should be raised. This unblocks `read_from_input` in 
    143        # `application.py`. 
    144        input_ready_callback() 
    145 
    146        return _dummy_context_manager() 
    147 
    148    def detach(self) -> ContextManager[None]: 
    149        return _dummy_context_manager() 
    150 
    151 
    152@contextmanager 
    153def _dummy_context_manager() -> Generator[None, None, None]: 
    154    yield