codekingpro/portable-devtools

1# Urwid raw display module2#    Copyright (C) 2004-2009  Ian Ward3#4#    This library is free software; you can redistribute it and/or5#    modify it under the terms of the GNU Lesser General Public6#    License as published by the Free Software Foundation; either7#    version 2.1 of the License, or (at your option) any later version.8#9#    This library is distributed in the hope that it will be useful,10#    but WITHOUT ANY WARRANTY; without even the implied warranty of11#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU12#    Lesser General Public License for more details.13#14#    You should have received a copy of the GNU Lesser General Public15#    License along with this library; if not, write to the Free Software16#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA17#18# Urwid web site: https://urwid.org/19 20 21"""22Direct terminal UI implementation23"""24 25from __future__ import annotations26 27import contextlib28import functools29import logging30import selectors31import socket32import sys33import threading34import typing35from ctypes import byref36from ctypes.wintypes import DWORD37 38from urwid import signals39 40from . import _raw_display_base, _win32, escape41from .common import INPUT_DESCRIPTORS_CHANGED42 43if typing.TYPE_CHECKING:44    from collections.abc import Callable45 46    from urwid.event_loop import EventLoop47 48 49class Screen(_raw_display_base.Screen):50    _term_input_file: socket.socket51 52    def __init__(53        self,54        input: socket.socket | None = None,  # noqa: A002  # pylint: disable=redefined-builtin55        output: typing.TextIO = sys.stdout,56    ) -> None:57        """Initialize a screen that directly prints escape codes to an output58        terminal.59        """60        if input is None:61            input, self._send_input = socket.socketpair()  # noqa: A00162 63        super().__init__(input, output)64 65    _dwOriginalOutMode = None66    _dwOriginalInMode = None67 68    def _start(self, alternate_buffer: bool = True) -> None:69        """70        Initialize the screen and input mode.71 72        alternate_buffer -- use alternate screen buffer73        """74        if alternate_buffer:75            self.write(escape.SWITCH_TO_ALTERNATE_BUFFER)76            self._rows_used = None77        else:78            self._rows_used = 079 80        handle_out = _win32.GetStdHandle(_win32.STD_OUTPUT_HANDLE)81        handle_in = _win32.GetStdHandle(_win32.STD_INPUT_HANDLE)82        self._dwOriginalOutMode = DWORD()83        self._dwOriginalInMode = DWORD()84        _win32.GetConsoleMode(handle_out, byref(self._dwOriginalOutMode))85        _win32.GetConsoleMode(handle_in, byref(self._dwOriginalInMode))86        # TODO: Restore on exit87 88        dword_out_mode = DWORD(89            self._dwOriginalOutMode.value90            | _win32.ENABLE_VIRTUAL_TERMINAL_PROCESSING91            | _win32.DISABLE_NEWLINE_AUTO_RETURN92        )93        dword_in_mode = DWORD(94            self._dwOriginalInMode.value | _win32.ENABLE_WINDOW_INPUT | _win32.ENABLE_VIRTUAL_TERMINAL_INPUT95        )96 97        if not (ok := _win32.SetConsoleMode(handle_out, dword_out_mode)):98            raise RuntimeError(f"ConsoleMode set failed for output. Err: {ok!r}")99 100        if not (ok := _win32.SetConsoleMode(handle_in, dword_in_mode)):101            raise RuntimeError(f"ConsoleMode set failed for input. Err: {ok!r}")102        self._alternate_buffer = alternate_buffer103        self._next_timeout = self.max_wait104 105        signals.emit_signal(self, INPUT_DESCRIPTORS_CHANGED)106        # restore mouse tracking to previous state107        self._mouse_tracking(self._mouse_tracking_enabled)108 109        return super()._start()110 111    def _stop(self) -> None:112        """113        Restore the screen.114        """115        self.clear()116 117        signals.emit_signal(self, INPUT_DESCRIPTORS_CHANGED)118 119        self._stop_mouse_restore_buffer()120 121        handle_out = _win32.GetStdHandle(_win32.STD_OUTPUT_HANDLE)122        handle_in = _win32.GetStdHandle(_win32.STD_INPUT_HANDLE)123 124        if not (ok := _win32.SetConsoleMode(handle_out, self._dwOriginalOutMode)):125            raise RuntimeError(f"ConsoleMode set failed for output. Err: {ok!r}")126 127        if not (ok := _win32.SetConsoleMode(handle_in, self._dwOriginalInMode)):128            raise RuntimeError(f"ConsoleMode set failed for input. Err: {ok!r}")129 130        super()._stop()131 132    def unhook_event_loop(self, event_loop: EventLoop) -> None:133        """134        Remove any hooks added by hook_event_loop.135        """136        if self._input_thread is not None:137            self._input_thread.should_exit = True138 139            with contextlib.suppress(RuntimeError):140                self._input_thread.join(5)141 142            self._input_thread = None143 144        for handle in self._current_event_loop_handles:145            event_loop.remove_watch_file(handle)146 147        if self._input_timeout:148            event_loop.remove_alarm(self._input_timeout)149            self._input_timeout = None150 151    def hook_event_loop(152        self,153        event_loop: EventLoop,154        callback: Callable[[list[str], list[int]], typing.Any],155    ) -> None:156        """157        Register the given callback with the event loop, to be called with new158        input whenever it's available.  The callback should be passed a list of159        processed keys and a list of unprocessed keycodes.160 161        Subclasses may wish to use parse_input to wrap the callback.162        """163        self._input_thread = ReadInputThread(self._send_input, lambda: self._sigwinch_handler(28))164        self._input_thread.start()165        if hasattr(self, "get_input_nonblocking"):166            wrapper = self._make_legacy_input_wrapper(event_loop, callback)167        else:168 169            @functools.wraps(callback)170            def wrapper() -> tuple[list[str], typing.Any] | None:171                return self.parse_input(event_loop, callback, self.get_available_raw_input())172 173        fds = self.get_input_descriptors()174        handles = [event_loop.watch_file(fd if isinstance(fd, int) else fd.fileno(), wrapper) for fd in fds]175        self._current_event_loop_handles = handles176 177    _input_thread: ReadInputThread | None = None178 179    def _read_raw_input(self, timeout: int) -> bytearray:180        ready = self._wait_for_input_ready(timeout)181 182        fd = self._input_fileno()183        chars = bytearray()184 185        if fd is None or fd not in ready:186            return chars187 188        with selectors.DefaultSelector() as selector:189            selector.register(fd, selectors.EVENT_READ)190 191            while selector.select(0):192                chars.extend(self._term_input_file.recv(1024))193 194            return chars195 196    def get_cols_rows(self) -> tuple[int, int]:197        """Return the terminal dimensions (num columns, num rows)."""198        y, x = super().get_cols_rows()199        with contextlib.suppress(OSError):  # Term size could not be determined200            if hasattr(self._term_output_file, "fileno"):201                if self._term_output_file != sys.stdout:202                    raise RuntimeError("Unexpected terminal output file")203                handle = _win32.GetStdHandle(_win32.STD_OUTPUT_HANDLE)204                info = _win32.CONSOLE_SCREEN_BUFFER_INFO()205 206                if _win32.GetConsoleScreenBufferInfo(handle, byref(info)):207                    # Fallback will be used in case of term size could not be determined208                    y, x = info.dwSize.Y, info.dwSize.X209 210        self.maxrow = y211        return x, y212 213 214class ReadInputThread(threading.Thread):215    name = "urwid Windows input reader"216    daemon = True217    should_exit: bool = False218 219    def __init__(220        self,221        input_socket: socket.socket,222        resize: Callable[[], typing.Any],223    ) -> None:224        self._input = input_socket225        self._resize = resize226        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)227        super().__init__()228 229    def run(self) -> None:230        hIn = _win32.GetStdHandle(_win32.STD_INPUT_HANDLE)231        MAX = 2048232 233        read = DWORD(0)234        arrtype = _win32.INPUT_RECORD * MAX235        input_records = arrtype()236 237        while True:238            _win32.ReadConsoleInputW(hIn, byref(input_records), MAX, byref(read))239            if self.should_exit:240                return241            for i in range(read.value):242                inp = input_records[i]243                if inp.EventType == _win32.EventType.KEY_EVENT:244                    if not inp.Event.KeyEvent.bKeyDown:245                        continue246 247                    input_data = inp.Event.KeyEvent.uChar.UnicodeChar248                    # On Windows atomic press/release of modifier keys produce phantom input with code NULL.249                    # This input cannot be decoded and should be handled as garbage.250                    input_bytes = input_data.encode("utf-8")251                    if input_bytes != b"\x00":252                        self._input.send(input_bytes)253 254                elif inp.EventType == _win32.EventType.WINDOW_BUFFER_SIZE_EVENT:255                    self._resize()256                else:257                    pass  # TODO: handle mouse events258 259 260def _test():261    import doctest262 263    doctest.testmod()264 265 266if __name__ == "__main__":267    _test()268