<commit_before>
import locale
import os
import signal
import select
import sys
import termios
import threading
import time
import tty

import logging
logger = logging.getLogger(__name__)


from .termhelpers import Nonblocking
from . import events

from typing import Callable, Type, TextIO, Optional, List, Union, Text, cast, Tuple, Any
from types import TracebackType, FrameType

PY3 = sys.version_info[0] >= 3

READ_SIZE = 1024
assert READ_SIZE >= events.MAX_KEYPRESS_SIZE
# if a keypress could require more bytes than we read to be identified,
# the paste logic that reads more data as needed might not work.


def is_main_thread():
    # type: () -> bool
    return isinstance(threading.current_thread(), threading._MainThread) # type: ignore


class ReplacedSigIntHandler(object):
    def __init__(self, handler):
        # type: (Callable) -> None
        self.handler = handler

    def __enter__(self):
        # type: () -> None
        self.orig_sigint_handler = signal.getsignal(signal.SIGINT)
        signal.signal(signal.SIGINT, self.handler)

    def __exit__(self, type=None, value=None, traceback=None):
        # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]) -> None
        signal.signal(signal.SIGINT, self.orig_sigint_handler)


class Input(object):
    """Keypress and control event generator"""
    def __init__(self, in_stream=None, keynames='curtsies',
                 paste_threshold=events.MAX_KEYPRESS_SIZE+1, sigint_event=False,
                 signint_callback_provider=None,
                 disable_terminal_start_stop=False):
        # type: (TextIO, str, int, bool, None, bool) -> None
        """Returns an Input instance.

        Args:
            in_stream (file): Defaults to sys.__stdin__
            keynames (string): How keypresses should be named - one of
              'curtsies', 'curses', or 'plain'.
            paste_threshold (int): How many bytes must be read in one
              os.read on the in_stream to trigger the keypresses they
              represent to be combined into a single paste event
            sigint_event (bool): Whether SIGINT signals from the OS
              should be intercepted and returned as SigIntEvent objects
            disable_terminal_start_stop (bool): If True, disable terminal
              start/stop using Ctrl-s/Ctrl-q, thus enabling these keys
              to be read as input by curtsies
        """
        if in_stream is None:
            in_stream = sys.__stdin__
        self.in_stream = in_stream
        self.unprocessed_bytes = []  # type: List[bytes]  # leftover from stdin, unprocessed yet
        self.keynames = keynames
        self.paste_threshold = paste_threshold
        self.sigint_event = sigint_event
        self.disable_terminal_start_stop = disable_terminal_start_stop
        self.sigints = []  # type: List[events.SigIntEvent]

        self.readers = []  # type: List[int]
        self.queued_interrupting_events = []  # type: List[Union[events.Event, Text]]
        self.queued_events = []  # type: List[events.Event]
        self.queued_scheduled_events = []  # type: List[Tuple[float, events.ScheduledEvent]]

    # prospective: this could be useful for an external select loop
    def fileno(self):
        # type: () -> int
        return self.in_stream.fileno()

    def __enter__(self):
        # type: () -> Input
        self.original_stty = termios.tcgetattr(self.in_stream)
        tty.setcbreak(self.in_stream, termios.TCSANOW)

        if self.disable_terminal_start_stop:
            attrs = termios.tcgetattr(self.in_stream)
            tty_cc = cast(List[Union[bytes, int]], attrs[-1])
            tty_cc[termios.VSTOP] = 0   # Ctrl-s
            tty_cc[termios.VSTART] = 0  # Ctrl-q
            termios.tcsetattr(self.in_stream, termios.TCSANOW, attrs)

        if sys.platform == 'darwin':
            attrs = termios.tcgetattr(self.in_stream)
            VDSUSP = termios.VSUSP + 1
            tty_cc = cast(List[Union[bytes, int]], attrs[-1])
            tty_cc[VDSUSP] = 0
            termios.tcsetattr(self.in_stream, termios.TCSANOW, attrs)

        if self.sigint_event and is_main_thread():
            self.orig_sigint_handler = signal.getsignal(signal.SIGINT)
            signal.signal(signal.SIGINT, self.sigint_handler)
        return self

    def __exit__(self, type=None, value=None, traceback=None):
        # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]) -> None
        if self.sigint_event and is_main_thread():
            signal.signal(signal.SIGINT, self.orig_sigint_handler)
        termios.tcsetattr(self.in_stream, termios.TCSANOW, self.original_stty)

    def sigint_handler(self, signum, frame):
        # type: (Union[signal.Signals, int], FrameType) -> None
        self.sigints.append(events.SigIntEvent())

    def __iter__(self):
        # type: () -> Input
        return self

    def next(self):
        # type: () -> Union[None, Text, events.Event]
        return self.send(None)

    __next__ = next

    def unget_bytes(self, string):
        # type: (bytes) -> None
        """Adds bytes to be internal buffer to be read

        This method is for reporting bytes from an in_stream read<commit_msg>Test if Input has an orig_sigint_handler before attempting to use it<commit_after>
import locale
import os
import signal
import select
import sys
import termios
import threading
import time
import tty

import logging
logger = logging.getLogger(__name__)


from .termhelpers import Nonblocking
from . import events

from typing import Callable, Type, TextIO, Optional, List, Union, Text, cast, Tuple, Any
from types import TracebackType, FrameType

PY3 = sys.version_info[0] >= 3

READ_SIZE = 1024
assert READ_SIZE >= events.MAX_KEYPRESS_SIZE
# if a keypress could require more bytes than we read to be identified,
# the paste logic that reads more data as needed might not work.


def is_main_thread():
    # type: () -> bool
    return isinstance(threading.current_thread(), threading._MainThread) # type: ignore


class ReplacedSigIntHandler(object):
    def __init__(self, handler):
        # type: (Callable) -> None
        self.handler = handler

    def __enter__(self):
        # type: () -> None
        self.orig_sigint_handler = signal.getsignal(signal.SIGINT)
        signal.signal(signal.SIGINT, self.handler)

    def __exit__(self, type=None, value=None, traceback=None):
        # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]) -> None
        signal.signal(signal.SIGINT, self.orig_sigint_handler)


class Input(object):
    """Keypress and control event generator"""
    def __init__(self, in_stream=None, keynames='curtsies',
                 paste_threshold=events.MAX_KEYPRESS_SIZE+1, sigint_event=False,
                 signint_callback_provider=None,
                 disable_terminal_start_stop=False):
        # type: (TextIO, str, int, bool, None, bool) -> None
        """Returns an Input instance.

        Args:
            in_stream (file): Defaults to sys.__stdin__
            keynames (string): How keypresses should be named - one of
              'curtsies', 'curses', or 'plain'.
            paste_threshold (int): How many bytes must be read in one
              os.read on the in_stream to trigger the keypresses they
              represent to be combined into a single paste event
            sigint_event (bool): Whether SIGINT signals from the OS
              should be intercepted and returned as SigIntEvent objects
            disable_terminal_start_stop (bool): If True, disable terminal
              start/stop using Ctrl-s/Ctrl-q, thus enabling these keys
              to be read as input by curtsies
        """
        if in_stream is None:
            in_stream = sys.__stdin__
        self.in_stream = in_stream
        self.unprocessed_bytes = []  # type: List[bytes]  # leftover from stdin, unprocessed yet
        self.keynames = keynames
        self.paste_threshold = paste_threshold
        self.sigint_event = sigint_event
        self.disable_terminal_start_stop = disable_terminal_start_stop
        self.sigints = []  # type: List[events.SigIntEvent]

        self.readers = []  # type: List[int]
        self.queued_interrupting_events = []  # type: List[Union[events.Event, Text]]
        self.queued_events = []  # type: List[events.Event]
        self.queued_scheduled_events = []  # type: List[Tuple[float, events.ScheduledEvent]]

    # prospective: this could be useful for an external select loop
    def fileno(self):
        # type: () -> int
        return self.in_stream.fileno()

    def __enter__(self):
        # type: () -> Input
        self.original_stty = termios.tcgetattr(self.in_stream)
        tty.setcbreak(self.in_stream, termios.TCSANOW)

        if self.disable_terminal_start_stop:
            attrs = termios.tcgetattr(self.in_stream)
            tty_cc = cast(List[Union[bytes, int]], attrs[-1])
            tty_cc[termios.VSTOP] = 0   # Ctrl-s
            tty_cc[termios.VSTART] = 0  # Ctrl-q
            termios.tcsetattr(self.in_stream, termios.TCSANOW, attrs)

        if sys.platform == 'darwin':
            attrs = termios.tcgetattr(self.in_stream)
            VDSUSP = termios.VSUSP + 1
            tty_cc = cast(List[Union[bytes, int]], attrs[-1])
            tty_cc[VDSUSP] = 0
            termios.tcsetattr(self.in_stream, termios.TCSANOW, attrs)

        if self.sigint_event and is_main_thread():
            self.orig_sigint_handler = signal.getsignal(signal.SIGINT)
            signal.signal(signal.SIGINT, self.sigint_handler)
        return self

    def __exit__(self, type=None, value=None, traceback=None):
        # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]) -> None
        if self.sigint_event and is_main_thread() and self.orig_sigint_handler is not None:
            signal.signal(signal.SIGINT, self.orig_sigint_handler)
        termios.tcsetattr(self.in_stream, termios.TCSANOW, self.original_stty)

    def sigint_handler(self, signum, frame):
        # type: (Union[signal.Signals, int], FrameType) -> None
        self.sigints.append(events.SigIntEvent())

    def __iter__(self):
        # type: () -> Input
        return self

    def next(self):
        # type: () -> Union[None, Text, events.Event]
        return self.send(None)

    __next__ = next

    def unget_bytes(self, string):
        # type: (bytes) -> None
        """Adds bytes to be internal buffer to be read

        This method is for reporting bytes from an in_stream read
{'fraction_non_alphanumeric': 0.09483895272179318, 'fraction_numerical': 0.0026370314560180824, 'mean_word_length': 2.735138937741822, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3662102', 'n_tokens_mistral': 3207, 'n_tokens_neox': 3092, 'n_words': 832}